<commit_before>package gpubsub

import (
	"net"
	"encoding/gob"
	"bytes"
	"io"
	"log"
)

type Publisher struct {
	conn net.Conn
	enc *gob.Encoder
	stop bool
}

type Subscriber struct {
	conn net.Conn
	stop bool
}

type SubscriberCallback func(event interface{})

func (p *Subscriber) Connect(url string) error {
	conn, err := net.Dial("tcp", url)
	if err != nil {
		return err
	}
	p.conn = conn
	p.stop = false
	return nil
}

func (p *Subscriber) Subscribe(topic string, sc SubscriberCallback, obj interface{}) error {

	var buff bytes.Buffer 
	enc := gob.NewEncoder(&buff)

	m := Message{Sub, topic, nil}
	err := enc.Encode(m)
	p.conn.Write(buff.Bytes())
	
	go func() {
		dec := gob.NewDecoder(p.conn)
		n := Message{}
		for ; !p.stop;  {
			//TODO: Put reader timetout to cancel more cleanly
			err := dec.Decode(&n)
			if err == io.EOF {
				break;
			} else if err != nil {
				enc := gob.NewDecoder(bytes.NewBuffer(n.Data));
				err = enc.Decode(obj)
				if err != nil {
					log.Fatal("decode error: ", err)
					break;
				}

				sc(obj)
			}
		}
		
		p.conn.Close();
	}()
	
	return err
}

func (p *Subscriber) Unsubscribe(topic string) {
	p.conn.Close();
	p.stop = true
}

func (p *Subscriber) Disconnect() {
	p.conn.Close()
}


/* ----------------------------------- */

func (p *Publisher) Connect(url string) error {
	conn, err := net.Dial("tcp", url)
	if err != nil {
		return err
	}
	p.conn = conn
	p.stop = false
	p.enc = gob.NewEncoder(p.conn)

	m := Message{Pub, "", nil}
	err = p.enc.Encode(m)
	return err
}

func (p *Publisher) Publish(topic string, event interface{}) error {
	var mbytes bytes.Buffer
	enc := gob.NewEncoder(&mbytes);
	enc.Encode(event)

	m := Message{Pub, topic, mbytes.Bytes()}
	err := p.enc.Encode(m)
	return err
}

func (p *Publisher) Disconnect() {
	p.conn.Close()
}<commit_msg>Using json because gob.NewDecoder is very slow!<commit_after>package gpubsub

import (
	"net"
	"encoding/gob"
	"encoding/json"
	"bytes"
	"io"
	"log"
)

type Publisher struct {
	conn net.Conn
	enc *gob.Encoder
	stop bool
}

type Subscriber struct {
	conn net.Conn
	stop bool
}

type SubscriberCallback func(event interface{})

func (p *Subscriber) Connect(url string) error {
	conn, err := net.Dial("tcp", url)
	if err != nil {
		return err
	}
	p.conn = conn
	p.stop = false
	return nil
}

func (p *Subscriber) Subscribe(topic string, sc SubscriberCallback, obj interface{}) error {

	var buff bytes.Buffer 
	enc := gob.NewEncoder(&buff)

	m := Message{Sub, topic, nil}
	err := enc.Encode(m)
	p.conn.Write(buff.Bytes())
	
	go func() {
		dec := gob.NewDecoder(p.conn)
		n := Message{}
		
		for ; !p.stop;  {
			//TODO: Put reader timetout to cancel more cleanly
			err := dec.Decode(&n)
			if err == io.EOF {
				break;
			} else if err != nil {
				err = json.Unmarshal(n.Data, obj)
				if err != nil {
					log.Fatal("decode error: ", err)
					break;
				}

				sc(obj)
			}
		}
		
		p.conn.Close();
	}()
	
	return err
}

func (p *Subscriber) Unsubscribe(topic string) {
	p.conn.Close();
	p.stop = true
}

func (p *Subscriber) Disconnect() {
	p.conn.Close()
}


/* ----------------------------------- */

func (p *Publisher) Connect(url string) error {
	conn, err := net.Dial("tcp", url)
	if err != nil {
		return err
	}
	p.conn = conn
	p.stop = false
	p.enc = gob.NewEncoder(p.conn)

	m := Message{Pub, "", nil}
	err = p.enc.Encode(m)
	return err
}

func (p *Publisher) Publish(topic string, event interface{}) error {
	
	ebytes, err := json.Marshal(event);
	if err != nil {
		return err
	}

	m := Message{Pub, topic, ebytes}
	err = p.enc.Encode(m)
	return err
}

func (p *Publisher) Disconnect() {
	p.conn.Close()
}
{'fraction_non_alphanumeric': 0.17189218921892188, 'fraction_numerical': 0.0, 'mean_word_length': 2.8773987206823026, 'pattern_counts': {'":': 0, '<': 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': '7563309', 'n_tokens_mistral': 1488, 'n_tokens_neox': 1356, 'n_words': 368}