<commit_before>package protocol

import (
	"github.com/AutogrowSystems/faye-go/utils"
	"reflect"
	"sync"
	"time"
)

type MsgStore interface {
	EnqueueMessages([]Message)
	GetAndClearMessages() []Message
}

// Connect requests starts a session
type Session struct {
	conn     Connection
	timeout  int
	response Message
	client   *Client
	started  time.Time
	logger   utils.Logger
}

func NewSession(client *Client, conn Connection, timeout int, response Message, logger utils.Logger) *Session {
	session := Session{conn, timeout, response, client, time.Now(), logger}
	if timeout > 0 {
		go func() {
			time.Sleep(time.Duration(timeout) * time.Millisecond)
			session.End()
		}()
	}
	return &session
}

func (s Session) End() {
	// c.mutex.Lock()
	// defer c.mutex.Unlock()

	if s.conn.IsConnected() {
		s.conn.Send([]Message{s.response})
	} else {
		s.logger.Debugf("No longer connected %s", s.client.clientId)
	}
}

type Client struct {
	clientId    string
	connection  Connection
	msgStore    MsgStore
	isConnected bool
	responseMsg Message
	mutex       sync.Mutex
	lastSession *Session
	created     time.Time
	logger      utils.Logger
}

func NewClient(clientId string, msgStore MsgStore, logger utils.Logger) Client {
	client := Client{
		clientId:    clientId,
		msgStore:    msgStore,
		isConnected: false,
		created:     time.Now(),
		logger:      logger,
	}

	return client
}

func (c Client) Id() string {
	return c.clientId
}

func (c *Client) Connect(timeout int, interval int, responseMsg Message, connection Connection) {
	c.mutex.Lock()
	defer c.mutex.Unlock()

	c.lastSession = NewSession(c, connection, timeout, responseMsg, c.logger)
	c.responseMsg = responseMsg

	c.flushMsgs()
}

func (c *Client) SetConnection(connection Connection) {
	c.mutex.Lock()
	defer c.mutex.Unlock()

	if c.connection == nil || connection.Priority() > c.connection.Priority() {
		c.connection = connection
		c.isConnected = true
	}
}

func (c Client) Queue(msg Message) {
	c.mutex.Lock()
	defer c.mutex.Unlock()

	c.msgStore.EnqueueMessages([]Message{msg})
	c.flushMsgs()
}

func (c Client) QueueMany(msgs []Message) {<commit_msg>made sure to set the client ID on any messages being queued on a client<commit_after>package protocol

import (
	"github.com/AutogrowSystems/faye-go/utils"
	"reflect"
	"sync"
	"time"
)

type MsgStore interface {
	EnqueueMessages([]Message)
	GetAndClearMessages() []Message
}

// Connect requests starts a session
type Session struct {
	conn     Connection
	timeout  int
	response Message
	client   *Client
	started  time.Time
	logger   utils.Logger
}

func NewSession(client *Client, conn Connection, timeout int, response Message, logger utils.Logger) *Session {
	session := Session{conn, timeout, response, client, time.Now(), logger}
	if timeout > 0 {
		go func() {
			time.Sleep(time.Duration(timeout) * time.Millisecond)
			session.End()
		}()
	}
	return &session
}

func (s Session) End() {
	// c.mutex.Lock()
	// defer c.mutex.Unlock()

	if s.conn.IsConnected() {
		s.conn.Send([]Message{s.response})
	} else {
		s.logger.Debugf("No longer connected %s", s.client.clientId)
	}
}

type Client struct {
	clientId    string
	connection  Connection
	msgStore    MsgStore
	isConnected bool
	responseMsg Message
	mutex       sync.Mutex
	lastSession *Session
	created     time.Time
	logger      utils.Logger
}

func NewClient(clientId string, msgStore MsgStore, logger utils.Logger) Client {
	client := Client{
		clientId:    clientId,
		msgStore:    msgStore,
		isConnected: false,
		created:     time.Now(),
		logger:      logger,
	}

	return client
}

func (c Client) Id() string {
	return c.clientId
}

func (c *Client) Connect(timeout int, interval int, responseMsg Message, connection Connection) {
	c.mutex.Lock()
	defer c.mutex.Unlock()

	c.lastSession = NewSession(c, connection, timeout, responseMsg, c.logger)
	c.responseMsg = responseMsg

	c.flushMsgs()
}

func (c *Client) SetConnection(connection Connection) {
	c.mutex.Lock()
	defer c.mutex.Unlock()

	if c.connection == nil || connection.Priority() > c.connection.Priority() {
		c.connection = connection
		c.isConnected = true
	}
}

func (c Client) Queue(msg Message) {
	c.mutex.Lock()
	defer c.mutex.Unlock()
	msg.SetClientId(c.clientId)
	c.msgStore.EnqueueMessages([]Message{msg})
	c.flushMsgs()
}

func (c Client) QueueMany(msgs []Message) {
{'fraction_non_alphanumeric': 0.12038745387453874, 'fraction_numerical': 0.00046125461254612545, 'mean_word_length': 3.9061085972850678, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 7, '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': '6770425', 'n_tokens_mistral': 1492, 'n_tokens_neox': 1446, 'n_words': 381}