<commit_before>package message

import (
	"encoding/json"
	"log"
	"sync"
	"time"

	"github.com/gorilla/websocket"
)

// Receiver can read Buttplug server messages from a websocket to multiple
// readers. Readers can subscribe/unsubscribe from receiving messages.
type Receiver struct {
	once sync.Once // Make sure Stop() is execute only once.
	conn *websocket.Conn
	hub  *hub
}

// NewReceiver creates a Receiver for the given websocket connection. Done
// channel is closed then receiver is done.
func NewReceiver(conn *websocket.Conn, done chan struct{}) *Receiver {
	r := &Receiver{
		conn: conn,
		hub:  newHub(),
	}
	go r.run(done)
	return r
}

// Run reads a message from the websocket connection and puts it on the hub.
func (rc *Receiver) run(done chan struct{}) {
	for {
		messageType, r, err := rc.conn.NextReader()
		if err != nil {
			rc.conn.Close()
			close(done)
			return
		}
		if messageType != websocket.TextMessage {
			log.Println("incoming message is not a text message")
			continue
		}
		var msgs IncomingMessages
		e := json.NewDecoder(r)
		if err := e.Decode(&msgs); err != nil {
			log.Printf("error unmarshaling message: %v", err)
			continue
		}
		for _, msg := range msgs {
			rc.hub.incoming <- msg
		}
	}

}

// Subscribe creates a new reader that receives messages. A consumer should
// call the Unsubscribe when it's done with the reader.
func (rc *Receiver) Subscribe() *Reader {
	r := &Reader{
		buf: make(chan IncomingMessage, 10),
	}
	rc.hub.subscribe <- r
	return r
}

// Unsubscribe removes the readers subscription and will no longer receive
// messages.
func (rc *Receiver) Unsubscribe(r *Reader) {
	select {
	case rc.hub.unsubscribe <- r:
	case <-time.After(time.Second * 10):
	}
}

// Stop the receiver from sending any messages.
func (rc *Receiver) Stop() {
	rc.once.Do(func() {
		rc.hub.stop <- true
	})
}

// Reader is receives messages from the Receiver subscription.
type Reader struct {
	// buffered channel for subscriber to read
	buf chan IncomingMessage<commit_msg>Grafefully handle unsubscribe from stopped receiver<commit_after>package message

import (
	"encoding/json"
	"log"
	"sync"

	"github.com/gorilla/websocket"
)

// Receiver can read Buttplug server messages from a websocket to multiple
// readers. Readers can subscribe/unsubscribe from receiving messages.
type Receiver struct {
	once sync.Once // Make sure Stop() is execute only once.
	conn *websocket.Conn
	hub  *hub
}

// NewReceiver creates a Receiver for the given websocket connection. Done
// channel is closed then receiver is done.
func NewReceiver(conn *websocket.Conn, done chan struct{}) *Receiver {
	r := &Receiver{
		conn: conn,
		hub:  newHub(),
	}
	go r.run(done)
	return r
}

// Run reads a message from the websocket connection and puts it on the hub.
func (rc *Receiver) run(done chan struct{}) {
	for {
		messageType, r, err := rc.conn.NextReader()
		if err != nil {
			rc.conn.Close()
			close(done)
			return
		}
		if messageType != websocket.TextMessage {
			log.Println("incoming message is not a text message")
			continue
		}
		var msgs IncomingMessages
		e := json.NewDecoder(r)
		if err := e.Decode(&msgs); err != nil {
			log.Printf("error unmarshaling message: %v", err)
			continue
		}
		for _, msg := range msgs {
			rc.hub.incoming <- msg
		}
	}

}

// Subscribe creates a new reader that receives messages. A consumer should
// call the Unsubscribe when it's done with the reader.
func (rc *Receiver) Subscribe() *Reader {
	r := &Reader{
		buf: make(chan IncomingMessage, 10),
	}
	rc.hub.subscribe <- r
	return r
}

// Unsubscribe removes the readers subscription and will no longer receive
// messages.
func (rc *Receiver) Unsubscribe(r *Reader) {
	select {
	case rc.hub.unsubscribe <- r:
	case <-rc.hub.stop:
	}
}

// Stop the receiver from sending any messages.
func (rc *Receiver) Stop() {
	rc.once.Do(func() {
		close(rc.hub.stop)
	})
}

// Reader is receives messages from the Receiver subscription.
type Reader struct {
	// buffered channel for subscriber to read
	buf chan IncomingMessage
{'fraction_non_alphanumeric': 0.11091897233201581, 'fraction_numerical': 0.0014822134387351778, 'mean_word_length': 3.464167585446527, 'pattern_counts': {'":': 0, '<': 12, '<?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': '5679822', 'n_tokens_mistral': 1385, 'n_tokens_neox': 1276, 'n_words': 465}