Repository: gammazero/nexus
Pull request #167 (2019-03-14 12:54:59)

Fix blocking with slow clients
This fixes issue #159.  When a client leaves the realm or is disconnected by the router, do not attempt to send more messages to the client.

Web and raw socker server's outbound per-client channel size is configurable.

Progressive results use backpressure to limit influx of results when results are being produced faster than they are being consumed.


Patch:
diff --git a/client/client.go b/client/client.go
index 7593d06..4d47804 100644
--- a/client/client.go
+++ b/client/client.go
@@ -156,7 +156,7 @@ type Client struct {
 	closed        int32
 	done          chan struct{}
 	routerGoodbye *wamp.Goodbye
-	idGen *wamp.SyncIDGen
+	idGen         *wamp.SyncIDGen
 }
 
 // NewClient takes a connected Peer, joins the realm specified in cfg, and if
@@ -284,7 +284,7 @@ func (c *Client) Subscribe(topic string, fn EventHandler, options wamp.Dict) err
 		}
 		<-sync
 	case *wamp.Error:
-		return fmt.Errorf("error subscribing to topic '%v': %s", topic,
+		return fmt.Errorf("subscribing to topic '%v': %s", topic,
 			wampErrorString(msg))
 	default:
 		return unexpectedMsgError(msg, wamp.SUBSCRIBED)
@@ -347,7 +347,7 @@ func (c *Client) Unsubscribe(topic string) error {
 		// Already deleted the event handler for the topic.
 		return nil
 	case *wamp.Error:
-		return fmt.Errorf("Error unsubscribing to '%s': %s", topic,
+		return fmt.Errorf("unsubscribing to '%s': %s", topic,
 			wampErrorString(msg))
 	}
 	return unexpectedMsgError(msg, wamp.UNSUBSCRIBED)
@@ -418,8 +418,7 @@ func (c *Client) Publish(topic string, options wamp.Dict, args wamp.List, kwargs
 	switch msg := msg.(type) {
 	case *wamp.Published:
 	case *wamp.Error:
-		return fmt.Errorf("error waiting for published message: %s",
-			wampErrorString(msg))
+		return fmt.Errorf("waiting for published message: %s", wampErrorString(msg))
 	default:
 		return unexpectedMsgError(msg, wamp.PUBLISHED)
 	}
@@ -486,7 +485,7 @@ func (c *Client) Register(procedure string, fn InvocationHandler, options wamp.D
 				msg.Registration)
 		}
 	case *wamp.Error:
-		return fmt.Errorf("Error registering procedure '%v': %s", procedure,
+		return fmt.Errorf("registering procedure '%v': %s", procedure,
 			wampErrorString(msg))
 	default:
 		return unexpectedMsgError(msg, wamp.REGISTERED)
@@ -548,7 +547,7 @@ func (c *Client) Unregister(procedure string) error {
 	case *wamp.Unregistered:
 		// Already deleted the invocation handler for the procedure.
 	case *wamp.Error:
-		return fmt.Errorf("error unregistering procedure '%s': %v", procedure,
+		return fmt.Errorf("unregistering procedure '%s': %v", procedure,
 			wampErrorString(msg))
 	default:
 		return unexpectedMsgError(msg, wamp.UNREGISTERED)
@@ -565,11 +564,18 @@ func (c *Client) Unregister(procedure string) error {
 //
 // Call Canceling
 //
-// The provided Context can be used to cancel a call, or to set a deadline that
-// cancels the call when the deadline expires.  There is no separate Cancel()
-// API to do this.  If the call is canceled before a result is received, then a
-// CANCEL message is sent to the router to cancel the call according to the
-// specified mode.
+// The provided Context allows the caller to cancel a call, or to set a
+// deadline that cancels the call when the deadline expires.  There is no
+// separate Cancel() API to do this.  If the call is canceled before a result
+// is received, then a CANCEL message is sent to the router to cancel the call
+// 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
+// initiated by the client (by canceling context), and cancellation initialed
+// elsewhere.
+//
+// Cancel Mode
 //
 // cancelMode must be one of the following: "kill", "killnowait', "skip".
 // Setting to "" specifies using the default value: "killnowait".  cancelMode
@@ -577,7 +583,7 @@ func (c *Client) Unregister(procedure string) error {
 // is specified as a parameter to the Call() API, and not a message option for
 // CALL.
 //
-// Cancellation behaves differently depending on the mode:
+// Cancele 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
@@ -720,7 +726,7 @@ type RPCError struct {
 // Error implements the error interface, returning an error string for the
 // RPCError.
 func (rpce RPCError) Error() string {
-	return fmt.Sprintf("error calling remote procedure '%s': %s",
+	return fmt.Sprintf("calling remote procedure '%s': %s",
 		rpce.Procedure, wampErrorString(rpce.Err))
 }
 
@@ -802,7 +808,7 @@ func (c *Client) SendProgress(ctx context.Context, args wamp.List, kwArgs wamp.D
 		// Caller is not accepting progressive results or call canceled.
 		return errors.New("caller not accepting progressive results")
 	}
-	return c.sess.Send(&wamp.Yield{
+	return c.sess.SendCtx(ctx, &wamp.Yield{
 		Request:     req,
 		Options:     wamp.Dict{wamp.OptProgress: true},
 		Arguments:   args,
@@ -887,9 +893,11 @@ func (c *Client) leaveRealm() {
 
 	// Wait for run() to exit, but do not wait longer that a normal response
 	// timeout.
+	timer := time.NewTimer(c.responseTimeout)
 	select {
 	case <-c.done:
-	case <-time.After(c.responseTimeout):
+		timer.Stop()
+	case <-timer.C:
 		close(c.actionChan) // force run() to exit
 		<-c.done
 	}
@@ -1024,14 +1032,16 @@ func (c *Client) waitForReply(id wamp.ID) (wamp.Message, error) {
 
 	var msg wamp.Message
 	var err error
+	timer := time.NewTimer(c.responseTimeout)
 	select {
 	case msg, ok = <-wait:
+		timer.Stop()
 		if !ok {
 			// Return directly here, since awaitingReply entry already deleted.
 			return nil, errors.New("client closed")
 		}
-	case <-time.After(c.responseTimeout):
-		err = errors.New("timeout while waiting for reply")
+	case <-timer.C:
+		err = errors.New("timeout waiting for reply")
 	}
 	c.replyLock.Lock()
 	delete(c.awaitingReply, id)
@@ -1065,13 +1075,14 @@ CollectResults:
 	select {
 	case msg, ok = <-wait:
 		if !ok {
-			// Return directly here, since awaitingReply entry already deleted.
+			// Return here, since awaitingReply entry already deleted.
 			return nil, errors.New("client closed")
 		}
 		// If this is a progressive result, put the Result message on the
 		// progress channel and go back to waiting for more results.
 		if progChan != nil {
-			if result, ok := msg.(*wamp.Result); ok {
+			var result *wamp.Result
+			if result, ok = msg.(*wamp.Result); ok {
 				if ok, _ = result.Details[wamp.OptProgress].(bool); ok {
 					progChan <- result
 					goto CollectResults
@@ -1079,19 +1090,33 @@ CollectResults:
 			}
 		}
 	case <-ctx.Done():
+		err = ctx.Err()
 		if c.debug {
-			c.log.Printf("Call to '%s' canceled (mode=%s)", procedure, mode)
+			c.log.Printf("Call to %q canceled by caller (mode=%s): %s",
+				procedure, mode, err)
 		}
 		c.sess.Send(&wamp.Cancel{
 			Request: id,
 			Options: wamp.SetOption(nil, wamp.OptMode, mode),
 		})
-		// If waiting for callee to cancel, then wait for the ERROR from
-		// the dealer.
+		// Wait for the ERROR from the dealer.
+		timer := time.NewTimer(c.responseTimeout)
+	waitCancel:
+		// Discard responses until ERROR or timeout
 		select {
-		case msg = <-wait:
-		case <-time.After(c.responseTimeout):
-			err = errors.New("timeout while waiting for reply after cancel")
+		case msg, ok = <-wait:
+			if !ok {
+				timer.Stop()
+				return nil, err
+			}
+			if _, ok = msg.(*wamp.Error); !ok {
+				// Discard message
+				goto waitCancel
+			}
+			timer.Stop()
+		case <-timer.C:
+			// Did not get expected response to cancel
+			err = errors.New("timeout waiting for reply after cancel")
 		}
 	}
 	// All done with this call, so not waiting for more replies.
diff --git a/examples/server/server.go b/examples/server/server.go
index aaad964..46e02d5 100644
--- a/examples/server/server.go
+++ b/examples/server/server.go
@@ -80,7 +80,7 @@ func main() {
 	wss.KeepAlive = 30 * time.Second
 
 	// Create rawsocket server.
-	rss := router.NewRawSocketServer(nxr, 0, 0)
+	rss := router.NewRawSocketServer(nxr)
 
 	// ---- Start servers ----
 
diff --git a/nexusd/config.go b/nexusd/config.go
index 92e230c..4884008 100644
--- a/nexusd/config.go
+++ b/nexusd/config.go
@@ -28,6 +28,8 @@ type Config struct {
 		// Allow origins that match these glob patterns when an origin header
 		// is present in the websocket upgrade request.
 		AllowOrigins []string `json:"allow_origins"`
+		// Limit on number of pending messages to send to each client.
+		OutQueueSize int `json:"out_queue_size"`
 	}
 
 	// RawSocket configuration parameters.
@@ -43,6 +45,8 @@ type Config struct {
 		// Files containing a certificate and matching private key.
 		CertFile string `json:"cert_file"`
 		KeyFile  string `json:"key_file"`
+		// Limit on number of pending messages to send to each client.
+		OutQueueSize int `json:"out_queue_size"`
 	}
 
 	// File to write log data to.  If not specified, log to stdout.
diff --git a/nexusd/main.go b/nexusd/main.go
index 7a84fb0..d285b30 100644
--- a/nexusd/main.go
+++ b/nexusd/main.go
@@ -85,6 +85,10 @@ func main() {
 			logger.Println("Allowing origins matching:",
 				strings.Join(conf.WebSocket.AllowOrigins, "|"))
 		}
+		if conf.WebSocket.OutQueueSize != 0 {
+			wss.OutQueueSize = conf.WebSocket.OutQueueSize
+			logger.Printf("Websocket outbound queue size: %d", wss.OutQueueSize)
+		}
 		var closer io.Closer
 		var sockDesc string
 		if conf.WebSocket.CertFile != "" && conf.WebSocket.KeyFile != "" {
@@ -106,9 +110,18 @@ func main() {
 	}
 	if conf.RawSocket.TCPAddress != "" || conf.RawSocket.UnixAddress != "" {
 		// Create a new rawsocket server with the router.
-		rss := router.NewRawSocketServer(r, conf.RawSocket.MaxMsgLen,
-			conf.RawSocket.TCPKeepAliveInterval)
+		rss := router.NewRawSocketServer(r)
+		rss.RecvLimit = conf.RawSocket.MaxMsgLen
+		if conf.RawSocket.OutQueueSize != 0 {
+			rss.OutQueueSize = conf.RawSocket.OutQueueSize
+			logger.Printf("raw socket outbound queue size: %d", rss.OutQueueSize)
+		}
 		if conf.RawSocket.TCPAddress != "" {
+			if conf.RawSocket.TCPKeepAliveInterval != 0 {
+				rss.KeepAlive = conf.RawSocket.TCPKeepAliveInterval
+				logger.Printf("tcp keep-alive interval: %s", rss.KeepAlive)
+			}
+
 			var closer io.Closer
 			var sockDesc string
 			if conf.RawSocket.CertFile != "" && conf.RawSocket.KeyFile != "" {
diff --git a/router/broker.go b/router/broker.go
index 0e336e4..fd5a81c 100644
--- a/router/broker.go
+++ b/router/broker.go
@@ -602,8 +602,8 @@ func (b *Broker) pubSubCreateMeta(topic wamp.URI, subSessID wamp.ID, sub *subscr
 }
 
 func (b *Broker) trySend(sess *session, msg wamp.Message) bool {
-	if err := sess.TrySend(msg); err != nil {
-		b.log.Println("!!! broker dropped", msg.MessageType(), "message:", err)
+	if err := sess.Session.TrySend(msg); err != nil {
+		b.log.Printf("!!! Dropped %s to session %s: %s", msg.MessageType(), sess, err)
 		return false
 	}
 	return true
diff --git a/router/dealer.go b/router/dealer.go
index 676c315..ac04dfa 100644
--- a/router/dealer.go
+++ b/router/dealer.go
@@ -23,6 +23,13 @@ const (
 	featureSharedReg        = "shared_registration"
 	featureRegMetaAPI       = "registration_meta_api"
 	featureTestamentMetaAPI = "testament_meta_api"
+
+	// sendResultDeadline is the amount of time until the dealer gives up
+	// trying to send a RESULT to a blocked caller.  This is different that the
+	// 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
 )
 
 // Role information for this broker.
@@ -57,9 +64,10 @@ type registration struct {
 
 // invocation tracks in-progress invocation
 type invocation struct {
-	callID   requestID
-	callee   *session
-	canceled bool
+	callID     requestID
+	callee     *session
+	canceled   bool
+	retryCount int
 }
 
 type requestID struct {
@@ -301,7 +309,6 @@ func (d *Dealer) Cancel(caller *session, msg *wamp.Cancel) {
 		})
 		return
 	}
-
 	d.actionChan <- func() {
 		d.cancel(caller, msg, mode, wamp.ErrCanceled)
 	}
@@ -309,12 +316,45 @@ 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
+// (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) {
 	if callee == nil || msg == nil {
 		panic("dealer.Yield with nil session or message")
 	}
+	var again bool
+	done := make(chan struct{})
 	d.actionChan <- func() {
-		d.yield(callee, msg)
+		again = d.yield(callee, msg, true)
+		done <- struct{}{}
+	}
+	<-done
+
+	// If blocked, retry
+	if again {
+		retry := true
+		delay := yieldRetryDelay
+		start := time.Now()
+		// Retry processing YIELD until caller gone or deadline reached
+		for {
+			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 {
+				retry = false
+			}
+			d.actionChan <- func() {
+				again = d.yield(callee, msg, retry)
+				done <- struct{}{}
+			}
+			<-done
+			if !again {
+				break
+			}
+			delay *= 2
+		}
 	}
 }
 
@@ -433,7 +473,7 @@ func (d *Dealer) register(callee *session, msg *wamp.Register, match, invokePoli
 		if reg.policy != invokePolicy {
 			d.log.Println("REGISTER for already registered procedure",
 				msg.Procedure, "with conflicting invocation policy (has",
-				reg.policy, "and", invokePolicy, "was requested")
+				reg.policy, "and requested", invokePolicy)
 			d.trySend(callee, &wamp.Error{
 				Type:    msg.MessageType(),
 				Request: msg.Request,
@@ -547,7 +587,8 @@ func (d *Dealer) matchProcedure(procedure wamp.URI) (*registration, bool) {
 				}
 			}
 		}
-		// according to the spec, we have to prefer prefix match over wildcard match:
+		// According to the spec, we have to prefer prefix match over wildcard
+		// match:
 		// https://wamp-proto.org/static/rfc/draft-oberstet-hybi-crossbar-wamp.html#rfc.section.14.3.8.1.4.2
 		if ok {
 			return reg, ok
@@ -642,13 +683,13 @@ func (d *Dealer) call(caller *session, msg *wamp.Call) {
 		if opt, _ := msg.Options[wamp.OptDiscloseMe].(bool); opt {
 			// Dealer MAY deny a Caller's request to disclose its identity.
 			if !d.allowDisclose {
+				// Do not continue a call when discloseMe was disallowed.
 				d.trySend(caller, &wamp.Error{
 					Type:    msg.MessageType(),
 					Request: msg.Request,
 					Details: wamp.Dict{},
 					Error:   wamp.ErrOptionDisallowedDiscloseMe,
 				})
-				// don't continue a call when discloseMe was disallowed.
 				return
 			}
 			if callee.HasFeature(roleCallee, featureCallerIdent) {
@@ -700,7 +741,7 @@ func (d *Dealer) call(caller *session, msg *wamp.Call) {
 			Request:   invocationID,
 			Details:   wamp.Dict{},
 			Error:     wamp.ErrNetworkFailure,
-			Arguments: wamp.List{"client blocked - cannot call procedure"},
+			Arguments: wamp.List{"callee blocked - cannot call procedure"},
 		})
 	}
 }
@@ -718,7 +759,7 @@ func (d *Dealer) cancel(caller *session, msg *wamp.Cancel, mode string, reason w
 
 	// Check if the caller of cancel is also the caller of the procedure.
 	if caller != procCaller {
-		// The caller it trying to cancel calls that it does not own.  It it
+		// The caller is trying to cancel calls that it does not own.  It it
 		// either confused or trying to do something bad.
 		d.log.Println("CANCEL received from caller", caller,
 			"for call owned by different session")
@@ -788,7 +829,7 @@ func (d *Dealer) cancel(caller *session, msg *wamp.Cancel, mode string, reason w
 	})
 }
 
-func (d *Dealer) yield(callee *session, msg *wamp.Yield) {
+func (d *Dealer) yield(callee *session, msg *wamp.Yield, canRetry bool) bool {
 	progress, _ := msg.Options[wamp.OptProgress].(bool)
 
 	// Find and delete pending invocation.
@@ -816,13 +857,13 @@ func (d *Dealer) yield(callee *session, msg *wamp.Yield) {
 			d.log.Println("YIELD received with unknown invocation request ID:",
 				msg.Request)
 		}
-		return
+		return false
 	}
 
 	// Make sure this yield was sent by the session that handled the call
 	if invk.callee != callee {
 		d.log.Println("Ignoring YIELD received from session", callee, "that does not own request", msg.Request)
-		return
+		return false
 	}
 
 	callID := invk.callID
@@ -831,15 +872,22 @@ func (d *Dealer) yield(callee *session, msg *wamp.Yield) {
 
 	details := wamp.Dict{}
 
-	if !progress {
-		delete(d.invocations, msg.Request)
-		// Delete callID -> invocation.
-		delete(d.invocationByCall, callID)
-		// Delete pending call since it is finished.
-		delete(d.calls, callID)
-	} else {
+	var keepInvocation bool
+	if progress {
 		// If this is a progressive response, then set progress=true.
 		details[wamp.OptProgress] = true
+	} else {
+		// Clean up the invocation, unless need to retry.
+		defer func() {
+			if keepInvocation {
+				return
+			}
+			delete(d.invocations, msg.Request)
+			// Delete callID -> invocation.
+			delete(d.invocationByCall, callID)
+			// Delete pending call since it is finished.
+			delete(d.calls, callID)
+		}()
 	}
 
 	// Did not find caller.
@@ -847,16 +895,30 @@ func (d *Dealer) yield(callee *session, msg *wamp.Yield) {
 		// Found invocation id that does not have any call id.
 		d.log.Println("!!! No matching caller for invocation from YIELD:",
 			msg.Request)
-		return
+		return false
 	}
 
-	// Send RESULT to the caller.  This forwards the YIELD from the callee.
-	d.trySend(caller, &wamp.Result{
+	// Send RESULT to the caller.  If the caller is blocked, then make the
+	// callee wait and retry sending this message again.  The caller may be
+	// blocked when the callee is generating progressive responses faster than
+	// the caller can handle them.
+	res := &wamp.Result{
 		Request:     callID.request,
 		Details:     details,
 		Arguments:   msg.Arguments,
 		ArgumentsKw: msg.ArgumentsKw,
-	})
+	}
+	err := caller.TrySend(res)
+	if err != nil {
+		if canRetry {
+			keepInvocation = true
+			return true
+		}
+		d.log.Printf("!!! Dropped %s to caller %s: %s", res.MessageType(), caller, err)
+		d.cancel(caller, &wamp.Cancel{Request: callID.request},
+			wamp.CancelModeKillNoWait, wamp.ErrCanceled)
+	}
+	return false
 }
 
 func (d *Dealer) error(msg *wamp.Error) {
@@ -1192,7 +1254,7 @@ func (d *Dealer) RegCountCallees(msg *wamp.Invocation) wamp.Message {
 
 func (d *Dealer) trySend(sess *session, msg wamp.Message) bool {
 	if err := sess.TrySend(msg); err != nil {
-		d.log.Println("!!! dealer dropped", msg.MessageType(), "message:", err)
+		d.log.Printf("!!! Dropped %s to session %s: %s", msg.MessageType(), sess, err)
 		return false
 	}
 	return true
diff --git a/router/rawsocketserver.go b/router/rawsocketserver.go
index d39e87e..9c2dfa6 100644
--- a/router/rawsocketserver.go
+++ b/router/rawsocketserver.go
@@ -7,26 +7,29 @@ import (
 	"net"
 	"time"
 
-	"github.com/gammazero/nexus/stdlog"
 	"github.com/gammazero/nexus/transport"
 )
 
 // RawSocketServer handles socket connections.
 type RawSocketServer struct {
-	router Router
+	// RecvLimit is the maximum length of messages the server is willing to
+	// receive.  Defaults to maximum allowed for protocol: 16M.
+	RecvLimit int
+
+	// KeepAlive is the TCP keep-alive period.  Default is disable keep-alive.
+	KeepAlive time.Duration
+
+	// OutQueueSize is the maximum number of pending outbound messages, per
+	// client.  The default is defaultOutQueueSize.
+	OutQueueSize int
 
-	log       stdlog.StdLog
-	recvLimit int
-	keepAlive time.Duration
+	router Router
 }
 
 // NewRawSocketServer takes a router instance and creates a new socket server.
-func NewRawSocketServer(r Router, recvLimit int, keepAlive time.Duration) *RawSocketServer {
+func NewRawSocketServer(r Router) *RawSocketServer {
 	return &RawSocketServer{
-		router:    r,
-		log:       r.Logger(),
-		recvLimit: recvLimit,
-		keepAlive: keepAlive,
+		router: r,
 	}
 }
 
@@ -35,7 +38,7 @@ func NewRawSocketServer(r Router, recvLimit int, keepAlive time.Duration) *RawSo
 func (s *RawSocketServer) ListenAndServe(network, address string) (io.Closer, error) {
 	l, err := net.Listen(network, address)
 	if err != nil {
-		s.log.Print(err)
+		s.router.Logger().Print(err)
 		return nil, err
 	}
 
@@ -68,7 +71,7 @@ func (s *RawSocketServer) ListenAndServeTLS(network, address string, tlscfg *tls
 
 	l, err := tls.Listen(network, address, tlscfg)
 	if err != nil {
-		s.log.Print(err)
+		s.router.Logger().Print(err)
 		return nil, err
 	}
 
@@ -87,9 +90,9 @@ func (s *RawSocketServer) requestHandler(l net.Listener) {
 			return
 		}
 		if tcpConn, ok := conn.(*net.TCPConn); ok {
-			if s.keepAlive != 0 {
+			if s.KeepAlive != 0 {
 				tcpConn.SetKeepAlive(true)
-				tcpConn.SetKeepAlivePeriod(s.keepAlive)
+				tcpConn.SetKeepAlivePeriod(s.KeepAlive)
 			} else {
 				tcpConn.SetKeepAlive(false)
 			}
@@ -102,13 +105,17 @@ func (s *RawSocketServer) requestHandler(l net.Listener) {
 // client handshake, creates a rawSocketPeer, and then attaches that peer to
 // the router.
 func (s *RawSocketServer) handleRawSocket(conn net.Conn) {
-	peer, err := transport.AcceptRawSocket(conn, s.log, s.recvLimit)
+	qsize := s.OutQueueSize
+	if qsize == 0 {
+		qsize = defaultOutQueueSize
+	}
+	peer, err := transport.AcceptRawSocket(conn, s.router.Logger(), s.RecvLimit, qsize)
 	if err != nil {
-		s.log.Println("Error accepting rawsocket client:", err)
+		s.router.Logger().Println("Error accepting rawsocket client:", err)
 		return
 	}
 
 	if err := s.router.Attach(peer); err != nil {
-		s.log.Println("Error attaching to router:", err)
+		s.router.Logger().Println("Error attaching to router:", err)
 	}
 }
diff --git a/router/websocketserver.go b/router/websocketserver.go
index cac5c06..d07d3a1 100644
--- a/router/websocketserver.go
+++ b/router/websocketserver.go
@@ -14,7 +14,6 @@ import (
 	"strings"
 	"time"
 
-	"github.com/gammazero/nexus/stdlog"
 	"github.com/gammazero/nexus/transport"
 	"github.com/gammazero/nexus/transport/serialize"
 	"github.com/gammazero/nexus/wamp"
@@ -25,6 +24,8 @@ const (
 	jsonWebsocketProtocol    = "wamp.2.json"
 	msgpackWebsocketProtocol = "wamp.2.msgpack"
 	cborWebsocketProtocol    = "wamp.2.cbor"
+
+	defaultOutQueueSize = 16
 )
 
 type protocol struct {
@@ -53,11 +54,6 @@ type WebsocketServer struct {
 	// Serializer for binary frames.  Defaults to MessagePackSerializer.
 	BinarySerializer serialize.Serializer
 
-	router Router
-
-	protocols map[string]protocol
-	log       stdlog.StdLog
-
 	// EnableTrackingCookie tells the server to send a random-value cookie to
 	// the websocket client.  A returning client may identify itself by sending
 	// a previously issued tracking cookie in a websocket request.  If a
@@ -95,6 +91,13 @@ type WebsocketServer struct {
 	// If a "pong" response is not received after 2 intervals have elapsed then
 	// the websocket connection is closed.
 	KeepAlive time.Duration
+
+	// OutQueueSize is the maximum number of pending outbound messages, per
+	// client.  The default is defaultOutQueueSize.
+	OutQueueSize int
+
+	router    Router
+	protocols map[string]protocol
 }
 
 // NewWebsocketServer takes a router instance and creates a new websocket
@@ -124,7 +127,6 @@ func NewWebsocketServer(r Router) *WebsocketServer {
 	s := &WebsocketServer{
 		router:    r,
 		protocols: map[string]protocol{},
-		log:       r.Logger(),
 	}
 	s.Upgrader = &websocket.Upgrader{}
 	s.addProtocol(jsonWebsocketProtocol, websocket.TextMessage,
@@ -233,7 +235,7 @@ func (s *WebsocketServer) SetConfig(wsCfg transport.WebsocketConfig) {
 func (s *WebsocketServer) ListenAndServe(address string) (io.Closer, error) {
 	l, err := net.Listen("tcp", address)
 	if err != nil {
-		s.log.Print(err)
+		s.router.Logger().Print(err)
 		return nil, err
 	}
 
@@ -271,7 +273,7 @@ func (s *WebsocketServer) ListenAndServeTLS(address string, tlscfg *tls.Config,
 
 	l, err := tls.Listen("tcp", address, tlscfg)
 	if err != nil {
-		s.log.Print(err)
+		s.router.Logger().Print(err)
 		return nil, err
 	}
 
@@ -331,7 +333,7 @@ func (s *WebsocketServer) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 
 	conn, err := s.Upgrader.Upgrade(w, r, w.Header())
 	if err != nil {
-		s.log.Println("Error upgrading to websocket connection:", err)
+		s.router.Logger().Println("Error upgrading to websocket connection:", err)
 		http.Error(w, err.Error(), http.StatusBadRequest)
 		return
 	}
@@ -380,9 +382,13 @@ func (s *WebsocketServer) handleWebsocket(conn *websocket.Conn, transportDetails
 
 	// Create a websocket peer from the websocket connection and attach the
 	// peer to the router.
-	peer := transport.NewWebsocketPeer(conn, serializer, payloadType, s.log, s.KeepAlive)
+	qsize := s.OutQueueSize
+	if qsize == 0 {
+		qsize = defaultOutQueueSize
+	}
+	peer := transport.NewWebsocketPeer(conn, serializer, payloadType, s.router.Logger(), s.KeepAlive, qsize)
 	if err := s.router.AttachClient(peer, transportDetails); err != nil {
-		s.log.Println("Error attaching to router:", err)
+		s.router.Logger().Println("Error attaching to router:", err)
 	}
 }
 
diff --git a/transport/localpeer.go b/transport/localpeer.go
index ad85d90..b140fcf 100644
--- a/transport/localpeer.go
+++ b/transport/localpeer.go
@@ -1,18 +1,12 @@
 package transport
 
 import (
-	"errors"
-	"time"
+	"context"
 
 	"github.com/gammazero/nexus/wamp"
 )
 
-const (
-	linkedPeersOutQueueSize = 16
-	// Time to wait to write message to send channel when send channel is
-	// blocked due to a burst of messages.
-	sendTimeout = time.Second
-)
+const linkedPeersOutQueueSize = 16
 
 // LinkedPeers creates two connected peers.  Messages sent to one peer appear
 // in the Recv of the other.  This is used for connecting client sessions to
@@ -55,18 +49,11 @@ func (p *localPeer) Recv() <-chan wamp.Message { return p.rd }
 
 // TrySend writes a message to the peer's outbound message channel.
 func (p *localPeer) TrySend(msg wamp.Message) error {
-	select {
-	case p.wr <- msg:
-		return nil
-	default:
-	}
+	return wamp.TrySend(p.wr, msg)
+}
 
-	select {
-	case p.wr <- msg:
-	case <-time.After(sendTimeout):
-		return errors.New("blocked")
-	}
-	return nil
+func (p *localPeer) SendCtx(ctx context.Context, msg wamp.Message) error {
+	return wamp.SendCtx(ctx, p.wr, msg)
 }
 
 // Send writes a message to the peer's outbound message channel.
diff --git a/transport/rawsocketpeer.go b/transport/rawsocketpeer.go
index 551af7b..c253bfb 100644
--- a/transport/rawsocketpeer.go
+++ b/transport/rawsocketpeer.go
@@ -1,6 +1,7 @@
 package transport
 
 import (
+	"context"
 	"crypto/tls"
 	"errors"
 	"fmt"
@@ -29,6 +30,9 @@ type rawSocketPeer struct {
 	rd chan wamp.Message
 	wr chan wamp.Message
 
+	cancelSender context.CancelFunc
+	ctxSender    context.Context
+
 	writerDone chan struct{}
 
 	log stdlog.StdLog
@@ -114,8 +118,8 @@ func ConnectTlsRawSocketPeer(network, address string, serialization serialize.Se
 // 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.
-func AcceptRawSocket(conn net.Conn, logger stdlog.StdLog, recvLimit int) (wamp.Peer, error) {
-	peer, err := serverHandshake(conn, logger, recvLimit)
+func AcceptRawSocket(conn net.Conn, logger stdlog.StdLog, recvLimit, outQueueSize int) (wamp.Peer, error) {
+	peer, err := serverHandshake(conn, logger, recvLimit, outQueueSize)
 	if err != nil {
 		conn.Close()
 		return nil, err
@@ -126,7 +130,7 @@ func AcceptRawSocket(conn net.Conn, logger stdlog.StdLog, recvLimit int) (wamp.P
 // newRawSocketPeer creates a rawsocket peer from an existing socket
 // connection.  This is used by clients connecting to the WAMP router, and by
 // servers to handle connections from clients.
-func newRawSocketPeer(conn net.Conn, serializer serialize.Serializer, logger stdlog.StdLog, sendLimit, recvLimit int) *rawSocketPeer {
+func newRawSocketPeer(conn net.Conn, serializer serialize.Serializer, logger stdlog.StdLog, sendLimit, recvLimit, outQueueSize int) *rawSocketPeer {
 	rs := &rawSocketPeer{
 		conn:       conn,
 		serializer: serializer,
@@ -149,6 +153,8 @@ func newRawSocketPeer(conn net.Conn, serializer serialize.Serializer, logger std
 
 		log: logger,
 	}
+	rs.ctxSender, rs.cancelSender = context.WithCancel(context.Background())
+
 	// Sending to and receiving from socket is handled concurrently.
 	go rs.recvHandler()
 	go rs.sendHandler()
@@ -159,23 +165,15 @@ func newRawSocketPeer(conn net.Conn, serializer serialize.Serializer, logger std
 func (rs *rawSocketPeer) Recv() <-chan wamp.Message { return rs.rd }
 
 func (rs *rawSocketPeer) TrySend(msg wamp.Message) error {
-	select {
-	case rs.wr <- msg:
-		return nil
-	default:
-	}
+	return wamp.TrySend(rs.wr, msg)
+}
 
-	select {
-	case rs.wr <- msg:
-	case <-time.After(time.Second):
-		return errors.New("blocked")
-	}
-	return nil
+func (rs *rawSocketPeer) SendCtx(ctx context.Context, msg wamp.Message) error {
+	return wamp.SendCtx(ctx, rs.wr, msg)
 }
 
 func (rs *rawSocketPeer) Send(msg wamp.Message) error {
-	rs.wr <- msg
-	return nil
+	return wamp.SendCtx(rs.ctxSender, rs.wr, msg)
 }
 
 // Close closes the rawsocket peer.  This closes the local send channel, and
@@ -184,10 +182,9 @@ func (rs *rawSocketPeer) Send(msg wamp.Message) error {
 //
 // *** Do not call Send after calling Close. ***
 func (rs *rawSocketPeer) Close() {
-	// Tell sendHandler to exit, allowing it to finish sending any queued
-	// messages.  Do not close wr channel in case there are incoming messages
-	// during close.
-	rs.wr <- nil
+	// Tell sendHandler to exit, and discard any queued messages.  Do not close
+	// wr channel in case there are incoming messages during close.
+	rs.cancelSender()
 	<-rs.writerDone
 
 	// Tell recvHandler to close.
@@ -202,33 +199,39 @@ func (rs *rawSocketPeer) Close() {
 // socket.
 func (rs *rawSocketPeer) sendHandler() {
 	defer close(rs.writerDone)
-	for msg := range rs.wr {
-		if msg == nil {
-			return
-		}
-		b, err := rs.serializer.Serialize(msg)
-		if err != nil {
-			rs.log.Print(err)
-			continue
-		}
-		if len(b) > rs.sendLimit {
-			rs.log.Println("Message size", len(b), "exceeds limit of",
-				rs.sendLimit)
-			continue
-		}
-		lenBytes := intToBytes(len(b))
-		header := []byte{0x0, lenBytes[0], lenBytes[1], lenBytes[2]}
-		if _, err = rs.conn.Write(header); err != nil {
-			if !wamp.IsGoodbyeAck(msg) {
-				rs.log.Println("Error writing header:", err)
+	defer rs.cancelSender()
+
+	senderDone := rs.ctxSender.Done()
+sendLoop:
+	for {
+		select {
+		case msg := <-rs.wr:
+			b, err := rs.serializer.Serialize(msg)
+			if err != nil {
+				rs.log.Print(err)
+				continue sendLoop
 			}
-			continue
-		}
-		if _, err = rs.conn.Write(b); err != nil {
-			if !wamp.IsGoodbyeAck(msg) {
-				rs.log.Println("Error writing message:", msg, err)
+			if len(b) > rs.sendLimit {
+				rs.log.Println("Message size", len(b), "exceeds limit of",
+					rs.sendLimit)
+				continue sendLoop
+			}
+			lenBytes := intToBytes(len(b))
+			header := []byte{0x0, lenBytes[0], lenBytes[1], lenBytes[2]}
+			if _, err = rs.conn.Write(header); err != nil {
+				if !wamp.IsGoodbyeAck(msg) {
+					rs.log.Println("Error writing header:", err)
+				}
+				continue sendLoop
 			}
-			continue
+			if _, err = rs.conn.Write(b); err != nil {
+				if !wamp.IsGoodbyeAck(msg) {
+					rs.log.Println("Error writing message:", msg, err)
+				}
+				continue sendLoop
+			}
+		case <-senderDone:
+			return
 		}
 	}
 }
@@ -251,9 +254,8 @@ MsgLoop:
 			default:
 				// Peer received control message to close.  Cause sendHandler
 				// to exit without closing the write channel (in case writes
-				// still happening) and allow it to finish sending any queued
-				// messages.
-				rs.wr <- nil
+				// still happening) and discard any queued messages.
+				rs.cancelSender()
 				<-rs.writerDone
 
 				// Close socket connection.
@@ -383,11 +385,11 @@ func clientHandshake(conn net.Conn, logger stdlog.StdLog, protocol byte, recvLim
 
 	sendLimit := byteToLength(buf[1] >> 4)
 	recvLimit = byteToLength(maxRecvLen)
-	return newRawSocketPeer(conn, serializer, logger, sendLimit, recvLimit), nil
+	return newRawSocketPeer(conn, serializer, logger, sendLimit, recvLimit, 0), nil
 }
 
 // serverHandshake handles the server-side of a RawSocket transport handshake.
-func serverHandshake(conn net.Conn, logger stdlog.StdLog, recvLimit int) (*rawSocketPeer, error) {
+func serverHandshake(conn net.Conn, logger stdlog.StdLog, recvLimit, outQueueSize int) (*rawSocketPeer, error) {
 	var buf [4]byte
 	if _, err := io.ReadFull(conn, buf[:]); err != nil {
 		return nil, err
@@ -426,10 +428,12 @@ func serverHandshake(conn net.Conn, logger stdlog.StdLog, recvLimit int) (*rawSo
 
 	sendLimit := byteToLength(buf[1] >> 4)
 	recvLimit = byteToLength(maxRecvLen)
-	return newRawSocketPeer(conn, serializer, logger, sendLimit, recvLimit), nil
+	return newRawSocketPeer(conn, serializer, logger, sendLimit, recvLimit, outQueueSize), nil
 }
 
-// fitRecvLimit finds the power of 2 that is greater than or equal to the specified receive limit.  This value is returned as the RawSocket transport byte representation of this value.
+// fitRecvLimit finds the power of 2 that is greater than or equal to the
+// specified receive limit.  This value is returned as the RawSocket transport
+// byte representation of this value.
 func fitRecvLimit(recvLimit int) byte {
 	if recvLimit > 0 {
 		for b := byte(0); b < 0xf; b++ {
diff --git a/transport/websocketpeer.go b/transport/websocketpeer.go
index a47a812..1b57f5c 100644
--- a/transport/websocketpeer.go
+++ b/transport/websocketpeer.go
@@ -1,8 +1,8 @@
 package transport
 
 import (
+	"context"
 	"crypto/tls"
-	"errors"
 	"fmt"
 	"net"
 	"net/http"
@@ -51,6 +51,9 @@ type websocketPeer struct {
 	rd chan wamp.Message
 	wr chan wamp.Message
 
+	cancelSender context.CancelFunc
+	ctxSender    context.Context
+
 	writerDone chan struct{}
 
 	log stdlog.StdLog
@@ -63,8 +66,7 @@ const (
 	msgpackWebsocketProtocol = "wamp.2.msgpack"
 	cborWebsocketProtocol    = "wamp.2.cbor"
 
-	outQueueSize = 16
-	ctrlTimeout  = 5 * time.Second
+	ctrlTimeout = 5 * time.Second
 )
 
 type DialFunc func(network, addr string) (net.Conn, error)
@@ -119,7 +121,7 @@ func ConnectWebsocketPeer(routerURL string, serialization serialize.Serializatio
 	if err != nil {
 		return nil, err
 	}
-	return NewWebsocketPeer(conn, serializer, payloadType, logger, 0), nil
+	return NewWebsocketPeer(conn, serializer, payloadType, logger, 0, 0), nil
 }
 
 // NewWebsocketPeer creates a websocket peer from an existing websocket
@@ -129,7 +131,7 @@ func ConnectWebsocketPeer(routerURL string, serialization serialize.Serializatio
 // A non-zero keepAlive value configures a websocket "ping/pong" heartbeat,
 // sendings websocket "pings" every keepAlive interval.  If a "pong" response
 // is not received after 2 intervals have elapsed then the websocket is closed.
-func NewWebsocketPeer(conn *websocket.Conn, serializer serialize.Serializer, payloadType int, logger stdlog.StdLog, keepAlive time.Duration) wamp.Peer {
+func NewWebsocketPeer(conn *websocket.Conn, serializer serialize.Serializer, payloadType int, logger stdlog.StdLog, keepAlive time.Duration, outQueueSize int) wamp.Peer {
 	w := &websocketPeer{
 		conn:        conn,
 		serializer:  serializer,
@@ -150,6 +152,8 @@ func NewWebsocketPeer(conn *websocket.Conn, serializer serialize.Serializer, pay
 
 		log: logger,
 	}
+	w.ctxSender, w.cancelSender = context.WithCancel(context.Background())
+
 	// Sending to and receiving from websocket is handled concurrently.
 	go w.recvHandler()
 	if keepAlive != 0 {
@@ -167,23 +171,15 @@ func NewWebsocketPeer(conn *websocket.Conn, serializer serialize.Serializer, pay
 func (w *websocketPeer) Recv() <-chan wamp.Message { return w.rd }
 
 func (w *websocketPeer) TrySend(msg wamp.Message) error {
-	select {
-	case w.wr <- msg:
-		return nil
-	default:
-	}
+	return wamp.TrySend(w.wr, msg)
+}
 
-	select {
-	case w.wr <- msg:
-	case <-time.After(sendTimeout):
-		return errors.New("blocked")
-	}
-	return nil
+func (w *websocketPeer) SendCtx(ctx context.Context, msg wamp.Message) error {
+	return wamp.SendCtx(ctx, w.wr, msg)
 }
 
 func (w *websocketPeer) Send(msg wamp.Message) error {
-	w.wr <- msg
-	return nil
+	return wamp.SendCtx(w.ctxSender, w.wr, msg)
 }
 
 // Close closes the websocket peer.  This closes the local send channel, and
@@ -192,10 +188,9 @@ func (w *websocketPeer) Send(msg wamp.Message) error {
 //
 // *** Do not call Send after calling Close. ***
 func (w *websocketPeer) Close() {
-	// Tell sendHandler to exit, allowing it to finish sending any queued
-	// messages.  Do not close wr channel in case there are incoming messages
-	// during close.
-	w.wr <- nil
+	// Tell sendHandler to exit and discard any queued messages.  Do not close
+	// wr channel in case there are incoming messages during close.
+	w.cancelSender()
 	<-w.writerDone
 
 	closeMsg := websocket.FormatCloseMessage(websocket.CloseNormalClosure,
@@ -215,20 +210,25 @@ func (w *websocketPeer) Close() {
 // websocket.
 func (w *websocketPeer) sendHandler() {
 	defer close(w.writerDone)
-	for msg := range w.wr {
-		if msg == nil {
-			return
-		}
-		b, err := w.serializer.Serialize(msg.(wamp.Message))
-		if err != nil {
-			w.log.Print(err)
-			continue
-		}
+	defer w.cancelSender()
 
-		if err = w.conn.WriteMessage(w.payloadType, b); err != nil {
-			if !wamp.IsGoodbyeAck(msg) {
+sendLoop:
+	for {
+		select {
+		case msg := <-w.wr:
+			b, err := w.serializer.Serialize(msg.(wamp.Message))
+			if err != nil {
 				w.log.Print(err)
+				continue sendLoop
 			}
+
+			if err = w.conn.WriteMessage(w.payloadType, b); err != nil {
+				if !wamp.IsGoodbyeAck(msg) {
+					w.log.Print(err)
+				}
+				return
+			}
+		case <-w.ctxSender.Done():
 			return
 		}
 	}
@@ -236,6 +236,7 @@ func (w *websocketPeer) sendHandler() {
 
 func (w *websocketPeer) sendHandlerKeepAlive(keepAlive time.Duration) {
 	defer close(w.writerDone)
+	defer w.cancelSender()
 
 	var pendingPongs int32
 	w.conn.SetPongHandler(func(msg string) error {
@@ -248,13 +249,11 @@ func (w *websocketPeer) sendHandlerKeepAlive(keepAlive time.Duration) {
 	defer ticker.Stop()
 	pingMsg := []byte("keepalive")
 
+	senderDone := w.ctxSender.Done()
 recvLoop:
 	for {
 		select {
-		case msg, open := <-w.wr:
-			if msg == nil || !open {
-				return
-			}
+		case msg := <-w.wr:
 			b, err := w.serializer.Serialize(msg.(wamp.Message))
 			if err != nil {
 				w.log.Print(err)
@@ -280,6 +279,8 @@ recvLoop:
 				return
 			}
 			atomic.AddInt32(&pendingPongs, 1)
+		case <-senderDone:
+			return
 		}
 	}
 }
@@ -301,9 +302,9 @@ func (w *websocketPeer) recvHandler() {
 			default:
 				// Peer received control message to close.  Cause sendHandler
 				// to exit without closing the write channel (in case writes
-				// still happening) and allow it to finish sending any queued
-				// messages.
-				w.wr <- nil
+				// still happening) and discard any queued messages.
+				w.cancelSender()
+				// Wait for writer to exit before closing websocket.
 				<-w.writerDone
 			}
 			// The error is only one of these errors.  It is generally not
diff --git a/wamp/peer.go b/wamp/peer.go
index 21d16aa..eb85227 100644
--- a/wamp/peer.go
+++ b/wamp/peer.go
@@ -1,6 +1,7 @@
 package wamp
 
 import (
+	"context"
 	"errors"
 	"time"
 )
@@ -10,6 +11,8 @@ type Peer interface {
 	// Sends the message to the peer.
 	Send(Message) error
 
+	SendCtx(context.Context, Message) error
+
 	// TrySend performs a non-blocking send.  Returns error if blocked.
 	TrySend(Message) error
 
@@ -32,3 +35,23 @@ func RecvTimeout(p Peer, t time.Duration) (Message, error) {
 		return nil, errors.New("timeout waiting for message")
 	}
 }
+
+func SendCtx(ctx context.Context, wr chan<- Message, msg Message) error {
+	select {
+	case <-ctx.Done():
+		return ctx.Err()
+	case wr <- msg:
+	}
+	return nil
+}
+
+// TrySend sends a message to the write-only channel and returns an error if
+// the channel blocks.
+func TrySend(wr chan<- Message, msg Message) error {
+	select {
+	case wr <- msg:
+	default:
+		return errors.New("blocked")
+	}
+	return nil
+}


Test patch:
diff --git a/aat/main_test.go b/aat/main_test.go
index 0277fd8..d55e049 100644
--- a/aat/main_test.go
+++ b/aat/main_test.go
@@ -31,6 +31,8 @@ const (
 
 	certFile = "cert.pem"
 	keyFile  = "rsakey.pem"
+
+	keepAliveInterval = time.Second * 5
 )
 
 var (
@@ -51,6 +53,9 @@ var (
 
 	// compress enables compression on both client and server config
 	compress bool
+
+	// size of server's per-client outbound message queue
+	outQueueSize int
 )
 
 type testAuthz struct{}
@@ -79,6 +84,7 @@ func TestMain(m *testing.M) {
 	flag.StringVar(&serType, "serialize", "",
 		"-serialize[json, msgpack, cbor] default is json")
 	flag.BoolVar(&compress, "compress", false, "enable compression")
+	flag.IntVar(&outQueueSize, "qsize", 0, "server's per-client outbound queue size")
 	flag.Parse()
 
 	if serType != "" && serType != "json" && serType != "msgpack" && serType != "cbor" {
@@ -161,7 +167,8 @@ func TestMain(m *testing.M) {
 		}
 		s.EnableTrackingCookie = true
 		s.EnableRequestCapture = true
-		s.KeepAlive = time.Second
+		s.KeepAlive = keepAliveInterval
+		s.OutQueueSize = outQueueSize
 		closer, err = s.ListenAndServe(tcpAddr)
 	case "https", "wss":
 		s := router.NewWebsocketServer(nxr)
@@ -172,19 +179,25 @@ func TestMain(m *testing.M) {
 		}
 		s.EnableTrackingCookie = true
 		s.EnableRequestCapture = true
-		s.KeepAlive = time.Second
+		s.KeepAlive = keepAliveInterval
+		s.OutQueueSize = outQueueSize
 		closer, err = s.ListenAndServeTLS(tcpAddr, nil, certPath, keyPath)
 	case "tcp":
-		s := router.NewRawSocketServer(nxr, 0, 0)
+		s := router.NewRawSocketServer(nxr)
+		s.KeepAlive = keepAliveInterval
 		closer, err = s.ListenAndServe(scheme, tcpAddr)
 		sockDesc = "TCP RAWSOCKETS"
 	case "tcps":
-		s := router.NewRawSocketServer(nxr, 0, 0)
+		s := router.NewRawSocketServer(nxr)
+		s.KeepAlive = keepAliveInterval
+		s.OutQueueSize = outQueueSize
 		closer, err = s.ListenAndServeTLS("tcp", tcpAddr, nil, certPath, keyPath)
 		sockDesc = "TCP RAWSOCKETS + TLS"
 	case "unix":
 		os.Remove(unixAddr)
-		s := router.NewRawSocketServer(nxr, 0, 0)
+		s := router.NewRawSocketServer(nxr)
+		s.KeepAlive = keepAliveInterval
+		s.OutQueueSize = outQueueSize
 		closer, err = s.ListenAndServe(scheme, unixAddr)
 		addr = unixAddr
 		sockDesc = "UNIX RAWSOCKETS"
diff --git a/aat/rpc_progress_test.go b/aat/rpc_progress_test.go
index cf59200..96bcae1 100644
--- a/aat/rpc_progress_test.go
+++ b/aat/rpc_progress_test.go
@@ -243,7 +243,6 @@ func TestRPCProgressiveCallInterrupt(t *testing.T) {
 }
 
 func TestProgressStress(t *testing.T) {
-	t.Skip("needs investigation with ci")
 	defer leaktest.Check(t)()
 
 	// Connect callee session.
@@ -319,11 +318,20 @@ func TestProgressStress(t *testing.T) {
 	progHandler := func(result *wamp.Result) {
 		// Received another chunk of data.
 		chunk := result.Arguments[0].(string)
-		// Uncomment to make the caller slow to receive responses.  This will
-		// cause the dealer to be blocked trying to send to this client.
-		//time.Sleep(20 * time.Millisecond)
 		recvLen += len(chunk)
 		recvCount++
+
+		// Simulate processing time.  This causes the callee to have to pause
+		// and resend the RESULT, which happens when a callee generates results
+		// faster than the caller can process them.  The pause is long enough
+		// to ensure there are multiple retries.
+		//
+		// Pause on the 13th RESULT or every third call.
+		if recvCount == 13 && result.Request%3 == 0 {
+			fmt.Println("Caller pausing to process", recvCount, "of result",
+				result.Request)
+			time.Sleep(time.Millisecond * 250)
+		}
 	}
 
 	// All results, for all calls, must be recieved by timeout.
@@ -336,7 +344,7 @@ func TestProgressStress(t *testing.T) {
 		result, err := caller.CallProgress(
 			ctx, chunkProc, nil, wamp.List{i}, nil, "", progHandler)
 		if err != nil {
-			t.Error("Failed to call procedure:", err)
+			t.Fatal(err)
 		}
 
 		// As a final result, the callee returns the total length the data.
diff --git a/aat/rpc_test.go b/aat/rpc_test.go
index f2b6cc8..75d7915 100644
--- a/aat/rpc_test.go
+++ b/aat/rpc_test.go
@@ -168,6 +168,7 @@ func TestRPCCancelCall(t *testing.T) {
 	if err != nil {
 		t.Fatal("Failed to connect client:", err)
 	}
+	defer callee.Close()
 
 	// Check for feature support in router.
 	const featureCallCanceling = "call_canceling"
@@ -215,11 +216,10 @@ func TestRPCCancelCall(t *testing.T) {
 	select {
 	case err = <-errChan:
 		if err == nil {
-			t.Fatal("Expected error from canceling call")
+			t.Fatal("expected error from canceling call")
 		}
-		if err.(client.RPCError).Err.Error != wamp.ErrCanceled {
-			t.Fatal("Wrong error for canceled call, got",
-				err.(client.RPCError).Err.Error, "want", wamp.ErrCanceled)
+		if err != context.Canceled {
+			t.Fatal("expected context.Canceled error")
 		}
 	case <-time.After(2 * time.Second):
 		t.Fatal("call should have been canceled")
@@ -232,13 +232,6 @@ func TestRPCCancelCall(t *testing.T) {
 		t.Fatal("invocation should have been canceled")
 	}
 
-	rpcError, ok := err.(client.RPCError)
-	if !ok {
-		t.Fatal("expected RPCError type of error")
-	}
-	if rpcError.Err.Error != wamp.ErrCanceled {
-		t.Fatal("expected canceled error, got:", err)
-	}
 	if err = callee.Unregister(procName); err != nil {
 		t.Fatal("failed to unregister procedure:", err)
 	}
diff --git a/client/client_test.go b/client/client_test.go
index cfbf4b9..ca00929 100644
--- a/client/client_test.go
+++ b/client/client_test.go
@@ -483,12 +483,8 @@ func TestTimeoutCancelRemoteProcedureCall(t *testing.T) {
 		t.Fatal("call should have been canceled")
 	}
 
-	rpcError, ok := err.(RPCError)
-	if !ok {
-		t.Fatal("expected RPCError type of error")
-	}
-	if rpcError.Err.Error != wamp.ErrCanceled {
-		t.Fatal("expected canceled error, got:", err)
+	if err != context.DeadlineExceeded {
+		t.Fatal("expected context.DeadlineExceeded error")
 	}
 	if err = callee.Unregister(procName); err != nil {
 		t.Fatal("failed to unregister procedure:", err)
@@ -541,12 +537,8 @@ func TestCancelRemoteProcedureCall(t *testing.T) {
 		t.Fatal("call should have been canceled")
 	}
 
-	rpcError, ok := err.(RPCError)
-	if !ok {
-		t.Fatal("expected RPCError type of error")
-	}
-	if rpcError.Err.Error != wamp.ErrCanceled {
-		t.Fatal("expected canceled error, got:", err)
+	if err != context.Canceled {
+		t.Fatal("expected context.Canceled error")
 	}
 	if err = callee.Unregister(procName); err != nil {
 		t.Fatal("failed to unregister procedure:", err)
diff --git a/router/broker_test.go b/router/broker_test.go
index 325b705..8636436 100644
--- a/router/broker_test.go
+++ b/router/broker_test.go
@@ -1,7 +1,7 @@
 package router
 
 import (
-	"errors"
+	"context"
 	"testing"
 	"time"
 
@@ -19,18 +19,18 @@ func newTestPeer() *testPeer {
 }
 
 func (p *testPeer) TrySend(msg wamp.Message) error {
-	select {
-	case p.in <- msg:
-	default:
-		return errors.New("try again")
-	}
-	return nil
+	return wamp.TrySend(p.in, msg)
 }
 
 func (p *testPeer) Send(msg wamp.Message) error {
 	p.in <- msg
 	return nil
 }
+
+func (p *testPeer) SendCtx(ctx context.Context, msg wamp.Message) error {
+	return wamp.SendCtx(ctx, p.in, msg)
+}
+
 func (p *testPeer) Recv() <-chan wamp.Message { return p.in }
 func (p *testPeer) Close()                    { return }
 
diff --git a/router/dealer_test.go b/router/dealer_test.go
index 3db8199..a8aecc5 100644
--- a/router/dealer_test.go
+++ b/router/dealer_test.go
@@ -1039,14 +1039,20 @@ func TestPatternBasedRegistration(t *testing.T) {
 	}
 }
 
-func TestRPCBlockedSlowClientCall(t *testing.T) {
+func TestRPCBlockedUnresponsiveCallee(t *testing.T) {
+	const (
+		rpcExecTime    = time.Second
+		timeoutMs      = 2000
+		shortTimeoutMs = 200
+	)
 	dealer, metaClient := newTestDealer()
 
 	// Register a procedure.
 	callee, rtr := transport.LinkedPeers()
 	calleeSess := newSession(rtr, 0, nil)
+	opts := wamp.Dict{wamp.OptTimeout: timeoutMs}
 	dealer.Register(calleeSess,
-		&wamp.Register{Request: 223, Procedure: testProcedure})
+		&wamp.Register{Request: 223, Procedure: testProcedure, Options: opts})
 	rsp := <-callee.Recv()
 	_, ok := rsp.(*wamp.Registered)
 	if !ok {
@@ -1060,16 +1066,28 @@ func TestRPCBlockedSlowClientCall(t *testing.T) {
 	caller, rtr := transport.LinkedPeers()
 	callerSession := newSession(rtr, 0, nil)
 
-	for i := 0; i < 20; i++ {
+	// Call unresponsive callee until call dropped.
+	var i int
+sendLoop:
+	for {
+		i++
 		fmt.Println("Calling", i)
 		// Test calling valid procedure
-		dealer.Call(callerSession,
-			&wamp.Call{Request: wamp.ID(i + 225), Procedure: testProcedure})
+		dealer.Call(callerSession, &wamp.Call{
+			Request:   wamp.ID(i + 225),
+			Procedure: testProcedure,
+			Options:   opts,
+		})
+		select {
+		case rsp = <-caller.Recv():
+			break sendLoop
+		default:
+		}
 	}
 
-	fmt.Println("Waiting for error")
+	callee.Close()
+
 	// Test that caller received an ERROR message.
-	rsp = <-caller.Recv()
 	rslt, ok := rsp.(*wamp.Error)
 	if !ok {
 		t.Fatal("expected ERROR, got:", rsp.MessageType())
diff --git a/router/rawsocketserver_test.go b/router/rawsocketserver_test.go
index 8c75946..ebc769a 100644
--- a/router/rawsocketserver_test.go
+++ b/router/rawsocketserver_test.go
@@ -19,7 +19,7 @@ func TestRSHandshakeJSON(t *testing.T) {
 		t.Fatal(err)
 	}
 	defer r.Close()
-	clsr, err := NewRawSocketServer(r, 0, 0).ListenAndServe("tcp", tcpAddr)
+	clsr, err := NewRawSocketServer(r).ListenAndServe("tcp", tcpAddr)
 	if err != nil {
 		t.Fatal(err)
 	}
@@ -51,7 +51,7 @@ func TestRSHandshakeMsgpack(t *testing.T) {
 		t.Fatal(err)
 	}
 	defer r.Close()
-	clsr, err := NewRawSocketServer(r, 0, 0).ListenAndServe("tcp", tcpAddr)
+	clsr, err := NewRawSocketServer(r).ListenAndServe("tcp", tcpAddr)
 	if err != nil {
 		t.Fatal(err)
 	}
diff --git a/transport/localpeer_test.go b/transport/localpeer_test.go
index cd08063..5110062 100644
--- a/transport/localpeer_test.go
+++ b/transport/localpeer_test.go
@@ -51,7 +51,7 @@ func TestDropOnBlockedClient(t *testing.T) {
 	}()
 	select {
 	case <-done:
-	case <-time.After(sendTimeout + 50*time.Millisecond):
+	case <-time.After(50 * time.Millisecond):
 		t.Fatal("Send should have dropped and not blocked")
 	}
 	if err == nil || err.Error() != "blocked" {
diff --git a/wamp/peer_test.go b/wamp/peer_test.go
index cc0fc02..79aabb0 100644
--- a/wamp/peer_test.go
+++ b/wamp/peer_test.go
@@ -1,6 +1,7 @@
 package wamp
 
 import (
+	"context"
 	"testing"
 	"time"
 )
@@ -10,15 +11,22 @@ type testPeer struct {
 }
 
 func newTestPeer() Peer {
-	return &testPeer{make(chan Message, 1)}
+	return &testPeer{make(chan Message)}
 }
 
-func (p *testPeer) TrySend(msg Message) error { return nil }
+func (p *testPeer) TrySend(msg Message) error {
+	return TrySend(p.in, msg)
+}
 
 func (p *testPeer) Send(msg Message) error {
 	p.in <- msg
 	return nil
 }
+
+func (p *testPeer) SendCtx(ctx context.Context, msg Message) error {
+	return SendCtx(ctx, p.in, msg)
+}
+
 func (p *testPeer) Recv() <-chan Message { return p.in }
 func (p *testPeer) Close()               { close(p.in) }
 
@@ -43,3 +51,42 @@ func TestRecvTimeout(t *testing.T) {
 		t.Fatal("Expected closed channel error")
 	}
 }
+
+func TestTrySend(t *testing.T) {
+	p := newTestPeer()
+	err := p.TrySend(&Hello{})
+	if err == nil {
+		t.Fatal("Expected error")
+	}
+
+	ready := make(chan struct{})
+	go func() {
+		close(ready)
+		<-p.Recv()
+	}()
+	<-ready
+
+	if err = p.TrySend(&Hello{}); err != nil {
+		t.Fatal("Failed to send message")
+	}
+
+	p.Close()
+}
+
+func TestSendCtx(t *testing.T) {
+	p := newTestPeer()
+	ctx, cancel := context.WithCancel(context.Background())
+	go func() {
+		<-p.Recv()
+	}()
+	err := p.SendCtx(ctx, &Hello{})
+	if err != nil {
+		t.Fatal(err)
+	}
+	cancel()
+	err = p.SendCtx(ctx, &Hello{})
+	if err != context.Canceled {
+		t.Fatal("Expected context.Canceled error")
+	}
+	p.Close()
+}