<commit_before>package mqttclient

import (
	"crypto/tls"
	"errors"
	"fmt"
	mqtt "github.com/clearblade/mqtt_parsing"
	"io"
	mrand "math/rand"
	"net"
	"sync"
	"time"
)

type Client struct {

	//this is all clearblade specific
	SystemKey    string
	SystemSecret string
	AuthToken    string
	Clientid     string
	//this will usually be occupied
	//by a net.Conn
	C                   io.ReadWriteCloser
	internalOutgoingBuf chan []byte
	Timeout             time.Duration
	//thou shalt type channels consumed by others
	ClientErrorBuffer chan error

	internalErrorBuffer chan *errWrap
	shutdown_reader     chan struct{}
	shutdown_writer     chan struct{}
	//introduce a sync write mode?
	last_timeout_reccd time.Time

	shutting_down bool

	msg_store                *storage
	subscriptions            *outgoing_topics
	waiting_for_subscription *subscription_store
	//TODO:redesign around a thread-local
	//rng
	rando    *mrand.Rand
	randomut *sync.RWMutex
}

var (
	Verbose bool
)

//Start connects to the mqtt broker. It does not send the connect packet. Use the SendConnect function for that.
func (c *Client) Start(addr string, ssl *tls.Config) error {
	var con net.Conn
	var err error
	if ssl != nil {
		con, err = tls.Dial("tcp", addr, ssl)
	} else {
		con, err = net.Dial("tcp", addr)
	}
	if err != nil {
		return err
	}
	c.C = con

	go c.connectionWriter()
	go c.connectionListener()
	go c.errorTree()
	return nil
}

//NewClient allocates a new client. It is supplied with the (in order of appearance)
//Token, SystemKey,SystemSecret,Clientid, and the mqtt timeout
//Note that the following combinations of (Token|SystemKey|SystemSecret) are allowed
//(Token && SystemKey), (SystemKey && SystemSecret)
func NewClient(tok, sk, ss, cid string, timeout int) *Client {
	client := &Client{
		msg_store:                newStorage(),
		waiting_for_subscription: newSubscriptionStore(),
		subscriptions:            newOutgoingTopics(),
		SystemSecret:             ss,
		SystemKey:                sk,
		AuthToken:                tok,
		Clientid:                 cid,
		Timeout:                  time.Duration(timeout) * time.Second,
		//internalOutgoingBuf:      make(chan []byte, 30),
		internalOutgoingBuf: make(chan []byte),
		ClientErrorBuffer:   make(chan error, 10),
		internalErrorBuffer: make(chan *errWrap, 2),
		shutdown_reader:     make(chan struct{}, 1),
		shutdown_writer:     make(chan struct{}, 1),
		rando:               mrand.New(mrand.NewSource(time.Now().UnixNano())),
		randomut:            new(sync.RWMutex),
	}
	return client
}

//sendMessage is an internal function that acts as a central point
//of failure for all of the message sending channels
//sort of like a fan-in, except this simply allows us to do
// all of the error handling logic in one place
func (c *Client) sendMessage(m mqtt.Message) error {
	_, err := c.C.Write(m.Encode())
	if err != nil {
		c.internalErrorBuffer <- &errWrap{
			err:      err,
			reciever: _CON_WRITER,
		}
		return err
	}
	return nil
}

//connectionWriter is an internal function. it essentially sits in a goroutine and writes to the connection
//whenever it recieves a message over the channel
func (c *Client) connectionWriter() {
	if c.C == nil {
		return
	}
	for {
		select {
		case out := <-c.internalOutgoingBuf:
			_, err := c.C.Write(out)
			if err != nil {
				c.internalErrorBuffer <- &errWrap{
					err:      err,
					reciever: _CON_WRITER,
				}
				return
			}
		case <-c.shutdown_writer:
			return
		}
	}
}

//connectionListener is another function that sits on a goroutine.
//DecodePacket blocks until it reads a complete mqtt packet
func (c *Client) connectionListener() {
	if c.C == nil {
		return
	}
	mch, ech, shutdown := make(chan mqtt.Message, 10), make(chan error, 1), false
	//we have to establish an internal chain of goroutines here
	//otherwise we couldn't shutdown the listener on demand
	//since it's really hard to coordinate all the shutting down
	//when a connection drops
	//we're waiting for the connection listener to simply fail
	//this allows us to handle it a bit more gracefully
	//in order to shut down via channels directly we'd have to
	//wait for the read to fail anyway.
	go func(m chan mqtt.Message, e chan error) {
		for {
			msg, err := mqtt.DecodePacket(c.C)
			if err != nil {
				shutdown = true
				e <- err
				return
			}
			m <- msg
		}
	}(mch, ech)
	sentPing := false
	for {
		select {

		case msg := <-mch:
			//dispatch the internal functions
			c.dispatch(msg)
			sentPing = false
		case e := <-ech:
			//an error was recieved from the listenr
			if !shutdown {
				c.internalErrorBuffer <- &errWrap{
					err:      e,
					reciever: _CON_READER,
				}
			}
		case <-time.After(c.Timeout):
			//keepalive
			if !sentPing {
				c.sendMessage(&mqtt.Pingreq{})
				sentPing = true
			} else if time.Since(c.last_timeout_reccd) > c.Timeout && sentPing {
				//there was a timeout
				c.Shutdown(true)
			}

		case <-c.shutdown_reader:
			shutdown = true
			return
		}
	}
}

//this is the "business logic" of the client. it decides how each packet mutates the
//internal state of the client
func (c *Client) dispatch(msg mqtt.Message) {
	//for example we make a decision if the client sees this request or not
	//or if we have to send another message in a flow
	c.last_timeout_reccd = time.Now()
	switch msg.Type() {
	case mqtt.CONNECT:
		//shouldn't happen?
	case mqtt.CONNACK:<commit_msg>broke out keep alive ping into it's own goroutine<commit_after>package mqttclient

import (
	"crypto/tls"
	"errors"
	"fmt"
	mqtt "github.com/clearblade/mqtt_parsing"
	"io"
	mrand "math/rand"
	"net"
	"sync"
	"time"
)

type Client struct {

	//this is all clearblade specific
	SystemKey    string
	SystemSecret string
	AuthToken    string
	Clientid     string
	//this will usually be occupied
	//by a net.Conn
	C                   io.ReadWriteCloser
	internalOutgoingBuf chan []byte
	Timeout             time.Duration
	//thou shalt type channels consumed by others
	ClientErrorBuffer chan error

	internalErrorBuffer chan *errWrap
	shutdown_reader     chan struct{}
	shutdown_writer     chan struct{}
	//introduce a sync write mode?
	last_timeout_reccd time.Time

	shutting_down bool

	msg_store                *storage
	subscriptions            *outgoing_topics
	waiting_for_subscription *subscription_store
	//TODO:redesign around a thread-local
	//rng
	rando    *mrand.Rand
	randomut *sync.RWMutex
}

var (
	Verbose bool
)

//Start connects to the mqtt broker. It does not send the connect packet. Use the SendConnect function for that.
func (c *Client) Start(addr string, ssl *tls.Config) error {
	var con net.Conn
	var err error
	if ssl != nil {
		con, err = tls.Dial("tcp", addr, ssl)
	} else {
		con, err = net.Dial("tcp", addr)
	}
	if err != nil {
		return err
	}
	c.C = con

	go c.connectionWriter()
	go c.connectionListener()
	go c.errorTree()
	return nil
}

//NewClient allocates a new client. It is supplied with the (in order of appearance)
//Token, SystemKey,SystemSecret,Clientid, and the mqtt timeout
//Note that the following combinations of (Token|SystemKey|SystemSecret) are allowed
//(Token && SystemKey), (SystemKey && SystemSecret)
func NewClient(tok, sk, ss, cid string, timeout int) *Client {
	client := &Client{
		msg_store:                newStorage(),
		waiting_for_subscription: newSubscriptionStore(),
		subscriptions:            newOutgoingTopics(),
		SystemSecret:             ss,
		SystemKey:                sk,
		AuthToken:                tok,
		Clientid:                 cid,
		Timeout:                  time.Duration(timeout) * time.Second,
		//internalOutgoingBuf:      make(chan []byte, 30),
		internalOutgoingBuf: make(chan []byte),
		ClientErrorBuffer:   make(chan error, 10),
		internalErrorBuffer: make(chan *errWrap, 2),
		shutdown_reader:     make(chan struct{}, 1),
		shutdown_writer:     make(chan struct{}, 1),
		rando:               mrand.New(mrand.NewSource(time.Now().UnixNano())),
		randomut:            new(sync.RWMutex),
	}
	return client
}

//sendMessage is an internal function that acts as a central point
//of failure for all of the message sending channels
//sort of like a fan-in, except this simply allows us to do
// all of the error handling logic in one place
func (c *Client) sendMessage(m mqtt.Message) error {
	_, err := c.C.Write(m.Encode())
	if err != nil {
		c.internalErrorBuffer <- &errWrap{
			err:      err,
			reciever: _CON_WRITER,
		}
		return err
	}
	return nil
}

//connectionWriter is an internal function. it essentially sits in a goroutine and writes to the connection
//whenever it recieves a message over the channel
func (c *Client) connectionWriter() {
	if c.C == nil {
		return
	}
	for {
		select {
		case out := <-c.internalOutgoingBuf:
			_, err := c.C.Write(out)
			if err != nil {
				c.internalErrorBuffer <- &errWrap{
					err:      err,
					reciever: _CON_WRITER,
				}
				return
			}
		case <-c.shutdown_writer:
			return
		}
	}
}

//connectionListener is another function that sits on a goroutine.
//DecodePacket blocks until it reads a complete mqtt packet
func (c *Client) connectionListener() {
	if c.C == nil {
		return
	}
	mch, ech, shutdown := make(chan mqtt.Message, 10), make(chan error, 1), false
	//we have to establish an internal chain of goroutines here
	//otherwise we couldn't shutdown the listener on demand
	//since it's really hard to coordinate all the shutting down
	//when a connection drops
	//we're waiting for the connection listener to simply fail
	//this allows us to handle it a bit more gracefully
	//in order to shut down via channels directly we'd have to
	//wait for the read to fail anyway.
	go func(m chan mqtt.Message, e chan error) {
		for {
			msg, err := mqtt.DecodePacket(c.C)
			if err != nil {
				shutdown = true
				e <- err
				return
			}
			m <- msg
		}
	}(mch, ech)

	heardFromServer := true
	go func(cli *Client) {
		for {
			time.Sleep(cli.Timeout)
			if heardFromServer {
				cli.sendMessage(&mqtt.Pingreq{})
				heardFromServer = false
			} else if time.Since(cli.last_timeout_reccd) > cli.Timeout && !heardFromServer {
				//there was a timeout
				cli.Shutdown(true)
			}
		}
	}(c)
	for {
		select {

		case msg := <-mch:
			//dispatch the internal functions
			c.dispatch(msg)
			heardFromServer = true
		case e := <-ech:
			//an error was recieved from the listenr
			if !shutdown {
				c.internalErrorBuffer <- &errWrap{
					err:      e,
					reciever: _CON_READER,
				}
			}
		case <-c.shutdown_reader:
			shutdown = true
			return
		}
	}
}

//this is the "business logic" of the client. it decides how each packet mutates the
//internal state of the client
func (c *Client) dispatch(msg mqtt.Message) {
	//for example we make a decision if the client sees this request or not
	//or if we have to send another message in a flow
	c.last_timeout_reccd = time.Now()
	switch msg.Type() {
	case mqtt.CONNECT:
		//shouldn't happen?
	case mqtt.CONNACK:
{'fraction_non_alphanumeric': 0.09831099687901597, 'fraction_numerical': 0.0018358729575913347, 'mean_word_length': 2.916247304097771, 'pattern_counts': {'":': 0, '<': 24, '<?xml version=': 0, '>': 5, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, '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': '6732622', 'n_tokens_mistral': 3905, 'n_tokens_neox': 3533, 'n_words': 1208}