client

func (*Client) Call ¶

func (c *Client) Call(ctx context.Context, procedure string, options wamp.Dict, args wamp.List, kwargs wamp.Dict, progcb ProgressHandler) (*wamp.Result, error)

Call calls the procedure corresponding to the given URI.

If an ERROR message is received from the router, the error value returned can be type asserted to RPCError to provide access to the returned ERROR message. This may be necessary for the client application to process error data from the RPC invocation.

Call Canceling ¶

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. The client's cancel mode can be set using SetCallCancelMode().

If the context is canceled or times out, then error returned will not be a RPCError. This allows the caller to distinguish between cancellation initiated by the client (by canceling context), and cancellation initialed elsewhere.

Call Timeout ¶

If a timeout is provided in the options, and the callee supports call timeout, then the timeout value is passed to the callee so that the invocation can be canceled by the callee if the timeout is reached before a response is returned. This is the behavior implemented by the nexus client in the callee role.

The nexus router also supports automatic call cancellation by the router, after the timeout specified in the call options. This way, the router will automatically cancel the call even if the callee does not support it.

To request automatic call timeout, by the router and callee, specify a timeout in milliseconds: options["timeout"] = 30000

Caller Identification ¶

A caller may request the disclosure of its identity (its WAMP session ID) to callees, if allowed by the dealer.

To request that this caller's identity disclosed to callees, set:

options["disclose_me"] = true

NOTE: Use consts defined in wamp/options.go instead of raw strings.

Progressive Call Results ¶

A caller indicates its willingness to receive progressive results by supplying a ProgressHandler function to handle progressive results that are returned before the final result. Call returns when the final result is returned by the callee. The progress handler is guaranteed not to be called after Call returns.

There is no need to set the "receive_progress" option, as this is automatically set if a progress callback is provided.

IMPORTANT: If the context has a timeout, then the amount of time needs to be sufficient for the caller to receive all progressive results as well as the final result.

// Configure and connect caller client.
ctx := context.Background()
cfg := Config{Realm: "nexus.realm1"}
caller, err := ConnectNet(ctx, "ws://localhost:8080/", cfg)
if err != nil {
	log.Fatal(err)
}
defer caller.Close()

// Create a context to cancel the call after 5 seconds if no response.
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
defer cancel()

// Test calling the "sum" procedure with args 1..10.
callArgs := wamp.List{1, 2, 3, 4, 5, 6, 7, 8, 9, 10}
result, err := caller.Call(ctx, "sum", nil, callArgs, nil, nil)
if err != nil {
	log.Fatal(err)
}

// Print the result.
sum, _ := wamp.AsInt64(result.Arguments[0])
log.Println("The sum is:", sum)

Output:

// Configure and connect caller client.
ctx := context.Background()
cfg := Config{Realm: "nexus.realm1"}
caller, err := ConnectNet(ctx, "ws://localhost:8080/", cfg)
if err != nil {
	log.Fatal(err)
}
defer caller.Close()

// The progress handler prints test output as it arrives.
progHandler := func(result *wamp.Result) {
	// Received more progress ingo from callee.
	percentDone, _ := wamp.AsInt64(result.Arguments[0])
	log.Printf("Test is %d%% done", percentDone)
}

// Create a context to cancel the call in one minute if not finished.
ctx, cancel := context.WithTimeout(context.Background(), time.Minute)
defer cancel()

// Call the example procedure "run.test".  The argument specifies how
// frequently to send progress updates.
result, err := caller.Call(
	ctx, "run.test", nil, wamp.List{time.Second * 2}, nil, progHandler)
if err != nil {
	log.Fatal("Failed to call procedure:", err)
}

// As a final result, in this example, the callee returns the boolean
// result of the test.
passFail, _ := result.Arguments[0].(bool)
if passFail {
	log.Println("Test passed")
} else {
	log.Println("Test failed")
}

Output:

func (*Client) Close ¶

func (c *Client) Close() error

Close causes the client to leave the realm it has joined, and closes the connection to the router.

func (*Client) Connected ¶

func (c *Client) Connected() bool

Connected returns true if the client is still connected to (receiving from) the router.

func (*Client) Done ¶

func (c *Client) Done() <-chan struct{}

Done returns a channel that signals when the client is no longer connected to a router and has shutdown.

func (*Client) HasFeature ¶

func (c *Client) HasFeature(role, feature string) bool

HasFeature returns true if the session has the specified feature for the specified role.

func (*Client) ID ¶

func (c *Client) ID() wamp.ID

ID returns the client's session ID which is assigned after attaching to a router and joining a realm.

func (*Client) Logger ¶

func (c *Client) Logger() stdlog.StdLog

Logger returns the clients logger that was provided by Config when the client was created, or the stdout logger if one was not provided in Config.

func (*Client) Publish ¶

func (c *Client) Publish(topic string, options wamp.Dict, args wamp.List, kwargs wamp.Dict) error

Publish publishes an EVENT to all subscribed clients.

Publish Options ¶

To receive a PUBLISHED response set:

options["acknowledge"] = true

To request subscriber blacklisting by subscriber, authid, or authrole, set:

options["exclude"] = [subscriberID, ...]
options["exclude_authid"] = ["authid", ..]
options["exclude_authrole"] = ["authrole", ..]

To request subscriber whitelisting by subscriber, authid, or authrole, set:

options["eligible"] = [subscriberID, ...]
options["eligible_authid"] = ["authid", ..]
options["eligible_authrole"] = ["authrole", ..]

When connecting to a nexus router, blacklisting and whitelisting can be used with any attribute assigned to the subscriber session, by setting:

options["exclude_xxx"] = [val1, val2, ..]

and

options["eligible_xxx"] = [val1, val2, ..]

where xxx is the name of any session attribute, typically supplied with the HELLO message.

To receive a published event, the subscriber session must not have any values that appear in a blacklist, and must have a value from each whitelist.

To request that this publisher's identity is disclosed to subscribers, set:

options["disclose_me"] = true

NOTE: Use consts defined in wamp/options.go instead of raw strings.

func (*Client) RealmDetails ¶

func (c *Client) RealmDetails() wamp.Dict

RealmDetails returns the realm information received in the WELCOME message.

func (*Client) Register ¶

func (c *Client) Register(procedure string, fn InvocationHandler, options wamp.Dict) error

Register registers the client to handle invocations of the specified procedure. The InvocationHandler is set to be called for each procedure call received.

If the registration handler wants to cancel the call without returning a result, then it should return InvocationCanceled.

Register Options ¶

To request a pattern-based registration set:

options["match"] = "prefix" or "wildcard"

To request a shared registration pattern set:

options["invoke"] = "single", "roundrobin", "random", "first", "last"

To request that caller identification is disclosed to this callee, set:

options["disclose_caller"] = true

NOTE: Use consts defined in wamp/options.go instead of raw strings.

// Configure and connect callee client.
ctx := context.Background()
cfg := Config{Realm: "nexus.realm1"}
callee, err := ConnectNet(ctx, "tcp://localhost:8080/", cfg)
if err != nil {
	log.Fatal(err)
}
defer callee.Close()

// Define function that is called to perform remote procedure.
sum := func(ctx context.Context, inv *wamp.Invocation) InvokeResult {
	var sum int64
	for _, arg := range inv.Arguments {
		n, _ := wamp.AsInt64(arg)
		sum += n
	}
	return InvokeResult{Args: wamp.List{sum}}
}

// Register procedure "sum"
if err = callee.Register("sum", sum, nil); err != nil {
	log.Fatal("Failed to register procedure:", err)
}

// Keep handling remote procedure calls until router exits.
<-callee.Done()

Output:

// Configure and connect callee client.
ctx := context.Background()
cfg := Config{Realm: "nexus.realm1"}
callee, err := ConnectNet(ctx, "tcp://localhost:8080/", cfg)
if err != nil {
	log.Fatal(err)
}
defer callee.Close()

// Define function that is called to perform remote procedure.
sendData := func(ctx context.Context, inv *wamp.Invocation) InvokeResult {
	// Get update interval from caller.
	interval, _ := wamp.AsInt64(inv.Arguments[0])

	// Send simulated progress every interval.
	ticker := time.NewTicker(time.Duration(interval))
	defer ticker.Stop()
	for percentDone := 0; percentDone < 100; {
		<-ticker.C
		percentDone += 20
		if e := callee.SendProgress(ctx, wamp.List{percentDone}, nil); e != nil {
			// If send failed, return error saying the call is canceled.
			return InvocationCanceled
		}
	}
	// Send true as final result.
	return InvokeResult{Args: wamp.List{true}}
}

// Register example procedure.
if err = callee.Register("system_test", sendData, nil); err != nil {
	log.Fatal("Failed to register procedure:", err)
}

// Keep handling remote procedure calls until router exits.
<-callee.Done()

Output:

func (*Client) RegistrationID ¶

func (c *Client) RegistrationID(procedure string) (regID wamp.ID, ok bool)

RegistrationID returns the registration ID for the specified procedure. If the client is not registered for the procedure, then returns false for second boolean return value.

func (*Client) RouterGoodbye ¶

func (c *Client) RouterGoodbye() *wamp.Goodbye

RouterGoodbye returns the GOODBYE message received from the router, if one was received. The client must be disconnected from the router first, so first check that the channel returned by client.Done() is closed before calling this function.

func (*Client) SendProgress ¶

func (c *Client) SendProgress(ctx context.Context, args wamp.List, kwArgs wamp.Dict) error

SendProgress is used by a Callee client to return progressive RPC results.

IMPORTANT: The context passed into SendProgress MUST be the same context that was passed into the invocation handler. This context is responsible for associating progressive results with the call in progress.

// Configure and connect callee client.
ctx := context.Background()
cfg := Config{Realm: "nexus.realm1"}
callee, err := ConnectNet(ctx, "ws://localhost:8080/", cfg)
if err != nil {
	log.Fatal(err)
}
defer callee.Close()

// Define function that is called to perform remote procedure.
sendData := func(ctx context.Context, inv *wamp.Invocation) InvokeResult {
	// Get update interval from caller.
	interval, _ := wamp.AsInt64(inv.Arguments[0])

	// Send simulated progress every interval.
	ticker := time.NewTicker(time.Duration(interval))
	defer ticker.Stop()
	for percentDone := 0; percentDone < 100; {
		<-ticker.C
		percentDone += 20
		if e := callee.SendProgress(ctx, wamp.List{percentDone}, nil); e != nil {
			// If send failed, return error saying the call is canceled.
			return InvocationCanceled
		}
	}
	// Send true as final result.
	return InvokeResult{Args: wamp.List{true}}
}

// Register example procedure.
if err = callee.Register("system_test", sendData, nil); err != nil {
	log.Fatal("Failed to register procedure:", err)
}

// Keep handling remote procedure calls until router exits.
<-callee.Done()

Output:

func (*Client) SetCallCancelMode ¶

func (c *Client) SetCallCancelMode(cancelMode string) error

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".

func (*Client) Subscribe ¶

func (c *Client) Subscribe(topic string, fn EventHandler, options wamp.Dict) error

Subscribe subscribes the client to the specified topic or topic pattern.

The specified EventHandler is registered to be called every time an event is received for the topic. The subscription can specify an exact event URI to match, or it can specify a URI pattern to match multiple events for the same handler by specifying the pattern type in options.

Subscribe Options ¶

To request a pattern-based subscription set:

options["match"] = "prefix" or "wildcard"

NOTE: Use consts defined in wamp/options.go instead of raw strings.

const (
	site1DbErrors  = "site1.db.errors"
	site2DbAny     = "site2.db."     // prefix match
	site1AnyAlerts = "site1..alerts" // wildcard match
)

// Configure and connect subscriber client.
ctx := context.Background()
cfg := Config{Realm: "nexus.realm1"}
subscriber, err := ConnectNet(ctx, "ws://localhost:8080/", cfg)
if err != nil {
	log.Fatal(err)
}
defer subscriber.Close()

// Define specific event handler.
handler := func(event *wamp.Event) {
	eventData, _ := wamp.AsString(event.Arguments[0])
	log.Printf("Event data for topic %s: %s", site1DbErrors, eventData)
}

// Subscribe to event.
err = subscriber.Subscribe(site1DbErrors, handler, nil)
if err != nil {
	log.Fatal("subscribe error:", err)
}

// Define pattern-based event handler.
patHandler := func(event *wamp.Event) {
	// Events received for pattern-based (prefix and wildcard)
	// subscriptions contain the matched topic in the details.
	eventTopic, _ := wamp.AsURI(event.Details["topic"])
	eventData, _ := wamp.AsString(event.Arguments[0])
	log.Printf("Event data for topic %s: %s", eventTopic, eventData)
}

// Subscribe to event with prefix match.
options := wamp.Dict{"match": "prefix"}
err = subscriber.Subscribe(site2DbAny, patHandler, options)
if err != nil {
	log.Fatal("subscribe error:", err)
}

// Subscribe to event with wildcard match.
options["match"] = "wildcard"
err = subscriber.Subscribe(site1AnyAlerts, patHandler, options)
if err != nil {
	log.Fatal("subscribe error:", err)
}

// Keep handling events until router exits.
<-subscriber.Done()

Output:

func (*Client) SubscribeChan ¶

func (c *Client) SubscribeChan(topic string, events chan<- *wamp.Event, options wamp.Dict) error

SubscribeChan subscribes the client to the specified topic or topic pattern. Events are written to the provided channel.

No other incoming messages can be processed while blocked on writing events to this channel. When waiting for a message that is not delivered to this channel, that message will never be seen if a previously received event cannot be written to this channel.

func (*Client) SubscriptionID ¶

func (c *Client) SubscriptionID(topic string) (subID wamp.ID, ok bool)

SubscriptionID returns the subscription ID for the specified topic. If the client does not have an active subscription to the topic, then returns false for second boolean return value.

func (*Client) Unregister ¶

func (c *Client) Unregister(procedure string) error

Unregister removes the registration of a procedure from the router.

func (*Client) Unsubscribe ¶

func (c *Client) Unsubscribe(topic string) error

Unsubscribe removes the registered EventHandler from the topic.

func (RPCError) Error ¶

func (rpce RPCError) Error() string

Error implements the error interface, returning an error string for the RPCError.