Breaking Change - All functions that took a Peer now take a *Session

README.md

@@ -1,4 +1,4 @@
-Turnpike [![Build Status](https://drone.io/github.com/jcelliott/turnpike/status.png)](https://drone.io/github.com/jcelliott/turnpike/latest) [![Coverage Status](https://coveralls.io/repos/jcelliott/turnpike/badge.svg?branch=v2)](https://coveralls.io/r/jcelliott/turnpike?branch=v2) [![GoDoc](https://godoc.org/gopkg.in/jcelliott/turnpike?status.svg)](http://godoc.org/gopkg.in/jcelliott/turnpike.v2)
+Turnpike [![Build Status](https://drone.io/github.com/jcelliott/turnpike/status.png)](https://drone.io/github.com/jcelliott/turnpike/latest) [![Coverage Status](https://coveralls.io/repos/jcelliott/turnpike/badge.svg?branch=v2)](https://coveralls.io/r/jcelliott/turnpike?branch=v2) [![GoDoc](https://godoc.org/gopkg.in/beatgammit/turnpike?status.svg)](http://godoc.org/gopkg.in/beatgammit/turnpike.v2)
 ===
 
 Go implementation of [WAMP](http://wamp.ws/) - The Web Application Messaging Protocol
@@ -18,25 +18,23 @@ basic stand-alone router. The router library can be used to embed a WAMP router
 in another application, or to build a custom router implementation. The client
 library can be used to communicate with any WAMP router.
 
-This version of Turnpike supports WAMP v2. For WAMP v1 support see the [v1 branch](https://github.com/jcelliott/turnpike/tree/v1).
-
 Status
 ---
 
 Turnpike v2 is still under development, but is getting close to a stable
 release. If you have any feedback or suggestions, please
-[open an issue](https://github.com/jcelliott/turnpike/issues/new).
+[open an issue](https://github.com/beatgammit/turnpike/issues/new).
 
 Installation
 ---
 
 Library:
 
-    go get -u gopkg.in/jcelliott/turnpike.v2
+    go get -u gopkg.in/beatgammit/turnpike.v2
 
 Stand-alone router:
 
-    go get -u gopkg.in/jcelliott/turnpike.v2/turnpike
+    go get -u gopkg.in/beatgammit/turnpike.v2/turnpike
 
 Client library usage
 ---
@@ -56,7 +54,7 @@ import (
 	"log"
 	"net/http"
 
-	"gopkg.in/jcelliott/turnpike.v2"
+	"gopkg.in/beatgammit/turnpike.v2"
 )
 
 func main() {

authorizer.go

@@ -6,7 +6,7 @@ package turnpike
 // Authorize takes the session and the message (request), and returns true if
 // the request is authorized, otherwise false.
 type Authorizer interface {
-	Authorize(session Session, msg Message) (bool, error)
+	Authorize(session *Session, msg Message) (bool, error)
 }
 
 // DefaultAuthorizer always returns authorized.
@@ -18,6 +18,6 @@ func NewDefaultAuthorizer() Authorizer {
 	return &defaultAuthorizer{}
 }
 
-func (da *defaultAuthorizer) Authorize(session Session, msg Message) (bool, error) {
+func (da *defaultAuthorizer) Authorize(session *Session, msg Message) (bool, error) {
 	return true, nil
 }

broker.go

@@ -1,20 +1,31 @@
 package turnpike
 
+import (
+	"sync"
+)
+
 // Broker is the interface implemented by an object that handles routing EVENTS
 // from Publishers to Subscribers.
 type Broker interface {
 	// Publishes a message to all Subscribers.
-	Publish(Sender, *Publish)
+	Publish(*Session, *Publish)
 	// Subscribes to messages on a URI.
-	Subscribe(Sender, *Subscribe)
+	Subscribe(*Session, *Subscribe)
 	// Unsubscribes from messages on a URI.
-	Unsubscribe(Sender, *Unsubscribe)
+	Unsubscribe(*Session, *Unsubscribe)
+	// Remove all of session's subscriptions.
+	RemoveSession(*Session)
 }
 
 // A super simple broker that matches URIs to Subscribers.
 type defaultBroker struct {
 	routes        map[URI]map[ID]Sender
 	subscriptions map[ID]URI
+	subscribers   map[*Session][]ID
+
+	lastRequestId ID
+
+	sync.RWMutex
 }
 
 // NewDefaultBroker initializes and returns a simple broker that matches URIs to
@@ -23,73 +34,139 @@ func NewDefaultBroker() Broker {
 	return &defaultBroker{
 		routes:        make(map[URI]map[ID]Sender),
 		subscriptions: make(map[ID]URI),
+		subscribers:   make(map[*Session][]ID),
 	}
 }
 
+func (br *defaultBroker) nextRequestId() ID {
+	br.lastRequestId++
+	if br.lastRequestId > MAX_REQUEST_ID {
+		br.lastRequestId = 1
+	}
+
+	return br.lastRequestId
+}
+
 // Publish sends a message to all subscribed clients except for the sender.
 //
 // If msg.Options["acknowledge"] == true, the publisher receives a Published event
 // after the message has been sent to all subscribers.
-func (br *defaultBroker) Publish(pub Sender, msg *Publish) {
+func (br *defaultBroker) Publish(sess *Session, msg *Publish) {
+	br.RLock()
+	defer br.RUnlock()
+
+	pub := sess.Peer
 	pubID := NewID()
 	evtTemplate := Event{
 		Publication: pubID,
 		Arguments:   msg.Arguments,
 		ArgumentsKw: msg.ArgumentsKw,
 		Details:     make(map[string]interface{}),
 	}
+
+	excludePublisher := true
+	if exclude, ok := msg.Options["exclude_me"].(bool); ok {
+		excludePublisher = exclude
+	}
+
 	for id, sub := range br.routes[msg.Topic] {
 		// shallow-copy the template
 		event := evtTemplate
 		event.Subscription = id
 		// don't send event to publisher
-		if sub != pub {
-			sub.Send(&event)
+		if sub != pub || !excludePublisher {
+			go sub.Send(&event)
 		}
 	}
 
 	// only send published message if acknowledge is present and set to true
 	if doPub, _ := msg.Options["acknowledge"].(bool); doPub {
-		pub.Send(&Published{Request: msg.Request, Publication: pubID})
+		go pub.Send(&Published{Request: msg.Request, Publication: pubID})
 	}
 }
 
 // Subscribe subscribes the client to the given topic.
-func (br *defaultBroker) Subscribe(sub Sender, msg *Subscribe) {
+func (br *defaultBroker) Subscribe(sess *Session, msg *Subscribe) {
+	br.Lock()
+	defer br.Unlock()
+
 	if _, ok := br.routes[msg.Topic]; !ok {
 		br.routes[msg.Topic] = make(map[ID]Sender)
 	}
-	id := NewID()
-	br.routes[msg.Topic][id] = sub
-
+	id := br.nextRequestId()
+	br.routes[msg.Topic][id] = sess.Peer
 	br.subscriptions[id] = msg.Topic
 
-	sub.Send(&Subscribed{Request: msg.Request, Subscription: id})
+	// subscribers
+	ids, ok := br.subscribers[sess]
+	if !ok {
+		ids = []ID{}
+	}
+	ids = append(ids, id)
+	br.subscribers[sess] = ids
+
+	go sess.Peer.Send(&Subscribed{Request: msg.Request, Subscription: id})
 }
 
-func (br *defaultBroker) Unsubscribe(sub Sender, msg *Unsubscribe) {
-	topic, ok := br.subscriptions[msg.Subscription]
-	if !ok {
+func (br *defaultBroker) RemoveSession(sess *Session) {
+	log.Printf("broker remove peer %p", sess)
+	br.Lock()
+	defer br.Unlock()
+
+	for _, id := range br.subscribers[sess] {
+		br.unsubscribe(sess, id)
+	}
+}
+
+func (br *defaultBroker) Unsubscribe(sess *Session, msg *Unsubscribe) {
+	br.Lock()
+	defer br.Unlock()
+
+	if !br.unsubscribe(sess, msg.Subscription) {
 		err := &Error{
 			Type:    msg.MessageType(),
 			Request: msg.Request,
 			Error:   ErrNoSuchSubscription,
 		}
-		sub.Send(err)
+		go sess.Peer.Send(err)
 		log.Printf("Error unsubscribing: no such subscription %v", msg.Subscription)
 		return
 	}
-	delete(br.subscriptions, msg.Subscription)
+
+	go sess.Peer.Send(&Unsubscribed{Request: msg.Request})
+}
+
+func (br *defaultBroker) unsubscribe(sess *Session, id ID) bool {
+	log.Printf("broker unsubscribing: %p, %d", &sess, id)
+	topic, ok := br.subscriptions[id]
+	if !ok {
+		return false
+	}
+	delete(br.subscriptions, id)
 
 	if r, ok := br.routes[topic]; !ok {
 		log.Printf("Error unsubscribing: unable to find routes for %s topic", topic)
-	} else if _, ok := r[msg.Subscription]; !ok {
-		log.Printf("Error unsubscribing: %s route does not exist for %v subscription", topic, msg.Subscription)
+	} else if _, ok := r[id]; !ok {
+		log.Printf("Error unsubscribing: %s route does not exist for %v subscription", topic, id)
 	} else {
-		delete(r, msg.Subscription)
+		delete(r, id)
 		if len(r) == 0 {
 			delete(br.routes, topic)
 		}
 	}
-	sub.Send(&Unsubscribed{Request: msg.Request})
+
+	// subscribers
+	ids := br.subscribers[sess][:0]
+	for _, id := range br.subscribers[sess] {
+		if id != id {
+			ids = append(ids, id)
+		}
+	}
+	if len(ids) == 0 {
+		delete(br.subscribers, sess)
+	} else {
+		br.subscribers[sess] = ids
+	}
+
+	return true
 }

broker_test.go

@@ -1,24 +1,46 @@
 package turnpike
 
 import (
+	"sync"
 	"testing"
 
 	. "github.com/smartystreets/goconvey/convey"
 )
 
-type TestSender struct {
+type TestPeer struct {
 	received Message
+	sent     Message
+
+	sync.RWMutex
+}
+
+func (s *TestPeer) Send(msg Message) error {
+	s.Lock()
+	defer s.Unlock()
+
+	s.received = msg
+	return nil
 }
 
-func (s *TestSender) Send(msg Message) error { s.received = msg; return nil }
+func (s *TestPeer) getReceived() Message {
+	s.RLock()
+	defer s.RUnlock()
+
+	return s.received
+}
+
+// TODO: implement me
+func (s *TestPeer) Receive() <-chan Message { return nil }
+func (s *TestPeer) Close() error            { return nil }
 
 func TestSubscribe(t *testing.T) {
 	Convey("Subscribing to a topic", t, func() {
 		broker := NewDefaultBroker().(*defaultBroker)
-		subscriber := &TestSender{}
+		subscriber := &TestPeer{}
+		sess := &Session{Peer: subscriber}
 		testTopic := URI("turnpike.test.topic")
 		msg := &Subscribe{Request: 123, Topic: testTopic}
-		broker.Subscribe(subscriber, msg)
+		broker.Subscribe(sess, msg)
 
 		Convey("The subscriber should have received a SUBSCRIBED message", func() {
 			sub := subscriber.received.(*Subscribed).Subscription
@@ -31,22 +53,58 @@ func TestSubscribe(t *testing.T) {
 			So(ok, ShouldBeTrue)
 			So(topic, ShouldEqual, testTopic)
 		})
+	})
+}
+
+func TestBrokerNextRequestId(t *testing.T) {
+	Convey("nextRequestId called multiple times", t, func() {
+		broker := NewDefaultBroker().(*defaultBroker)
+		id1 := broker.nextRequestId()
+		id2 := broker.nextRequestId()
 
-		// TODO: multiple subscribe requests?
+		So(id1, ShouldNotEqual, id2)
+	})
+
+	Convey("nextRequestId should roll over", t, func() {
+		broker := NewDefaultBroker().(*defaultBroker)
+		broker.lastRequestId = MAX_REQUEST_ID
+		id := broker.nextRequestId()
+
+		So(id, ShouldEqual, 1)
+	})
+}
+
+func TestMultipleSubscribe(t *testing.T) {
+	const SUBSCRIBERS = 2
+	const TOPIC = URI("turnpike.test.topic")
+
+	Convey("Multiple subscribers to a topic", t, func() {
+		broker := NewDefaultBroker().(*defaultBroker)
+		for i := 0; i < SUBSCRIBERS; i++ {
+			subscriber := &TestPeer{}
+			sess := &Session{Peer: subscriber, Id: NewID()}
+			msg := &Subscribe{Request: sess.NextRequestId(), Topic: TOPIC}
+			broker.Subscribe(sess, msg)
+		}
+
+		Convey("There should be a map entry for each subscriber", func() {
+			So(len(broker.routes[TOPIC]), ShouldEqual, SUBSCRIBERS)
+		})
 	})
 }
 
 func TestUnsubscribe(t *testing.T) {
 	broker := NewDefaultBroker().(*defaultBroker)
-	subscriber := &TestSender{}
+	subscriber := &TestPeer{}
 	testTopic := URI("turnpike.test.topic")
 	msg := &Subscribe{Request: 123, Topic: testTopic}
-	broker.Subscribe(subscriber, msg)
+	sess := &Session{Peer: subscriber}
+	broker.Subscribe(sess, msg)
 	sub := subscriber.received.(*Subscribed).Subscription
 
 	Convey("Unsubscribing from a topic", t, func() {
 		msg := &Unsubscribe{Request: 124, Subscription: sub}
-		broker.Unsubscribe(subscriber, msg)
+		broker.Unsubscribe(sess, msg)
 
 		Convey("The peer should have received an UNSUBSCRIBED message", func() {
 			unsub := subscriber.received.(*Unsubscribed).Request