Implementing Streams via Progressive Call Results/Invocations

[ecorm]

Here is a collection of notes from implementing a streaming API in CppWAMP via the Progressive Call Results/Invocations feature. I'm posting this in the hopes that other implementations would implement a compatible streaming API in their libraries.

• Data in streams is sent/received via Chunks.
• From the user's perspective, a stream Chunk consists of an [args, kwargs] payload, as well as a boolean flag indicating if the chunk is a final one.
• A callee can register a Stream endpoint associated with an URI, via a REGISTER message. Stream URIs share the same namespace as procedure URIs.
• A caller uses an RPC to send an Invitation to open a Channel for a new stream associated with an URI. This operation is what's referred to as "signalling" in the literature (such as in WebRTC). The invitation is sent to a callee via a CALL message. Note that I don't like to use term "signal" in this context (especially in code), because I immediately think of POSIX signals when I see that term.
• A Channel has an ephemeral integer ID which is identical to the request id of the invitation. Channel ids therefore share the same number space as other WAMP requests.
• The invitation contains an optional [args, kwargs] payload, as well as a desired Streaming Mode:

Streaming Mode	CALL.Options.progress	CALL.Options.receive_progress	CALL Count	RESULT Count
simple_call*	false	false	1	1
caller_to_callee	false	true	1	1 or more
callee_to_caller	true	false	1 or more	1
bidirectional	true	true	1 or more	1 or more

* Note that when both progressive results and invocations are disabled, this decays into a "simple" remote procedure call with a single invocation and a single result.

• A callee receives a streaming invitation (i.e. a signal) via an INVOCATION message. It deduces the desired streaming mode via the INVOCATION.Details.progress and INVOCATION.Details.receive_progress options.
• A callee can interpret the payload of the invitation as either:
  • metadata about the invitation, or,
  • the first Chunk of incoming data.
• If the mode is callee_to_caller or bidirectional, a callee may explicitly accept the invitation by sending an RSVP via an initial progressive YIELD. Otherwise, the invitation is assumed as accepted if the callee does not return an ERROR.
• The caller and callee must have a mutual, application-level understanding concerning the interpretation of the initial payloads as invitations and RSVPs. NOTE: The WAMP protocol could help here by defining corresponding CALL and YIELD options for this purpose.
• A callee may reject an invitation by returning an ERROR message.
• If the streaming mode is caller_to_callee or bidirectional, after sending an invitation, the caller may send chunks to the callee via progressive CALL messages. The stream is closed from the caller's perspective upon sending a final chunk with CALL.Options.progress=false.
• If the streaming mode is callee_to_caller or bidirectional, the callee may send intermediary chunks to the caller via progressive YIELD messages. The stream is closed from the callee's perspective upon sending a final chunk with YIELD.Details.progress=false.
• A callee may send an ERROR message to close the stream and report back an error condition.
• A caller may send a CANCEL message if it needs to close the stream before receiving the final chunk from the callee.
• In bidirectional mode, a caller can communicate its desire to end the stream gracefully (without CANCEL) via chunk payload arguments. It is up to the application to define the payload schema which allows this intent to be communicated. NOTE: The WAMP protocol could help here by defining a CALL option for this purpose. This would allow client implementations to perform this "graceful closing" behavior.
• In all streaming modes, a callee must always end up yielding a final chunk or an ERROR (except for cancellation with the killnowait mode, where the callee is not required to yield an ERROR.
• Throttling is done at the application level. It is up to the application to define the payload schema which allows throttling directives to be communicated.

The above allows for interoperability with client implementations that do not "understand" the streaming concept. The above streaming idea can be done with the existing progressive results/invocations features, but adding the following to the WAMP protocol would ease implementation:

• CALL.Options.is_invitation|bool as a hint for the callee to interpret the first progressive invocation as a stream invitation (i.e. stream signal).
• CALL.Options.expects_rsvp|bool as a directive for the callee send an RSVP if it accepts the streaming request.
• CALL.Options.stop|bool to communicate the caller's desire to end the stream gracefully in bidirectional mode.
• YIELD.Details.is_rsvp as a hint for the caller to interpret the first progressive result as an RSVP to the invitation.

I will add sequence diagrams later for each streaming mode, including the invitation/RSVP exchanges. I will also add a suggested streaming API in pseudocode.

[oberstet]

just one quick note

and the GitHub discussions feature is disabled for this repo

I think I've enabled it #464 .. not sure, but it should work to start new discussions

[Jopie64]

The above streaming idea can be done with the existing progressive results/invocations features, but adding the following to the WAMP protocol would ease implementation:

• CALL.Options.is_invitation|bool as a hint for the callee to interpret the first progressive invocation as a stream invitation (i.e. stream signal).
• CALL.Options.expects_rsvp|bool as a directive for the callee send an RSVP if it accepts the streaming request.
• CALL.Options.stop|bool to communicate the caller's desire to end the stream gracefully in bidirectional mode.
• YIELD.Details.is_rsvp as a hint for the caller to interpret the first progressive result as an RSVP to the invitation.

Hmmm... Reading this actually makes me more convinced that this indeed is all application (or library) specific signalling and leads me to believe that this should not be specified on WAMP level after all.

Both sides need to have a mutual understanding what's going on there. Also, as how WAMP progressive calls is currently defined, you can specify even another application (or library) on top of this progressive calls which have different requirements and signalling. All that signalling doesn't have any effect on WAMP level... At least, I think it shouldn't, so to keep the progressive call feature more generic.

That makes me feel it can be seen as a protocol that can be defined as an extra layer on top of WAMP.
The upside of specifying it like that is that you can also implement or specify the same protocol on raw web sockets or GRPC (protobuf) or whatever raw bidirectional message based transport.

Instead of putting all those options in the WAMP specific options dict, couldn't you define or specify those options to be in the application specific kwargs field for this specific streaming protocol?

Or do you have another reason to define all that in the WAMP options dict?

Also I was thinking, maybe add an application specific section in the WAMP options dict? Like X_stuff headers in http, but then as just one application specific multipurpose node in options dict. Or is the kwargs dict enough already?

[Jopie64]

I do like the idea for this application/use case btw. I just think WAMP features should not aim for a specific use case, but should stay generic to support other use cases as well.

[oberstet]

yes, I agree with @Jopie64, this is application specific and shouldn't be defined in the WAMP protocol itself, but simply proprietary to the application.

[ecorm]

Instead of putting all those options in the WAMP specific options dict, couldn't you define or specify those options to be in the application specific kwargs field for this specific streaming protocol?

I don't want my CppWAMP client library (which is not application specific) to clobber the kwargs payload and remove/hinder the possibility of the user making use of the kwargs for their own purpose.

Anyway, I can make it work without those extra WAMP options, it's just they would have been "nice to have".

[Jopie64]

Streaming like that may not be application specific, but it is use case specific.
As said you could implement the streaming abstraction on top of other transport layers. And also tbh I think you should be able to use CppWAMP for other purposes then just streaming. For ease of use it could expose a WAMP implementation of the streaming abstraction of course. But of course it's up to you how you want to build your own lib 😜

[ecorm]

Of course, you can still use regular RPCs and sub/pub in CppWAMP. The streaming API, like you said, is an abstraction of progressive results/invocations, but it doesn't add a layer on top of it. Everything you could do before with "raw" progressive results/invocations you can still do with the streaming interface.

[ecorm]

Also I was thinking, maybe add an application specific section in the WAMP options dict? Like X_stuff headers in http, but then as just one application specific multipurpose node in options dict. Or is the kwargs dict enough already?

The kwargs dict is intended for the user payload. A general-purpose WAMP client library cannot add items to the kwargs dict without possibly breaking user apps.

[Jopie64]

The kwargs dict is intended for the user payload. A general-purpose WAMP client library cannot add items to the kwargs dict without possibly breaking user apps.

But if the streaming API is a specification built on top of WAMP (but is not part of WAMP), it can use the kwargs right? Since then kwargs and even args become part of the streaming API layer and are abstracted away for the application using streaming.

I just think streaming is one of the specific use cases for the more general purpose progressive call feature, even when streaming itself is also generic. Other APIs can be built on the progressive call feature that can not necessarily be built with (or need the extra features of) the streaming API.

[ecorm]

As I wrote in your other reply, the streaming API I propose doesn't need to add a layer on top of WAMP, but rather provides an abstraction for the progressive results/invocations feature that is (hopefully) more user-friendly.

If it were to add another layer, then it could use the WAMP payload args/kwargs however it sees fit, and bundle the application's args/kwargs within that outer payload. Just like other layered protocols with [lower_layer_header, [upper_layer_header, payload]].

[ecorm]

Anyway, I see now that nobody is fond of the idea of adding these extra "hint" options, which is fine by me (and perhaps for the better). This just means the user will have to make extra sure both ends of the stream behave in a mutually understandable way, and that the WAMP client library can't do much about it.

[oberstet]

Anyway, I see now that nobody is fond of the idea of adding these extra "hint" options

it seems so, but let me still give you a bit more feedback which might be useful (or not;) to you:

• I think what this aims for is add specific "signaling" for "full duplex channels" based on WAMP Progressive Invocations + Results
• of course that is useful, but quite "application specific", and hence I don't think it should be in the WAMP spec
• rgd your practical itch of needing specific options for above without touching args/kwargs (keep those freely usable at the app level) makes sense, but
• can be done using custom x_ attributes Custom message attributes #345
• AND, even that is then only needed if you want to add the signaling in-band (to the one RPC also then carrying the streaming channel)
• an alternative is to have signaling setup first using a separate RPC which then returns a reference to a set up streaming RPC used for the channel
• both design choices (in-band vs out-of-band signaling) might make sense, I haven't thought much about tradeoffs (and there will be such)

anyways, just my 2cts;)

[KSDaemon]

Quick note (as i'm afk):
We discussed custom options feature. It is exactly for that purpose.

[oberstet]

yes, this #345 - missing in the spec currently