<commit_before>// Package turnpike provides a Websocket Application Messaging Protocol (WAMP) server and client
package turnpike

import (
	"turnpike/wamp"
	"code.google.com/p/go.net/websocket"
	"github.com/nu7hatch/gouuid"
	"log"
	"encoding/json"
)

var backlog = 10

type WAMPClient struct {
	Out chan<- []byte
	conn *websocket.Conn
}

type Turnpike struct {
	clients map[string]WAMPClient
}

// HandleWebsocket is a Go1.0 shim to imitate Turnpike#HandleWebsocket
func HandleWebsocket(t *Turnpike) func(*websocket.Conn) {
	return func (conn *websocket.Conn) {
		t.HandleWebsocket(conn)
	}
}

func (t *Turnpike) handlePrefix(id string, msg wamp.PrefixMsg) {
	panic("not implemented")
}

func (t *Turnpike) handleCall(id string, msg wamp.CallMsg) {
	panic("not implemented")
}

func (t *Turnpike) handleSubscribe(id string, msg wamp.SubscribeMsg) {
	panic("not implemented")
}

func (t *Turnpike) handleUnsubscribe(id string, msg wamp.UnsubscribeMsg) {
	panic("not implemented")
}

func (t *Turnpike) handlePublish(id string, msg wamp.PublishMsg) {
	panic("not implemented")
}

func (t *Turnpike) HandleWebsocket(conn *websocket.Conn) {
	defer conn.Close()

	tid, err := uuid.NewV4()
	if err != nil {
		panic(err)
	}
	id := tid.String()

	arr, err := wamp.Welcome(id)
	if err != nil {
		log.Println("Error encoding welcome message")
		return
	}
	err = websocket.Message.Send(conn, arr)
	if err != nil {
		log.Println("Error sending welcome message, aborting connection")
		return
	}

	c := make(chan []byte, backlog)
	t.clients[id] = WAMPClient{c, conn}

	go func () {
		for msg := range c {
			err := websocket.Message.Send(conn, msg)
			if err != nil {
				log.Println("Error sending message:", err)
			}
		}
	}()

	for {
		var data []byte
		err := websocket.Message.Receive(conn, &data)
		if err != nil {
			log.Println("Error receiving message, aborting connection")
			break
		}

		switch typ := wamp.ParseType(data); typ {
		case wamp.TYPE_ID_PREFIX:
			var msg wamp.PrefixMsg
			err := json.Unmarshal(data, &msg)
			if err != nil {<commit_msg>Implemented pubsub (not tested)<commit_after>// Package turnpike provides a Websocket Application Messaging Protocol (WAMP) server and client
package turnpike

import (
	"code.google.com/p/go.net/websocket"
	"encoding/json"
	"github.com/nu7hatch/gouuid"
	"log"
	"sync"
	"turnpike/wamp"
)

var backlog = 10

type WAMPClient struct {
	Out  chan<- []byte
	conn *websocket.Conn
}

type Turnpike struct {
	clients map[string]WAMPClient
	// this is a map because it cheaply prevents a client from subscribing multiple times
	// the nice side-effect here is it's easy to unsubscribe
	subscriptions map[string]map[string]bool
	subLock       *sync.Mutex
}

func New() *Turnpike {
	return &Turnpike{
		clients:       make(map[string]WAMPClient),
		subscriptions: make(map[string]map[string]bool),
		subLock:       new(sync.Mutex)}
}

// HandleWebsocket is a Go1.0 shim to imitate Turnpike#HandleWebsocket
func HandleWebsocket(t *Turnpike) func(*websocket.Conn) {
	return func(conn *websocket.Conn) {
		t.HandleWebsocket(conn)
	}
}

func (t *Turnpike) handlePrefix(id string, msg wamp.PrefixMsg) {
	panic("not implemented")
}

func (t *Turnpike) handleCall(id string, msg wamp.CallMsg) {
	panic("not implemented")
}

func (t *Turnpike) handleSubscribe(id string, msg wamp.SubscribeMsg) {
	t.subLock.Lock()
	evt := msg.TopicURI
	if _, ok := t.subscriptions[evt]; !ok {
		t.subscriptions[evt] = make(map[string]bool)
	}
	t.subscriptions[evt][id] = true
	t.subLock.Unlock()
}

func (t *Turnpike) handleUnsubscribe(id string, msg wamp.UnsubscribeMsg) {
	t.subLock.Lock()
	if tmap, ok := t.subscriptions[msg.TopicURI]; ok {
		delete(tmap, id)
	}
	t.subLock.Unlock()
}

func (t *Turnpike) handlePublish(id string, msg wamp.PublishMsg) {
	tmap, ok := t.subscriptions[msg.TopicURI]
	if !ok {
		return
	}

	out, err := wamp.Event(msg.TopicURI, msg.Event)
	if err != nil {
		log.Println("Error creating event message:", err)
		return
	}

	var sendTo []string
	if len(msg.ExcludeList) > 0 || len(msg.EligibleList) > 0 {
		// this is super ugly, but I couldn't think of a better way...
		for tid := range tmap {
			include := true
			for _, _tid := range msg.ExcludeList {
				if tid == _tid {
					include = false
					break
				}
			}
			if include {
				sendTo = append(sendTo, tid)
			}
		}

		for _, tid := range msg.EligibleList {
			include := true
			for _, _tid := range sendTo {
				if _tid == tid {
					include = false
					break
				}
			}
			if include {
				sendTo = append(sendTo, tid)
			}
		}
	} else {
		for tid := range tmap {
			if tid == id && msg.ExcludeMe {
				continue
			}
			sendTo = append(sendTo, tid)
		}
	}

	for _, tid := range sendTo {
		// we're not locking anything, so we need
		// to make sure the client didn't disconnecct in the
		// last few nanoseconds...
		if client, ok := t.clients[tid]; ok {
			client.Out <- out
		}
	}
}

func (t *Turnpike) HandleWebsocket(conn *websocket.Conn) {
	defer conn.Close()

	tid, err := uuid.NewV4()
	if err != nil {
		panic(err)
	}
	id := tid.String()

	arr, err := wamp.Welcome(id)
	if err != nil {
		log.Println("Error encoding welcome message")
		return
	}
	err = websocket.Message.Send(conn, arr)
	if err != nil {
		log.Println("Error sending welcome message, aborting connection")
		return
	}

	c := make(chan []byte, backlog)
	t.clients[id] = WAMPClient{c, conn}

	go func() {
		for msg := range c {
			err := websocket.Message.Send(conn, msg)
			if err != nil {
				log.Println("Error sending message:", err)
			}
		}
	}()

	for {
		var data []byte
		err := websocket.Message.Receive(conn, &data)
		if err != nil {
			log.Println("Error receiving message, aborting connection")
			break
		}

		switch typ := wamp.ParseType(data); typ {
		case wamp.TYPE_ID_PREFIX:
			var msg wamp.PrefixMsg
			err := json.Unmarshal(data, &msg)
			if err != nil {
{'fraction_non_alphanumeric': 0.13219470538001707, 'fraction_numerical': 0.0023911187019641333, 'mean_word_length': 3.55011655011655, 'pattern_counts': {'":': 0, '<': 6, '<?xml version=': 0, '>': 5, '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': '5642527', 'n_tokens_mistral': 2322, 'n_tokens_neox': 2111, 'n_words': 566}