Current review

[oberstet]

This is a meta issue to collect some remarks to the current state of the implementation.

CC @om26er

• on-ready
• request IDs
• session state
• call implementation. part 1
• call implementation. part 2
• completable future usage
• state machine transitions and session lifecycle

on-ready

When the session receives a WELCOME message, it correctly notifies all "on-join" listeners, and the user code for each of the listeners will run (from here):

mOnJoinListeners.forEach(onJoinListener -> onJoinListener.onJoin(details));

However, these notification handlers may return futures (or regular results), and we want to collect all these and run some code when all the returned futures have triggered.

The code that we want to run is notifying all "on-ready" listeners and run the user code attached to these.

[oberstet]

request IDs

The request IDs for outgoing messages are currently generated using a random number generator

long requestID = getRandomNumber();

This is in conflict with the latest WAMP spec, which is followed by AutobahnPython (but not yet AutobahnJS!). We should follow the spec in AutobahnJava, because the change was introduced for a couple of subtle reasons. Means, believe me, we want this;)

Here is what AutobahnPython does:

from autobahn.util import public, IdGenerator, ObservableMixin

self._request_id_gen = IdGenerator()

request_id = self._request_id_gen.next()

Eg here and here.

The request ID generator generates ..

WAMP request IDs are sequential per WAMP session, starting at 1 and
wrapping around at 253 (both value are inclusive [1, 253]).
The upper bound 253** is chosen since it is the maximum integer that can be
represented as a IEEE double such that all smaller integers are representable as well.
Hence, IDs can be safely used with languages that use IEEE double as their
main (or only) number type (JavaScript, Lua, etc).

[oberstet]

session state

The session object needs to keep track of it's state, and it needs to store detail information about requests it has sent (and is hence expecting to receive some message for) - at least in some of these states.

So for tracking state, session could have a member attribute

private int state;

with a couple of allowed values like 1 = HELLO-SENT, ... = JOINED

Then session needs to keep track of requests:

        # outstanding requests
        self._publish_reqs = {}
        self._subscribe_reqs = {}
        self._unsubscribe_reqs = {}
        self._call_reqs = {}
        self._register_reqs = {}
        self._unregister_reqs = {}

and it needs to keep track of subscriptions and registrations it has set up, as well as invocations it has to answer to:

       # subscriptions in place
        self._subscriptions = {}

        # registrations in place
        self._registrations = {}

        # incoming invocations
       self._invocations = {}

[oberstet]

call implementation: part 1

My recommendation would be to start with the call pattern. So this implementation already does a couple of necessary things:

• create a future to be return to user code
• generate a next request id
• send out a CALL message

What it also needs to do is:

• remember the future returned in a "call requests" map indexed by request id
• probably remember more about the call, and hence probably the need for a CallRequest class
• forward the call options so they get included in the CALL message
• safe guard against "no transport attached" (because then we obviously cannot do a call)

Then we need to process incoming CALL_RESULT and ERROR messages, but this is part 2.

[oberstet]

call implementation: part 2

The second part of implementing calls in session needs to process the incoming CALL_RESULT and process incoming ERROR messages.

For processing incoming CALL_RESULTs, the session object needs to ITransportHandler.onMessage message switch extended, and:

Leaving progressive calls and end-to-end encryption out for the moment (both are advanced, optional features), then essentially the code needs to lookup the original call request by request ID (in the map), and then fire the future store therein with an instance of CallResult. Similar for ERROR messages. Lookup original call request to get at the future, and the fire the future with the ERROR instead (errback the future).

[oberstet]

executor service

The executor service that is doing the background magic to keep the completable future illusion rolling corresponds to the Twisted reactor or asyncio event loop thing. It is the heart of the clock work.

The Client instance likely needs access to that executor service, and hence we should add a parameter to the Client constructor that takes an executor.

Here is a rough sketch of the idea:

private void test() {

    ExecutorService executor = Executors.newSingleThreadExecutor();

    Client client(executor, ...);
    ...
}

public static void main(String[] args) {
    App app = new App();
    app.test();
}

I think the single thread executor is what we are looking for the demos/PoC.

CompletableFuture usage

Actually use these;)

References

• http://www.baeldung.com/java-completablefuture
• http://codingjunkie.net/completable-futures-part1/
• http://blog.indrek.io/articles/java-8-completablefuture/
• https://docs.oracle.com/javase/8/docs/api/java/util/concurrent/CompletableFuture.html
• https://stackoverflow.com/questions/36981432/why-completablefuture-s-thenaccept-not-running-on-the-main-thread
• https://headcrashing.wordpress.com/2015/07/20/iobound-completablefuture/

[oberstet]

state machine transitions and session lifecycle

The states that session needs to keep track of are described on an abstract level (finite state machine, FSM) in the WAMP spec here: Session Statechart.

This state diagram not only show the states a session can be in, but also the possible state transitions, as well as the messages sent or received when transitioning states.

This is what we should implement ultimately - I am not saying that we need the full blown thing right away;) But I think it is good to have the full picture in mind right from the beginning.

Notifying of observers on the session lifecycle events happens as part of state transitions.

Theoretically, this state machinery could be implemented on top of a generic finite-state-machine (FSM) library, but I'm not sure it is worth, so what we practically do here is implementing an embedded, ad-hoc state machine within session.

[oberstet]

To summarize, session is really independent in its lifecycle. From transports, and from anything else.

This will allow us to do things like the following in a future release.

Let Session buffer up published events internally when currently no transport is attached. Later, when a transport does become attached, then session will rejoin or resume to a realm (short-cutting on the initial WAMP opening handshake by providing a resume ticket in the WAMP HELLO message), and then catch up with actual publishing the buffered events.

So this is kinda heavy on the WAMP client library side: it needs to buffer the events. Probably even persistently to local device storage. But it would allow a new level of QoS publishing.

In an extreme form, the client will first persist an event published with such a QoS level to persistent memory, and then publish the event over the network. If the network disappear, the event will be in the persistent buffer for republishing. Probably even with a 2-PC exchange to make sure to avoid dups at the same time.

[om26er]

However, these notification handlers may return futures (or regular results), and we want to collect all these and run some code when all the returned futures have triggered.

The code that we want to run is notifying all "on-ready" listeners and run the user code attached to these.

In ABJ implementation we don't have onReady observer, shall we add that ?

[om26er]

on-ready

When the session receives a WELCOME message, it correctly notifies all "on-join" listeners, and the user code for each of the listeners will run (from here):

mOnJoinListeners.forEach(onJoinListener -> onJoinListener.onJoin(details));
However, these notification handlers may return futures (or regular results), and we want to collect all these and run some code when all the returned futures have triggered.

The code that we want to run is notifying all "on-ready" listeners and run the user code attached to these.

Done, please check the commit here: 2d3a6d1

request IDs

Fixed, already merged to master.

session state

Partially implementation landed in master a few hours ago.

call implementation: part 1

My recommendation would be to start with the call pattern. So this implementation already does a couple of necessary things:

create a future to be return to user code
generate a next request id
send out a CALL message
What it also needs to do is:

remember the future returned in a "call requests" map indexed by request id
probably remember more about the call, and hence probably the need for a CallRequest class
forward the call options so they get included in the CALL message
safe guard against "no transport attached" (because then we obviously cannot do a call)
Then we need to process incoming CALL_RESULT and ERROR messages, but this is part 2.

That is done. Some functionality landed in master, others are proposed here: #186

call implementation: part 2

The second part of implementing calls in session needs to process the incoming CALL_RESULT and process incoming ERROR messages.

For processing incoming CALL_RESULTs, the session object needs to ITransportHandler.onMessage message switch extended, and:

Leaving progressive calls and end-to-end encryption out for the moment (both are advanced, optional features), then essentially the code needs to lookup the original call request by request ID (in the map), and then fire the future store therein with an instance of CallResult. Similar for ERROR messages. Lookup original call request to get at the future, and the fire the future with the ERROR instead (errback the future).

Call Result processing is already in master, error processing still needs to be implemented.

[om26er]

executor service

The executor service that is doing the background magic to keep the completable future illusion rolling corresponds to the Twisted reactor or asyncio event loop thing. It is the heart of the clock work.

The Client instance likely needs access to that executor service, and hence we should add a parameter to the Client constructor that takes an executor.

Here is a rough sketch of the idea:

private void test() {

ExecutorService executor = Executors.newSingleThreadExecutor();

Client client(executor, ...);
...

}

public static void main(String[] args) {
App app = new App();
app.test();
}
I think the single thread executor is what we are looking for the demos/PoC.

I think we need some kind of illustration on how that executor is passed on to the Session as that's going to be the one to actually call pub/sub register/call etc and uses CompleteableFuture. Calling the executor to each method call might not be fun, but I am not sure.

[oberstet]

@om26er actually, I think this issue is pretty much "done" in as all of the items of the list at the top are implemented. Awesome!! Great work.

I will file follow up issues on the remaining open things ..

[oberstet]

In ABJ implementation we don't have onReady observer, shall we add that ?

Yep, pls see #188

The on-ready observers are notified after all the on-join observer are finished (means, the futures returned from calling them have triggered).

[oberstet]

So in summary, the follow up issues are:

• ExecutorService based example #189
• on-ready observer #188
• Call ERROR message processing #187