<commit_before>package susigo

import (
	"crypto/tls"
	"encoding/json"
	"errors"
	"log"
	"net"
	"regexp"
	"strconv"
	"time"
)

type Callback func(*Event)

type Susi struct {
	cert             tls.Certificate
	addr             string
	connected        bool
	conn             net.Conn
	encoder          *json.Encoder
	decoder          *json.Decoder
	callbacks        map[string]Callback
	consumers        map[string]map[int64]Callback
	processors       map[string]map[int64]Callback
	publishProcesses map[string][]Callback
}

type Event struct {
	Topic     string              `json:"topic"`
	Payload   interface{}         `json:"payload"`
	Headers   []map[string]string `json:"headers"`
	ID        string              `json:"id"`
	SessionID string              `json:"sessionid"`
}

type Message struct {
	Type string `json:"type"`
	Data *Event `json:"data"`
}

func NewSusi(addr, certFile, keyFile string) (*Susi, error) {
	susi := new(Susi)
	cert, err := tls.LoadX509KeyPair(certFile, keyFile)
	if err != nil {
		return nil, err
	}
	susi.cert = cert
	susi.callbacks = make(map[string]Callback)
	susi.addr = addr
	susi.consumers = make(map[string]map[int64]Callback)
	susi.processors = make(map[string]map[int64]Callback)
	susi.publishProcesses = make(map[string][]Callback)
	susi.connected = false
	go susi.backend()
	return susi, nil
}

func (susi *Susi) Publish(event Event, callback Callback) error {
	id := strconv.FormatInt(time.Now().UnixNano(), 10)
	event.ID = id
	packet := map[string]interface{}{
		"type": "publish",
		"data": event,
	}
	susi.callbacks[id] = callback
	return susi.encoder.Encode(packet)
}

func (susi *Susi) RegisterConsumer(topic string, callback Callback) (int64, error) {
	if susi.consumers[topic] == nil {
		susi.consumers[topic] = make(map[int64]Callback)
	}
	consumers := susi.consumers[topic]
	id := time.Now().UnixNano()
	consumers[id] = callback
	if len(consumers) == 1 {
		packet := map[string]interface{}{
			"type": "registerConsumer",
			"data": map[string]interface{}{
				"topic": topic,
			},
		}
		return id, susi.encoder.Encode(packet)
	}<commit_msg>readded keeping of event ids;<commit_after>package susigo

import (
	"crypto/tls"
	"encoding/json"
	"errors"
	"log"
	"net"
	"regexp"
	"strconv"
	"time"
)

type Callback func(*Event)

type Susi struct {
	cert             tls.Certificate
	addr             string
	connected        bool
	conn             net.Conn
	encoder          *json.Encoder
	decoder          *json.Decoder
	callbacks        map[string]Callback
	consumers        map[string]map[int64]Callback
	processors       map[string]map[int64]Callback
	publishProcesses map[string][]Callback
}

type Event struct {
	Topic     string              `json:"topic"`
	Payload   interface{}         `json:"payload"`
	Headers   []map[string]string `json:"headers"`
	ID        string              `json:"id"`
	SessionID string              `json:"sessionid"`
}

type Message struct {
	Type string `json:"type"`
	Data *Event `json:"data"`
}

func NewSusi(addr, certFile, keyFile string) (*Susi, error) {
	susi := new(Susi)
	cert, err := tls.LoadX509KeyPair(certFile, keyFile)
	if err != nil {
		return nil, err
	}
	susi.cert = cert
	susi.callbacks = make(map[string]Callback)
	susi.addr = addr
	susi.consumers = make(map[string]map[int64]Callback)
	susi.processors = make(map[string]map[int64]Callback)
	susi.publishProcesses = make(map[string][]Callback)
	susi.connected = false
	go susi.backend()
	return susi, nil
}

func (susi *Susi) Publish(event Event, callback Callback) error {
	var id = event.ID
	if event.ID == "" {
		id = strconv.FormatInt(time.Now().UnixNano(), 10)
		event.ID = id
	}
	event.ID = id
	packet := map[string]interface{}{
		"type": "publish",
		"data": event,
	}
	susi.callbacks[id] = callback
	return susi.encoder.Encode(packet)
}

func (susi *Susi) RegisterConsumer(topic string, callback Callback) (int64, error) {
	if susi.consumers[topic] == nil {
		susi.consumers[topic] = make(map[int64]Callback)
	}
	consumers := susi.consumers[topic]
	id := time.Now().UnixNano()
	consumers[id] = callback
	if len(consumers) == 1 {
		packet := map[string]interface{}{
			"type": "registerConsumer",
			"data": map[string]interface{}{
				"topic": topic,
			},
		}
		return id, susi.encoder.Encode(packet)
	}
{'fraction_non_alphanumeric': 0.13756737754862902, 'fraction_numerical': 0.00843684087180689, 'mean_word_length': 3.516402116402116, 'pattern_counts': {'":': 10, '<': 3, '<?xml version=': 0, '>': 3, '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': '1107537', 'n_tokens_mistral': 1584, 'n_tokens_neox': 1489, 'n_words': 329}