func (c *NetClient) controlOutQueue() error {
	c.log.Debugf("Queue controller started")
	for {
		select {
		case <-c.haltedCh:
			c.log.Infof("Halting controlOutQueue")
			return nil
		case realPacket := <-c.outQueue:
			response, err := c.send(realPacket, c.Provider.Host, c.Provider.Port)
			if err != nil {
				c.log.Errorf("Could not send real packet: %v", err)
			}
			c.log.Debugf("Real packet was sent")
			c.log.Debugf("Received response: %v", response)
		default:
			if !c.cfg.Debug.RateCompliantCoverMessagesDisabled {
				dummyPacket, err := c.createLoopCoverMessage()
				if err != nil {
					return err
				}
				response, err := c.send(dummyPacket, c.Provider.Host, c.Provider.Port)
				if err != nil {
					c.log.Errorf("Could not send dummy packet: %v", err)
				}
				c.log.Debugf("Dummy packet was sent")
				c.log.Debugf("Received response: %v", response)
			}
		}
		err := delayBeforeContinue(c.cfg.Debug.MessageSendingRate)
		if err != nil {
			return err
		}
	}
}
controlOutQueue controls the outgoing queue of the client.
If a message awaits in the queue, it is sent. Otherwise a
drop cover message is sent instead.
package client
import (
	"encoding/base64"
	"errors"
	"fmt"
	"io/ioutil"
	"math"
	"net"
	"os"
	"sync"
	"time"
	"github.com/golang/protobuf/proto"
	"github.com/nymtech/nym-directory/models"
	clientConfig "github.com/nymtech/nym-mixnet/client/config"
	"github.com/nymtech/nym-mixnet/clientcore"
	"github.com/nymtech/nym-mixnet/config"
	"github.com/nymtech/nym-mixnet/constants"
	"github.com/nymtech/nym-mixnet/flags"
	"github.com/nymtech/nym-mixnet/helpers"
	"github.com/nymtech/nym-mixnet/helpers/topology"
	"github.com/nymtech/nym-mixnet/logger"
	"github.com/nymtech/nym-mixnet/networker"
	"github.com/nymtech/nym-mixnet/sphinx"
	"github.com/sirupsen/logrus"
)
const (
	loopLoad = "LoopCoverMessage"
)
type Client interface {
	networker.NetworkClient
	networker.NetworkServer
	Start() error
	SendMessage(message []byte, recipient config.ClientConfig) error
	ReadInNetworkFromTopology(pkiName string) error
}
type ReceivedMessages struct {
	sync.Mutex
	messages [][]byte
}
type NetClient struct {
	*clientcore.CryptoClient
	cfg              *clientConfig.Config
	config           config.ClientConfig
	token            []byte  
	outQueue         chan []byte
	haltedCh         chan struct{}
	haltOnce         sync.Once
	log              *logrus.Logger
	receivedMessages ReceivedMessages
}
func (c *NetClient) GetReceivedMessages() [][]byte {
	c.receivedMessages.Lock()
	defer c.receivedMessages.Unlock()
	msgsPtr := c.receivedMessages.messages
	c.receivedMessages.messages = make([][]byte, 0, 20)
	return msgsPtr
}
func (c *NetClient) addNewMessage(msg []byte) {
	c.receivedMessages.Lock()
	defer c.receivedMessages.Unlock()
	c.receivedMessages.messages = append(c.receivedMessages.messages, msg)
}
func (c *NetClient) OutQueue() chan<- []byte {
	return c.outQueue
}
func getProvider(presences []models.MixProviderPresence, pubKey string) (models.MixProviderPresence, error) {
	var pres models.MixProviderPresence
	for _, presence := range presences {
		if presence.PubKey == pubKey {
			return presence, nil
		}
	}
	return pres, fmt.Errorf("no provider with the given public key exists in the current network")
}
func (c *NetClient) Start() error {
	c.outQueue = make(chan []byte)
	initialTopology, err := topology.GetNetworkTopology(c.cfg.Client.DirectoryServerTopologyEndpoint)
	if err != nil {
		return err
	}
	if err := c.ReadInNetworkFromTopology(initialTopology); err != nil {
		return err
	}
	var providerPresence models.MixProviderPresence
	if providerPresence, err = getProvider(initialTopology.MixProviderNodes, c.cfg.Client.ProviderID); err != nil {
		return fmt.Errorf("specified provider does not seem to be online: %v", c.cfg.Client.ProviderID)
	}
	provider, err := topology.ProviderPresenceToConfig(providerPresence)
	if err != nil {
		return err
	}
	c.Provider = provider
	for {
		if err := c.sendRegisterMessageToProvider(); err != nil {
			c.log.Errorf("Error during registration to provider: %v", err)
			time.Sleep(5 * time.Second)
		} else {
			c.log.Debug("Registration done!")
			break
		}
	}
	for {
		initialTopology, err := topology.GetNetworkTopology(c.cfg.Client.DirectoryServerTopologyEndpoint)
		if err != nil {
			return err
		}
		if err := c.ReadInNetworkFromTopology(initialTopology); err != nil {
			return err
		}
		if len(c.Network.Clients) > 0 {
			break
		}
		c.log.Debug("No registered clients available. Waiting for a second before retrying.")
		time.Sleep(time.Second)
	}
	c.log.Info("Obtained valid network topology")
	c.startTraffic()
	return nil
}
func (c *NetClient) Wait() {
	<-c.haltedCh
}
func (c *NetClient) Shutdown() {
	c.haltOnce.Do(func() { c.halt() })
}
func (c *NetClient) halt() {
	c.log.Infof("Starting graceful shutdown")
	close(c.haltedCh)
}
func (c *NetClient) UpdateNetworkView() error {
	newTopology, err := topology.GetNetworkTopology(c.cfg.Client.DirectoryServerTopologyEndpoint)
	if err != nil {
		c.log.Errorf("error while reading network topology: %v", err)
		return err
	}
	if err := c.ReadInNetworkFromTopology(newTopology); err != nil {
		c.log.Errorf("error while trying to update topology: %v", err)
		return err
	}
	return nil
}
func (c *NetClient) checkTopology() error {
	if c.Network.ShouldUpdate() {
		return c.UpdateNetworkView()
	}
	return nil
}
func (c *NetClient) GetOwnDetails() *config.ClientConfig {
	return &c.config
}
func (c *NetClient) GetAllPossibleRecipients() []*config.ClientConfig {
	if c.UpdateNetworkView() != nil {
		return nil
	}
	clients := make([]*config.ClientConfig, len(c.Network.Clients))
	for i := range c.Network.Clients {
		clients[i] = &c.Network.Clients[i]
	}
	return clients
}
func (c *NetClient) SendMessage(message []byte, recipient config.ClientConfig) error {
	if err := c.checkTopology(); err != nil {
		c.log.Errorf("error in updating topology: %v", err)
		return err
	}
	packet, err := c.encodeMessage(message, recipient)
	if err != nil {
		c.log.Errorf("Error in sending message - encode message returned error: %v", err)
		return err
	}
	c.outQueue <- packet
	return nil
}
func (c *NetClient) encodeMessage(message []byte, recipient config.ClientConfig) ([]byte, error) {
	sphinxPacket, err := c.EncodeMessage(message, recipient)
	if err != nil {
		c.log.Errorf("Error in sending message - create sphinx packet returned an error: %v", err)
		return nil, err
	}
	packetBytes, err := config.WrapWithFlag(flags.CommFlag, sphinxPacket)
	if err != nil {
		c.log.Errorf("Error in sending message - wrap with flag returned an error: %v", err)
		return nil, err
	}
	return packetBytes, nil
}
func (c *NetClient) send(packet []byte, host string, port string) (config.ProviderResponse, error) {
	conn, err := net.Dial("tcp", net.JoinHostPort(host, port))
	if err != nil {
		c.log.Errorf("Error in send - dial returned an error: %v", err)
		return config.ProviderResponse{}, err
	}
	defer conn.Close()
	if _, err := conn.Write(packet); err != nil {
		c.log.Errorf("Failed to write to connection: %v", err)
		return config.ProviderResponse{}, err
	}
	buff, err := ioutil.ReadAll(conn)
	if err != nil {
		c.log.Errorf("Failed to read response: %v", err)
		return config.ProviderResponse{}, err
	}
	var resPacket config.ProviderResponse
	if err = proto.Unmarshal(buff, &resPacket); err != nil {
		c.log.Errorf("Error while unmarshalling received packet: %v", err)
		return config.ProviderResponse{}, err
	}
	return resPacket, nil
}
func (c *NetClient) registerToken(token []byte) {
	c.token = token
	c.log.Debugf("Registered token %s", c.token)
}
func (c *NetClient) processPacket(packet []byte) ([]byte, error) {
	if len(packet) > 38 {
		return packet[38:], nil
	}
	return packet, nil
}
func (c *NetClient) startTraffic() {
	go func() {
		err := c.controlOutQueue()
		if err != nil {
			c.log.Fatalf("Error in the controller of the outgoing packets queue. Possible security threat.: %v", err)
		}
	}()
	if c.cfg.Debug.LoopCoverTrafficRate > 0.0 {
		c.turnOnLoopCoverTraffic()
	}
	if c.cfg.Debug.FetchMessageRate > 0.0 {
		go func() {
			c.controlMessagingFetching()
		}()
	}
}
func (c *NetClient) sendRegisterMessageToProvider() error {
	c.log.Debugf("Sending request to provider to register")
	confBytes, err := proto.Marshal(&c.config)
	if err != nil {
		c.log.Errorf("Error in register provider - marshal of provider config returned an error: %v", err)
		return err
	}
	pktBytes, err := config.WrapWithFlag(flags.AssignFlag, confBytes)
	if err != nil {
		c.log.Errorf("Error in register provider - wrap with flag returned an error: %v", err)
		return err
	}
	response, err := c.send(pktBytes, c.Provider.Host, c.Provider.Port)
	if err != nil {
		c.log.Errorf("Error in register provider - send registration packet returned an error: %v", err)
		return err
	}
	packets, err := config.UnmarshalProviderResponse(response)
	if err != nil || len(packets) != 1 {
		c.log.Errorf("error in register provider - failed to unmarshal response: %v", err)
	}
	c.registerToken(packets[0].Data)
	return nil
}
func (c *NetClient) getMessagesFromProvider() error {
	pullRqs := config.PullRequest{ClientPublicKey: c.GetPublicKey().Bytes(), Token: c.token}
	pullRqsBytes, err := proto.Marshal(&pullRqs)
	if err != nil {
		c.log.Errorf("Error in register provider - marshal of pull request returned an error: %v", err)
		return err
	}
	pktBytes, err := config.WrapWithFlag(flags.PullFlag, pullRqsBytes)
	if err != nil {
		c.log.Errorf("Error in register provider - marshal of provider config returned an error: %v", err)
		return err
	}
	response, err := c.send(pktBytes, c.Provider.Host, c.Provider.Port)
	if err != nil {
		return err
	}
	packets, err := config.UnmarshalProviderResponse(response)
	if err != nil {
		c.log.Errorf("error in register provider - failed to unmarshal response: %v", err)
	}
	for _, packet := range packets {
		packetData, err := c.processPacket(packet.Data)
		if err != nil {
			c.log.Errorf("Error in processing received packet: %v", err)
		}
		packetDataStr := string(packetData)
		switch packetDataStr {
		case loopLoad:
			c.log.Debugf("Received loop cover message %v", packetDataStr)
		default:
			c.log.Infof("Received new message: %v", packetDataStr)
			c.addNewMessage(packetData)
		}
	}
	return nil
}