Repository: gammazero/nexus
Pull request #193 (2019-08-10 07:36:18)

Client sync
- Refactor client synchronization for simplicity
- Session roles and features cannot be changed after session attached to router
- Remove need for concurrency safety (mutex lock) when looking up features
- Remove number of channels in main select look for each session 


Patch:
diff --git a/client/client.go b/client/client.go
index f06bfc8..992cde0 100644
--- a/client/client.go
+++ b/client/client.go
@@ -8,7 +8,6 @@ package client
 
 import (
 	"context"
-	"crypto/tls"
 	"errors"
 	"fmt"
 	"strings"
@@ -17,70 +16,13 @@ import (
 	"time"
 
 	"github.com/gammazero/nexus/stdlog"
-	"github.com/gammazero/nexus/transport"
 	"github.com/gammazero/nexus/transport/serialize"
 	"github.com/gammazero/nexus/wamp"
 )
 
 const (
-	helloAuthmethods = "authmethods"
-	helloRoles       = "roles"
-)
-
-// Deprecated: replaced by Config
-//
-// ClientConfig is a type alias for the deprecated ClientConfig.
-type ClientConfig = Config
-
-// Config configures a client with everything needed to begin a session
-// with a WAMP router.
-type Config struct {
-	// Realm is the URI of the realm the client will join.
-	Realm string
-
-	// HelloDetails contains details about the client.  The client provides the
-	// roles, unless already supplied by the user.
-	HelloDetails wamp.Dict
-
-	// AuthHandlers is a map of authmethod to AuthFunc.  All authmethod keys
-	// from this map are automatically added to HelloDetails["authmethods"]
-	AuthHandlers map[string]AuthFunc
-
-	// ResponseTimeout specifies the amount of time that the client will block
-	// waiting for a response from the router.  A value of 0 uses the default.
-	ResponseTimeout time.Duration
-
-	// Enable debug logging for client.
-	Debug bool
-
-	// Set to JSON or MSGPACK.  Default (zero-value) is JSON.
-	Serialization serialize.Serialization
-
-	// Provide a tls.Config to connect the client using TLS.  The zero
-	// configuration specifies using defaults.  A nil tls.Config means do not
-	// use TLS.
-	TlsCfg *tls.Config
-
-	// Supplies alternate Dial function for the websocket dialer.
-	// See https://godoc.org/github.com/gorilla/websocket#Dialer
-	Dial transport.DialFunc
-
-	// Client receive limit for use with RawSocket transport.
-	// If recvLimit is > 0, then the client will not receive messages with size
-	// larger than the nearest power of 2 greater than or equal to recvLimit.
-	// If recvLimit is <= 0, then the default of 16M is used.
-	RecvLimit int
-
-	// Logger for client to use.  If not set, client logs to os.Stderr.
-	Logger stdlog.StdLog
-
-	// Websocket transport configuration.
-	WsCfg transport.WebsocketConfig
-}
-
-// Define serialization consts in client package so that client code does not
-// need to import the serialize package to get the consts.
-const (
+	// Define serialization consts in client package so that client code does
+	// not need to import the serialize package to get the consts.
 	JSON    = serialize.JSON
 	MSGPACK = serialize.MSGPACK
 	CBOR    = serialize.CBOR
@@ -129,11 +71,10 @@ type InvokeResult struct {
 
 // A Client routes messages to/from a WAMP router.
 type Client struct {
-	sess wamp.Session
+	sess *wamp.Session
 
 	responseTimeout time.Duration
 	awaitingReply   map[wamp.ID]chan wamp.Message
-	replyLock       sync.Mutex
 	authHandlers    map[string]AuthFunc
 
 	eventHandlers map[wamp.ID]EventHandler
@@ -143,9 +84,6 @@ type Client struct {
 	nameProcID     map[string]wamp.ID
 	invHandlerKill map[wamp.ID]context.CancelFunc
 	progGate       map[context.Context]wamp.ID
-	progGateLock   sync.Mutex
-
-	actionChan chan func()
 
 	stopping          chan struct{}
 	activeInvHandlers sync.WaitGroup
@@ -175,13 +113,16 @@ func NewClient(p wamp.Peer, cfg Config) (*Client, error) {
 		p.Close()
 		return nil, err
 	}
+	sess := wamp.NewSession(p, welcome.ID, welcome.Details, welcome.Details)
+
+	// Check that router has at least one supported role.
+	if !sess.HasRole("broker") && !sess.HasRole("dealer") {
+		p.Close()
+		return nil, errors.New("router did not announce any supported roles")
+	}
 
 	c := &Client{
-		sess: wamp.Session{
-			Peer:    p,
-			ID:      welcome.ID,
-			Details: welcome.Details,
-		},
+		sess: sess,
 
 		responseTimeout: cfg.ResponseTimeout,
 		awaitingReply:   map[wamp.ID]chan wamp.Message{},
@@ -194,9 +135,8 @@ func NewClient(p wamp.Peer, cfg Config) (*Client, error) {
 		invHandlerKill: map[wamp.ID]context.CancelFunc{},
 		progGate:       map[context.Context]wamp.ID{},
 
-		actionChan: make(chan func()),
-		stopping:   make(chan struct{}),
-		done:       make(chan struct{}),
+		stopping: make(chan struct{}),
+		done:     make(chan struct{}),
 
 		log:   cfg.Logger,
 		debug: cfg.Debug,
@@ -276,10 +216,11 @@ func (c *Client) Subscribe(topic string, fn EventHandler, options wamp.Dict) err
 	switch msg := msg.(type) {
 	case *wamp.Subscribed:
 		// Register the event handler for this subscription.
-		return c.runAction(func() {
-			c.eventHandlers[msg.Subscription] = fn
-			c.topicSubID[topic] = msg.Subscription
-		})
+		c.sess.Lock()
+		c.eventHandlers[msg.Subscription] = fn
+		c.topicSubID[topic] = msg.Subscription
+		c.sess.Unlock()
+		return nil
 	case *wamp.Error:
 		return fmt.Errorf("subscribing to topic '%v': %s", topic,
 			wampErrorString(msg))
@@ -292,36 +233,27 @@ func (c *Client) Subscribe(topic string, fn EventHandler, options wamp.Dict) err
 // client does not have an active subscription to the topic, then returns false
 // for second boolean return value.
 func (c *Client) SubscriptionID(topic string) (subID wamp.ID, ok bool) {
-	c.runAction(func() {
-		subID, ok = c.topicSubID[topic]
-	})
+	c.sess.Lock()
+	subID, ok = c.topicSubID[topic]
+	c.sess.Unlock()
 	return
 }
 
 // Unsubscribe removes the registered EventHandler from the topic.
 func (c *Client) Unsubscribe(topic string) error {
-	var subID wamp.ID
-	var err error
-	runErr := c.runAction(func() {
-		var ok bool
-		subID, ok = c.topicSubID[topic]
-		if !ok {
-			err = errors.New("not subscribed to: " + topic)
-		} else {
-			// Delete the subscription anyway, regardless of whether or not the
-			// the router succeeds or fails to unsubscribe.  If the client
-			// called Unsubscribe() then it has no interest in receiving any
-			// more events for the topic, and may expect any.
-			delete(c.topicSubID, topic)
-			delete(c.eventHandlers, subID)
-		}
-	})
-	if runErr != nil {
-		return runErr
-	}
-	if err != nil {
-		return err
+	c.sess.Lock()
+	subID, ok := c.topicSubID[topic]
+	if !ok {
+		c.sess.Unlock()
+		return errors.New("not subscribed to: " + topic)
 	}
+	// Delete the subscription anyway, regardless of whether or not the the
+	// router succeeds or fails to unsubscribe.  If the client called
+	// Unsubscribe() then it has no interest in receiving any more events for
+	// the topic, and may expect any.
+	delete(c.topicSubID, topic)
+	delete(c.eventHandlers, subID)
+	c.sess.Unlock()
 
 	id := c.idGen.Next()
 	c.expectReply(id)
@@ -466,39 +398,20 @@ func (c *Client) Register(procedure string, fn InvocationHandler, options wamp.D
 	switch msg := msg.(type) {
 	case *wamp.Registered:
 		// Register the event handler for this registration.
-		return c.runAction(func() {
-			c.invHandlers[msg.Registration] = fn
-			c.nameProcID[procedure] = msg.Registration
-
-			if c.debug {
-				c.log.Println("Registered", procedure, "as registration",
-					msg.Registration)
-			}
-		})
+		c.sess.Lock()
+		c.invHandlers[msg.Registration] = fn
+		c.nameProcID[procedure] = msg.Registration
+		c.sess.Unlock()
+		if c.debug {
+			c.log.Println("Registered", procedure, "as registration",
+				msg.Registration)
+		}
 	case *wamp.Error:
 		return fmt.Errorf("registering procedure '%v': %s", procedure,
 			wampErrorString(msg))
 	default:
 		return unexpectedMsgError(msg, wamp.REGISTERED)
 	}
-}
-
-// runAction runs the action on the main run() goroutine blocking waiting for a
-// response
-func (c *Client) runAction(action func()) error {
-	sync := make(chan struct{})
-	select {
-	case c.actionChan <- func() {
-		action()
-		close(sync)
-	}:
-	case <-c.done:
-		if c.debug {
-			c.log.Println("Action cancelled client is closed")
-		}
-		return errors.New("client closed")
-	}
-	<-sync
 	return nil
 }
 
@@ -506,36 +419,27 @@ func (c *Client) runAction(action func()) error {
 // the client is not registered for the procedure, then returns false for
 // second boolean return value.
 func (c *Client) RegistrationID(procedure string) (regID wamp.ID, ok bool) {
-	c.runAction(func() {
-		regID, ok = c.nameProcID[procedure]
-	})
+	c.sess.Lock()
+	regID, ok = c.nameProcID[procedure]
+	c.sess.Unlock()
 	return
 }
 
 // Unregister removes the registration of a procedure from the router.
 func (c *Client) Unregister(procedure string) error {
-	var procID wamp.ID
-	var err error
-	runErr := c.runAction(func() {
-		var ok bool
-		procID, ok = c.nameProcID[procedure]
-		if !ok {
-			err = errors.New("not registered to handle procedure " + procedure)
-		} else {
-			// Delete the registration anyway, regardless of whether or not the
-			// the router succeeds or fails to unregister.  If the client
-			// called Unregister() then it has no interest in receiving any
-			// more invocations for the procedure, and may not expect any.
-			delete(c.nameProcID, procedure)
-			delete(c.invHandlers, procID)
-		}
-	})
-	if runErr != nil {
-		return runErr
-	}
-	if err != nil {
-		return err
+	c.sess.Lock()
+	procID, ok := c.nameProcID[procedure]
+	if !ok {
+		c.sess.Unlock()
+		return errors.New("not registered to handle procedure " + procedure)
 	}
+	// Delete the registration anyway, regardless of whether or not the the
+	// router succeeds or fails to unregister.  If the client called
+	// Unregister() then it has no interest in receiving any more invocations
+	// for the procedure, and may not expect any.
+	delete(c.nameProcID, procedure)
+	delete(c.invHandlers, procID)
+	c.sess.Unlock()
 
 	id := c.idGen.Next()
 	c.expectReply(id)
@@ -577,7 +481,7 @@ func (c *Client) Unregister(procedure string) error {
 // according to the specified mode.
 //
 // If the context is canceled or times out, then error returned will not be a
-// RPCError.  This allows the caller to distinquish between cancellation
+// RPCError.  This allows the caller to distinguish between cancellation
 // initiated by the client (by canceling context), and cancellation initialed
 // elsewhere.
 //
@@ -589,7 +493,7 @@ func (c *Client) Unregister(procedure string) error {
 // is specified as a parameter to the Call() API, and not a message option for
 // CALL.
 //
-// Cancele Mode Behavior
+// Cancel Mode Behavior
 //
 // "skip": The pending call is canceled and ERROR is sent immediately back to
 // the caller.  No INTERRUPT is sent to the callee and the result is discarded
@@ -743,25 +647,25 @@ func (c *Client) Close() error {
 		return errors.New("already closed")
 	}
 
-	// Cancel any running invocation handlers and wait for them to finish.
-	// This makes it safe to close the actionChan.  Do this before leaving the
-	// realm so that any invocation handlers do not hang waiting to send to the
-	// router after it has stopped receiving from the client.
+	// Cancel any running invocation handlers and wait for them to finish.  Do
+	// this before leaving the realm so that any invocation handlers do not
+	// hang waiting to send to the router after it has stopped receiving from
+	// the client.
 	close(c.stopping)
 	c.activeInvHandlers.Wait()
 
 	// Stop waiting for replies and clear the reply channel of pending writes.
 	var awaitingReply map[wamp.ID]chan wamp.Message
-	c.replyLock.Lock()
+	c.sess.Lock()
 	awaitingReply = c.awaitingReply
 	c.awaitingReply = nil
+	c.sess.Unlock()
 	for _, ch := range awaitingReply {
 		select {
 		case <-ch:
 		default:
 		}
 	}
-	c.replyLock.Unlock()
 
 	// Leave the realm and stop the client's main goroutine.
 	c.leaveRealm()
@@ -769,17 +673,10 @@ func (c *Client) Close() error {
 	// The run goroutine is guaranteed to have exited when leaveRealm()
 	// returns, so there will be nothing trying to write to the reply channel.
 	// Closing the channels dismisses any possible readers.
-	if len(awaitingReply) != 0 {
-		c.replyLock.Lock()
-		for _, ch := range awaitingReply {
-			close(ch)
-		}
-		c.replyLock.Unlock()
+	for _, ch := range awaitingReply {
+		close(ch)
 	}
 
-	// c.sess.Peer is accessed from several go routines and would require
-	// synchronisation if updated here.  Peer is closed in leaveRealm() so we
-	// do not need to set it to nil.
 	return nil
 }
 
@@ -808,9 +705,9 @@ func (c *Client) SendProgress(ctx context.Context, args wamp.List, kwArgs wamp.D
 	// invocation handler has been closed because the call was canceled.
 	var req wamp.ID
 	var ok bool
-	c.progGateLock.Lock()
+	c.sess.Lock()
 	req, ok = c.progGate[ctx]
-	c.progGateLock.Unlock()
+	c.sess.Unlock()
 
 	if !ok {
 		// Caller is not accepting progressive results or call canceled.
@@ -829,6 +726,10 @@ func (c *Client) SendProgress(ctx context.Context, args wamp.List, kwArgs wamp.D
 // names to functions that handle each auth type.  This can be nil if router is
 // expected to allow anonymous authentication.
 func joinRealm(peer wamp.Peer, cfg Config) (*wamp.Welcome, error) {
+	const (
+		helloAuthmethods = "authmethods"
+		helloRoles       = "roles"
+	)
 	if cfg.Realm == "" {
 		return nil, errors.New("realm not specified")
 	}
@@ -906,7 +807,7 @@ func (c *Client) leaveRealm() {
 	case <-c.done:
 		timer.Stop()
 	case <-timer.C:
-		close(c.actionChan) // force run() to exit
+		c.sess.EndRecv(nil) // force run() to exit
 		<-c.done
 	}
 }
@@ -1018,11 +919,11 @@ func unexpectedMsgError(msg wamp.Message, expected wamp.MessageType) error {
 
 func (c *Client) expectReply(id wamp.ID) {
 	wait := make(chan wamp.Message)
-	c.replyLock.Lock()
+	c.sess.Lock()
 	if c.awaitingReply != nil { // Client has already been closed
 		c.awaitingReply[id] = wait
 	}
-	c.replyLock.Unlock()
+	c.sess.Unlock()
 }
 
 // waitForReply waits for an expected reply from the router.
@@ -1033,9 +934,9 @@ func (c *Client) expectReply(id wamp.ID) {
 func (c *Client) waitForReply(id wamp.ID) (wamp.Message, error) {
 	var wait chan wamp.Message
 	var ok bool
-	c.replyLock.Lock()
+	c.sess.Lock()
 	wait, ok = c.awaitingReply[id]
-	c.replyLock.Unlock()
+	c.sess.Unlock()
 	if !ok {
 		return nil, fmt.Errorf("not expecting reply for ID: %v", id)
 	}
@@ -1053,9 +954,9 @@ func (c *Client) waitForReply(id wamp.ID) (wamp.Message, error) {
 	case <-timer.C:
 		err = errors.New("timeout waiting for reply")
 	}
-	c.replyLock.Lock()
+	c.sess.Lock()
 	delete(c.awaitingReply, id)
-	c.replyLock.Unlock()
+	c.sess.Unlock()
 
 	return msg, err
 }
@@ -1072,9 +973,9 @@ func (c *Client) waitForReplyWithCancel(ctx context.Context, id wamp.ID, mode, p
 	}
 	var wait chan wamp.Message
 	var ok bool
-	c.replyLock.Lock()
+	c.sess.Lock()
 	wait, ok = c.awaitingReply[id]
-	c.replyLock.Unlock()
+	c.sess.Unlock()
 	if !ok {
 		return nil, fmt.Errorf("not expecting reply for ID: %v", id)
 	}
@@ -1130,9 +1031,9 @@ CollectResults:
 		}
 	}
 	// All done with this call, so not waiting for more replies.
-	c.replyLock.Lock()
+	c.sess.Lock()
 	delete(c.awaitingReply, id)
-	c.replyLock.Unlock()
+	c.sess.Unlock()
 
 	return msg, err
 }
@@ -1144,15 +1045,14 @@ func (c *Client) run() {
 	if c.debug {
 		defer c.log.Println("Client", c.sess, "closed")
 	}
-	recvChan := c.sess.Recv()
+
+	recv := c.sess.Recv()
+	done := c.sess.RecvDone()
 	for {
 		select {
-		case action, ok := <-c.actionChan:
-			if !ok {
-				return
-			}
-			action()
-		case msg, ok := <-recvChan:
+		case <-done:
+			return
+		case msg, ok := <-recv:
 			if !ok {
 				return
 			}
@@ -1215,7 +1115,9 @@ func (c *Client) runReceiveFromRouter(msg wamp.Message) bool {
 // as the messages are received in.  This could not be guaranteed if executing
 // concurrently in separate goroutines.
 func (c *Client) runHandleEvent(msg *wamp.Event) {
+	c.sess.Lock()
 	handler, ok := c.eventHandlers[msg.Subscription]
+	c.sess.Unlock()
 	if !ok {
 		c.log.Println("No handler registered for subscription:",
 			msg.Subscription)
@@ -1227,15 +1129,20 @@ func (c *Client) runHandleEvent(msg *wamp.Event) {
 // runHandleInvocation processes an INVOCATION message from the router
 // requesting a call to a registered RPC procedure.
 func (c *Client) runHandleInvocation(msg *wamp.Invocation) {
+	timeout, _ := wamp.AsInt64(msg.Details[wamp.OptTimeout])
+	progResOK, _ := msg.Details[wamp.OptReceiveProgress].(bool)
+
+	c.sess.Lock()
 	handler, ok := c.invHandlers[msg.Registration]
 	if !ok {
-		errMsg := fmt.Sprintf("Client has no handler for registration %v",
+		c.sess.Unlock()
+		errMsg := fmt.Sprintf("client has no handler for registration %v",
 			msg.Registration)
 		// The dealer has a procedure registered to this client, but this
-		// client does not recognize the registration ID.  This is not
-		// reported as ErrNoSuchProcedure, since the dealer has a procedure
-		// registered.  It is reported as ErrInvalidArgument to denote that
-		// the client has a problem with the registration ID argument.
+		// client does not recognize the registration ID.  This is not reported
+		// as ErrNoSuchProcedure, since the dealer has a procedure registered.
+		// It is reported as ErrInvalidArgument to denote that the client has a
+		// problem with the registration ID argument.
 		c.sess.Send(&wamp.Error{
 			Type:      wamp.INVOCATION,
 			Request:   msg.Request,
@@ -1250,7 +1157,6 @@ func (c *Client) runHandleInvocation(msg *wamp.Invocation) {
 	// Create a kill switch so that invocation can be canceled.
 	var cancel context.CancelFunc
 	var ctx context.Context
-	timeout, _ := wamp.AsInt64(msg.Details[wamp.OptTimeout])
 	if timeout > 0 {
 		// The caller specified a timeout, in milliseconds.
 		ctx, cancel = context.WithTimeout(context.Background(),
@@ -1261,13 +1167,12 @@ func (c *Client) runHandleInvocation(msg *wamp.Invocation) {
 	c.invHandlerKill[msg.Request] = cancel
 	c.activeInvHandlers.Add(1)
 
-	// If caller is accepting progressive results, create map entry to
-	// allow progress to be sent.
-	if ok, _ = msg.Details[wamp.OptReceiveProgress].(bool); ok {
-		c.progGateLock.Lock()
+	// If caller is accepting progressive results, create map entry to allow
+	// progress to be sent.
+	if progResOK {
 		c.progGate[ctx] = msg.Request
-		c.progGateLock.Unlock()
 	}
+	c.sess.Unlock()
 
 	// Start a goroutine to run the user-defined invocation handler.
 	go func() {
@@ -1287,14 +1192,11 @@ func (c *Client) runHandleInvocation(msg *wamp.Invocation) {
 
 		// Remove the kill switch when done processing invocation.
 		defer func() {
-			c.progGateLock.Lock()
+			c.sess.Lock()
 			delete(c.progGate, ctx)
-			c.progGateLock.Unlock()
-
-			c.runAction(func() {
-				delete(c.invHandlerKill, msg.Request)
-				c.activeInvHandlers.Done()
-			})
+			delete(c.invHandlerKill, msg.Request)
+			c.sess.Unlock()
+			c.activeInvHandlers.Done()
 		}()
 
 		// Wait for the handler to finish or for the call be to canceled.
@@ -1349,7 +1251,9 @@ func (c *Client) runHandleInvocation(msg *wamp.Invocation) {
 // requesting that a pending call be canceled.
 func (c *Client) runHandleInterrupt(msg *wamp.Interrupt) {
 	logMsg := "Received INTERRUPT for INVOCATION"
+	c.sess.Lock()
 	cancel, ok := c.invHandlerKill[msg.Request]
+	c.sess.Unlock()
 	if !ok {
 		c.log.Println(logMsg, msg.Request, "that no longer exists")
 		return
@@ -1365,9 +1269,9 @@ func (c *Client) runHandleInterrupt(msg *wamp.Interrupt) {
 func (c *Client) runSignalReply(msg wamp.Message, requestID wamp.ID) {
 	var w chan wamp.Message
 	var ok bool
-	c.replyLock.Lock()
+	c.sess.Lock()
 	w, ok = c.awaitingReply[requestID]
-	c.replyLock.Unlock()
+	c.sess.Unlock()
 	if !ok {
 		c.log.Println("Received", msg.MessageType(), requestID,
 			"that client is no longer waiting for")
diff --git a/client/config.go b/client/config.go
new file mode 100644
index 0000000..30de769
--- /dev/null
+++ b/client/config.go
@@ -0,0 +1,62 @@
+package client
+
+import (
+	"crypto/tls"
+	"time"
+
+	"github.com/gammazero/nexus/stdlog"
+	"github.com/gammazero/nexus/transport"
+	"github.com/gammazero/nexus/transport/serialize"
+	"github.com/gammazero/nexus/wamp"
+)
+
+// Config configures a client with everything needed to begin a session
+// with a WAMP router.
+type Config struct {
+	// Realm is the URI of the realm the client will join.
+	Realm string
+
+	// HelloDetails contains details about the client.  The client provides the
+	// roles, unless already supplied by the user.
+	HelloDetails wamp.Dict
+
+	// AuthHandlers is a map of authmethod to AuthFunc.  All authmethod keys
+	// from this map are automatically added to HelloDetails["authmethods"]
+	AuthHandlers map[string]AuthFunc
+
+	// ResponseTimeout specifies the amount of time that the client will block
+	// waiting for a response from the router.  A value of 0 uses the default.
+	ResponseTimeout time.Duration
+
+	// Enable debug logging for client.
+	Debug bool
+
+	// Set to JSON or MSGPACK.  Default (zero-value) is JSON.
+	Serialization serialize.Serialization
+
+	// Provide a tls.Config to connect the client using TLS.  The zero
+	// configuration specifies using defaults.  A nil tls.Config means do not
+	// use TLS.
+	TlsCfg *tls.Config
+
+	// Supplies alternate Dial function for the websocket dialer.
+	// See https://godoc.org/github.com/gorilla/websocket#Dialer
+	Dial transport.DialFunc
+
+	// Client receive limit for use with RawSocket transport.
+	// If recvLimit is > 0, then the client will not receive messages with size
+	// larger than the nearest power of 2 greater than or equal to recvLimit.
+	// If recvLimit is <= 0, then the default of 16M is used.
+	RecvLimit int
+
+	// Logger for client to use.  If not set, client logs to os.Stderr.
+	Logger stdlog.StdLog
+
+	// Websocket transport configuration.
+	WsCfg transport.WebsocketConfig
+}
+
+// Deprecated: replaced by Config
+//
+// ClientConfig is a type alias for the deprecated ClientConfig.
+type ClientConfig = Config
diff --git a/router/broker.go b/router/broker.go
index fd5a81c..5197b0f 100644
--- a/router/broker.go
+++ b/router/broker.go
@@ -38,7 +38,7 @@ type subscription struct {
 	topic       wamp.URI // topic URI
 	match       string   // match policy
 	created     string   // when subscription was created
-	subscribers map[*session]struct{}
+	subscribers map[*wamp.Session]struct{}
 }
 
 // FilterFactory is a function which creates a PublishFilter from a publication
@@ -54,7 +54,7 @@ type Broker struct {
 	subscriptions map[wamp.ID]*subscription
 
 	// Session -> subscription ID set
-	sessionSubIDSet map[*session]map[wamp.ID]struct{}
+	sessionSubIDSet map[*wamp.Session]map[wamp.ID]struct{}
 
 	actionChan chan func()
 
@@ -83,7 +83,7 @@ func NewBroker(logger stdlog.StdLog, strictURI, allowDisclose, debug bool, publi
 		wcTopicSubscription:  map[wamp.URI]*subscription{},
 
 		subscriptions:   map[wamp.ID]*subscription{},
-		sessionSubIDSet: map[*session]map[wamp.ID]struct{}{},
+		sessionSubIDSet: map[*wamp.Session]map[wamp.ID]struct{}{},
 
 		// The action handler should be nearly always runable, since it is the
 		// critical section that does the only routing.  So, and unbuffered
@@ -118,7 +118,7 @@ func (b *Broker) Role() wamp.Dict {
 //
 // The Subscriber can detect the delivery of that same event on multiple
 // subscriptions via EVENT.PUBLISHED.Publication, which will be identical.
-func (b *Broker) Publish(pub *session, msg *wamp.Publish) {
+func (b *Broker) Publish(pub *wamp.Session, msg *wamp.Publish) {
 	if pub == nil || msg == nil {
 		panic("broker.Publish with nil session or message")
 	}
@@ -196,7 +196,7 @@ func (b *Broker) Publish(pub *session, msg *wamp.Publish) {
 // Subscriber might want to subscribe to topics based on a pattern.  If the
 // Broker and the Subscriber support pattern-based subscriptions, this matching
 // can happen by prefix-matching policy or wildcard-matching policy.
-func (b *Broker) Subscribe(sub *session, msg *wamp.Subscribe) {
+func (b *Broker) Subscribe(sub *wamp.Session, msg *wamp.Subscribe) {
 	if sub == nil || msg == nil {
 		panic("broker.Subscribe with nil session or message")
 	}
@@ -224,7 +224,7 @@ func (b *Broker) Subscribe(sub *session, msg *wamp.Subscribe) {
 }
 
 // Unsubscribe removes the requested subscription.
-func (b *Broker) Unsubscribe(sub *session, msg *wamp.Unsubscribe) {
+func (b *Broker) Unsubscribe(sub *wamp.Session, msg *wamp.Unsubscribe) {
 	if sub == nil || msg == nil {
 		panic("broker.Unsubscribe with nil session or message")
 	}
@@ -237,7 +237,7 @@ func (b *Broker) Unsubscribe(sub *session, msg *wamp.Unsubscribe) {
 // when a client leaves the realm by sending a GOODBYE message or by
 // disconnecting from the router.  If there are any subscriptions for this
 // session a wamp.subscription.on_delete meta event is published for each.
-func (b *Broker) RemoveSession(sess *session) {
+func (b *Broker) RemoveSession(sess *wamp.Session) {
 	if sess == nil {
 		return
 	}
@@ -260,7 +260,7 @@ func (b *Broker) run() {
 	}
 }
 
-func (b *Broker) publish(pub *session, msg *wamp.Publish, pubID wamp.ID, excludePub, disclose bool, filter PublishFilter) {
+func (b *Broker) publish(pub *wamp.Session, msg *wamp.Publish, pubID wamp.ID, excludePub, disclose bool, filter PublishFilter) {
 	// Publish to subscribers with exact match.
 	if sub, ok := b.topicSubscription[msg.Topic]; ok {
 		b.pubEvent(pub, msg, pubID, sub, excludePub, false, disclose, filter)
@@ -281,17 +281,17 @@ func (b *Broker) publish(pub *session, msg *wamp.Publish, pubID wamp.ID, exclude
 	}
 }
 
-func (b *Broker) newSubscription(subscriber *session, topic wamp.URI, match string) *subscription {
+func (b *Broker) newSubscription(subscriber *wamp.Session, topic wamp.URI, match string) *subscription {
 	return &subscription{
 		id:          b.idGen.Next(),
 		topic:       topic,
 		match:       match,
 		created:     wamp.NowISO8601(),
-		subscribers: map[*session]struct{}{subscriber: struct{}{}},
+		subscribers: map[*wamp.Session]struct{}{subscriber: struct{}{}},
 	}
 }
 
-func (b *Broker) subscribe(subscriber *session, msg *wamp.Subscribe, match string) {
+func (b *Broker) subscribe(subscriber *wamp.Session, msg *wamp.Subscribe, match string) {
 	var sub *subscription
 	var existingSub bool
 
@@ -376,7 +376,7 @@ func (b *Broker) delSubscription(sub *subscription) {
 }
 
 // unsibsubscribe removes the subscriber from the specified subscription.
-func (b *Broker) unsubscribe(subscriber *session, msg *wamp.Unsubscribe) {
+func (b *Broker) unsubscribe(subscriber *wamp.Session, msg *wamp.Unsubscribe) {
 	subID := msg.Subscription
 	sub, ok := b.subscriptions[subID]
 	if !ok {
@@ -428,7 +428,7 @@ func (b *Broker) unsubscribe(subscriber *session, msg *wamp.Unsubscribe) {
 }
 
 // removeSession removed all subscriptions for the session.
-func (b *Broker) removeSession(subscriber *session) {
+func (b *Broker) removeSession(subscriber *wamp.Session) {
 	subIDSet, ok := b.sessionSubIDSet[subscriber]
 	if !ok {
 		return
@@ -457,7 +457,7 @@ func (b *Broker) removeSession(subscriber *session) {
 	}
 }
 
-func allowPublish(sub *session, filter PublishFilter) bool {
+func allowPublish(sub *wamp.Session, filter PublishFilter) bool {
 	if filter == nil {
 		return true
 	}
@@ -466,15 +466,15 @@ func allowPublish(sub *session, filter PublishFilter) bool {
 		ID:      sub.ID,
 		Details: sub.Details,
 	}
-	sub.rLock()
+	sub.Lock()
 	ok := filter.Allowed(&safeSession)
-	sub.rUnlock()
+	sub.Unlock()
 	return ok
 }
 
 // pubEvent sends an event to all subscribers that are not excluded from
 // receiving the event.
-func (b *Broker) pubEvent(pub *session, msg *wamp.Publish, pubID wamp.ID, sub *subscription, excludePublisher, sendTopic, disclose bool, filter PublishFilter) {
+func (b *Broker) pubEvent(pub *wamp.Session, msg *wamp.Publish, pubID wamp.ID, sub *subscription, excludePublisher, sendTopic, disclose bool, filter PublishFilter) {
 	for subscriber, _ := range sub.subscribers {
 		// Do not send event to publisher.
 		if subscriber == pub && excludePublisher {
@@ -601,8 +601,8 @@ func (b *Broker) pubSubCreateMeta(topic wamp.URI, subSessID wamp.ID, sub *subscr
 	b.pubMeta(wamp.MetaEventSubOnCreate, sendMeta)
 }
 
-func (b *Broker) trySend(sess *session, msg wamp.Message) bool {
-	if err := sess.Session.TrySend(msg); err != nil {
+func (b *Broker) trySend(sess *wamp.Session, msg wamp.Message) bool {
+	if err := sess.TrySend(msg); err != nil {
 		b.log.Printf("!!! Dropped %s to session %s: %s", msg.MessageType(), sess, err)
 		return false
 	}
@@ -610,17 +610,17 @@ func (b *Broker) trySend(sess *session, msg wamp.Message) bool {
 }
 
 // disclosePublisher adds publisher identity information to EVENT.Details.
-func disclosePublisher(pub *session, details wamp.Dict) {
+func disclosePublisher(pub *wamp.Session, details wamp.Dict) {
 	details[rolePub] = pub.ID
 	// These values are not required by the specification, but are here for
 	// compatibility with Crossbar.
-	pub.rLock()
+	pub.Lock()
 	for _, f := range []string{"authid", "authrole"} {
 		if val, ok := pub.Details[f]; ok {
 			details[fmt.Sprintf("%s_%s", rolePub, f)] = val
 		}
 	}
-	pub.rUnlock()
+	pub.Unlock()
 }
 
 // ----- Subscription Meta Procedure Handlers -----
diff --git a/router/dealer.go b/router/dealer.go
index 576f414..cfccb68 100644
--- a/router/dealer.go
+++ b/router/dealer.go
@@ -29,7 +29,7 @@ const (
 	// CALL timeout which spedifies how long the callee may take to answer.
 	sendResultDeadline = time.Minute
 	// yieldRetryDelay is the initial delay before reprocessin a blocked yield
-	yieldRetryDelay = 200 * time.Millisecond
+	yieldRetryDelay = time.Millisecond
 )
 
 // Role information for this broker.
@@ -59,13 +59,13 @@ type registration struct {
 
 	// Multiple sessions can register as callees depending on invocation policy
 	// resulting in multiple procedures for the same registration ID.
-	callees []*session
+	callees []*wamp.Session
 }
 
 // invocation tracks in-progress invocation
 type invocation struct {
 	callID     requestID
-	callee     *session
+	callee     *wamp.Session
 	canceled   bool
 	retryCount int
 }
@@ -86,7 +86,7 @@ type Dealer struct {
 	registrations map[wamp.ID]*registration
 
 	// call ID -> caller session
-	calls map[requestID]*session
+	calls map[requestID]*wamp.Session
 
 	// invocation ID -> {call ID, callee, canceled}
 	invocations map[wamp.ID]*invocation
@@ -96,7 +96,7 @@ type Dealer struct {
 
 	// callee session -> registration ID set.
 	// Used to lookup registrations when removing a callee session.
-	calleeRegIDSet map[*session]map[wamp.ID]struct{}
+	calleeRegIDSet map[*wamp.Session]map[wamp.ID]struct{}
 
 	actionChan chan func()
 
@@ -133,10 +133,10 @@ func NewDealer(logger stdlog.StdLog, strictURI, allowDisclose, debug bool) *Deal
 
 		registrations: map[wamp.ID]*registration{},
 
-		calls:            map[requestID]*session{},
+		calls:            map[requestID]*wamp.Session{},
 		invocations:      map[wamp.ID]*invocation{},
 		invocationByCall: map[requestID]wamp.ID{},
-		calleeRegIDSet:   map[*session]map[wamp.ID]struct{}{},
+		calleeRegIDSet:   map[*wamp.Session]map[wamp.ID]struct{}{},
 
 		// The action handler should be nearly always runable, since it is the
 		// critical section that does the only routing.  So, and unbuffered
@@ -174,7 +174,7 @@ func (d *Dealer) Role() wamp.Dict {
 // If the shared_registration feature is supported, and if allowed by the
 // invocation policy, multiple callees may register to handle the same
 // procedure.
-func (d *Dealer) Register(callee *session, msg *wamp.Register) {
+func (d *Dealer) Register(callee *wamp.Session, msg *wamp.Register) {
 	if callee == nil || msg == nil {
 		panic("dealer.Register with nil session or message")
 	}
@@ -217,9 +217,9 @@ func (d *Dealer) Register(callee *session, msg *wamp.Register) {
 	disclose, _ := msg.Options[wamp.OptDiscloseCaller].(bool)
 	// allow disclose for trusted clients
 	if !d.allowDisclose && disclose {
-		callee.rLock()
+		callee.Lock()
 		authrole, _ := wamp.AsString(callee.Details["authrole"])
-		callee.rUnlock()
+		callee.Unlock()
 		if authrole != "trusted" {
 			d.trySend(callee, &wamp.Error{
 				Type:    msg.MessageType(),
@@ -231,21 +231,6 @@ func (d *Dealer) Register(callee *session, msg *wamp.Register) {
 		}
 	}
 
-	// If the callee supports progressive call results, but does not support
-	// call canceling, then disable the callee's progressive call results
-	// feature.  Call canceling is necessary to stop progressive results if
-	// the caller session is closed during progressive result delivery.
-	if callee.HasFeature(roleCallee, featureProgCallResults) {
-		if !callee.HasFeature(roleCallee, featureCallCanceling) {
-			dict := wamp.DictChild(callee.Details, "roles")
-			dict = wamp.DictChild(dict, roleCallee)
-			dict = wamp.DictChild(dict, "features")
-			delete(dict, featureProgCallResults)
-			d.log.Println("disabling", featureProgCallResults, "for callee",
-				callee, "that does not support", featureCallCanceling)
-		}
-	}
-
 	invoke, _ := wamp.AsString(msg.Options[wamp.OptInvoke])
 	var metaPubs []*wamp.Publish
 	done := make(chan struct{})
@@ -260,7 +245,7 @@ func (d *Dealer) Register(callee *session, msg *wamp.Register) {
 }
 
 // Unregister removes a remote procedure previously registered by the callee.
-func (d *Dealer) Unregister(callee *session, msg *wamp.Unregister) {
+func (d *Dealer) Unregister(callee *wamp.Session, msg *wamp.Unregister) {
 	if callee == nil || msg == nil {
 		panic("dealer.Unregister with nil session or message")
 	}
@@ -278,7 +263,7 @@ func (d *Dealer) Unregister(callee *session, msg *wamp.Unregister) {
 }
 
 // Call invokes a registered remote procedure.
-func (d *Dealer) Call(caller *session, msg *wamp.Call) {
+func (d *Dealer) Call(caller *wamp.Session, msg *wamp.Call) {
 	if caller == nil || msg == nil {
 		panic("dealer.Call with nil session or message")
 	}
@@ -305,7 +290,7 @@ func (d *Dealer) Call(caller *session, msg *wamp.Call) {
 // invocation or interrupt from the callee is discarded when received.
 //
 // If the callee does not support call canceling, then behavior is "skip".
-func (d *Dealer) Cancel(caller *session, msg *wamp.Cancel) {
+func (d *Dealer) Cancel(caller *wamp.Session, msg *wamp.Cancel) {
 	if caller == nil || msg == nil {
 		panic("dealer.Cancel with nil session or message")
 	}
@@ -332,10 +317,10 @@ func (d *Dealer) Cancel(caller *session, msg *wamp.Cancel) {
 // Yield handles the result of successfully processing and finishing the
 // execution of a call, send from callee to dealer.
 //
-// If the RESULT could not be sent to the caller because he caller was blocked
+// If the RESULT could not be sent to the caller because the caller was blocked
 // (send queue full), then retry sending until timeout.  If timeout while
 // trying to send RESULT, then cancel call.
-func (d *Dealer) Yield(callee *session, msg *wamp.Yield) {
+func (d *Dealer) Yield(callee *wamp.Session, msg *wamp.Yield) {
 	if callee == nil || msg == nil {
 		panic("dealer.Yield with nil session or message")
 	}
@@ -354,7 +339,9 @@ func (d *Dealer) Yield(callee *session, msg *wamp.Yield) {
 		start := time.Now()
 		// Retry processing YIELD until caller gone or deadline reached
 		for {
-			d.log.Println("Retry sending RESULT after", delay)
+			if d.debug {
+				d.log.Println("Retry sending RESULT after", delay)
+			}
 			<-time.After(delay)
 			// Do not retry if the elapsed time exceeds deadline
 			if time.Since(start) >= sendResultDeadline {
@@ -387,7 +374,7 @@ func (d *Dealer) Error(msg *wamp.Error) {
 // realm by sending a GOODBYE message or by disconnecting from the router.  If
 // there are any registrations for this session wamp.registration.on_unregister
 // and wamp.registration.on_delete meta events are published for each.
-func (d *Dealer) RemoveSession(sess *session) {
+func (d *Dealer) RemoveSession(sess *wamp.Session) {
 	if sess == nil {
 		// No session specified, no session removed.
 		return
@@ -422,7 +409,7 @@ func (d *Dealer) run() {
 	}
 }
 
-func (d *Dealer) register(callee *session, msg *wamp.Register, match, invokePolicy string, disclose, wampURI bool) []*wamp.Publish {
+func (d *Dealer) register(callee *wamp.Session, msg *wamp.Register, match, invokePolicy string, disclose, wampURI bool) []*wamp.Publish {
 	var metaPubs []*wamp.Publish
 	var reg *registration
 	switch match {
@@ -448,7 +435,7 @@ func (d *Dealer) register(callee *session, msg *wamp.Register, match, invokePoli
 			match:     match,
 			policy:    invokePolicy,
 			disclose:  disclose,
-			callees:   []*session{callee},
+			callees:   []*wamp.Session{callee},
 		}
 		d.registrations[regID] = reg
 		switch match {
@@ -546,7 +533,7 @@ func (d *Dealer) register(callee *session, msg *wamp.Register, match, invokePoli
 	return metaPubs
 }
 
-func (d *Dealer) unregister(callee *session, msg *wamp.Unregister) []*wamp.Publish {
+func (d *Dealer) unregister(callee *wamp.Session, msg *wamp.Unregister) []*wamp.Publish {
 	var metaPubs []*wamp.Publish
 	// Delete the registration ID from the callee's set of registrations.
 	if _, ok := d.calleeRegIDSet[callee]; ok {
@@ -637,7 +624,7 @@ func (d *Dealer) matchProcedure(procedure wamp.URI) (*registration, bool) {
 	return reg, ok
 }
 
-func (d *Dealer) call(caller *session, msg *wamp.Call) {
+func (d *Dealer) call(caller *wamp.Session, msg *wamp.Call) {
 	reg, ok := d.matchProcedure(msg.Procedure)
 	if !ok || len(reg.callees) == 0 {
 		// If no registered procedure, send error.
@@ -650,7 +637,7 @@ func (d *Dealer) call(caller *session, msg *wamp.Call) {
 		return
 	}
 
-	var callee *session
+	var callee *wamp.Session
 
 	// If there are multiple callees, then select a callee based invocation
 	// policy.
@@ -731,17 +718,20 @@ func (d *Dealer) call(caller *session, msg *wamp.Call) {
 	// A Caller indicates its willingness to receive progressive results by
 	// setting CALL.Options.receive_progress|bool := true
 	if opt, _ := msg.Options[wamp.OptReceiveProgress].(bool); opt {
-		// If the Callee supports progressive calls, the Dealer will
-		// forward the Caller's willingness to receive progressive
-		// results by setting.
-		if callee.HasFeature(roleCallee, featureProgCallResults) {
+		// If the Callee supports progressive calls, the Dealer will forward
+		// the Caller's willingness to receive progressive results by setting.
+		//
+		// The Callee must all call canceling, as this is necessary to stop
+		// progressive results if the caller session is closed during
+		// progressive result delivery.
+		if callee.HasFeature(roleCallee, featureProgCallResults) && callee.HasFeature(roleCallee, featureCallCanceling) {
 			details[wamp.OptReceiveProgress] = true
 		}
 	}
 
 	if reg.match != wamp.MatchExact {
-		// According to the spec, a router must provide the actual
-		// procedure to the client.
+		// According to the spec, a router must provide the actual procedure to
+		// the client.
 		details[wamp.OptProcedure] = msg.Procedure
 	}
 
@@ -776,7 +766,7 @@ func (d *Dealer) call(caller *session, msg *wamp.Call) {
 	}
 }
 
-func (d *Dealer) cancel(caller *session, msg *wamp.Cancel, mode string, reason wamp.URI) {
+func (d *Dealer) cancel(caller *wamp.Session, msg *wamp.Cancel, mode string, reason wamp.URI) {
 	reqID := requestID{
 		session: caller.ID,
 		request: msg.Request,
@@ -859,7 +849,7 @@ func (d *Dealer) cancel(caller *session, msg *wamp.Cancel, mode string, reason w
 	})
 }
 
-func (d *Dealer) yield(callee *session, msg *wamp.Yield, canRetry bool) bool {
+func (d *Dealer) yield(callee *wamp.Session, msg *wamp.Yield, canRetry bool) bool {
 	progress, _ := msg.Options[wamp.OptProgress].(bool)
 
 	// Find and delete pending invocation.
@@ -987,7 +977,7 @@ func (d *Dealer) error(msg *wamp.Error) {
 	})
 }
 
-func (d *Dealer) removeSession(sess *session) []*wamp.Publish {
+func (d *Dealer) removeSession(sess *wamp.Session) []*wamp.Publish {
 	var metaPubs []*wamp.Publish
 	// Remove any remaining registrations for the removed session.
 	for regID := range d.calleeRegIDSet[sess] {
@@ -1046,7 +1036,7 @@ func (d *Dealer) removeSession(sess *session) []*wamp.Publish {
 // If there are no more callees for the registration, then the registration is
 // removed and true is returned to indicate that the last registration was
 // deleted.
-func (d *Dealer) delCalleeReg(callee *session, regID wamp.ID) (bool, error) {
+func (d *Dealer) delCalleeReg(callee *wamp.Session, regID wamp.ID) (bool, error) {
 	reg, ok := d.registrations[regID]
 	if !ok {
 		// The registration doesn't exist
@@ -1284,7 +1274,7 @@ func (d *Dealer) RegCountCallees(msg *wamp.Invocation) wamp.Message {
 	}
 }
 
-func (d *Dealer) trySend(sess *session, msg wamp.Message) bool {
+func (d *Dealer) trySend(sess *wamp.Session, msg wamp.Message) bool {
 	if err := sess.TrySend(msg); err != nil {
 		d.log.Printf("!!! Dropped %s to session %s: %s", msg.MessageType(), sess, err)
 		return false
@@ -1293,15 +1283,15 @@ func (d *Dealer) trySend(sess *session, msg wamp.Message) bool {
 }
 
 // discloseCaller adds caller identity information to INVOCATION.Details.
-func discloseCaller(caller *session, details wamp.Dict) {
+func discloseCaller(caller *wamp.Session, details wamp.Dict) {
 	details[roleCaller] = caller.ID
 	// These values are not required by the specification, but are here for
 	// compatibility with Crossbar.
-	caller.rLock()
+	caller.Lock()
 	for _, f := range []string{"authid", "authrole"} {
 		if val, ok := caller.Details[f]; ok {
 			details[fmt.Sprintf("%s_%s", roleCaller, f)] = val
 		}
 	}
-	caller.rUnlock()
+	caller.Unlock()
 }
diff --git a/router/realm.go b/router/realm.go
index 05b636e..e9ae5e0 100644
--- a/router/realm.go
+++ b/router/realm.go
@@ -92,13 +92,12 @@ type realm struct {
 	authenticators map[string]auth.Authenticator
 
 	// session ID -> Session
-	clients map[wamp.ID]*session
+	clients map[wamp.ID]*wamp.Session
 	// session ID -> testament
 	testaments map[wamp.ID]testamentBucket
-	clientStop chan struct{}
 
 	metaPeer  wamp.Peer
-	metaSess  *session
+	metaSess  *wamp.Session
 	metaIDGen *wamp.IDGen
 
 	actionChan chan func()
@@ -126,6 +125,13 @@ type realm struct {
 	enableMetaModify bool
 }
 
+var (
+	shutdownGoodbye = &wamp.Goodbye{
+		Reason:  wamp.ErrSystemShutdown,
+		Details: wamp.Dict{},
+	}
+)
+
 // newRealm creates a new realm with the given RealmConfig, broker and dealer.
 func newRealm(config *RealmConfig, broker *Broker, dealer *Dealer, logger stdlog.StdLog, debug bool) (*realm, error) {
 	if !config.URI.ValidURI(config.StrictURI, "") {
@@ -137,9 +143,8 @@ func newRealm(config *RealmConfig, broker *Broker, dealer *Dealer, logger stdlog
 		broker:      broker,
 		dealer:      dealer,
 		authorizer:  config.Authorizer,
-		clients:     map[wamp.ID]*session{},
+		clients:     map[wamp.ID]*wamp.Session{},
 		testaments:  map[wamp.ID]testamentBucket{},
-		clientStop:  make(chan struct{}),
 		actionChan:  make(chan func()),
 		metaIDGen:   new(wamp.IDGen),
 		metaDone:    make(chan struct{}),
@@ -229,9 +234,16 @@ func (r *realm) close() {
 	// running, before closing.
 	r.waitReady()
 
-	// Kick all clients off.  Clients will not generate meta events when the
-	// ErrSystesShutdown reason is given.
-	close(r.clientStop)
+	// Kick all clients off.  Sending shutdownGoodbye causes client message
+	// handlers to exit without sending meta events.
+	sync := make(chan struct{})
+	r.actionChan <- func() {
+		for _, c := range r.clients {
+			c.EndRecv(shutdownGoodbye)
+		}
+		close(sync)
+	}
+	<-sync
 
 	// Wait until each client's handleInboundMessages() has exited.  No new
 	// messages can be generated once sessions are closed.
@@ -241,7 +253,7 @@ func (r *realm) close() {
 	// the meta client receives GOODBYE from the meta session, the meta
 	// session is done and will not try to publish anything more to the
 	// broker, and it is finally safe to exit and close the broker.
-	r.metaSess.kill(nil)
+	r.metaSess.EndRecv(shutdownGoodbye)
 	<-r.metaDone
 
 	// handleInboundMessages() and metaProcedureHandler() are the only things
@@ -306,9 +318,9 @@ func (r *realm) run() {
 // createMetaSession creates and starts a session that runs in this realm, and
 // bridges two peers.  One peer, the r.metaSess, is associated with the router
 // and handles meta session requests.  The other, r.metaPeer, is the remote
-// side of the router uplink and is used as the interface to send meta session
-// messages to.  Sending a PUBLISH message to it will result in the router
-// publishing the event to any subscribers.
+// (client) side of the router uplink and is used as the interface to send meta
+// session messages to.  Sending a PUBLISH message to it will result in the
+// router publishing the event to any subscribers.
 func (r *realm) createMetaSession() {
 	cli, rtr := transport.LinkedPeers()
 	r.metaPeer = cli
@@ -316,10 +328,10 @@ func (r *realm) createMetaSession() {
 	r.dealer.SetMetaPeer(cli)
 
 	// This session is the local leg of the router uplink.
-	r.metaSess = newSession(rtr, metaID, wamp.Dict{"authrole": "trusted"})
+	r.metaSess = wamp.NewSession(rtr, metaID, wamp.Dict{"authrole": "trusted"}, nil)
 
 	// Run the handler for messages from the meta session.
-	go r.handleInboundMessages(r.metaSess, nil)
+	go r.handleInboundMessages(r.metaSess)
 	if r.debug {
 		r.log.Println("Started meta-session", r.metaSess)
 	}
@@ -330,7 +342,7 @@ func (r *realm) createMetaSession() {
 //
 // Note: onJoin() is called from handleSession, not handleInboundMessages, so
 // that it is not called for the meta client.
-func (r *realm) onJoin(sess *session) {
+func (r *realm) onJoin(sess *wamp.Session) {
 	r.waitHandlers.Add(1)
 	sync := make(chan struct{})
 	r.actionChan <- func() {
@@ -345,9 +357,9 @@ func (r *realm) onJoin(sess *session) {
 	// WAMP spec only specifies publishing "session", "authid", "authrole",
 	// "authmethod", "authprovider", "transport".  This implementation
 	// publishes all details except transport.auth.
-	sess.rLock()
+	sess.Lock()
 	output := r.cleanSessionDetails(sess.Details)
-	sess.rUnlock()
+	sess.Unlock()
 	r.metaPeer.Send(&wamp.Publish{
 		Request:   wamp.GlobalID(),
 		Topic:     wamp.MetaEventSessionOnJoin,
@@ -368,7 +380,7 @@ func (r *realm) onJoin(sess *session) {
 //
 // Note: onLeave() must be called from outside handleInboundMessages so that it
 // is not called for the meta client.
-func (r *realm) onLeave(sess *session, shutdown, killAll bool) {
+func (r *realm) onLeave(sess *wamp.Session, shutdown, killAll bool) {
 	var testaments testamentBucket
 	var hasTstm bool
 	sync := make(chan struct{})
@@ -422,7 +434,7 @@ func (r *realm) onLeave(sess *session, shutdown, killAll bool) {
 // HandleSession starts a session attached to this realm.
 //
 // Routing occurs only between WAMP Sessions that have joined the same Realm.
-func (r *realm) handleSession(sess *session) error {
+func (r *realm) handleSession(sess *wamp.Session) error {
 	// The lock is held in mutual exclusion with the closing of the realm.
 	// This ensures that no new session handler can start once the realm is
 	// closing, during which the realm waits for all existing session handlers
@@ -442,7 +454,7 @@ func (r *realm) handleSession(sess *session) error {
 		r.log.Println("Started session", sess)
 	}
 	go func() {
-		shutdown, killAll, err := r.handleInboundMessages(sess, r.clientStop)
+		shutdown, killAll, err := r.handleInboundMessages(sess)
 		if err != nil {
 			abortMsg := wamp.Abort{
 				Reason:  wamp.ErrProtocolViolation,
@@ -460,39 +472,27 @@ func (r *realm) handleSession(sess *session) error {
 
 // handleInboundMessages handles the messages sent from a client session to
 // the router.
-func (r *realm) handleInboundMessages(sess *session, stopChan <-chan struct{}) (bool, bool, error) {
+func (r *realm) handleInboundMessages(sess *wamp.Session) (bool, bool, error) {
 	if r.debug {
 		defer r.log.Println("Ended session", sess)
 	}
-	killChan := sess.killChan
-	recvChan := sess.Recv()
 	for {
 		var msg wamp.Message
 		var open bool
 		select {
-		case msg, open = <-recvChan:
+		case msg, open = <-sess.Recv():
 			if !open {
 				r.log.Println("Lost", sess)
 				return false, false, nil
 			}
-		case <-stopChan:
-			if r.debug {
-				r.log.Printf("Stop session %s: system shutdown", sess)
-			}
-			sess.TrySend(&wamp.Goodbye{
-				Reason:  wamp.ErrSystemShutdown,
-				Details: wamp.Dict{},
-			})
-			return true, false, nil
-		case goodbye, open := <-killChan:
-			if !open {
+		case <-sess.RecvDone():
+			goodbye := sess.Goodbye()
+			switch goodbye {
+			case shutdownGoodbye, wamp.NoGoodbye:
 				if r.debug {
 					r.log.Printf("Stop session %s: system shutdown", sess)
 				}
-				sess.TrySend(&wamp.Goodbye{
-					Reason:  wamp.ErrSystemShutdown,
-					Details: wamp.Dict{},
-				})
+				sess.TrySend(goodbye)
 				return true, false, nil
 			}
 			if r.debug {
@@ -566,7 +566,7 @@ func (r *realm) handleInboundMessages(sess *session, stopChan <-chan struct{}) (
 // authzMessage checks if the session is authorized to send the message.  If
 // authorization fails or if the session is not authorized, then an error
 // response is returned to the client, and this method returns false.
-func (r *realm) authzMessage(sess *session, msg wamp.Message) bool {
+func (r *realm) authzMessage(sess *wamp.Session, msg wamp.Message) bool {
 	// If the client is local, then do not check authorization, unless
 	// requested in config.
 	if transport.IsLocal(sess.Peer) && !r.localAuthz {
@@ -574,15 +574,16 @@ func (r *realm) authzMessage(sess *session, msg wamp.Message) bool {
 	}
 
 	// Create a safe session to prevent access to the session.Peer.
-	safeSession := wamp.Session{
+	safeSession := &wamp.Session{
 		ID:      sess.ID,
 		Details: sess.Details,
 	}
+
 	// Write-lock the session, becuase there is no telling what the Authorizer
 	// will do to the session details.
-	sess.lock()
-	isAuthz, err := r.authorizer.Authorize(&safeSession, msg)
-	sess.unlock()
+	sess.Lock()
+	isAuthz, err := r.authorizer.Authorize(safeSession, msg)
+	sess.Unlock()
 
 	if !isAuthz {
 		skipResponse := false
@@ -817,9 +818,9 @@ func (r *realm) sessionCount(msg *wamp.Invocation) wamp.Message {
 		r.actionChan <- func() {
 			var nclients int
 			for _, sess := range r.clients {
-				sess.rLock()
+				sess.Lock()
 				authrole, _ := wamp.AsString(sess.Details["authrole"])
-				sess.rUnlock()
+				sess.Unlock()
 				for j := range filter {
 					if filter[j] == authrole {
 						nclients++
@@ -875,9 +876,9 @@ func (r *realm) sessionList(msg *wamp.Invocation) wamp.Message {
 		r.actionChan <- func() {
 			var ids []wamp.ID
 			for sid, sess := range r.clients {
-				sess.rLock()
+				sess.Lock()
 				authrole, _ := wamp.AsString(sess.Details["authrole"])
-				sess.rUnlock()
+				sess.Unlock()
 				for j := range filter {
 					if filter[j] == authrole {
 						ids = append(ids, sid)
@@ -904,7 +905,7 @@ func (r *realm) sessionGet(msg *wamp.Invocation) wamp.Message {
 		return makeError(msg.Request, wamp.ErrNoSuchSession)
 	}
 
-	retChan := make(chan *session)
+	retChan := make(chan *wamp.Session)
 	r.actionChan <- func() {
 		sess, _ := r.clients[sid]
 		retChan <- sess
@@ -918,9 +919,9 @@ func (r *realm) sessionGet(msg *wamp.Invocation) wamp.Message {
 	// "authmethod", "authprovider", and "transport".  All details are returned
 	// in this implementation, except transport.auth, unless Config.MetaStrict
 	// is set to true.
-	sess.rLock()
+	sess.Lock()
 	output := r.cleanSessionDetails(sess.Details)
-	sess.rUnlock()
+	sess.Unlock()
 
 	return &wamp.Yield{
 		Request:   msg.Request,
@@ -1067,7 +1068,7 @@ func (r *realm) sessionModifyDetails(msg *wamp.Invocation) wamp.Message {
 
 	done := make(chan struct{})
 	r.actionChan <- func() {
-		var sess *session
+		var sess *wamp.Session
 		if sess, ok = r.clients[sid]; ok {
 			r.modifySessionDetails(sess, delta)
 		}
@@ -1275,7 +1276,7 @@ func (r *realm) killSession(sid wamp.ID, reason wamp.URI, message string) error
 			errChan <- errors.New("no such session")
 			return
 		}
-		sess.kill(goodbye)
+		sess.EndRecv(goodbye)
 		close(errChan)
 	}
 	return <-errChan
@@ -1294,14 +1295,14 @@ func (r *realm) killSessionsByDetail(key, value string, reason wamp.URI, message
 				continue
 			}
 
-			sess.rLock()
+			sess.Lock()
 			val, ok := wamp.AsString(sess.Details[key])
-			sess.rUnlock()
+			sess.Unlock()
 
 			if !ok || val != value {
 				continue
 			}
-			if sess.kill(goodbye) {
+			if sess.EndRecv(goodbye) {
 				kills++
 			}
 		}
@@ -1325,7 +1326,7 @@ func (r *realm) killAllSessions(reason wamp.URI, message string, exclude wamp.ID
 			if sid == exclude {
 				continue
 			}
-			if sess.kill(goodbye) {
+			if sess.EndRecv(goodbye) {
 				kills++
 			}
 		}
@@ -1340,9 +1341,9 @@ func (r *realm) killAllSessions(reason wamp.URI, message string, exclude wamp.ID
 // An item with a non-nil value in the delta wamp.Dict specifies adding or
 // updating that item in the session details.  An item with a nil value in the
 // delta wamp.Dict specifies deleting that item from the session details.
-func (r *realm) modifySessionDetails(sess *session, delta wamp.Dict) {
-	sess.lock()
-	defer sess.unlock()
+func (r *realm) modifySessionDetails(sess *wamp.Session, delta wamp.Dict) {
+	sess.Lock()
+	defer sess.Unlock()
 
 	for k, v := range delta {
 		if v == nil {
diff --git a/router/router.go b/router/router.go
index 4454cb9..33651f0 100644
--- a/router/router.go
+++ b/router/router.go
@@ -216,33 +216,26 @@ func (r *router) AttachClient(client wamp.Peer, transportDetails wamp.Dict) erro
 	}
 
 	hello.Details = wamp.NormalizeDict(hello.Details)
+	sid := wamp.GlobalID()
+
+	// Create new session.
+	sess := wamp.NewSession(client, sid, nil, hello.Details)
 
 	// A Client must announce the roles it supports via
 	// Hello.Details.roles|dict, where the keys can be: publisher, subscriber,
-	// caller, callee.  If the client announces any roles, to list specific
-	// features for the role, then check that the role is something this router
-	// recognizes.
-	_roleVals, err := wamp.DictValue(hello.Details, []string{"roles"})
-	if err != nil {
-		err = errors.New("no client roles specified")
-		sendAbort(wamp.ErrNoSuchRole, err)
-		return err
+	// caller, callee.  Check that client has at least one supported role.
+	var rolesOK bool
+	for _, role := range []string{"publisher", "subscriber", "caller", "callee"} {
+		if sess.HasRole(role) {
+			rolesOK = true
+			break
+		}
 	}
-	roleVals, ok := _roleVals.(wamp.Dict)
-	if !ok || len(roleVals) == 0 {
-		err = errors.New("no client roles specified")
+	if !rolesOK {
+		err = errors.New("client did not announce any supported roles")
 		sendAbort(wamp.ErrNoSuchRole, err)
 		return err
 	}
-	for roleName := range roleVals {
-		switch roleName {
-		case "publisher", "subscriber", "caller", "callee":
-		default:
-			err = errors.New("invalid client role specified: " + roleName)
-			sendAbort(wamp.ErrNoSuchRole, err)
-			return err
-		}
-	}
 
 	// Include any transport details with HELLO.Details.
 	if len(transportDetails) != 0 {
@@ -253,7 +246,6 @@ func (r *router) AttachClient(client wamp.Peer, transportDetails wamp.Dict) erro
 	// message or an error.
 	//
 	// Authentication may take some time.
-	sid := wamp.GlobalID()
 	welcome, err := realm.authClient(sid, client, hello.Details)
 	if err != nil {
 		sendAbort(wamp.ErrAuthenticationFailed, err)
@@ -267,7 +259,7 @@ func (r *router) AttachClient(client wamp.Peer, transportDetails wamp.Dict) erro
 	// only.
 	sessDetails := make(wamp.Dict, len(hello.Details)+len(welcome.Details))
 	for k, v := range hello.Details {
-		if k == "authmethods" {
+		if k == "authmethods" || k == "roles" {
 			continue
 		}
 		sessDetails[k] = v
@@ -280,8 +272,7 @@ func (r *router) AttachClient(client wamp.Peer, transportDetails wamp.Dict) erro
 	}
 	sessDetails["session"] = sid
 
-	// Create new session.
-	sess := newSession(client, sid, sessDetails)
+	sess.Details = sessDetails
 
 	if err := realm.handleSession(sess); err != nil {
 		// Any error returned here is a shutdown error.
diff --git a/router/session.go b/router/session.go
deleted file mode 100644
index dac3a05..0000000
--- a/router/session.go
+++ /dev/null
@@ -1,70 +0,0 @@
-package router
-
-import (
-	"sync"
-
-	"github.com/gammazero/nexus/wamp"
-)
-
-// session is a wrapper around a wamp.Session to provide the router with a
-// lockable killable session.
-type session struct {
-	wamp.Session
-
-	killChan chan *wamp.Goodbye
-	rwlock   sync.RWMutex
-}
-
-// newSession creates a new lockable session.
-func newSession(peer wamp.Peer, sid wamp.ID, details wamp.Dict) *session {
-	if sid == 0 {
-		sid = wamp.GlobalID()
-	}
-
-	return &session{
-		Session: wamp.Session{
-			Peer:    peer,
-			ID:      sid,
-			Details: details,
-		},
-		killChan: make(chan *wamp.Goodbye),
-	}
-}
-
-func (s *session) rLock()   { s.rwlock.RLock() }
-func (s *session) rUnlock() { s.rwlock.RUnlock() }
-func (s *session) lock()    { s.rwlock.Lock() }
-func (s *session) unlock()  { s.rwlock.Unlock() }
-
-func (s *session) kill(goodbye *wamp.Goodbye) bool {
-	if s.killChan == nil {
-		return false
-	}
-	if goodbye == nil {
-		close(s.killChan)
-	} else {
-		s.killChan <- goodbye
-	}
-	s.killChan = nil // prevent subsequent kill from using chan
-	return true
-}
-
-// String returns the session ID as a string.
-func (s *session) String() string { return s.Session.String() }
-
-// HasRole returns true if the session supports the specified role.
-func (s *session) HasRole(role string) bool {
-	s.rwlock.RLock()
-	ok := s.Session.HasRole(role)
-	s.rwlock.RUnlock()
-	return ok
-}
-
-// HasFeature returns true if the session has the specified feature for the
-// specified role.
-func (s *session) HasFeature(role, feature string) bool {
-	s.rwlock.RLock()
-	ok := s.Session.HasFeature(role, feature)
-	s.rwlock.RUnlock()
-	return ok
-}
diff --git a/wamp/peer.go b/wamp/peer.go
index eb85227..528d472 100644
--- a/wamp/peer.go
+++ b/wamp/peer.go
@@ -16,7 +16,7 @@ type Peer interface {
 	// TrySend performs a non-blocking send.  Returns error if blocked.
 	TrySend(Message) error
 
-	// Closes the peer connection and and channel returned from Recv().
+	// Closes the peer connection and the channel returned from Recv().
 	Close()
 
 	// Recv returns a channel of messages from the peer.
diff --git a/wamp/session.go b/wamp/session.go
index bd56fce..ae271b1 100644
--- a/wamp/session.go
+++ b/wamp/session.go
@@ -1,6 +1,9 @@
 package wamp
 
-import "fmt"
+import (
+	"fmt"
+	"sync"
+)
 
 // Session is an active WAMP session.  It associates a session ID and details
 // with a connected Peer, which is the remote side of the session.  So, if the
@@ -12,20 +15,150 @@ type Session struct {
 	ID ID
 	// Details about session.
 	Details Dict
+
+	// Roles and features supported by peer.
+	roles map[string]map[string]struct{}
+
+	mu      sync.Mutex
+	done    chan struct{}
+	goodbye *Goodbye
+}
+
+var (
+	// NoGoodbye indicates that no Goodbye message was sent out
+	NoGoodbye = &Goodbye{}
+)
+
+// NewSession creates a new session.  The greetDetails is the details from the
+// HELLO or WELCOME message, from which roles and features are extracted.
+func NewSession(peer Peer, id ID, details Dict, greetDetails Dict) *Session {
+	s := &Session{
+		Peer:    peer,
+		ID:      id,
+		Details: details,
+	}
+	s.setRoles(greetDetails)
+	return s
 }
 
+// Lock locks the session to protect against concurrent updates.
+func (s *Session) Lock() { s.mu.Lock() }
+
+// Unlock unlocks the session.
+func (s *Session) Unlock() { s.mu.Unlock() }
+
 // String returns the session ID as a string.
-func (s Session) String() string { return fmt.Sprintf("%d", s.ID) }
+func (s *Session) String() string { return fmt.Sprintf("%d", s.ID) }
 
 // HasRole returns true if the session supports the specified role.
-func (s Session) HasRole(role string) bool {
-	_, err := DictValue(s.Details, []string{"roles", role})
-	return err == nil
+func (s *Session) HasRole(role string) bool {
+	_, ok := s.roles[role]
+	return ok
 }
 
 // HasFeature returns true if the session has the specified feature for the
 // specified role.
-func (s Session) HasFeature(role, feature string) bool {
-	b, _ := DictFlag(s.Details, []string{"roles", role, "features", feature})
-	return b
+func (s *Session) HasFeature(role, feature string) bool {
+	features, ok := s.roles[role]
+	if !ok {
+		return false
+	}
+	_, ok = features[feature]
+	return ok
+}
+
+// RecvDone returns a channel that is closed when this session has been ended
+// by calling EndRecv.
+func (s *Session) RecvDone() <-chan struct{} {
+	s.mu.Lock()
+	if s.done == nil {
+		s.done = make(chan struct{})
+	}
+	d := s.done
+	s.mu.Unlock()
+	return d
+}
+
+// If RecvDone is not yet closed, Goodbye returns nil.
+// If RecvDone is closed, Goodbye returns the GOODBYE message that was supplied
+// when RecvEnd was called.
+func (s *Session) Goodbye() *Goodbye {
+	s.mu.Lock()
+	g := s.goodbye
+	s.mu.Unlock()
+	return g
+}
+
+// EndRecv tells the session to signal messages handlers to stop receiving
+// messages from this session.  An optional goodbye message may be provided for
+// the message handler to send to the peer.  It is the responsibility of the
+// message handler to send the goodbye message, so that this can be coordinated
+// with exiting the message handler for other reasons.
+func (s *Session) EndRecv(goodbye *Goodbye) bool {
+	s.mu.Lock()
+	if s.goodbye != nil {
+		s.mu.Unlock()
+		return false // already ended
+	}
+
+	if goodbye == nil {
+		s.goodbye = NoGoodbye
+	} else {
+		s.goodbye = goodbye
+	}
+
+	if s.done == nil {
+		s.done = make(chan struct{})
+	}
+	close(s.done)
+
+	s.mu.Unlock()
+	return true
+}
+
+// setRoles extracts the specified roles from HELLO or WELCOME details, and
+// configures the session with the roles and features for each role.
+func (s *Session) setRoles(details Dict) {
+	_roles, ok := details["roles"]
+	if !ok {
+		s.roles = nil // no roles
+		return
+	}
+	roles, ok := AsDict(_roles)
+	if !ok || len(roles) == 0 {
+		s.roles = nil // no roles
+		return
+	}
+
+	roleMap := make(map[string]map[string]struct{})
+	for role, _roleDict := range roles {
+		roleMap[role] = nil
+		roleDict, ok := _roleDict.(Dict)
+		if !ok {
+			roleDict = NormalizeDict(_roleDict)
+			if roleDict == nil {
+				continue
+			}
+		}
+		_features, ok := roleDict["features"]
+		if !ok {
+			continue
+		}
+		features, ok := _features.(Dict)
+		if !ok {
+			features = NormalizeDict(_features)
+			if features == nil {
+				continue
+			}
+		}
+		featMap := make(map[string]struct{})
+		for feature, iface := range features {
+			if b, _ := iface.(bool); !b {
+				continue
+			}
+			featMap[feature] = struct{}{}
+		}
+		roleMap[role] = featMap
+	}
+	s.roles = roleMap
 }


Test patch:
diff --git a/aat/auth_test.go b/aat/auth_test.go
index bc666b9..bcc0c32 100644
--- a/aat/auth_test.go
+++ b/aat/auth_test.go
@@ -76,8 +76,8 @@ func TestJoinRealmWithCRCookieAuth(t *testing.T) {
 	if err != nil {
 		t.Fatal("Failed to connect client:", err)
 	}
-	cli.Close()
 	details := cli.RealmDetails()
+	cli.Close()
 
 	// Client should not have be authenticated by cookie first time.
 	if ok, _ := wamp.AsBool(details["authbycookie"]); ok {
@@ -88,8 +88,8 @@ func TestJoinRealmWithCRCookieAuth(t *testing.T) {
 	if err != nil {
 		t.Fatal("Failed to connect client:", err)
 	}
-	cli.Close()
 	details = cli.RealmDetails()
+	cli.Close()
 
 	// If websocket, then should be authenticated by cookie this time.
 	if cfg.WsCfg.Jar != nil {
diff --git a/aat/session_meta_test.go b/aat/session_meta_test.go
index bfb522c..c6fff01 100644
--- a/aat/session_meta_test.go
+++ b/aat/session_meta_test.go
@@ -135,6 +135,8 @@ func TestMetaEventOnLeave(t *testing.T) {
 		}
 	}
 
+	sid := sess.ID()
+
 	// Disconnect client to generate wamp.session.on_leave event.
 	err = sess.Close()
 	if err != nil {
@@ -154,9 +156,9 @@ func TestMetaEventOnLeave(t *testing.T) {
 		t.Fatal("expected 3 event args, got:", len(eventArgs))
 	}
 	onLeaveID, _ := wamp.AsID(eventArgs[0])
-	if onLeaveID != sess.ID() {
+	if onLeaveID != sid {
 		t.Fatal(metaOnLeave, "meta event had wrong session ID, got", onLeaveID,
-			"want", sess.ID())
+			"want", sid)
 	}
 	authid, _ := wamp.AsString(eventArgs[1])
 	if len(authid) == 0 {
@@ -354,6 +356,8 @@ func TestMetaProcSessionList(t *testing.T) {
 	}
 	firstLen := len(list)
 
+	sid := sess.ID()
+
 	err = sess.Close()
 	if err != nil {
 		t.Fatal("Failed to disconnect client:", err)
@@ -379,7 +383,7 @@ func TestMetaProcSessionList(t *testing.T) {
 	found = false
 	for i := range list {
 		id, _ := wamp.AsID(list[i])
-		if id == sess.ID() {
+		if id == sid {
 			found = true
 			break
 		}
@@ -463,6 +467,8 @@ func TestMetaProcSessionGet(t *testing.T) {
 		}
 	}
 
+	sid := sess.ID()
+
 	err = sess.Close()
 	if err != nil {
 		t.Fatal("Failed to disconnect client:", err)
@@ -471,7 +477,7 @@ func TestMetaProcSessionGet(t *testing.T) {
 	<-sync
 
 	// Call meta procedure to get session list.
-	result, err = caller.Call(ctx, metaGet, nil, wamp.List{sess.ID()}, nil, "")
+	result, err = caller.Call(ctx, metaGet, nil, wamp.List{sid}, nil, "")
 	if err == nil {
 		t.Fatal("Expected error")
 	}
diff --git a/client/client_test.go b/client/client_test.go
index 6438921..4f470bf 100644
--- a/client/client_test.go
+++ b/client/client_test.go
@@ -837,3 +837,63 @@ func TestClientRace(t *testing.T) {
 	killer.Close()
 	callee.Close()
 }
+
+// Test that if the router disconnects the client, while the client is running
+// an invocation handler, that the handler still gets marked as done when it
+// completes.
+func TestInvocationHandlerMissedDone(t *testing.T) {
+	//defer leaktest.Check(t)()
+
+	// Connect two clients to the same server
+	callee, caller, r, err := connectedTestClients()
+	if err != nil {
+		t.Fatal("failed to connect test clients:", err)
+	}
+
+	calledChan := make(chan struct{})
+
+	// Register procedure.
+	handler := func(ctx context.Context, args wamp.List, kwargs, details wamp.Dict) *InvokeResult {
+		close(calledChan)
+		<-ctx.Done()
+		time.Sleep(2 * time.Second)
+		return &InvokeResult{Args: wamp.List{args[0].(int) * 37}}
+	}
+	procName := "myproc"
+	if err = callee.Register(procName, handler, nil); err != nil {
+		t.Fatal("failed to register procedure:", err)
+	}
+
+	// Call procedure
+	callArgs := wamp.List{73}
+	ctx := context.Background()
+
+	go caller.Call(ctx, procName, nil, callArgs, nil, "")
+
+	<-calledChan
+
+	killArgs := wamp.List{callee.ID()}
+	killKwArgs := wamp.Dict{"reason": "com.session.kill", "message": "because i can"}
+	var result *wamp.Result
+	result, err = caller.Call(ctx, string(wamp.MetaProcSessionKill), nil, killArgs, killKwArgs, "")
+	if err != nil {
+		t.Log("Kill new client failed:", err)
+	}
+	if result == nil {
+		t.Log("Kill new client returned no result")
+	}
+
+	caller.Close()
+
+	done := make(chan struct{})
+	go func() {
+		callee.Close()
+		close(done)
+	}()
+	select {
+	case <-done:
+	case <-time.After(2 * time.Second):
+		t.Fatal("Timed out waiting to close client")
+	}
+	r.Close()
+}
diff --git a/router/broker_test.go b/router/broker_test.go
index 8636436..f16d3dd 100644
--- a/router/broker_test.go
+++ b/router/broker_test.go
@@ -38,7 +38,7 @@ func TestBasicSubscribe(t *testing.T) {
 	// Test subscribing to a topic.
 	broker := NewBroker(logger, false, true, debug, nil)
 	subscriber := newTestPeer()
-	sess := newSession(subscriber, 0, nil)
+	sess := wamp.NewSession(subscriber, 0, nil, nil)
 	testTopic := wamp.URI("nexus.test.topic")
 	broker.Subscribe(sess, &wamp.Subscribe{Request: 123, Topic: testTopic})
 
@@ -147,14 +147,14 @@ func TestUnsubscribe(t *testing.T) {
 
 	// Subscribe session1 to topic
 	subscriber := newTestPeer()
-	sess1 := newSession(subscriber, 0, nil)
+	sess1 := wamp.NewSession(subscriber, 0, nil, nil)
 	broker.Subscribe(sess1, &wamp.Subscribe{Request: 123, Topic: testTopic})
 	rsp := <-sess1.Recv()
 	subID := rsp.(*wamp.Subscribed).Subscription
 
 	// Subscribe session2 to topic
 	subscriber2 := newTestPeer()
-	sess2 := newSession(subscriber2, 0, nil)
+	sess2 := wamp.NewSession(subscriber2, 0, nil, nil)
 	broker.Subscribe(sess2, &wamp.Subscribe{Request: 567, Topic: testTopic})
 	rsp = <-sess2.Recv()
 	subID2 := rsp.(*wamp.Subscribed).Subscription
@@ -252,7 +252,7 @@ func TestRemove(t *testing.T) {
 	// Subscribe to topic
 	broker := NewBroker(logger, false, true, debug, nil)
 	subscriber := newTestPeer()
-	sess := newSession(subscriber, 0, nil)
+	sess := wamp.NewSession(subscriber, 0, nil, nil)
 	testTopic := wamp.URI("nexus.test.topic")
 	broker.Subscribe(sess, &wamp.Subscribe{Request: 123, Topic: testTopic})
 	rsp := <-sess.Recv()
@@ -267,7 +267,7 @@ func TestRemove(t *testing.T) {
 
 	// Wait for another subscriber as a way to wait for the RemoveSession to
 	// complete.
-	sess2 := newSession(subscriber, 0, nil)
+	sess2 := wamp.NewSession(subscriber, 0, nil, nil)
 	broker.Subscribe(sess2,
 		&wamp.Subscribe{Request: 789, Topic: wamp.URI("nexus.test.sync")})
 	rsp = <-sess2.Recv()
@@ -294,7 +294,7 @@ func TestRemove(t *testing.T) {
 func TestBasicPubSub(t *testing.T) {
 	broker := NewBroker(logger, false, true, debug, nil)
 	subscriber := newTestPeer()
-	sess := newSession(subscriber, 0, nil)
+	sess := wamp.NewSession(subscriber, 0, nil, nil)
 	testTopic := wamp.URI("nexus.test.topic")
 	msg := &wamp.Subscribe{
 		Request: 123,
@@ -310,7 +310,7 @@ func TestBasicPubSub(t *testing.T) {
 	}
 
 	publisher := newTestPeer()
-	pubSess := newSession(publisher, 0, nil)
+	pubSess := wamp.NewSession(publisher, 0, nil, nil)
 	broker.Publish(pubSess, &wamp.Publish{Request: 124, Topic: testTopic,
 		Arguments: wamp.List{"hello world"}})
 	rsp = <-sess.Recv()
@@ -333,7 +333,7 @@ func TestPrefxPatternBasedSubscription(t *testing.T) {
 	// Test match=prefix
 	broker := NewBroker(logger, false, true, debug, nil)
 	subscriber := newTestPeer()
-	sess := newSession(subscriber, 0, nil)
+	sess := wamp.NewSession(subscriber, 0, nil, nil)
 	testTopic := wamp.URI("nexus.test.topic")
 	testTopicPfx := wamp.URI("nexus.test.")
 	msg := &wamp.Subscribe{
@@ -375,7 +375,7 @@ func TestPrefxPatternBasedSubscription(t *testing.T) {
 	}
 
 	publisher := newTestPeer()
-	pubSess := newSession(publisher, 0, nil)
+	pubSess := wamp.NewSession(publisher, 0, nil, nil)
 	broker.Publish(pubSess, &wamp.Publish{Request: 124, Topic: testTopic})
 	rsp = <-sess.Recv()
 	evt, ok := rsp.(*wamp.Event)
@@ -396,7 +396,7 @@ func TestWildcardPatternBasedSubscription(t *testing.T) {
 	// Test match=prefix
 	broker := NewBroker(logger, false, true, debug, nil)
 	subscriber := newTestPeer()
-	sess := newSession(subscriber, 0, nil)
+	sess := wamp.NewSession(subscriber, 0, nil, nil)
 	testTopic := wamp.URI("nexus.test.topic")
 	testTopicWc := wamp.URI("nexus..topic")
 	msg := &wamp.Subscribe{
@@ -447,7 +447,7 @@ func TestWildcardPatternBasedSubscription(t *testing.T) {
 	}
 
 	publisher := newTestPeer()
-	pubSess := newSession(publisher, 0, nil)
+	pubSess := wamp.NewSession(publisher, 0, nil, nil)
 	broker.Publish(pubSess, &wamp.Publish{Request: 124, Topic: testTopic})
 	rsp = <-sess.Recv()
 	evt, ok := rsp.(*wamp.Event)
@@ -471,7 +471,7 @@ func TestSubscriberBlackwhiteListing(t *testing.T) {
 		"authid":   "jdoe",
 		"authrole": "admin",
 	}
-	sess := newSession(subscriber, wamp.GlobalID(), details)
+	sess := wamp.NewSession(subscriber, wamp.GlobalID(), details, nil)
 	testTopic := wamp.URI("nexus.test.topic")
 
 	broker.Subscribe(sess, &wamp.Subscribe{Request: 123, Topic: testTopic})
@@ -494,7 +494,7 @@ func TestSubscriberBlackwhiteListing(t *testing.T) {
 			},
 		},
 	}
-	pubSess := newSession(publisher, 0, details)
+	pubSess := wamp.NewSession(publisher, 0, details, nil)
 
 	// Test whilelist
 	broker.Publish(pubSess, &wamp.Publish{
@@ -574,7 +574,7 @@ func TestSubscriberBlackwhiteListing(t *testing.T) {
 func TestPublisherExclusion(t *testing.T) {
 	broker := NewBroker(logger, false, true, debug, nil)
 	subscriber := newTestPeer()
-	sess := newSession(subscriber, 0, nil)
+	sess := wamp.NewSession(subscriber, 0, nil, nil)
 	testTopic := wamp.URI("nexus.test.topic")
 
 	broker.Subscribe(sess, &wamp.Subscribe{Request: 123, Topic: testTopic})
@@ -600,7 +600,7 @@ func TestPublisherExclusion(t *testing.T) {
 			},
 		},
 	}
-	pubSess := newSession(publisher, 0, details)
+	pubSess := wamp.NewSession(publisher, 0, nil, details)
 
 	// Subscribe the publish session also.
 	broker.Subscribe(pubSess, &wamp.Subscribe{Request: 123, Topic: testTopic})
@@ -658,7 +658,7 @@ func TestPublisherIdentification(t *testing.T) {
 			},
 		},
 	}
-	sess := newSession(subscriber, 0, details)
+	sess := wamp.NewSession(subscriber, 0, nil, details)
 
 	testTopic := wamp.URI("nexus.test.topic")
 
@@ -672,7 +672,7 @@ func TestPublisherIdentification(t *testing.T) {
 	}
 
 	publisher := newTestPeer()
-	pubSess := newSession(publisher, wamp.GlobalID(), nil)
+	pubSess := wamp.NewSession(publisher, wamp.GlobalID(), nil, nil)
 	broker.Publish(pubSess, &wamp.Publish{
 		Request: 124,
 		Topic:   testTopic,
diff --git a/router/dealer_test.go b/router/dealer_test.go
index a8aecc5..462783d 100644
--- a/router/dealer_test.go
+++ b/router/dealer_test.go
@@ -45,7 +45,7 @@ func TestBasicRegister(t *testing.T) {
 
 	// Register callee
 	callee := newTestPeer()
-	sess := newSession(callee, 0, nil)
+	sess := wamp.NewSession(callee, 0, nil, nil)
 	dealer.Register(sess, &wamp.Register{Request: 123, Procedure: testProcedure})
 
 	rsp := <-callee.Recv()
@@ -100,7 +100,7 @@ func TestUnregister(t *testing.T) {
 
 	// Register a procedure.
 	callee := newTestPeer()
-	sess := newSession(callee, 0, nil)
+	sess := wamp.NewSession(callee, 0, nil, nil)
 	dealer.Register(sess, &wamp.Register{Request: 123, Procedure: testProcedure})
 	rsp := <-callee.Recv()
 	regID := rsp.(*wamp.Registered).Registration
@@ -150,7 +150,7 @@ func TestBasicCall(t *testing.T) {
 
 	// Register a procedure.
 	callee := newTestPeer()
-	calleeSess := newSession(callee, 0, nil)
+	calleeSess := wamp.NewSession(callee, 0, nil, nil)
 	dealer.Register(calleeSess,
 		&wamp.Register{Request: 123, Procedure: testProcedure})
 	var rsp wamp.Message
@@ -171,7 +171,7 @@ func TestBasicCall(t *testing.T) {
 	}
 
 	caller := newTestPeer()
-	callerSession := newSession(caller, 0, nil)
+	callerSession := wamp.NewSession(caller, 0, nil, nil)
 
 	// Test calling invalid procedure
 	dealer.Call(callerSession,
@@ -240,7 +240,7 @@ func TestRemovePeer(t *testing.T) {
 
 	// Register a procedure.
 	callee := newTestPeer()
-	sess := newSession(callee, 0, nil)
+	sess := wamp.NewSession(callee, 0, nil, nil)
 	msg := &wamp.Register{Request: 123, Procedure: testProcedure}
 	dealer.Register(sess, msg)
 	rsp := <-callee.Recv()
@@ -264,7 +264,7 @@ func TestRemovePeer(t *testing.T) {
 	dealer.RemoveSession(sess)
 
 	// Register as a way to sync with dealer.
-	sess2 := newSession(callee, 0, nil)
+	sess2 := wamp.NewSession(callee, 0, nil, nil)
 	dealer.Register(sess2,
 		&wamp.Register{Request: 789, Procedure: wamp.URI("nexus.test.p2")})
 	rsp = <-callee.Recv()
@@ -302,7 +302,7 @@ func TestCancelCallModeKill(t *testing.T) {
 
 	// Register a procedure.
 	callee := newTestPeer()
-	calleeSess := newSession(callee, 0, calleeRoles)
+	calleeSess := wamp.NewSession(callee, 0, nil, calleeRoles)
 	dealer.Register(calleeSess,
 		&wamp.Register{Request: 123, Procedure: testProcedure})
 	rsp := <-callee.Recv()
@@ -316,7 +316,7 @@ func TestCancelCallModeKill(t *testing.T) {
 	}
 
 	caller := newTestPeer()
-	callerSession := newSession(caller, 0, nil)
+	callerSession := wamp.NewSession(caller, 0, nil, nil)
 
 	// Test calling valid procedure
 	dealer.Call(callerSession,
@@ -383,7 +383,7 @@ func TestCancelCallModeKillNoWait(t *testing.T) {
 
 	// Register a procedure.
 	callee := newTestPeer()
-	calleeSess := newSession(callee, 0, calleeRoles)
+	calleeSess := wamp.NewSession(callee, 0, nil, calleeRoles)
 	dealer.Register(calleeSess,
 		&wamp.Register{Request: 123, Procedure: testProcedure})
 	rsp := <-callee.Recv()
@@ -397,7 +397,7 @@ func TestCancelCallModeKillNoWait(t *testing.T) {
 	}
 
 	caller := newTestPeer()
-	callerSession := newSession(caller, 0, nil)
+	callerSession := wamp.NewSession(caller, 0, nil, nil)
 
 	// Test calling valid procedure
 	dealer.Call(callerSession,
@@ -461,7 +461,7 @@ func TestCancelCallModeSkip(t *testing.T) {
 		},
 	}
 
-	calleeSess := newSession(callee, 0, calleeRoles)
+	calleeSess := wamp.NewSession(callee, 0, nil, calleeRoles)
 	dealer.Register(calleeSess,
 		&wamp.Register{Request: 123, Procedure: testProcedure})
 	rsp := <-callee.Recv()
@@ -475,7 +475,7 @@ func TestCancelCallModeSkip(t *testing.T) {
 	}
 
 	caller := newTestPeer()
-	callerSession := newSession(caller, 0, nil)
+	callerSession := wamp.NewSession(caller, 0, nil, nil)
 
 	// Test calling valid procedure
 	dealer.Call(callerSession,
@@ -525,7 +525,7 @@ func TestSharedRegistrationRoundRobin(t *testing.T) {
 
 	// Register callee1 with roundrobin shared registration
 	callee1 := newTestPeer()
-	calleeSess1 := newSession(callee1, 0, calleeRoles)
+	calleeSess1 := wamp.NewSession(callee1, 0, nil, calleeRoles)
 	dealer.Register(calleeSess1, &wamp.Register{
 		Request:   123,
 		Procedure: testProcedure,
@@ -547,7 +547,7 @@ func TestSharedRegistrationRoundRobin(t *testing.T) {
 
 	// Register callee2 with roundrobin shared registration
 	callee2 := newTestPeer()
-	calleeSess2 := newSession(callee2, 0, calleeRoles)
+	calleeSess2 := wamp.NewSession(callee2, 0, nil, calleeRoles)
 	dealer.Register(calleeSess2, &wamp.Register{
 		Request:   124,
 		Procedure: testProcedure,
@@ -569,7 +569,7 @@ func TestSharedRegistrationRoundRobin(t *testing.T) {
 
 	// Test calling valid procedure
 	caller := newTestPeer()
-	callerSession := newSession(caller, 0, nil)
+	callerSession := wamp.NewSession(caller, 0, nil, nil)
 	dealer.Call(callerSession,
 		&wamp.Call{Request: 125, Procedure: testProcedure})
 
@@ -644,7 +644,7 @@ func TestSharedRegistrationFirst(t *testing.T) {
 
 	// Register callee1 with first shared registration
 	callee1 := newTestPeer()
-	calleeSess1 := newSession(callee1, 1111, calleeRoles)
+	calleeSess1 := wamp.NewSession(callee1, 1111, nil, calleeRoles)
 	dealer.Register(calleeSess1, &wamp.Register{
 		Request:   123,
 		Procedure: testProcedure,
@@ -666,7 +666,7 @@ func TestSharedRegistrationFirst(t *testing.T) {
 
 	// Register callee2 with roundrobin shared registration
 	callee2 := newTestPeer()
-	calleeSess2 := newSession(callee2, 2222, calleeRoles)
+	calleeSess2 := wamp.NewSession(callee2, 2222, nil, calleeRoles)
 	dealer.Register(calleeSess2, &wamp.Register{
 		Request:   1233,
 		Procedure: testProcedure,
@@ -704,7 +704,7 @@ func TestSharedRegistrationFirst(t *testing.T) {
 
 	// Test calling valid procedure
 	caller := newTestPeer()
-	callerSession := newSession(caller, 333, nil)
+	callerSession := wamp.NewSession(caller, 333, nil, nil)
 	dealer.Call(callerSession,
 		&wamp.Call{Request: 125, Procedure: testProcedure})
 
@@ -833,7 +833,7 @@ func TestSharedRegistrationLast(t *testing.T) {
 
 	// Register callee1 with last shared registration
 	callee1 := newTestPeer()
-	calleeSess1 := newSession(callee1, 0, calleeRoles)
+	calleeSess1 := wamp.NewSession(callee1, 0, nil, calleeRoles)
 	dealer.Register(calleeSess1, &wamp.Register{
 		Request:   123,
 		Procedure: testProcedure,
@@ -854,7 +854,7 @@ func TestSharedRegistrationLast(t *testing.T) {
 
 	// Register callee2 with last shared registration
 	callee2 := newTestPeer()
-	calleeSess2 := newSession(callee2, 0, calleeRoles)
+	calleeSess2 := wamp.NewSession(callee2, 0, nil, calleeRoles)
 	dealer.Register(calleeSess2, &wamp.Register{
 		Request:   124,
 		Procedure: testProcedure,
@@ -870,7 +870,7 @@ func TestSharedRegistrationLast(t *testing.T) {
 
 	// Test calling valid procedure
 	caller := newTestPeer()
-	callerSession := newSession(caller, 0, nil)
+	callerSession := wamp.NewSession(caller, 0, nil, nil)
 	dealer.Call(callerSession,
 		&wamp.Call{Request: 125, Procedure: testProcedure})
 
@@ -983,7 +983,7 @@ func TestPatternBasedRegistration(t *testing.T) {
 
 	// Register a procedure with wildcard match.
 	callee := newTestPeer()
-	calleeSess := newSession(callee, 0, calleeRoles)
+	calleeSess := wamp.NewSession(callee, 0, nil, calleeRoles)
 	dealer.Register(calleeSess,
 		&wamp.Register{
 			Request:   123,
@@ -1005,7 +1005,7 @@ func TestPatternBasedRegistration(t *testing.T) {
 	}
 
 	caller := newTestPeer()
-	callerSession := newSession(caller, 0, nil)
+	callerSession := wamp.NewSession(caller, 0, nil, nil)
 
 	// Test calling valid procedure with full name.  Widlcard should match.
 	dealer.Call(callerSession,
@@ -1049,7 +1049,7 @@ func TestRPCBlockedUnresponsiveCallee(t *testing.T) {
 
 	// Register a procedure.
 	callee, rtr := transport.LinkedPeers()
-	calleeSess := newSession(rtr, 0, nil)
+	calleeSess := wamp.NewSession(rtr, 0, nil, nil)
 	opts := wamp.Dict{wamp.OptTimeout: timeoutMs}
 	dealer.Register(calleeSess,
 		&wamp.Register{Request: 223, Procedure: testProcedure, Options: opts})
@@ -1064,7 +1064,7 @@ func TestRPCBlockedUnresponsiveCallee(t *testing.T) {
 	}
 
 	caller, rtr := transport.LinkedPeers()
-	callerSession := newSession(rtr, 0, nil)
+	callerSession := wamp.NewSession(rtr, 0, nil, nil)
 
 	// Call unresponsive callee until call dropped.
 	var i int
@@ -1114,7 +1114,7 @@ func TestCallerIdentification(t *testing.T) {
 
 	// Register a procedure, set option to request disclosing caller.
 	callee := newTestPeer()
-	calleeSess := newSession(callee, 0, calleeRoles)
+	calleeSess := wamp.NewSession(callee, 0, nil, calleeRoles)
 	dealer.Register(calleeSess,
 		&wamp.Register{
 			Request:   123,
@@ -1135,7 +1135,7 @@ func TestCallerIdentification(t *testing.T) {
 
 	caller := newTestPeer()
 	callerID := wamp.ID(11235813)
-	callerSession := newSession(caller, callerID, nil)
+	callerSession := wamp.NewSession(caller, callerID, nil, nil)
 
 	// Test calling valid procedure with full name.  Widlcard should match.
 	dealer.Call(callerSession,
@@ -1160,7 +1160,7 @@ func TestWrongYielder(t *testing.T) {
 
 	// Register a procedure.
 	callee := newTestPeer()
-	calleeSess := newSession(callee, 7777, nil)
+	calleeSess := wamp.NewSession(callee, 7777, nil, nil)
 	dealer.Register(calleeSess,
 		&wamp.Register{Request: 4321, Procedure: testProcedure})
 	rsp := <-callee.Recv()
@@ -1171,11 +1171,11 @@ func TestWrongYielder(t *testing.T) {
 
 	// Create caller
 	caller := newTestPeer()
-	callerSession := newSession(caller, 0, nil)
+	callerSession := wamp.NewSession(caller, 0, nil, nil)
 
 	// Create imposter callee
 	badCallee := newTestPeer()
-	badCalleeSess := newSession(badCallee, 1313, nil)
+	badCalleeSess := wamp.NewSession(badCallee, 1313, nil, nil)
 
 	// Call the procedure
 	dealer.Call(callerSession,
diff --git a/router/publishfilter_test.go b/router/publishfilter_test.go
index 14e6d20..3174553 100644
--- a/router/publishfilter_test.go
+++ b/router/publishfilter_test.go
@@ -37,34 +37,34 @@ func TestFilterBlacklist(t *testing.T) {
 		"authrole": allowedAuthrole,
 		"misc":     "other",
 	}
-	sess := newSession(nil, allowedID, details)
-	if !pf.Allowed(&sess.Session) {
+	sess := &wamp.Session{ID: allowedID, Details: details}
+	if !pf.Allowed(sess) {
 		t.Error(shouldAllowMsg)
 	}
 
-	sess = newSession(nil, blacklistID, details)
+	sess = &wamp.Session{ID: blacklistID, Details: details}
 	// Check that session is denied by ID.
-	if pf.Allowed(&sess.Session) {
+	if pf.Allowed(sess) {
 		t.Error(shouldDenyMsg)
 	}
 
-	sess = newSession(nil, allowedID, details)
+	sess = &wamp.Session{ID: allowedID, Details: details}
 	// Check that session is denied by authid.
 	sess.Details["authid"] = blacklistAuthid
-	if pf.Allowed(&sess.Session) {
+	if pf.Allowed(sess) {
 		t.Error(shouldDenyMsg)
 	}
 
 	// Check that session is denied by authrole.
 	sess.Details["authid"] = allowedAuthid
 	sess.Details["authrole"] = blacklistAuthrole
-	if pf.Allowed(&sess.Session) {
+	if pf.Allowed(sess) {
 		t.Error(shouldDenyMsg)
 	}
 
 	// Check that session is allowed by not having value in blacklist.
 	delete(sess.Details, "authrole")
-	if !pf.Allowed(&sess.Session) {
+	if !pf.Allowed(sess) {
 		t.Error(shouldDenyMsg)
 	}
 }
@@ -100,34 +100,34 @@ func TestFilterWhitelist(t *testing.T) {
 		"authrole": allowedAuthrole,
 		"misc":     "other",
 	}
-	sess := newSession(nil, allowedID, details)
-	if !pf.Allowed(&sess.Session) {
+	sess := &wamp.Session{ID: allowedID, Details: details}
+	if !pf.Allowed(sess) {
 		t.Error(shouldAllowMsg)
 	}
 
-	sess = newSession(nil, deniedID, details)
+	sess = &wamp.Session{ID: deniedID, Details: details}
 	// Check that session is denied by ID.
-	if pf.Allowed(&sess.Session) {
+	if pf.Allowed(sess) {
 		t.Error(shouldDenyMsg)
 	}
 
-	sess = newSession(nil, allowedID, details)
+	sess = &wamp.Session{ID: allowedID, Details: details}
 	// Check that session is denied by authid.
 	sess.Details["authid"] = deniedAuthid
-	if pf.Allowed(&sess.Session) {
+	if pf.Allowed(sess) {
 		t.Error(shouldDenyMsg)
 	}
 
 	// Check that session is denied by authrole.
 	sess.Details["authid"] = allowedAuthid
 	sess.Details["authrole"] = deniedAuthrole
-	if pf.Allowed(&sess.Session) {
+	if pf.Allowed(sess) {
 		t.Error(shouldDenyMsg)
 	}
 
 	// Check that session is denied by not having value in whitelise.
 	delete(sess.Details, "authrole")
-	if pf.Allowed(&sess.Session) {
+	if pf.Allowed(sess) {
 		t.Error(shouldDenyMsg)
 	}
 }
@@ -167,21 +167,21 @@ func TestFilterBlackWhitelistPrecedence(t *testing.T) {
 		"misc":     "other",
 	}
 
-	sess := newSession(nil, allowedID, details)
-	if !pf.Allowed(&sess.Session) {
+	sess := &wamp.Session{ID: allowedID, Details: details}
+	if !pf.Allowed(sess) {
 		t.Error(shouldAllowMsg)
 	}
 
-	sess = newSession(nil, blacklistID, details)
+	sess = &wamp.Session{ID: blacklistID, Details: details}
 	// Check that session is denied by ID even thought ID is also in whitelist.
-	if pf.Allowed(&sess.Session) {
+	if pf.Allowed(sess) {
 		t.Error(shouldDenyMsg)
 	}
 
-	sess = newSession(nil, allowedID, details)
+	sess = &wamp.Session{ID: allowedID, Details: details}
 	// Check that session is denied by authid even though also whitelisted.
 	sess.Details["authid"] = blacklistAuthid
-	if pf.Allowed(&sess.Session) {
+	if pf.Allowed(sess) {
 		t.Error(shouldDenyMsg)
 	}
 }
diff --git a/wamp/dict_test.go b/wamp/dict_test.go
index 94280bd..e8db04d 100644
--- a/wamp/dict_test.go
+++ b/wamp/dict_test.go
@@ -15,7 +15,7 @@ func hasFeature(d Dict, role, feature string) bool {
 	return b
 }
 
-func checkRoles(sess Session) error {
+func checkRoles(sess *Session) error {
 	if !sess.HasRole("caller") {
 		return errors.New("session does not have caller role")
 	}
@@ -61,11 +61,11 @@ func TestHasRoleFeatureLookup(t *testing.T) {
 	clientRoles["caller"]["features"] = boolMap
 	dict["roles"] = clientRoles
 
-	if err := checkRoles(Session{Details: dict}); err != nil {
+	if err := checkRoles(NewSession(nil, 0, nil, dict)); err != nil {
 		t.Fatal(err)
 	}
 
-	sess := Session{ID: ID(193847264)}
+	sess := &Session{ID: ID(193847264)}
 	if sess.String() != "193847264" {
 		t.Fatal("Got unexpected session ID string")
 	}
@@ -89,7 +89,7 @@ func TestHasRoleFeatureLookup(t *testing.T) {
 	}
 
 	// Check again after conversion.
-	if err := checkRoles(Session{Details: dict}); err != nil {
+	if err := checkRoles(NewSession(nil, 0, nil, dict)); err != nil {
 		t.Fatal(err)
 	}
 
@@ -110,7 +110,7 @@ func TestHasRoleFeatureLookup(t *testing.T) {
 		},
 		"authmethods": []string{"anonymous", "ticket"},
 	}
-	if err := checkRoles(Session{Details: dict}); err != nil {
+	if err := checkRoles(NewSession(nil, 0, nil, dict)); err != nil {
 		t.Fatal(err)
 	}
 }
@@ -221,7 +221,7 @@ func BenchmarkNormalized(b *testing.B) {
 
 	b.ResetTimer()
 	dict = NormalizeDict(dict)
-	sess := Session{Details: dict}
+	sess := &Session{Details: dict}
 	for i := 0; i < b.N; i++ {
 		checkRoles(sess)
 	}
@@ -242,7 +242,7 @@ func BenchmarkNotNormalized(b *testing.B) {
 	dict["roles"] = clientRoles
 
 	b.ResetTimer()
-	sess := Session{Details: dict}
+	sess := &Session{Details: dict}
 	for i := 0; i < b.N; i++ {
 		checkRoles(sess)
 	}