Repository: gammazero/nexus
Pull request #302 (2023-09-08 21:15:53)

Send API returns a write-only channel

<!--- Provide a general summary of your changes in the Title above -->

### Description, Motivation and Context
This PR simplifies APIs by removing `TrySend` and `SendCtx`, and allows the caller of `Send` more flexability. More flexibility is provided by a channel as it allows the caller to choose to use it in a select, range loop, etc. when writing a message.

Addresses Issue #291

### What is the current behavior?
A message is passed into the blocking `Send` or non-blocking `TrySend`, or a message and context is passed into `SendCtx`.

### What is the new behavior?
The `Send` function returns a write-only message channel. The `TrySend` and `SendCtx` functions disappear.

### What kind of change does this PR introduce?
- [x] Bug fix (non-breaking change which fixes an issue)
- [ ] New feature (non-breaking change which adds functionality)
- [ ] Breaking change (fix or feature that would cause existing functionality to not work as expected)

_This might not be considered a breaking change if `Send` is considered an internal API used when writing your own router or client implementation using the `wamp` package. You can see this is the case since all of the [examples](https://github.com/gammazero/nexus/tree/v3/examples) do not need to change._

### Checklist:
- [x] My code follows the code style of this project.
- [ ] I have added tests to cover my changes.
- [x] Overall test coverage is not decreased.
- [x] All new and existing tests passed.
- [ ] My change requires a change to the documentation.
  - [ ] I have updated the documentation accordingly.



Patch:
diff --git a/client/client.go b/client/client.go
index 74a95c8..218f2f3 100644
--- a/client/client.go
+++ b/client/client.go
@@ -313,11 +313,11 @@ func (c *Client) Subscribe(topic string, fn EventHandler, options wamp.Dict) err
 	}
 	id := c.idGen.Next()
 	c.expectReply(id)
-	c.sess.Send(&wamp.Subscribe{
+	c.sess.Send() <- &wamp.Subscribe{
 		Request: id,
 		Options: options,
 		Topic:   wamp.URI(topic),
-	})
+	}
 
 	// Wait to receive SUBSCRIBED message.
 	msg, err := c.waitForReply(id)
@@ -384,10 +384,10 @@ func (c *Client) Unsubscribe(topic string) error {
 
 	id := c.idGen.Next()
 	c.expectReply(id)
-	c.sess.Send(&wamp.Unsubscribe{
+	c.sess.Send() <- &wamp.Unsubscribe{
 		Request:      id,
 		Subscription: subID,
-	})
+	}
 
 	// Wait to receive UNSUBSCRIBED message.
 	msg, err := c.waitForReply(id)
@@ -509,7 +509,7 @@ func (c *Client) Publish(topic string, options wamp.Dict, args wamp.List, kwargs
 		message.ArgumentsKw = kwargs
 	}
 
-	c.sess.Send(message)
+	c.sess.Send() <- message
 
 	if !pubAck {
 		return nil
@@ -573,11 +573,11 @@ func (c *Client) Register(procedure string, fn InvocationHandler, options wamp.D
 	if options == nil {
 		options = wamp.Dict{}
 	}
-	c.sess.Send(&wamp.Register{
+	c.sess.Send() <- &wamp.Register{
 		Request:   id,
 		Options:   options,
 		Procedure: wamp.URI(procedure),
-	})
+	}
 
 	// Wait to receive REGISTERED message.
 	msg, err := c.waitForReply(id)
@@ -636,10 +636,10 @@ func (c *Client) Unregister(procedure string) error {
 
 	id := c.idGen.Next()
 	c.expectReply(id)
-	c.sess.Send(&wamp.Unregister{
+	c.sess.Send() <- &wamp.Unregister{
 		Request:      id,
 		Registration: procID,
-	})
+	}
 
 	// Wait to receive UNREGISTERED message.
 	msg, err := c.waitForReply(id)
@@ -769,7 +769,7 @@ func (c *Client) Call(ctx context.Context, procedure string, options wamp.Dict,
 		return nil, err
 	}
 
-	c.sess.Send(message)
+	c.sess.Send() <- message
 
 	// Wait to receive RESULT message.
 	msg, err := c.waitForReplyWithCancel(ctx, id, procedure, progChan)
@@ -791,7 +791,7 @@ func (c *Client) Call(ctx context.Context, procedure string, options wamp.Dict,
 
 		if err != nil {
 			if abortMsg != nil {
-				c.sess.Peer.Send(abortMsg)
+				c.sess.Peer.Send() <- abortMsg
 				c.sess.Peer.Close()
 			}
 
@@ -868,7 +868,7 @@ func (c *Client) CallProgressive(ctx context.Context, procedure string, sendProg
 		return nil, err
 	}
 
-	c.sess.Send(message)
+	c.sess.Send() <- message
 
 	callInProgress, _ := options[wamp.OptProgress].(bool)
 
@@ -880,10 +880,10 @@ func (c *Client) CallProgressive(ctx context.Context, procedure string, sendProg
 				options, args, kwargs, err = sendProg(ctx)
 
 				if err != nil {
-					c.sess.Send(&wamp.Cancel{
+					c.sess.Send() <- &wamp.Cancel{
 						Request: id,
 						Options: wamp.SetOption(nil, wamp.OptMode, wamp.CancelModeKillNoWait),
-					})
+					}
 					return
 				}
 
@@ -902,14 +902,14 @@ func (c *Client) CallProgressive(ctx context.Context, procedure string, sendProg
 				}
 
 				if err := c.prepareCallPayloadMessage(message, options, args, kwargs); err != nil {
-					c.sess.Send(&wamp.Cancel{
+					c.sess.Send() <- &wamp.Cancel{
 						Request: id,
 						Options: wamp.SetOption(nil, wamp.OptMode, wamp.CancelModeKillNoWait),
-					})
+					}
 					return
 				}
 
-				c.sess.Send(message)
+				c.sess.Send() <- message
 			}
 		}()
 	}
@@ -935,7 +935,7 @@ func (c *Client) CallProgressive(ctx context.Context, procedure string, sendProg
 
 		if err != nil {
 			if abortMsg != nil {
-				c.sess.Peer.Send(abortMsg)
+				c.sess.Peer.Send() <- abortMsg
 				c.sess.Peer.Close()
 			}
 
@@ -1018,11 +1018,14 @@ func (c *Client) Close() error {
 		sendCtx, cancel := context.WithTimeout(context.Background(), 2*c.responseTimeout)
 		defer cancel()
 
-		var stopped bool
-		if c.sess.SendCtx(sendCtx, &wamp.Goodbye{
+		gmMsg := &wamp.Goodbye{
 			Details: wamp.Dict{},
 			Reason:  wamp.CloseRealm,
-		}) == nil {
+		}
+
+		var stopped bool
+		select {
+		case c.sess.Send() <- gmMsg:
 			// The router should respond with a GOODBYE message, which causes
 			// run() to exit.  Wait for run() to exit, but only wait for
 			// whatever time remains on the context.
@@ -1031,6 +1034,7 @@ func (c *Client) Close() error {
 				stopped = true
 			case <-sendCtx.Done():
 			}
+		case <-sendCtx.Done():
 		}
 
 		if !stopped {
@@ -1085,12 +1089,15 @@ func (c *Client) SendProgress(ctx context.Context, args wamp.List, kwArgs wamp.D
 		// Caller is not accepting progressive results or call canceled.
 		return ErrCallerNoProg
 	}
-	if c.sess.SendCtx(ctx, &wamp.Yield{
+	yieldMsg := &wamp.Yield{
 		Request:     req,
 		Options:     wamp.Dict{wamp.OptProgress: true},
 		Arguments:   args,
 		ArgumentsKw: kwArgs,
-	}) != nil {
+	}
+	select {
+	case c.sess.Send() <- yieldMsg:
+	case <-ctx.Done():
 		select {
 		case <-c.Done():
 			return ErrNotConn
@@ -1131,7 +1138,7 @@ func joinRealm(peer wamp.Peer, cfg Config) (*wamp.Welcome, error) {
 		details[helloAuthmethods] = authmethods
 	}
 
-	peer.Send(&wamp.Hello{Realm: wamp.URI(cfg.Realm), Details: details})
+	peer.Send() <- &wamp.Hello{Realm: wamp.URI(cfg.Realm), Details: details}
 	msg, err := wamp.RecvTimeout(peer, cfg.ResponseTimeout)
 	if err != nil {
 		return nil, err
@@ -1167,14 +1174,14 @@ func handleCRAuth(peer wamp.Peer, challenge *wamp.Challenge, authHandlers map[st
 		// know how to deal with.  In response to a CHALLENGE message, the
 		// Client MUST send an AUTHENTICATE message.  So, send empty
 		// AUTHENTICATE since client does not know what to put in it.
-		peer.Send(&wamp.Authenticate{})
+		peer.Send() <- &wamp.Authenticate{}
 	} else {
 		// Create signature and send AUTHENTICATE.
 		signature, authDetails := authFunc(challenge)
-		peer.Send(&wamp.Authenticate{
+		peer.Send() <- &wamp.Authenticate{
 			Signature: signature,
 			Extra:     authDetails,
-		})
+		}
 	}
 	msg, err := wamp.RecvTimeout(peer, rspTimeout)
 	if err != nil {
@@ -1352,10 +1359,10 @@ CollectResults:
 			c.log.Printf("Call to %q canceled by caller (mode=%s): %s",
 				procedure, c.cancelMode, err)
 		}
-		c.sess.Send(&wamp.Cancel{
+		c.sess.Send() <- &wamp.Cancel{
 			Request: id,
 			Options: wamp.SetOption(nil, wamp.OptMode, c.cancelMode),
-		})
+		}
 		// Wait for the ERROR from the dealer.
 		timer := time.NewTimer(c.responseTimeout)
 	waitCancel:
@@ -1531,13 +1538,13 @@ func (c *Client) runHandleInvocation(msg *wamp.Invocation) {
 		// 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{
+		c.sess.Send() <- &wamp.Error{
 			Type:      wamp.INVOCATION,
 			Request:   reqID,
 			Details:   wamp.Dict{},
 			Error:     wamp.ErrInvalidArgument,
 			Arguments: wamp.List{errMsg},
-		})
+		}
 		c.log.Print(errMsg)
 		return
 	}
@@ -1547,13 +1554,13 @@ func (c *Client) runHandleInvocation(msg *wamp.Invocation) {
 	if pptScheme, _ := msg.Details[wamp.OptPPTScheme].(string); pptScheme != "" {
 
 		if !isPPTSchemeValid(pptScheme) {
-			c.sess.Send(&wamp.Error{
+			c.sess.Send() <- &wamp.Error{
 				Type:      wamp.INVOCATION,
 				Request:   reqID,
 				Details:   wamp.Dict{},
 				Error:     wamp.ErrInvalidArgument,
 				Arguments: wamp.List{ErrPPTSchemeInvalid.Error()},
-			})
+			}
 			c.log.Printf("cannot process invocation with invalid ppt schema %q: %v", pptScheme, ErrPPTSchemeInvalid)
 			return
 		}
@@ -1571,13 +1578,13 @@ func (c *Client) runHandleInvocation(msg *wamp.Invocation) {
 		}
 
 		if err != nil {
-			c.sess.Send(&wamp.Error{
+			c.sess.Send() <- &wamp.Error{
 				Type:      wamp.INVOCATION,
 				Request:   reqID,
 				Details:   wamp.Dict{},
 				Error:     wamp.ErrInvalidArgument,
 				Arguments: wamp.List{err.Error()},
-			})
+			}
 			c.log.Printf("cannot unpack invocation message: %v", err)
 			return
 		}
@@ -1707,14 +1714,18 @@ func (c *Client) runHandleInvocation(msg *wamp.Invocation) {
 			}
 
 			if result.Err != "" {
-				c.sess.SendCtx(c.ctx, &wamp.Error{
+				errMsg := &wamp.Error{
 					Type:        wamp.INVOCATION,
 					Request:     reqID,
 					Details:     wamp.Dict{},
 					Arguments:   result.Args,
 					ArgumentsKw: result.Kwargs,
 					Error:       result.Err,
-				})
+				}
+				select {
+				case c.sess.Send() <- errMsg:
+				case <-c.ctx.Done():
+				}
 				return
 			}
 
@@ -1742,13 +1753,13 @@ func (c *Client) runHandleInvocation(msg *wamp.Invocation) {
 							wamp.OptMessage: ErrPPTNotSupportedByPeer.Error(),
 						},
 					}
-					c.sess.Peer.Send(&abortMsg)
+					c.sess.Peer.Send() <- &abortMsg
 					c.sess.Peer.Close()
 					return
 				}
 
 				if !isPPTSchemeValid(pptScheme) {
-					c.sess.SendCtx(c.ctx, &wamp.Error{
+					errMsg := &wamp.Error{
 						Type:    wamp.INVOCATION,
 						Request: reqID,
 						Details: wamp.Dict{
@@ -1757,7 +1768,11 @@ func (c *Client) runHandleInvocation(msg *wamp.Invocation) {
 						Arguments:   result.Args,
 						ArgumentsKw: result.Kwargs,
 						Error:       wamp.ErrInvalidArgument,
-					})
+					}
+					select {
+					case c.sess.Send() <- errMsg:
+					case <-c.ctx.Done():
+					}
 					return
 				}
 
@@ -1773,7 +1788,7 @@ func (c *Client) runHandleInvocation(msg *wamp.Invocation) {
 				}
 
 				if err != nil {
-					c.sess.SendCtx(c.ctx, &wamp.Error{
+					errMsg := &wamp.Error{
 						Type:    wamp.INVOCATION,
 						Request: reqID,
 						Details: wamp.Dict{
@@ -1782,7 +1797,11 @@ func (c *Client) runHandleInvocation(msg *wamp.Invocation) {
 						Arguments:   result.Args,
 						ArgumentsKw: result.Kwargs,
 						Error:       wamp.ErrInvalidArgument,
-					})
+					}
+					select {
+					case c.sess.Send() <- errMsg:
+					case <-c.ctx.Done():
+					}
 					return
 				}
 
@@ -1793,7 +1812,10 @@ func (c *Client) runHandleInvocation(msg *wamp.Invocation) {
 				message.ArgumentsKw = result.Kwargs
 			}
 
-			c.sess.SendCtx(c.ctx, message)
+			select {
+			case c.sess.Send() <- message:
+			case <-c.ctx.Done():
+			}
 		}()
 	}
 }
diff --git a/router/auth/crauth.go b/router/auth/crauth.go
index b687447..d659508 100644
--- a/router/auth/crauth.go
+++ b/router/auth/crauth.go
@@ -89,12 +89,14 @@ func (cr *CRAuthenticator) Authenticate(sid wamp.ID, details wamp.Dict, client w
 	}
 
 	// Challenge response needed.  Send CHALLENGE message to client.
-	err = client.Send(&wamp.Challenge{
+	chalMsg := &wamp.Challenge{
 		AuthMethod: cr.AuthMethod(),
 		Extra:      extra,
-	})
-	if err != nil {
-		return nil, err
+	}
+	select {
+	case client.Send() <- chalMsg:
+	default:
+		return nil, errors.New("cannot send challenge to client: blocked")
 	}
 
 	// Read AUTHENTICATE response from client.
diff --git a/router/auth/cryptosign.go b/router/auth/cryptosign.go
index 4b30a2c..6460100 100644
--- a/router/auth/cryptosign.go
+++ b/router/auth/cryptosign.go
@@ -72,14 +72,15 @@ func (cr *CryptoSignAuthenticator) Authenticate(sid wamp.ID, details wamp.Dict,
 
 	extra := wamp.Dict{"challenge": hex.EncodeToString(challenge)}
 
-	// Challenge response needed.  Send CHALLENGE message to client.
-	err = client.Send(&wamp.Challenge{
+	chalMsg := &wamp.Challenge{
 		AuthMethod: cr.AuthMethod(),
 		Extra:      extra,
-	})
-
-	if err != nil {
-		return nil, err
+	}
+	// Challenge response needed.  Send CHALLENGE message to client.
+	select {
+	case client.Send() <- chalMsg:
+	default:
+		return nil, errors.New("cannot send challenge to client: blocked")
 	}
 
 	// Read AUTHENTICATE response from client.
diff --git a/router/auth/ticket.go b/router/auth/ticket.go
index 4748c8e..4687f94 100644
--- a/router/auth/ticket.go
+++ b/router/auth/ticket.go
@@ -74,14 +74,16 @@ func (t *TicketAuthenticator) Authenticate(sid wamp.ID, details wamp.Dict, clien
 		ticket = nil
 	}
 
-	// Challenge Extra map is empty since the ticket challenge only asks for a
-	// ticket (using authmethod) and provides no additional challenge info.
-	err = client.Send(&wamp.Challenge{
+	chalMsg := &wamp.Challenge{
 		AuthMethod: t.AuthMethod(),
 		Extra:      wamp.Dict{},
-	})
-	if err != nil {
-		return nil, err
+	}
+	// Challenge Extra map is empty since the ticket challenge only asks for a
+	// ticket (using authmethod) and provides no additional challenge info.
+	select {
+	case client.Send() <- chalMsg:
+	default:
+		return nil, errors.New("Cannot send challenge to client: blocked")
 	}
 
 	// Read AUTHENTICATE response from client.
diff --git a/router/broker.go b/router/broker.go
index b7481d0..6ada303 100644
--- a/router/broker.go
+++ b/router/broker.go
@@ -208,7 +208,7 @@ func (b *broker) publish(pub *wamp.Session, msg *wamp.Publish) {
 			abortMsg := wamp.Abort{Reason: wamp.ErrProtocolViolation}
 			abortMsg.Details = wamp.Dict{}
 			abortMsg.Details[wamp.OptMessage] = ErrPPTNotSupportedByPeer.Error()
-			_ = pub.TrySend(&abortMsg)
+			b.trySend(pub, &abortMsg)
 			pub.Close()
 
 			return
@@ -737,12 +737,12 @@ func (b *broker) syncPubSubCreateMeta(topic wamp.URI, subSessID wamp.ID, sub *su
 	})
 }
 
-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
+func (b *broker) trySend(sess *wamp.Session, msg wamp.Message) {
+	select {
+	case sess.Send() <- msg:
+	default:
+		b.log.Printf("!!! Dropped %s to session %s: blocked", msg.MessageType(), sess)
 	}
-	return true
 }
 
 func prepareEvent(pub *wamp.Session, msg *wamp.Publish, pubID wamp.ID, sub *subscription, sendTopic, disclose bool, eventDetails wamp.Dict, subscriber *wamp.Session) *wamp.Event { //nolint:lll
diff --git a/router/dealer.go b/router/dealer.go
index 9806ab6..567c9eb 100644
--- a/router/dealer.go
+++ b/router/dealer.go
@@ -234,7 +234,7 @@ func (d *dealer) register(callee *wamp.Session, msg *wamp.Register) {
 	}
 	<-done
 	for _, pub := range metaPubs {
-		d.metaPeer.Send(pub)
+		d.metaPeer.Send() <- pub
 	}
 }
 
@@ -251,7 +251,7 @@ func (d *dealer) unregister(callee *wamp.Session, msg *wamp.Unregister) {
 	}
 	<-done
 	for _, pub := range metaPubs {
-		d.metaPeer.Send(pub)
+		d.metaPeer.Send() <- pub
 	}
 
 }
@@ -390,7 +390,7 @@ func (d *dealer) removeSession(sess *wamp.Session) {
 	}
 	<-done
 	for _, pub := range metaPubs {
-		d.metaPeer.Send(pub)
+		d.metaPeer.Send() <- pub
 	}
 }
 
@@ -657,9 +657,8 @@ func (d *dealer) syncCall(caller *wamp.Session, msg *wamp.Call) {
 		abortMsg.Details = wamp.Dict{}
 		abortMsg.Details[wamp.OptMessage] = "Peer is trying to use Progressive Call Invocations while it was not " +
 			"announced during HELLO handshake"
-		_ = caller.TrySend(&abortMsg)
+		d.trySend(caller, &abortMsg)
 		caller.Close()
-
 		return
 	}
 
@@ -761,9 +760,8 @@ func (d *dealer) syncCall(caller *wamp.Session, msg *wamp.Call) {
 			abortMsg.Details = wamp.Dict{}
 			abortMsg.Details[wamp.OptMessage] = "Peer is trying to use Payload PassThru Mode while it was not " +
 				"announced during HELLO handshake"
-			_ = caller.TrySend(&abortMsg)
+			d.trySend(caller, &abortMsg)
 			caller.Close()
-
 			return
 		}
 
@@ -835,13 +833,16 @@ func (d *dealer) syncCall(caller *wamp.Session, msg *wamp.Call) {
 
 	// Send INVOCATION to the endpoint that has registered the requested
 	// procedure.
-	if !d.trySend(callee, &wamp.Invocation{
+	invMsg := &wamp.Invocation{
 		Request:      invocationID,
 		Registration: reg.id,
 		Details:      details,
 		Arguments:    msg.Arguments,
 		ArgumentsKw:  msg.ArgumentsKw,
-	}) {
+	}
+	select {
+	case callee.Send() <- invMsg:
+	default:
 		d.syncError(&wamp.Error{
 			Type:      wamp.INVOCATION,
 			Request:   invocationID,
@@ -929,10 +930,12 @@ func (d *dealer) syncCancel(caller *wamp.Session, msg *wamp.Cancel, mode string,
 			d.log.Println("Callee", invk.callee, "does not support call canceling")
 		} else {
 			// Send INTERRUPT message to callee.
-			if d.trySend(invk.callee, &wamp.Interrupt{
+			intrMsg := &wamp.Interrupt{
 				Request: invocationID,
 				Options: wamp.Dict{wamp.OptReason: reason, wamp.OptMode: mode},
-			}) {
+			}
+			select {
+			case invk.callee.Send() <- intrMsg:
 				d.log.Println("Dealer sent INTERRUPT to cancel invocation",
 					invocationID, "for call", msg.Request, "mode:", mode)
 
@@ -942,6 +945,7 @@ func (d *dealer) syncCancel(caller *wamp.Session, msg *wamp.Cancel, mode string,
 				if mode == wamp.CancelModeKill {
 					return
 				}
+			default:
 			}
 		}
 	}
@@ -983,12 +987,15 @@ func (d *dealer) syncYield(callee *wamp.Session, msg *wamp.Yield, progress, canR
 			// callee's progressive call results feature would have been
 			// disabled at registration time if the callee did not support call
 			// canceling.
-			if d.trySend(callee, &wamp.Interrupt{
+			intrMsg := &wamp.Interrupt{
 				Request: msg.Request,
 				Options: wamp.Dict{wamp.OptMode: wamp.CancelModeKillNoWait},
-			}) {
+			}
+			select {
+			case callee.Send() <- intrMsg:
 				d.log.Println("Dealer sent INTERRUPT to cancel progressive",
 					"results for request", msg.Request, "to callee", callee)
+			default:
 			}
 		} else {
 			// WAMP does not allow sending INTERRUPT in response to normal or
@@ -1061,9 +1068,8 @@ func (d *dealer) syncYield(callee *wamp.Session, msg *wamp.Yield, progress, canR
 			abortMsg := wamp.Abort{Reason: wamp.ErrProtocolViolation}
 			abortMsg.Details = wamp.Dict{}
 			abortMsg.Details[wamp.OptMessage] = ErrPPTNotSupportedByPeer.Error()
-			_ = callee.TrySend(&abortMsg)
+			d.trySend(callee, &abortMsg)
 			callee.Close()
-
 			return false
 		}
 
@@ -1104,13 +1110,14 @@ func (d *dealer) syncYield(callee *wamp.Session, msg *wamp.Yield, progress, canR
 		Arguments:   msg.Arguments,
 		ArgumentsKw: msg.ArgumentsKw,
 	}
-	err := caller.TrySend(res)
-	if err != nil {
+	select {
+	case caller.Send() <- res:
+	default:
 		if canRetry {
 			keepInvocation = true
 			return true
 		}
-		d.log.Printf("!!! Dropped %s to caller %s: %s", res.MessageType(), caller, err)
+		d.log.Printf("!!! Dropped %s to caller %s: blocked", res.MessageType(), caller)
 		d.syncCancel(caller, &wamp.Cancel{Request: callID.request},
 			wamp.CancelModeKillNoWait, wamp.ErrCanceled, nil)
 	}
@@ -1487,12 +1494,12 @@ func (d *dealer) regCountCallees(msg *wamp.Invocation) wamp.Message {
 	}
 }
 
-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
+func (d *dealer) trySend(sess *wamp.Session, msg wamp.Message) {
+	select {
+	case sess.Send() <- msg:
+	default:
+		d.log.Printf("!!! Dropped %s to session %s: blocked", msg.MessageType(), sess)
 	}
-	return true
 }
 
 // discloseCaller adds caller identity information to INVOCATION.Details.
diff --git a/router/realm.go b/router/realm.go
index 75125d9..4520b77 100644
--- a/router/realm.go
+++ b/router/realm.go
@@ -300,11 +300,11 @@ func (r *realm) onJoin(sess *wamp.Session) {
 	sess.Lock()
 	output := r.cleanSessionDetails(sess.Details)
 	sess.Unlock()
-	r.metaPeer.Send(&wamp.Publish{
+	r.metaPeer.Send() <- &wamp.Publish{
 		Request:   wamp.GlobalID(),
 		Topic:     wamp.MetaEventSessionOnJoin,
 		Arguments: wamp.List{output},
-	})
+	}
 }
 
 // onLeave is called when a non-meta session leaves this realm.  The session is
@@ -347,26 +347,26 @@ func (r *realm) onLeave(sess *wamp.Session, shutdown, killAll bool) {
 	if hasTstm {
 		sendTestaments := func(testaments []testament) {
 			for i := range testaments {
-				r.metaPeer.Send(&wamp.Publish{
+				r.metaPeer.Send() <- &wamp.Publish{
 					Request:     wamp.GlobalID(),
 					Topic:       testaments[i].topic,
 					Arguments:   testaments[i].args,
 					ArgumentsKw: testaments[i].kwargs,
 					Options:     testaments[i].options,
-				})
+				}
 			}
 		}
 		sendTestaments(testaments.detached)
 		sendTestaments(testaments.destroyed)
 	}
-	r.metaPeer.Send(&wamp.Publish{
+	r.metaPeer.Send() <- &wamp.Publish{
 		Request: wamp.GlobalID(),
 		Topic:   wamp.MetaEventSessionOnLeave,
 		Arguments: wamp.List{
 			sess.ID,
 			sess.Details["authid"],
 			sess.Details["authrole"]},
-	})
+	}
 }
 
 // HandleSession starts a session attached to this realm.
@@ -401,7 +401,10 @@ func (r *realm) handleSession(sess *wamp.Session) error {
 				Details: wamp.Dict{wamp.OptMessage: err.Error()},
 			}
 			r.log.Println("Aborting session", sess, ":", err)
-			sess.TrySend(&abortMsg)
+			select {
+			case sess.Send() <- &abortMsg:
+			default:
+			}
 		}
 		r.onLeave(sess, shutdown, killAll)
 		sess.Close()
@@ -435,7 +438,10 @@ func (r *realm) handleInboundMessages(sess *wamp.Session) (bool, bool, error) {
 				if r.debug {
 					r.log.Printf("Stop session %s: system shutdown", sess)
 				}
-				sess.TrySend(goodbye)
+				select {
+				case sess.Send() <- goodbye:
+				default:
+				}
 				return true, false, nil
 			}
 			if r.debug {
@@ -445,7 +451,10 @@ func (r *realm) handleInboundMessages(sess *wamp.Session) (bool, bool, error) {
 			if _, ok := goodbye.Details["all"]; ok {
 				killAll = true
 			}
-			sess.TrySend(goodbye)
+			select {
+			case sess.Send() <- goodbye:
+			default:
+			}
 			return false, killAll, nil
 		}
 
@@ -489,10 +498,14 @@ func (r *realm) handleInboundMessages(sess *wamp.Session) (bool, bool, error) {
 
 		case *wamp.Goodbye:
 			// Handle client leaving realm.
-			sess.TrySend(&wamp.Goodbye{
+			gmMsg := &wamp.Goodbye{
 				Reason:  wamp.ErrGoodbyeAndOut,
 				Details: wamp.Dict{},
-			})
+			}
+			select {
+			case sess.Send() <- gmMsg:
+			default:
+			}
 			if r.debug {
 				r.log.Println("GOODBYE from session", sess, "reason:",
 					msg.Reason)
@@ -567,8 +580,9 @@ func (r *realm) authzMessage(sess *wamp.Session, msg wamp.Message) bool {
 			r.log.Println("Client", sess, msg.MessageType(), "not authorized")
 		}
 		if !skipResponse {
-			err = sess.TrySend(errRsp)
-			if err != nil {
+			select {
+			case sess.Send() <- errRsp:
+			default:
 				r.log.Println("!!! client blocked, could not send authz error")
 			}
 		}
@@ -665,13 +679,13 @@ func (r *realm) getAuthenticator(methods []string) (auth auth.Authenticator, aut
 func (r *realm) registerMetaProcedure(procedure wamp.URI, f func(*wamp.Invocation) wamp.Message) {
 	// Register the meta procedure.  The "disclose_caller" option must be
 	// enabled for the testament API and the meta session API.
-	r.metaPeer.Send(&wamp.Register{
+	r.metaPeer.Send() <- &wamp.Register{
 		Request: r.metaIDGen.Next(),
 		Options: wamp.Dict{
 			"disclose_caller": true,
 		},
 		Procedure: procedure,
-	})
+	}
 	msg := <-r.metaPeer.Recv()
 	if msg == nil {
 		// This would only happen if the meta client was closed before or
@@ -708,12 +722,12 @@ func (r *realm) metaProcedureHandler() {
 		case *wamp.Invocation:
 			metaProcHandler, ok := r.metaProcMap[msg.Registration]
 			if !ok {
-				r.metaPeer.Send(&wamp.Error{
+				r.metaPeer.Send() <- &wamp.Error{
 					Type:    msg.MessageType(),
 					Request: msg.Request,
 					Details: wamp.Dict{},
 					Error:   wamp.ErrNoSuchProcedure,
-				})
+				}
 				continue
 			}
 			rsp = metaProcHandler(msg)
@@ -725,7 +739,7 @@ func (r *realm) metaProcedureHandler() {
 		default:
 			r.log.Println("Meta procedure received unexpected", msg.MessageType())
 		}
-		r.metaPeer.Send(rsp)
+		r.metaPeer.Send() <- rsp
 	}
 }
 
diff --git a/router/router.go b/router/router.go
index f7598fb..9b87046 100644
--- a/router/router.go
+++ b/router/router.go
@@ -145,7 +145,7 @@ func (r *router) AttachClient(client wamp.Peer, transportDetails wamp.Dict) erro
 			abortMsg.Details[wamp.OptMessage] = abortErr.Error()
 			r.log.Println("Aborting client connection:", abortErr)
 		}
-		client.Send(&abortMsg) // Blocking OK; this is session goroutine.
+		client.Send() <- &abortMsg // Blocking OK; this is session goroutine.
 		client.Close()
 	}
 
@@ -282,7 +282,7 @@ func (r *router) AttachClient(client wamp.Peer, transportDetails wamp.Dict) erro
 		return err
 	}
 
-	client.Send(welcome) // Blocking OK; this is session goroutine.
+	client.Send() <- welcome // Blocking OK; this is session goroutine.
 	if r.debug {
 		r.log.Println("Finished attaching session:", sid)
 	}
diff --git a/transport/localpeer.go b/transport/localpeer.go
index b358435..0d88d0e 100644
--- a/transport/localpeer.go
+++ b/transport/localpeer.go
@@ -1,8 +1,6 @@
 package transport
 
 import (
-	"context"
-
 	"github.com/gammazero/nexus/v3/wamp"
 )
 
@@ -53,30 +51,9 @@ func (p *localPeer) IsLocal() bool { return true }
 // Recv returns the channel this peer reads incoming messages from.
 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 {
-	return wamp.TrySend(p.wr, msg)
-}
-
-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.
-// Typically called by clients, since it is OK for the router to block a client
-// since this will not block other clients.
-func (p *localPeer) Send(msg wamp.Message) error {
-	p.wr <- msg
-	return nil
-}
+// Send returns the peer's outbound message channel.
+func (p *localPeer) Send() chan<- wamp.Message { return p.wr }
 
 // Close closes the outgoing channel, waking any readers waiting on data from
 // this peer.
 func (p *localPeer) Close() { close(p.wr) }
-
-// DEPRECATED
-// IsLocal returns true is the wamp.Peer is a localPeer.  These do not need
-// authentication since they are part of the same process.
-func IsLocal(p wamp.Peer) bool {
-	return p.IsLocal()
-}
diff --git a/transport/rawsocketpeer.go b/transport/rawsocketpeer.go
index b253e98..7ff9475 100644
--- a/transport/rawsocketpeer.go
+++ b/transport/rawsocketpeer.go
@@ -173,17 +173,7 @@ 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 {
-	return wamp.TrySend(rs.wr, msg)
-}
-
-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 {
-	return wamp.SendCtx(rs.ctxSender, rs.wr, msg)
-}
+func (rs *rawSocketPeer) Send() chan<- wamp.Message { return rs.wr }
 
 func (rs *rawSocketPeer) IsLocal() bool { return false }
 
@@ -197,6 +187,9 @@ func (rs *rawSocketPeer) Close() {
 	// wr channel in case there are incoming messages during close.
 	rs.cancelSender()
 	<-rs.writerDone
+	close(rs.wr)
+	for range rs.wr {
+	}
 
 	// Tell recvHandler to close.
 	close(rs.closed)
diff --git a/transport/websocketpeer.go b/transport/websocketpeer.go
index ce534f4..04ca137 100644
--- a/transport/websocketpeer.go
+++ b/transport/websocketpeer.go
@@ -223,17 +223,7 @@ func NewWebsocketPeer(conn WebsocketConnection, serializer serialize.Serializer,
 
 func (w *websocketPeer) Recv() <-chan wamp.Message { return w.rd }
 
-func (w *websocketPeer) TrySend(msg wamp.Message) error {
-	return wamp.TrySend(w.wr, msg)
-}
-
-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 {
-	return wamp.SendCtx(w.ctxSender, w.wr, msg)
-}
+func (w *websocketPeer) Send() chan<- wamp.Message { return w.wr }
 
 func (w *websocketPeer) IsLocal() bool { return false }
 
@@ -247,6 +237,9 @@ func (w *websocketPeer) Close() {
 	// wr channel in case there are incoming messages during close.
 	w.cancelSender()
 	<-w.writerDone
+	close(w.wr)
+	for range w.wr {
+	}
 
 	closeMsg := websocket.FormatCloseMessage(websocket.CloseNormalClosure,
 		"goodbye")
diff --git a/wamp/peer.go b/wamp/peer.go
index 9a42286..b8d9932 100644
--- a/wamp/peer.go
+++ b/wamp/peer.go
@@ -1,64 +1,37 @@
 package wamp
 
 import (
-	"context"
 	"errors"
 	"time"
 )
 
 // Peer is the interface implemented by endpoints communicating via WAMP.
 type Peer interface {
-	// Sends the message to the peer.
-	Send(Message) error
-
-	// SendCtx sends the message to the peer, and uses a context to cancel or
-	// timeout sending the message when blocked waiting to write to the peer.
-	SendCtx(context.Context, Message) error
-
-	// TrySend performs a non-blocking send.  Returns error if blocked.
-	TrySend(Message) error
-
 	// Closes the peer connection and the channel returned from Recv().
 	Close()
 
+	// IsLocal returns true if the session is local.
+	IsLocal() bool
+
 	// Recv returns a channel of messages from the peer.
 	Recv() <-chan Message
 
-	// IsLocal returns true if the session is local.
-	IsLocal() bool
+	// Send returns the peer's outgoing message channel.
+	Send() chan<- Message
 }
 
 // RecvTimeout receives a message from a peer within the specified time.
-func RecvTimeout(p Peer, t time.Duration) (Message, error) {
+func RecvTimeout(p Peer, timeout time.Duration) (Message, error) {
+	to := time.NewTimer(timeout)
+	defer to.Stop()
+
 	select {
 	case msg, open := <-p.Recv():
 		if !open {
 			return nil, errors.New("receive channel closed")
 		}
 		return msg, nil
-	case <-time.After(t):
+	case <-to.C:
 		return nil, errors.New("timeout waiting for message")
 	}
 }
-
-// SendCtx sends a message to the write-only channel, using a context to cancel
-// sending if blocked.
-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/router/auth/crauth_test.go b/router/auth/crauth_test.go
index da569b2..e55be40 100644
--- a/router/auth/crauth_test.go
+++ b/router/auth/crauth_test.go
@@ -94,10 +94,10 @@ func cliRsp(p wamp.Peer) {
 			continue
 		}
 		signature, authDetails := clientAuthFunc(ch)
-		p.Send(&wamp.Authenticate{
+		p.Send() <- &wamp.Authenticate{
 			Signature: signature,
 			Extra:     authDetails,
-		})
+		}
 	}
 }
 
diff --git a/router/authorizer_test.go b/router/authorizer_test.go
index 02b4b2e..d91f551 100644
--- a/router/authorizer_test.go
+++ b/router/authorizer_test.go
@@ -59,13 +59,13 @@ func TestAuthorizer(t *testing.T) {
 
 	// Test that authorizer forbids denyTopic.
 	subscribeID := wamp.GlobalID()
-	sub.Send(&wamp.Subscribe{Request: subscribeID, Topic: denyTopic})
+	sub.Send() <- &wamp.Subscribe{Request: subscribeID, Topic: denyTopic}
 	msg := <-sub.Recv()
 	_, ok := msg.(*wamp.Error)
 	require.True(t, ok, "Expected ERROR")
 
 	// Test that authorizer allows allowTopic.
-	sub.Send(&wamp.Subscribe{Request: subscribeID, Topic: allowTopic})
+	sub.Send() <- &wamp.Subscribe{Request: subscribeID, Topic: allowTopic}
 	msg = <-sub.Recv()
 	_, ok = msg.(*wamp.Subscribed)
 	require.True(t, ok, "Expected SUBSCRIBED")
@@ -90,7 +90,7 @@ func TestAuthorizerBypassLocal(t *testing.T) {
 	sub := testClient(t, r)
 
 	subscribeID := wamp.GlobalID()
-	sub.Send(&wamp.Subscribe{Request: subscribeID, Topic: denyTopic})
+	sub.Send() <- &wamp.Subscribe{Request: subscribeID, Topic: denyTopic}
 	msg := <-sub.Recv()
 	_, ok := msg.(*wamp.Subscribed)
 	require.True(t, ok, "Expected SUBSCRIBED")
@@ -113,11 +113,11 @@ func TestAuthorizerModify(t *testing.T) {
 
 	cli := testClient(t, r)
 	for i := 1; i <= 10; i++ {
-		cli.Send(&wamp.Call{
+		cli.Send() <- &wamp.Call{
 			Request:   wamp.ID(i),
 			Procedure: wamp.MetaProcSessionGet,
 			Arguments: wamp.List{cli.ID},
-		})
+		}
 		msg, err := wamp.RecvTimeout(cli, time.Second)
 		require.NoError(t, err)
 		result, ok := msg.(*wamp.Result)
@@ -157,7 +157,7 @@ func TestAuthorizerRace(t *testing.T) {
 	// Test that authorizer forbids denyTopic.
 	subscribeID := wamp.GlobalID()
 	const topic = "whatever.topic"
-	sub.Send(&wamp.Subscribe{Request: subscribeID, Topic: topic})
+	sub.Send() <- &wamp.Subscribe{Request: subscribeID, Topic: topic}
 	msg, err := wamp.RecvTimeout(sub, time.Second)
 	require.NoError(t, err)
 	_, ok := msg.(*wamp.Subscribed)
@@ -166,39 +166,39 @@ func TestAuthorizerRace(t *testing.T) {
 	done := make(chan struct{})
 	go func() {
 		for i := 0; i < 100; i++ {
-			pub.Send(&wamp.Publish{
+			pub.Send() <- &wamp.Publish{
 				Request: wamp.ID(i),
 				Topic:   topic,
 				Options: wamp.Dict{"eligible_xyzzy": wamp.List{"plugh", "baz"}},
-			})
+			}
 		}
 		done <- struct{}{}
 	}()
 
 	go func() {
 		for i := 0; i < 100; i++ {
-			cli.Send(&wamp.Call{
+			cli.Send() <- &wamp.Call{
 				Request:   wamp.ID(100 + i),
 				Procedure: wamp.MetaProcSessionGet,
 				Arguments: wamp.List{pub.ID},
-			})
+			}
 			<-cli.Recv()
-			cli.Send(&wamp.Call{
+			cli.Send() <- &wamp.Call{
 				Request:   wamp.ID(200 + i),
 				Procedure: wamp.MetaProcSessionGet,
 				Arguments: wamp.List{sub.ID},
-			})
+			}
 			<-cli.Recv()
 		}
 		done <- struct{}{}
 	}()
 
 	for i := 0; i < 100; i++ {
-		sub.Send(&wamp.Publish{
+		sub.Send() <- &wamp.Publish{
 			Request: wamp.ID(500 + i),
 			Topic:   topic,
 			Options: wamp.Dict{"eligible_xyzzy": wamp.List{"plugh"}},
-		})
+		}
 	}
 	<-done
 	<-done
diff --git a/router/broker_test.go b/router/broker_test.go
index 76dab69..50b696e 100644
--- a/router/broker_test.go
+++ b/router/broker_test.go
@@ -1,7 +1,6 @@
 package router
 
 import (
-	"context"
 	"testing"
 	"time"
 
@@ -19,20 +18,8 @@ func newTestPeer() *testPeer {
 	}
 }
 
-func (p *testPeer) TrySend(msg wamp.Message) error {
-	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) Send() chan<- wamp.Message { return p.in }
 func (p *testPeer) Close()                    {}
 
 func (p *testPeer) IsLocal() bool { return true }
diff --git a/router/rawsocketserver_test.go b/router/rawsocketserver_test.go
index 85a0d96..380f02f 100644
--- a/router/rawsocketserver_test.go
+++ b/router/rawsocketserver_test.go
@@ -27,7 +27,7 @@ func TestRSHandshakeJSON(t *testing.T) {
 	require.NoError(t, err)
 	defer client.Close()
 
-	client.Send(&wamp.Hello{Realm: testRealm, Details: clientRoles})
+	client.Send() <- &wamp.Hello{Realm: testRealm, Details: clientRoles}
 	msg, ok := <-client.Recv()
 	require.True(t, ok, "recv chan closed")
 
@@ -50,7 +50,7 @@ func TestRSHandshakeMsgpack(t *testing.T) {
 	require.NoError(t, err)
 	defer client.Close()
 
-	client.Send(&wamp.Hello{Realm: testRealm, Details: clientRoles})
+	client.Send() <- &wamp.Hello{Realm: testRealm, Details: clientRoles}
 	msg, ok := <-client.Recv()
 	require.True(t, ok, "Receive buffer closed")
 
diff --git a/router/router_test.go b/router/router_test.go
index 357fc44..87f3c3d 100644
--- a/router/router_test.go
+++ b/router/router_test.go
@@ -91,8 +91,9 @@ func testClientInRealm(t *testing.T, r Router, realm wamp.URI) *wamp.Session {
 	details := clientRoles
 	details["authid"] = "user1"
 	details["xyzzy"] = "plugh"
-	//go client.Send(&wamp.Hello{Realm: realm, Details: clientRoles})
-	go client.Send(&wamp.Hello{Realm: realm, Details: details})
+	go func() {
+		client.Send() <- &wamp.Hello{Realm: realm, Details: details}
+	}()
 	err := r.Attach(server)
 	require.NoError(t, err)
 
@@ -117,7 +118,7 @@ func TestHandshake(t *testing.T) {
 	r := newTestRouter(t)
 
 	cli := testClient(t, r)
-	cli.Send(&wamp.Goodbye{})
+	cli.Send() <- &wamp.Goodbye{}
 	msg, err := wamp.RecvTimeout(cli, time.Second)
 	require.NoError(t, err, "no goodbye message after sending goodbye")
 	_, ok := msg.(*wamp.Goodbye)
@@ -128,7 +129,9 @@ func TestHandshakeBadRealm(t *testing.T) {
 	checkGoLeaks(t)
 	r := newTestRouter(t)
 	client, server := transport.LinkedPeers()
-	go client.Send(&wamp.Hello{Realm: "does.not.exist"})
+	go func() {
+		client.Send() <- &wamp.Hello{Realm: "does.not.exist"}
+	}()
 	err := r.Attach(server)
 	require.Error(t, err)
 
@@ -145,7 +148,7 @@ func TestProtocolViolation(t *testing.T) {
 	cli := testClient(t, r)
 
 	// Send HELLO message after session established.
-	cli.Send(&wamp.Hello{Realm: testRealm, Details: clientRoles})
+	cli.Send() <- &wamp.Hello{Realm: testRealm, Details: clientRoles}
 	msg, err := wamp.RecvTimeout(cli, time.Second)
 	require.NoError(t, err, "timed out waiting for ABORT")
 	abort, ok := msg.(*wamp.Abort)
@@ -156,7 +159,9 @@ func TestProtocolViolation(t *testing.T) {
 	client, server := transport.LinkedPeers()
 	// Run as goroutine since Send will block until message read by router, if
 	// client uses unbuffered channel.
-	go client.Send(&wamp.Subscribe{Request: wamp.GlobalID(), Topic: wamp.URI("some.uri")})
+	go func() {
+		client.Send() <- &wamp.Subscribe{Request: wamp.GlobalID(), Topic: wamp.URI("some.uri")}
+	}()
 	err = r.Attach(server)
 	require.Error(t, err)
 
@@ -174,7 +179,7 @@ func TestRouterSubscribe(t *testing.T) {
 	sub := testClient(t, r)
 
 	subscribeID := wamp.GlobalID()
-	sub.Send(&wamp.Subscribe{Request: subscribeID, Topic: testTopic})
+	sub.Send() <- &wamp.Subscribe{Request: subscribeID, Topic: testTopic}
 	msg, err := wamp.RecvTimeout(sub, time.Second)
 	require.NoError(t, err, "Timed out waiting for SUBSCRIBED")
 	subMsg, ok := msg.(*wamp.Subscribed)
@@ -184,7 +189,7 @@ func TestRouterSubscribe(t *testing.T) {
 
 	pub := testClient(t, r)
 	pubID := wamp.GlobalID()
-	pub.Send(&wamp.Publish{Request: pubID, Topic: testTopic})
+	pub.Send() <- &wamp.Publish{Request: pubID, Topic: testTopic}
 
 	msg, err = wamp.RecvTimeout(sub, time.Second)
 	require.NoError(t, err, "Timed out waiting for EVENT")
@@ -199,10 +204,11 @@ func TestPublishAcknowledge(t *testing.T) {
 	client := testClient(t, r)
 
 	id := wamp.GlobalID()
-	client.Send(&wamp.Publish{
+	client.Send() <- &wamp.Publish{
 		Request: id,
 		Options: wamp.Dict{"acknowledge": true},
-		Topic:   "some.uri"})
+		Topic:   "some.uri",
+	}
 
 	msg, err := wamp.RecvTimeout(client, time.Second)
 	require.NoError(t, err, "sent acknowledge=true, timed out waiting for PUBLISHED")
@@ -216,10 +222,11 @@ func TestPublishFalseAcknowledge(t *testing.T) {
 	client := testClient(t, r)
 
 	id := wamp.GlobalID()
-	client.Send(&wamp.Publish{
+	client.Send() <- &wamp.Publish{
 		Request: id,
 		Options: wamp.Dict{"acknowledge": false},
-		Topic:   "some.uri"})
+		Topic:   "some.uri",
+	}
 
 	msg, err := wamp.RecvTimeout(client, 200*time.Millisecond)
 	if err == nil {
@@ -233,7 +240,7 @@ func TestPublishNoAcknowledge(t *testing.T) {
 	client := testClient(t, r)
 
 	id := wamp.GlobalID()
-	client.Send(&wamp.Publish{Request: id, Topic: "some.uri"})
+	client.Send() <- &wamp.Publish{Request: id, Topic: "some.uri"}
 	msg, err := wamp.RecvTimeout(client, 200*time.Millisecond)
 	if err == nil {
 		_, ok := msg.(*wamp.Published)
@@ -248,7 +255,7 @@ func TestRouterCall(t *testing.T) {
 
 	registerID := wamp.GlobalID()
 	// Register remote procedure
-	callee.Send(&wamp.Register{Request: registerID, Procedure: testProcedure})
+	callee.Send() <- &wamp.Register{Request: registerID, Procedure: testProcedure}
 
 	msg, err := wamp.RecvTimeout(callee, time.Second)
 	require.NoError(t, err, "Timed out waiting for REGISTERED")
@@ -260,7 +267,7 @@ func TestRouterCall(t *testing.T) {
 	caller := testClient(t, r)
 	callID := wamp.GlobalID()
 	// Call remote procedure
-	caller.Send(&wamp.Call{Request: callID, Procedure: testProcedure})
+	caller.Send() <- &wamp.Call{Request: callID, Procedure: testProcedure}
 
 	msg, err = wamp.RecvTimeout(callee, time.Second)
 	require.NoError(t, err, "Timed out waiting for INVOCATION")
@@ -270,7 +277,7 @@ func TestRouterCall(t *testing.T) {
 	invocationID := invocation.Request
 
 	// Returns result of remove procedure
-	callee.Send(&wamp.Yield{Request: invocationID})
+	callee.Send() <- &wamp.Yield{Request: invocationID}
 
 	msg, err = wamp.RecvTimeout(caller, time.Second)
 	require.NoError(t, err, "Timed out waiting for RESULT")
@@ -287,7 +294,7 @@ func TestSessionCountMetaProcedure(t *testing.T) {
 
 	// Call wamp.MetaProcSessionCount
 	req := &wamp.Call{Request: wamp.GlobalID(), Procedure: wamp.MetaProcSessionCount}
-	caller.Send(req)
+	caller.Send() <- req
 	msg, err := wamp.RecvTimeout(caller, time.Second)
 	require.NoError(t, err)
 	result, ok := msg.(*wamp.Result)
@@ -304,7 +311,7 @@ func TestSessionCountMetaProcedure(t *testing.T) {
 		Procedure: wamp.MetaProcSessionCount,
 		Arguments: wamp.List{"should-be-a-list"},
 	}
-	caller.Send(req)
+	caller.Send() <- req
 	msg, err = wamp.RecvTimeout(caller, time.Second)
 	require.NoError(t, err)
 	errResult, ok := msg.(*wamp.Error)
@@ -318,7 +325,7 @@ func TestSessionCountMetaProcedure(t *testing.T) {
 		Procedure: wamp.MetaProcSessionCount,
 		Arguments: wamp.List{filter},
 	}
-	caller.Send(req)
+	caller.Send() <- req
 	msg, err = wamp.RecvTimeout(caller, time.Second)
 	require.NoError(t, err)
 	result, ok = msg.(*wamp.Result)
@@ -335,7 +342,7 @@ func TestSessionCountMetaProcedure(t *testing.T) {
 		Procedure: wamp.MetaProcSessionCount,
 		Arguments: wamp.List{filter},
 	}
-	caller.Send(req)
+	caller.Send() <- req
 	msg, err = wamp.RecvTimeout(caller, time.Second)
 	require.NoError(t, err)
 	result, ok = msg.(*wamp.Result)
@@ -358,7 +365,7 @@ func TestListSessionMetaProcedures(t *testing.T) {
 		Request:   wamp.GlobalID(),
 		Procedure: wamp.MetaProcSessionList,
 	}
-	caller.Send(req)
+	caller.Send() <- req
 	msg, err := wamp.RecvTimeout(caller, time.Second)
 	require.NoError(t, err)
 	result, ok := msg.(*wamp.Result)
@@ -377,7 +384,7 @@ func TestListSessionMetaProcedures(t *testing.T) {
 		Procedure: wamp.MetaProcSessionList,
 		Arguments: wamp.List{filter},
 	}
-	caller.Send(req)
+	caller.Send() <- req
 	msg, err = wamp.RecvTimeout(caller, time.Second)
 	require.NoError(t, err)
 	result, ok = msg.(*wamp.Result)
@@ -399,11 +406,11 @@ func TestGetSessionMetaProcedures(t *testing.T) {
 
 	// Call session meta-procedure with bad session ID
 	callID := wamp.GlobalID()
-	caller.Send(&wamp.Call{
+	caller.Send() <- &wamp.Call{
 		Request:   callID,
 		Procedure: wamp.MetaProcSessionGet,
 		Arguments: wamp.List{wamp.ID(123456789)},
-	})
+	}
 	msg, err := wamp.RecvTimeout(caller, time.Second)
 	require.NoError(t, err)
 	errRsp, ok := msg.(*wamp.Error)
@@ -412,11 +419,11 @@ func TestGetSessionMetaProcedures(t *testing.T) {
 
 	// Call session meta-procedure to get session information.
 	callID = wamp.GlobalID()
-	caller.Send(&wamp.Call{
+	caller.Send() <- &wamp.Call{
 		Request:   callID,
 		Procedure: wamp.MetaProcSessionGet,
 		Arguments: wamp.List{sessID},
-	})
+	}
 	msg, err = wamp.RecvTimeout(caller, time.Second)
 	require.NoError(t, err)
 	result, ok := msg.(*wamp.Result)
@@ -436,7 +443,7 @@ func TestRegistrationMetaProcedures(t *testing.T) {
 
 	// ----- Test wamp.registration.list meta procedure -----
 	callID := wamp.GlobalID()
-	caller.Send(&wamp.Call{Request: callID, Procedure: wamp.MetaProcRegList})
+	caller.Send() <- &wamp.Call{Request: callID, Procedure: wamp.MetaProcRegList}
 	msg, err := wamp.RecvTimeout(caller, time.Second)
 	require.NoError(t, err, "Timed out waiting for RESULT")
 	result, ok := msg.(*wamp.Result)
@@ -456,7 +463,7 @@ func TestRegistrationMetaProcedures(t *testing.T) {
 	sessID := callee.ID
 	// Register remote procedure
 	registerID := wamp.GlobalID()
-	callee.Send(&wamp.Register{Request: registerID, Procedure: testProcedure})
+	callee.Send() <- &wamp.Register{Request: registerID, Procedure: testProcedure}
 
 	msg, err = wamp.RecvTimeout(callee, time.Second)
 	require.NoError(t, err, "Timed out waiting for REGISTERED")
@@ -466,18 +473,18 @@ func TestRegistrationMetaProcedures(t *testing.T) {
 	registrationID := registered.Registration
 
 	// Register remote procedure
-	callee.Send(&wamp.Register{
+	callee.Send() <- &wamp.Register{
 		Request:   wamp.GlobalID(),
 		Procedure: testProcedureWC,
 		Options:   wamp.Dict{"match": "wildcard"},
-	})
+	}
 	msg = <-callee.Recv()
 	_, ok = msg.(*wamp.Registered)
 	require.True(t, ok, "expected REGISTERED")
 
 	// Call session meta-procedure to get session count.
 	callID = wamp.GlobalID()
-	caller.Send(&wamp.Call{Request: callID, Procedure: wamp.MetaProcRegList})
+	caller.Send() <- &wamp.Call{Request: callID, Procedure: wamp.MetaProcRegList}
 	msg, err = wamp.RecvTimeout(caller, time.Second)
 	require.NoError(t, err, "Timed out waiting for RESULT")
 	result, ok = msg.(*wamp.Result)
@@ -505,11 +512,11 @@ func TestRegistrationMetaProcedures(t *testing.T) {
 
 	// ----- Test wamp.registration.lookup meta procedure -----
 	callID = wamp.GlobalID()
-	caller.Send(&wamp.Call{
+	caller.Send() <- &wamp.Call{
 		Request:   callID,
 		Procedure: wamp.MetaProcRegLookup,
 		Arguments: wamp.List{testProcedure},
-	})
+	}
 	msg, err = wamp.RecvTimeout(caller, time.Second)
 	require.NoError(t, err, "Timed out waiting for RESULT")
 	result, ok = msg.(*wamp.Result)
@@ -522,11 +529,11 @@ func TestRegistrationMetaProcedures(t *testing.T) {
 
 	// ----- Test wamp.registration.match meta procedure -----
 	callID = wamp.GlobalID()
-	caller.Send(&wamp.Call{
+	caller.Send() <- &wamp.Call{
 		Request:   callID,
 		Procedure: wamp.MetaProcRegMatch,
 		Arguments: wamp.List{testProcedure},
-	})
+	}
 	msg, err = wamp.RecvTimeout(caller, time.Second)
 	require.NoError(t, err, "Timed out waiting for RESULT")
 	result, ok = msg.(*wamp.Result)
@@ -539,11 +546,11 @@ func TestRegistrationMetaProcedures(t *testing.T) {
 
 	// ----- Test wamp.registration.get meta procedure -----
 	callID = wamp.GlobalID()
-	caller.Send(&wamp.Call{
+	caller.Send() <- &wamp.Call{
 		Request:   callID,
 		Procedure: wamp.MetaProcRegGet,
 		Arguments: wamp.List{registrationID},
-	})
+	}
 	msg, err = wamp.RecvTimeout(caller, time.Second)
 	require.NoError(t, err, "Timed out waiting for RESULT")
 	result, ok = msg.(*wamp.Result)
@@ -559,11 +566,11 @@ func TestRegistrationMetaProcedures(t *testing.T) {
 
 	// ----- Test wamp.registration.list_callees meta procedure -----
 	callID = wamp.GlobalID()
-	caller.Send(&wamp.Call{
+	caller.Send() <- &wamp.Call{
 		Request:   callID,
 		Procedure: wamp.MetaProcRegListCallees,
 		Arguments: wamp.List{registrationID},
-	})
+	}
 	msg, err = wamp.RecvTimeout(caller, time.Second)
 	require.NoError(t, err, "Timed out waiting for RESULT")
 	result, ok = msg.(*wamp.Result)
@@ -577,11 +584,11 @@ func TestRegistrationMetaProcedures(t *testing.T) {
 
 	// ----- Test wamp.registration.count_callees meta procedure -----
 	callID = wamp.GlobalID()
-	caller.Send(&wamp.Call{
+	caller.Send() <- &wamp.Call{
 		Request:   callID,
 		Procedure: wamp.MetaProcRegCountCallees,
 		Arguments: wamp.List{registrationID},
-	})
+	}
 	msg, err = wamp.RecvTimeout(caller, time.Second)
 	require.NoError(t, err, "Timed out waiting for RESULT")
 	result, ok = msg.(*wamp.Result)
@@ -600,7 +607,7 @@ func TestSubscriptionMetaProcedures(t *testing.T) {
 
 	// ----- Test wamp.subscription.list meta procedure -----
 	callID := wamp.GlobalID()
-	caller.Send(&wamp.Call{Request: callID, Procedure: wamp.MetaProcSubList})
+	caller.Send() <- &wamp.Call{Request: callID, Procedure: wamp.MetaProcSubList}
 	msg, err := wamp.RecvTimeout(caller, time.Second)
 	require.NoError(t, err, "Timed out waiting for RESULT")
 	result, ok := msg.(*wamp.Result)
@@ -620,7 +627,7 @@ func TestSubscriptionMetaProcedures(t *testing.T) {
 	sessID := subscriber.ID
 	// Subscribe to topic
 	reqID := wamp.GlobalID()
-	subscriber.Send(&wamp.Subscribe{Request: reqID, Topic: testTopic})
+	subscriber.Send() <- &wamp.Subscribe{Request: reqID, Topic: testTopic}
 	msg, err = wamp.RecvTimeout(subscriber, time.Second)
 	require.NoError(t, err, "Timed out waiting for SUBSCRIBED")
 	subscribed, ok := msg.(*wamp.Subscribed)
@@ -629,11 +636,11 @@ func TestSubscriptionMetaProcedures(t *testing.T) {
 	subscriptionID := subscribed.Subscription
 
 	// Subscriber to wildcard topic
-	subscriber.Send(&wamp.Subscribe{
+	subscriber.Send() <- &wamp.Subscribe{
 		Request: wamp.GlobalID(),
 		Topic:   testTopicWC,
 		Options: wamp.Dict{"match": "wildcard"},
-	})
+	}
 	msg = <-subscriber.Recv()
 	subscribed, ok = msg.(*wamp.Subscribed)
 	require.True(t, ok, "expected SUBSCRIBED")
@@ -641,7 +648,7 @@ func TestSubscriptionMetaProcedures(t *testing.T) {
 
 	// Call subscription meta-procedure to get subscriptions.
 	callID = wamp.GlobalID()
-	caller.Send(&wamp.Call{Request: callID, Procedure: wamp.MetaProcSubList})
+	caller.Send() <- &wamp.Call{Request: callID, Procedure: wamp.MetaProcSubList}
 	msg, err = wamp.RecvTimeout(caller, time.Second)
 	require.NoError(t, err, "Timed out waiting for RESULT")
 	result, ok = msg.(*wamp.Result)
@@ -669,11 +676,11 @@ func TestSubscriptionMetaProcedures(t *testing.T) {
 
 	// ----- Test wamp.subscription.lookup meta procedure -----
 	callID = wamp.GlobalID()
-	caller.Send(&wamp.Call{
+	caller.Send() <- &wamp.Call{
 		Request:   callID,
 		Procedure: wamp.MetaProcSubLookup,
 		Arguments: wamp.List{testTopic},
-	})
+	}
 	msg, err = wamp.RecvTimeout(caller, time.Second)
 	require.NoError(t, err, "Timed out waiting for RESULT")
 	result, ok = msg.(*wamp.Result)
@@ -686,11 +693,11 @@ func TestSubscriptionMetaProcedures(t *testing.T) {
 
 	// ----- Test wamp.subscription.match meta procedure -----
 	callID = wamp.GlobalID()
-	caller.Send(&wamp.Call{
+	caller.Send() <- &wamp.Call{
 		Request:   callID,
 		Procedure: wamp.MetaProcSubMatch,
 		Arguments: wamp.List{testTopic},
-	})
+	}
 	msg, err = wamp.RecvTimeout(caller, time.Second)
 	require.NoError(t, err, "Timed out waiting for RESULT")
 	result, ok = msg.(*wamp.Result)
@@ -705,11 +712,11 @@ func TestSubscriptionMetaProcedures(t *testing.T) {
 
 	// ----- Test wamp.subscription.get meta procedure -----
 	callID = wamp.GlobalID()
-	caller.Send(&wamp.Call{
+	caller.Send() <- &wamp.Call{
 		Request:   callID,
 		Procedure: wamp.MetaProcSubGet,
 		Arguments: wamp.List{subscriptionID},
-	})
+	}
 	msg, err = wamp.RecvTimeout(caller, time.Second)
 	require.NoError(t, err, "Timed out waiting for RESULT")
 	result, ok = msg.(*wamp.Result)
@@ -725,11 +732,11 @@ func TestSubscriptionMetaProcedures(t *testing.T) {
 
 	// ----- Test wamp.subscription.list_subscribers meta procedure -----
 	callID = wamp.GlobalID()
-	caller.Send(&wamp.Call{
+	caller.Send() <- &wamp.Call{
 		Request:   callID,
 		Procedure: wamp.MetaProcSubListSubscribers,
 		Arguments: wamp.List{subscriptionID},
-	})
+	}
 	msg, err = wamp.RecvTimeout(caller, time.Second)
 	require.NoError(t, err, "Timed out waiting for RESULT")
 	result, ok = msg.(*wamp.Result)
@@ -743,11 +750,11 @@ func TestSubscriptionMetaProcedures(t *testing.T) {
 
 	// ----- Test wamp.subscription.count_subscribers meta procedure -----
 	callID = wamp.GlobalID()
-	caller.Send(&wamp.Call{
+	caller.Send() <- &wamp.Call{
 		Request:   callID,
 		Procedure: wamp.MetaProcSubCountSubscribers,
 		Arguments: wamp.List{subscriptionID},
-	})
+	}
 	msg, err = wamp.RecvTimeout(caller, time.Second)
 	require.NoError(t, err, "Timed out waiting for RESULT")
 	result, ok = msg.(*wamp.Result)
@@ -772,7 +779,7 @@ func TestDynamicRealmChange(t *testing.T) {
 	require.NoError(t, err)
 
 	cli := testClientInRealm(t, dr, testRealm2)
-	cli.Send(&wamp.Goodbye{})
+	cli.Send() <- &wamp.Goodbye{}
 	msg, err := wamp.RecvTimeout(cli, time.Second)
 	require.NoError(t, err, "no goodbye message after sending goodbye")
 	_, ok := msg.(*wamp.Goodbye)
diff --git a/router/sessionkill_test.go b/router/sessionkill_test.go
index d2fa12b..765f5c4 100644
--- a/router/sessionkill_test.go
+++ b/router/sessionkill_test.go
@@ -20,11 +20,12 @@ func TestSessionKill(t *testing.T) {
 	message := "this is a test"
 
 	// Request that client-3 be killed.
-	cli1.Send(&wamp.Call{
+	cli1.Send() <- &wamp.Call{
 		Request:     wamp.GlobalID(),
 		Procedure:   wamp.MetaProcSessionKill,
 		Arguments:   wamp.List{cli3.ID},
-		ArgumentsKw: wamp.Dict{"reason": reason, "message": message}})
+		ArgumentsKw: wamp.Dict{"reason": reason, "message": message},
+	}
 
 	msg, err := wamp.RecvTimeout(cli1, time.Second)
 	require.NoError(t, err)
@@ -45,11 +46,12 @@ func TestSessionKill(t *testing.T) {
 	require.Error(t, err, "Expected timeout")
 
 	// Test that killing self gets error.
-	cli1.Send(&wamp.Call{
+	cli1.Send() <- &wamp.Call{
 		Request:     wamp.GlobalID(),
 		Procedure:   wamp.MetaProcSessionKill,
 		Arguments:   wamp.List{cli1.ID},
-		ArgumentsKw: nil})
+		ArgumentsKw: nil,
+	}
 
 	msg, err = wamp.RecvTimeout(cli1, time.Second)
 	require.NoError(t, err)
@@ -69,10 +71,11 @@ func TestSessionKillAll(t *testing.T) {
 	reason := wamp.URI("foo.bar.baz")
 	message := "this is a test"
 
-	cli1.Send(&wamp.Call{
+	cli1.Send() <- &wamp.Call{
 		Request:     wamp.GlobalID(),
 		Procedure:   wamp.MetaProcSessionKillAll,
-		ArgumentsKw: wamp.Dict{"reason": reason, "message": message}})
+		ArgumentsKw: wamp.Dict{"reason": reason, "message": message},
+	}
 
 	msg, err := wamp.RecvTimeout(cli1, time.Second)
 	require.NoError(t, err)
@@ -112,11 +115,12 @@ func TestSessionKillByAuthid(t *testing.T) {
 
 	// All clients have the same authid, so killing by authid should kill all
 	// except the requesting client.
-	cli1.Send(&wamp.Call{
+	cli1.Send() <- &wamp.Call{
 		Request:     wamp.GlobalID(),
 		Procedure:   wamp.MetaProcSessionKillByAuthid,
 		Arguments:   wamp.List{cli1.Details["authid"]},
-		ArgumentsKw: wamp.Dict{"reason": reason, "message": message}})
+		ArgumentsKw: wamp.Dict{"reason": reason, "message": message},
+	}
 
 	msg, err := wamp.RecvTimeout(cli1, time.Second)
 	require.NoError(t, err)
@@ -156,11 +160,11 @@ func TestSessionModifyDetails(t *testing.T) {
 	// Call session meta-procedure to get session information.
 	callID := wamp.GlobalID()
 	delta := wamp.Dict{"xyzzy": nil, "pi": 3.14, "authid": "bob"}
-	caller.Send(&wamp.Call{
+	caller.Send() <- &wamp.Call{
 		Request:   callID,
 		Procedure: wamp.MetaProcSessionModifyDetails,
 		Arguments: wamp.List{caller.ID, delta},
-	})
+	}
 	msg, err := wamp.RecvTimeout(caller, time.Second)
 	require.NoError(t, err)
 	result, ok := msg.(*wamp.Result)
@@ -169,11 +173,11 @@ func TestSessionModifyDetails(t *testing.T) {
 
 	// Call session meta-procedure to get session information.
 	callID = wamp.GlobalID()
-	caller.Send(&wamp.Call{
+	caller.Send() <- &wamp.Call{
 		Request:   callID,
 		Procedure: wamp.MetaProcSessionGet,
 		Arguments: wamp.List{sessID},
-	})
+	}
 	msg, err = wamp.RecvTimeout(caller, time.Second)
 	require.NoError(t, err)
 	result, ok = msg.(*wamp.Result)
diff --git a/router/testaments_test.go b/router/testaments_test.go
index 7aed751..b3a9fd3 100644
--- a/router/testaments_test.go
+++ b/router/testaments_test.go
@@ -14,13 +14,13 @@ func TestSessionTestaments(t *testing.T) {
 
 	sub := testClient(t, r)
 	subscribeID := wamp.GlobalID()
-	sub.Send(&wamp.Subscribe{Request: subscribeID, Topic: "testament.test1"})
+	sub.Send() <- &wamp.Subscribe{Request: subscribeID, Topic: "testament.test1"}
 	msg, err := wamp.RecvTimeout(sub, time.Second)
 	require.NoError(t, err)
 	_, ok := msg.(*wamp.Subscribed)
 	require.True(t, ok, "expected RESULT")
 
-	sub.Send(&wamp.Subscribe{Request: subscribeID, Topic: "testament.test2"})
+	sub.Send() <- &wamp.Subscribe{Request: subscribeID, Topic: "testament.test2"}
 	msg, err = wamp.RecvTimeout(sub, time.Second)
 	require.NoError(t, err)
 	require.NotNil(t, msg)
@@ -29,7 +29,7 @@ func TestSessionTestaments(t *testing.T) {
 	caller2 := testClient(t, r)
 
 	callID := wamp.GlobalID()
-	caller1.Send(&wamp.Call{
+	caller1.Send() <- &wamp.Call{
 		Request:   callID,
 		Procedure: wamp.MetaProcSessionAddTestament,
 		Arguments: wamp.List{
@@ -37,7 +37,7 @@ func TestSessionTestaments(t *testing.T) {
 			wamp.List{"foo"},
 			wamp.Dict{},
 		},
-	})
+	}
 
 	msg, err = wamp.RecvTimeout(caller1, time.Second)
 	require.NoError(t, err)
@@ -45,7 +45,7 @@ func TestSessionTestaments(t *testing.T) {
 	require.True(t, ok, "expected RESULT")
 	require.Equal(t, callID, result.Request, "wrong result ID")
 
-	caller2.Send(&wamp.Call{
+	caller2.Send() <- &wamp.Call{
 		Request:   wamp.GlobalID(),
 		Procedure: wamp.MetaProcSessionAddTestament,
 		Arguments: wamp.List{
@@ -53,7 +53,7 @@ func TestSessionTestaments(t *testing.T) {
 			wamp.List{"foo"},
 			wamp.Dict{},
 		},
-	})
+	}
 
 	msg, err = wamp.RecvTimeout(caller2, time.Second)
 	require.NoError(t, err)
diff --git a/router/websocketserver_test.go b/router/websocketserver_test.go
index f446fdd..7cf2d20 100644
--- a/router/websocketserver_test.go
+++ b/router/websocketserver_test.go
@@ -50,7 +50,7 @@ func TestWSHandshakeJSON(t *testing.T) {
 	require.NoError(t, err)
 	defer client.Close()
 
-	client.Send(&wamp.Hello{Realm: testRealm, Details: clientRoles})
+	client.Send() <- &wamp.Hello{Realm: testRealm, Details: clientRoles}
 	msg, ok := <-client.Recv()
 	require.True(t, ok, "recv chan closed")
 
@@ -74,7 +74,7 @@ func TestWSHandshakeMsgpack(t *testing.T) {
 	require.NoError(t, err)
 	defer client.Close()
 
-	client.Send(&wamp.Hello{Realm: testRealm, Details: clientRoles})
+	client.Send() <- &wamp.Hello{Realm: testRealm, Details: clientRoles}
 	msg, ok := <-client.Recv()
 	require.True(t, ok, "Receive buffer closed")
 
diff --git a/transport/localpeer_test.go b/transport/localpeer_test.go
index 8e9c279..9623a36 100644
--- a/transport/localpeer_test.go
+++ b/transport/localpeer_test.go
@@ -1,7 +1,7 @@
 package transport
 
 import (
-	"runtime"
+	"errors"
 	"testing"
 	"time"
 
@@ -12,14 +12,16 @@ import (
 func TestSendRecv(t *testing.T) {
 	c, r := LinkedPeers()
 
-	go c.Send(&wamp.Hello{})
+	go func() {
+		c.Send() <- &wamp.Hello{}
+	}()
 	select {
 	case <-r.Recv():
 	case <-time.After(time.Second):
 		require.FailNow(t, "Router peer did not receive msg")
 	}
 
-	r.Send(&wamp.Welcome{})
+	r.Send() <- &wamp.Welcome{}
 	select {
 	case <-c.Recv():
 	default:
@@ -41,12 +43,19 @@ func TestDropOnBlockedClient(t *testing.T) {
 
 	// Check that r -> c drops when full
 	for i := 0; i < qsize; i++ {
-		r.TrySend(&wamp.Publish{})
+		select {
+		case r.Send() <- &wamp.Publish{}:
+		default:
+		}
 	}
 	done := make(chan struct{})
 	var err error
 	go func() {
-		err = r.TrySend(&wamp.Publish{})
+		select {
+		case r.Send() <- &wamp.Publish{}:
+		default:
+			err = errors.New("blocked")
+		}
 		close(done)
 	}()
 	select {
@@ -63,7 +72,7 @@ func TestBlockOnBlockedRouter(t *testing.T) {
 	done := make(chan struct{})
 	go func() {
 		for i := 0; i < cap(r.Recv())+1; i++ {
-			c.Send(&wamp.Publish{})
+			c.Send() <- &wamp.Publish{}
 		}
 		close(done)
 	}()
@@ -82,7 +91,7 @@ func BenchmarkClientToRouter(b *testing.B) {
 	b.ResetTimer()
 	go func() {
 		for i := 0; i < b.N; i++ {
-			c.Send(&wamp.Hello{})
+			c.Send() <- &wamp.Hello{}
 		}
 	}()
 	for i := 0; i < b.N; i++ {
@@ -96,11 +105,7 @@ func BenchmarkRouterToClient(b *testing.B) {
 	b.ResetTimer()
 	go func() {
 		for i := 0; i < b.N; i++ {
-			err := r.Send(&wamp.Hello{})
-			for err != nil {
-				runtime.Gosched()
-				err = r.Send(&wamp.Hello{})
-			}
+			r.Send() <- &wamp.Hello{}
 		}
 	}()
 	for i := 0; i < b.N; i++ {
diff --git a/wamp/peer_test.go b/wamp/peer_test.go
index 57c94b3..0e867af 100644
--- a/wamp/peer_test.go
+++ b/wamp/peer_test.go
@@ -1,7 +1,6 @@
 package wamp
 
 import (
-	"context"
 	"testing"
 	"time"
 
@@ -16,19 +15,7 @@ func newTestPeer() Peer {
 	return &testPeer{make(chan Message)}
 }
 
-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) Send() chan<- Message { return p.in }
 func (p *testPeer) Recv() <-chan Message { return p.in }
 func (p *testPeer) Close()               { close(p.in) }
 
@@ -41,7 +28,7 @@ func TestRecvTimeout(t *testing.T) {
 	require.Nil(t, msg)
 
 	go func() {
-		p.Send(&Hello{})
+		p.Send() <- &Hello{}
 	}()
 	msg, err = RecvTimeout(p, time.Millisecond)
 	require.NoError(t, err)
@@ -51,36 +38,3 @@ func TestRecvTimeout(t *testing.T) {
 	_, err = RecvTimeout(p, time.Millisecond)
 	require.EqualError(t, err, "receive channel closed")
 }
-
-func TestTrySend(t *testing.T) {
-	p := newTestPeer()
-	err := p.TrySend(&Hello{})
-	require.Error(t, err)
-
-	ready := make(chan struct{})
-	go func() {
-		close(ready)
-		<-p.Recv()
-	}()
-	<-ready
-
-	err = p.TrySend(&Hello{})
-	require.NoError(t, err)
-
-	p.Close()
-}
-
-func TestSendCtx(t *testing.T) {
-	p := newTestPeer()
-	ctx, cancel := context.WithCancel(context.Background())
-	go func() {
-		<-p.Recv()
-	}()
-	err := p.SendCtx(ctx, &Hello{})
-	require.NoError(t, err)
-
-	cancel()
-	err = p.SendCtx(ctx, &Hello{})
-	require.ErrorIs(t, err, context.Canceled)
-	p.Close()
-}