func (c *Client) SetCallCancelMode(cancelMode string) error {
	switch cancelMode {
	case wamp.CancelModeKill, wamp.CancelModeKillNoWait, wamp.CancelModeSkip:
	case "":
		cancelMode = wamp.CancelModeKillNoWait
	default:
		return fmt.Errorf("cancel mode not one of: %q, %q, %q",
			wamp.CancelModeKill, wamp.CancelModeKillNoWait, wamp.CancelModeSkip)
	}
	c.cancelMode = cancelMode
	return nil
}
SetCallCancelMode sets the client's call cancel mode to one of the
following: "kill", "killnowait', "skip".  Setting to "" specifies using the
default value: "killnowait".  The cancel mode is an option that is sent in a
CANCEL message when a CALL is canceled.

Cancel Mode Behavior

"skip": The pending call is canceled and ERROR is sent immediately back to
the caller.  No INTERRUPT is sent to the callee and the result is discarded
when received.

"kill": INTERRUPT is sent to the client, but ERROR is not returned to the
caller until after the callee has responded to the canceled call.  In this
case the caller may receive RESULT or ERROR depending whether the callee
finishes processing the invocation or the interrupt first.

"killnowait": The pending call is canceled and ERROR is sent immediately
back to the caller.  INTERRUPT is sent to the callee and any response to the
invocation or interrupt from the callee is discarded when received.

If the callee does not support call canceling, then behavior is "skip".
package client
import (
	"context"
	"errors"
	"fmt"
	"strings"
	"sync"
	"time"
	"github.com/gammazero/nexus/v3/stdlog"
	"github.com/gammazero/nexus/v3/transport/serialize"
	"github.com/gammazero/nexus/v3/wamp"
)
const (
	JSON    = serialize.JSON
	MSGPACK = serialize.MSGPACK
	CBOR    = serialize.CBOR
)
type Client struct {
	sess *wamp.Session
	responseTimeout time.Duration
	awaitingReply   map[wamp.ID]chan wamp.Message
	authHandlers    map[string]AuthFunc
	eventHandlers map[wamp.ID]EventHandler
	topicSubID    map[string]wamp.ID
	invHandlers    map[wamp.ID]InvocationHandler
	nameProcID     map[string]wamp.ID
	invHandlerKill map[wamp.ID]context.CancelFunc
	progGate       map[context.Context]wamp.ID
	activeInvHandlers sync.WaitGroup
	log   stdlog.StdLog
	debug bool
	cancel     context.CancelFunc
	ctx        context.Context
	cancelMode string
	closed bool
	routerGoodbye *wamp.Goodbye
	idGen         *wamp.SyncIDGen
}
type InvokeResult struct {
	Args   wamp.List
	Kwargs wamp.Dict
	Err    wamp.URI
}
var InvocationCanceled = InvokeResult{Err: wamp.ErrCanceled}
func NewClient(p wamp.Peer, cfg Config) (*Client, error) {
	if cfg.ResponseTimeout == 0 {
		cfg.ResponseTimeout = defaultResponseTimeout
	}
	welcome, err := joinRealm(p, cfg)
	if err != nil {
		p.Close()
		return nil, err
	}
	sess := wamp.NewSession(p, welcome.ID, welcome.Details, welcome.Details)
	if !sess.HasRole(wamp.RoleBroker) && !sess.HasRole(wamp.RoleDealer) {
		p.Close()
		return nil, ErrRouterNoRoles
	}
	c := &Client{
		sess: sess,
		responseTimeout: cfg.ResponseTimeout,
		awaitingReply:   map[wamp.ID]chan wamp.Message{},
		eventHandlers: map[wamp.ID]EventHandler{},
		topicSubID:    map[string]wamp.ID{},
		invHandlers:    map[wamp.ID]InvocationHandler{},
		nameProcID:     map[string]wamp.ID{},
		invHandlerKill: map[wamp.ID]context.CancelFunc{},
		progGate:       map[context.Context]wamp.ID{},
		log:        cfg.Logger,
		debug:      cfg.Debug,
		cancelMode: wamp.CancelModeKillNoWait,
		idGen:      new(wamp.SyncIDGen),
	}
	c.ctx, c.cancel = context.WithCancel(context.Background())
	go c.run()  
	return c, nil
}
func (c *Client) Done() <-chan struct{} { return c.ctx.Done() }
func (c *Client) Connected() bool { return c.ctx.Err() == nil }
func (c *Client) ID() wamp.ID { return c.sess.ID }
func (c *Client) Logger() stdlog.StdLog { return c.log }
func (c *Client) RealmDetails() wamp.Dict { return c.sess.Details }
func (c *Client) HasFeature(role, feature string) bool {
	return c.sess.HasFeature(role, feature)
}
type EventHandler func(event *wamp.Event)
func (c *Client) Subscribe(topic string, fn EventHandler, options wamp.Dict) error {
	if !c.Connected() {
		return ErrNotConn
	}
	if options == nil {
		options = wamp.Dict{}
	}
	id := c.idGen.Next()
	c.expectReply(id)
	c.sess.Send(&wamp.Subscribe{
		Request: id,
		Options: options,
		Topic:   wamp.URI(topic),
	})
	msg, err := c.waitForReply(id)
	if err != nil {
		return err
	}
	switch msg := msg.(type) {
	case *wamp.Subscribed:
		c.sess.Lock()
		c.eventHandlers[msg.Subscription] = fn
		c.topicSubID[topic] = msg.Subscription
		c.sess.Unlock()
		return nil
	case *wamp.Error:
		return fmt.Errorf("subscribing to topic '%v': %s", topic,
			wampErrorString(msg))
	default:
		return unexpectedMsgError(msg, wamp.SUBSCRIBED)
	}
}
func (c *Client) SubscribeChan(topic string, events chan<- *wamp.Event, options wamp.Dict) error {
	handler := func(ev *wamp.Event) { events <- ev }
	return c.Subscribe(topic, handler, options)
}
func (c *Client) SubscriptionID(topic string) (subID wamp.ID, ok bool) {
	c.sess.Lock()
	subID, ok = c.topicSubID[topic]
	c.sess.Unlock()
	return
}
func (c *Client) Unsubscribe(topic string) error {
	c.sess.Lock()
	subID, ok := c.topicSubID[topic]
	if !ok {
		c.sess.Unlock()
		return ErrNotSubscribed
	}
	delete(c.topicSubID, topic)
	delete(c.eventHandlers, subID)
	c.sess.Unlock()
	if !c.Connected() {
		return ErrNotConn
	}
	id := c.idGen.Next()
	c.expectReply(id)
	c.sess.Send(&wamp.Unsubscribe{
		Request:      id,
		Subscription: subID,
	})
	msg, err := c.waitForReply(id)
	if err != nil {
		return err
	}
	switch msg := msg.(type) {
	case *wamp.Unsubscribed:
		return nil
	case *wamp.Error:
		return fmt.Errorf("unsubscribing to '%s': %s", topic,
			wampErrorString(msg))
	}
	return unexpectedMsgError(msg, wamp.UNSUBSCRIBED)
}
func (c *Client) Publish(topic string, options wamp.Dict, args wamp.List, kwargs wamp.Dict) error {
	if !c.Connected() {
		return ErrNotConn
	}
	id := c.idGen.Next()
	var pubAck bool
	if options == nil {
		options = wamp.Dict{}
	} else {
		pubAck, _ = options[wamp.OptAcknowledge].(bool)
		if pubAck {
			c.expectReply(id)
		}
	}
	c.sess.Send(&wamp.Publish{
		Request:     id,
		Options:     options,
		Topic:       wamp.URI(topic),
		Arguments:   args,
		ArgumentsKw: kwargs,
	})
	if !pubAck {
		return nil
	}
	msg, err := c.waitForReply(id)
	if err != nil {
		return err
	}
	switch msg := msg.(type) {
	case *wamp.Published:
	case *wamp.Error:
		return fmt.Errorf("waiting for published message: %s", wampErrorString(msg))
	default:
		return unexpectedMsgError(msg, wamp.PUBLISHED)
	}
	return nil
}
type InvocationHandler func(context.Context, *wamp.Invocation) InvokeResult
func (c *Client) Register(procedure string, fn InvocationHandler, options wamp.Dict) error {
	if !c.Connected() {
		return ErrNotConn
	}
	id := c.idGen.Next()
	c.expectReply(id)
	if options == nil {
		options = wamp.Dict{}
	}
	c.sess.Send(&wamp.Register{
		Request:   id,
		Options:   options,
		Procedure: wamp.URI(procedure),
	})
	msg, err := c.waitForReply(id)
	if err != nil {
		return err
	}
	switch msg := msg.(type) {
	case *wamp.Registered:
		c.sess.Lock()
		c.invHandlers[msg.Registration] = fn
		c.nameProcID[procedure] = msg.Registration
		c.sess.Unlock()
		if c.debug {
			c.log.Println("Registered", procedure, "as registration",
				msg.Registration)
		}
	case *wamp.Error:
		return fmt.Errorf("registering procedure '%v': %s", procedure,
			wampErrorString(msg))
	default:
		return unexpectedMsgError(msg, wamp.REGISTERED)
	}
	return nil
}
func (c *Client) RegistrationID(procedure string) (regID wamp.ID, ok bool) {
	c.sess.Lock()
	regID, ok = c.nameProcID[procedure]
	c.sess.Unlock()
	return
}
func (c *Client) Unregister(procedure string) error {
	c.sess.Lock()
	procID, ok := c.nameProcID[procedure]
	if !ok {
		c.sess.Unlock()
		return ErrNotRegistered
	}
	delete(c.nameProcID, procedure)
	delete(c.invHandlers, procID)
	c.sess.Unlock()
	if !c.Connected() {
		return ErrNotConn
	}
	id := c.idGen.Next()
	c.expectReply(id)
	c.sess.Send(&wamp.Unregister{
		Request:      id,
		Registration: procID,
	})
	msg, err := c.waitForReply(id)
	if err != nil {
		return err
	}
	switch msg := msg.(type) {
	case *wamp.Unregistered:
	case *wamp.Error:
		return fmt.Errorf("unregistering procedure '%s': %v", procedure,
			wampErrorString(msg))
	default:
		return unexpectedMsgError(msg, wamp.UNREGISTERED)
	}
	return nil
}
type ProgressHandler func(*wamp.Result)
func (c *Client) Call(ctx context.Context, procedure string, options wamp.Dict, args wamp.List, kwargs wamp.Dict, progcb ProgressHandler) (*wamp.Result, error) {
	if !c.Connected() {
		return nil, ErrNotConn
	}
	if options == nil {
		options = wamp.Dict{}
	}
	var progChan chan *wamp.Result
	var progDone chan struct{}
	if progcb != nil {
		options[wamp.OptReceiveProgress] = true
		progChan = make(chan *wamp.Result)
		progDone = make(chan struct{})
		go func() {
			for result := range progChan {
				progcb(result)
			}
			close(progDone)
		}()
	}
	id := c.idGen.Next()
	c.expectReply(id)
	c.sess.Send(&wamp.Call{
		Request:     id,
		Procedure:   wamp.URI(procedure),
		Options:     options,
		Arguments:   args,
		ArgumentsKw: kwargs,
	})
	msg, err := c.waitForReplyWithCancel(ctx, id, procedure, progChan)
	if progcb != nil {
		close(progChan)
		<-progDone
	}
	if err != nil {
		return nil, err
	}
	switch msg := msg.(type) {
	case *wamp.Result:
		return msg, nil
	case *wamp.Error:
		return nil, RPCError{msg, procedure}
	default:
		return nil, unexpectedMsgError(msg, wamp.RESULT)
	}
}