<commit_before>/*
   Copyright (c) 2014, Percona LLC and/or its affiliates. All rights reserved.

   This program is free software: you can redistribute it and/or modify
   it under the terms of the GNU Affero General Public License as published by
   the Free Software Foundation, either version 3 of the License, or
   (at your option) any later version.

   This program is distributed in the hope that it will be useful,
   but WITHOUT ANY WARRANTY; without even the implied warranty of
   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
   GNU Affero General Public License for more details.

   You should have received a copy of the GNU Affero General Public License
   along with this program.  If not, see <http://www.gnu.org/licenses/>
*/

package data

import (
	"fmt"
	"github.com/percona/cloud-protocol/proto"
	"github.com/percona/percona-agent/pct"
	"math/rand"
	"time"
)

type Sender struct {
	logger *pct.Logger
	client pct.WebsocketClient
	// --
	spool      Spooler
	tickerChan <-chan time.Time
	blackhole  bool
	sync       *pct.SyncChan
	status     *pct.Status
}

func NewSender(logger *pct.Logger, client pct.WebsocketClient) *Sender {
	s := &Sender{
		logger: logger,
		client: client,
		sync:   pct.NewSyncChan(),
		status: pct.NewStatus([]string{"data-sender"}),
	}
	return s
}

func (s *Sender) Start(spool Spooler, tickerChan <-chan time.Time, blackhole bool) error {
	s.spool = spool
	s.tickerChan = tickerChan
	s.blackhole = blackhole
	go s.run()
	s.logger.Info("Started")
	return nil
}

func (s *Sender) Stop() error {
	s.sync.Stop()
	s.sync.Wait()
	s.spool = nil
	s.tickerChan = nil
	s.logger.Info("Stopped")
	return nil
}

func (s *Sender) Status() map[string]string {
	return s.status.Merge(s.client.Status())
}

/////////////////////////////////////////////////////////////////////////////
// Implementation
/////////////////////////////////////////////////////////////////////////////

// @goroutine[1]
func (s *Sender) run() {
	defer func() {
		if s.sync.IsGraceful() {
			s.logger.Info("Stop")
			s.status.Update("data-sender", "Stopped")
		} else {
			s.logger.Error("Crash")
			s.status.Update("data-sender", "Crashed")
		}
		s.sync.Done()
	}()

	s.logger.Info("Start")
	s.status.Update("data-sender", "Idle")
	for {
		select {
		case <-s.tickerChan:
			s.send()
		case <-s.sync.StopChan:
			s.sync.Graceful()
			return
		}
	}
}

func (s *Sender) send() {
	s.logger.Debug("send:call")
	defer s.logger.Debug("send:return")

	// Try a few times to connect to the API.
	sentOK := 0
	s.status.Update("data-sender", "Connecting")
	defer s.status.Update("data-sender", fmt.Sprintf("Idle (last sent %d files at %s)", sentOK, time.Now()))
	connected := false
	var apiErr error
	for i := 1; i <= 3; i++ {
		if apiErr = s.client.ConnectOnce(); apiErr != nil {
			s.logger.Warn("Connect API failed:", apiErr)
			t := int(5*rand.Float64() + 1) // [1, 5] seconds
			time.Sleep(time.Duration(t) * time.Second)
		} else {
			connected = true
			// client.WebsocketClient expects caller to recv on ConenctChan(),
			// even though in this case we're not using the async channels.
			// todo: fix this poor design assumption/coupling in ws/client.go
			defer func() {
				s.status.Update("data-sender", "Disconnecting")
				s.client.Disconnect()
				<-s.client.ConnectChan()
			}()
			break
		}
	}
	if !connected {<commit_msg>Need func literal for closure on sentOK.<commit_after>/*
   Copyright (c) 2014, Percona LLC and/or its affiliates. All rights reserved.

   This program is free software: you can redistribute it and/or modify
   it under the terms of the GNU Affero General Public License as published by
   the Free Software Foundation, either version 3 of the License, or
   (at your option) any later version.

   This program is distributed in the hope that it will be useful,
   but WITHOUT ANY WARRANTY; without even the implied warranty of
   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
   GNU Affero General Public License for more details.

   You should have received a copy of the GNU Affero General Public License
   along with this program.  If not, see <http://www.gnu.org/licenses/>
*/

package data

import (
	"fmt"
	"github.com/percona/cloud-protocol/proto"
	"github.com/percona/percona-agent/pct"
	"math/rand"
	"time"
)

type Sender struct {
	logger *pct.Logger
	client pct.WebsocketClient
	// --
	spool      Spooler
	tickerChan <-chan time.Time
	blackhole  bool
	sync       *pct.SyncChan
	status     *pct.Status
}

func NewSender(logger *pct.Logger, client pct.WebsocketClient) *Sender {
	s := &Sender{
		logger: logger,
		client: client,
		sync:   pct.NewSyncChan(),
		status: pct.NewStatus([]string{"data-sender"}),
	}
	return s
}

func (s *Sender) Start(spool Spooler, tickerChan <-chan time.Time, blackhole bool) error {
	s.spool = spool
	s.tickerChan = tickerChan
	s.blackhole = blackhole
	go s.run()
	s.logger.Info("Started")
	return nil
}

func (s *Sender) Stop() error {
	s.sync.Stop()
	s.sync.Wait()
	s.spool = nil
	s.tickerChan = nil
	s.logger.Info("Stopped")
	return nil
}

func (s *Sender) Status() map[string]string {
	return s.status.Merge(s.client.Status())
}

/////////////////////////////////////////////////////////////////////////////
// Implementation
/////////////////////////////////////////////////////////////////////////////

// @goroutine[1]
func (s *Sender) run() {
	defer func() {
		if s.sync.IsGraceful() {
			s.logger.Info("Stop")
			s.status.Update("data-sender", "Stopped")
		} else {
			s.logger.Error("Crash")
			s.status.Update("data-sender", "Crashed")
		}
		s.sync.Done()
	}()

	s.logger.Info("Start")
	s.status.Update("data-sender", "Idle")
	for {
		select {
		case <-s.tickerChan:
			s.send()
		case <-s.sync.StopChan:
			s.sync.Graceful()
			return
		}
	}
}

func (s *Sender) send() {
	s.logger.Debug("send:call")
	defer s.logger.Debug("send:return")

	// Try a few times to connect to the API.
	s.status.Update("data-sender", "Connecting")
	sentOK := 0
	defer func() {
		s.status.Update("data-sender", fmt.Sprintf("Idle (last sent %d files at %s)", sentOK, time.Now()))
	}()
	connected := false
	var apiErr error
	for i := 1; i <= 3; i++ {
		if apiErr = s.client.ConnectOnce(); apiErr != nil {
			s.logger.Warn("Connect API failed:", apiErr)
			t := int(5*rand.Float64() + 1) // [1, 5] seconds
			time.Sleep(time.Duration(t) * time.Second)
		} else {
			connected = true
			// client.WebsocketClient expects caller to recv on ConenctChan(),
			// even though in this case we're not using the async channels.
			// todo: fix this poor design assumption/coupling in ws/client.go
			defer func() {
				s.status.Update("data-sender", "Disconnecting")
				s.client.Disconnect()
				<-s.client.ConnectChan()
			}()
			break
		}
	}
	if !connected {
{'fraction_non_alphanumeric': 0.17497411625499187, 'fraction_numerical': 0.004437213429965981, 'mean_word_length': 2.7776536312849163, 'pattern_counts': {'":': 0, '<': 17, '<?xml version=': 0, '>': 5, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '7410559', 'n_tokens_mistral': 2432, 'n_tokens_neox': 2250, 'n_words': 663}