<commit_before>package moh

import (
	"code.google.com/p/go.net/websocket"
	"log"
	"net/http"
)

// Publisher is the counterpart for Subscriber.
// It is a HTTP server accepting websocket connections.
type Publisher struct {
	// Registered filters, holds pointers to open connections.
	// All clients are registered to the "all" key by default for allowing broadcasting.
	// Modifier operations on this type is made by registrar() function.
	filters *Filters
}

// This is the magic subscription key for broadcast events.
// Hoping that it is unique enough to not collide with another key.
const all = "4658f005d49885355f4e771ed9dace10cca9563e"

// NewPublisher creates a new Publisher and returns a pointer to it.  The
// publisher will listen on addr and accept websocket connections from
// Subscribers.
func NewPublisher() *Publisher {
	return &Publisher{filters: NewFilters()}
}

// Publish sends a message to registered Subscribers with the key.
func (p *Publisher) Publish(key string, message []byte) {
	// log.Println("Sending message to send channel", string(message))
	for c := range p.filters.Get(key) {
		select {
		case c.send <- message:
			// log.Println("Message sent to send channel")
		default:
			// Buffer is full, writer() is not fast enough to send all published messages .
			// Drop the websocket client and let it synchronize by re-connecting.
			log.Println("Websocket buffer is full. Dropping socket")
			go c.ws.Close()
		}
	}
}

// Broadcast sends a message to all of the connected Subscribers.
func (p *Publisher) Broadcast(message []byte) {
	p.Publish(all, message)
}

// ServeHTTP implements the http.Handler interface for a WebSocket.
func (p *Publisher) ServeHTTP(w http.ResponseWriter, req *http.Request) {
	websocket.Handler(p.websocketHandler).ServeHTTP(w, req)
}

func (p *Publisher) websocketHandler(ws *websocket.Conn) {
	c := &connection{
		ws:   ws,
		send: make(chan []byte, 256),
		keys: make(map[string]bool),
	}

	p.filters.Add(c, all)
	defer func() {
		p.filters.RemoveAll(c)
		close(c.send)
	}()

	go c.writer()
	c.reader(p.filters)
}

// connection represents a connected Subscriber in Publisher.<commit_msg>Publisher is websocket.Server<commit_after>package moh

import (
	"code.google.com/p/go.net/websocket"
	"log"
)

// Publisher is the counterpart for Subscriber.
// It is a HTTP server accepting websocket connections.
type Publisher struct {
	// Registered filters, holds pointers to open connections.
	// All clients are registered to the "all" key by default for allowing broadcasting.
	// Modifier operations on this type is made by registrar() function.
	filters *Filters

	websocket.Server // implements http.Handler interface
}

// This is the magic subscription key for broadcast events.
// Hoping that it is unique enough to not collide with another key.
const all = "4658f005d49885355f4e771ed9dace10cca9563e"

// NewPublisher creates a new Publisher and returns a pointer to it.  The
// publisher will listen on addr and accept websocket connections from
// Subscribers.
func NewPublisher() *Publisher {
	p := &Publisher{
		filters: NewFilters(),
		Server:  websocket.Server{},
	}
	p.Server.Handler = p.handleWebsocketConn
	return p
}

// Publish sends a message to registered Subscribers with the key.
func (p *Publisher) Publish(key string, message []byte) {
	// log.Println("Sending message to send channel", string(message))
	for c := range p.filters.Get(key) {
		select {
		case c.send <- message:
			// log.Println("Message sent to send channel")
		default:
			// Buffer is full, writer() is not fast enough to send all published messages .
			// Drop the websocket client and let it synchronize by re-connecting.
			log.Println("Websocket buffer is full. Dropping socket")
			go c.ws.Close()
		}
	}
}

// Broadcast sends a message to all of the connected Subscribers.
func (p *Publisher) Broadcast(message []byte) {
	p.Publish(all, message)
}

func (p *Publisher) handleWebsocketConn(ws *websocket.Conn) {
	c := &connection{
		ws:   ws,
		send: make(chan []byte, 256),
		keys: make(map[string]bool),
	}

	p.filters.Add(c, all)
	defer func() {
		p.filters.RemoveAll(c)
		close(c.send)
	}()

	go c.writer()
	c.reader(p.filters)
}

// connection represents a connected Subscriber in Publisher.
{'fraction_non_alphanumeric': 0.10581886438291882, 'fraction_numerical': 0.01360863444392304, 'mean_word_length': 3.872, 'pattern_counts': {'":': 0, '<': 5, '<?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': '2387626', 'n_tokens_mistral': 1435, 'n_tokens_neox': 1315, 'n_words': 476}