New ABJS / WAMP client library feature: Registration/subscription booking and auto-processing

[markope]

When working with the autoban-js client (and other clients) one of the major headaches was the management of session in my experience. In particular those parts of the code, that used a session to publish or call would need to be notified in case there was a new session. IMO sessions are an implementation detail from the user space point of view. i.e. I would prefer to have aWAMPConnection class, that encapsulates session management. Users would publish / subscribe / call / register on the connection, not on the session.

This would also make it much easier to implement functions like autoresubscribe in case a session needed to be swapped for a new one.

Here is a draft version of this to get a look and feel:

import autobahn from "autobahn"
import type { IPublication, IRegistration, RegisterEndpoint, SubscribeHandler } from "autobahn"

export class WAMPConnection {
    private session?: autobahn.Session
    private aconn?: autobahn.Connection
    private connectionOptions?: autobahn.IConnectionOptions
    private subscriptions: autobahn.Subscription[] = []
    private registrations: autobahn.Registration[] = []
    private firstConnection: Promise<void>
    private firstResolver: (value?: void | PromiseLike<void>) => void = () => undefined
    private firstRejecter: (reason?: any) => void = () => undefined
    private connectionDropWaitTime: number = 2000
    realm?: string

    constructor(config: IConnectionOptions) {
        this.firstConnection = new Promise((resolve, reject) => {
            this.firstResolver = resolve
            this.firstRejecter = reject
        })
        const defaultOptions = {
            realm: CBREALM,
            url: CROSSBAR_URI,
            use_es6_promises: true,
            authmethods: ['wampcra'],
            authid: 'system',
            max_retries: -1,
            max_retry_delay: 1,
            initial_retry_delay: 1,
            autoping_interval: 2,
            autoping_timeout: 10,
            onchallenge: (session, method, extra) => {
                return autobahn.auth_cra.sign(SYSTEM_SECRET, extra.challenge)
            },
        } as IConnectionOptions

        this.connectionOptions = {...defaultOptions, ...config}
        console.log('configured', this.connectionOptions)
    }

    private onOpen = (session: autobahn.Session) => {
        this.session = session
        this.resubscribeAll()
        console.log(`Connection to realm '${session.realm}' established`)
        this.firstResolver()
        this.firstResolver = () => undefined
    }

    private onClose = (reason: string, details: any): boolean => {
        this.session = undefined
        this.firstRejecter()
        this.firstRejecter = () => undefined
        console.warn(`Connection to realm ${CBREALM} closed: ${reason}`, details)
        return false // false for auto reconnect
    }

    private liveCheck = () => {
        while (true) {
            if (this.session) {
                this.session?.call('my.backend.health.endpoint')
                    .then(() => console.log('Live check successful'))
                    .catch(() => {
                        console.error('Live check failed')
                        this.session = undefined
                    })
            }
            setTimeout(() => this.liveCheck(), 30000)
        }
    }

    private sessionWait = async () => {
        const startTime = new Date().getTime()
        while (!this.session) {
            if (new Date().getTime() - startTime < this.connectionDropWaitTime)
                throw new Error("Timeout waiting for session")
            await new Promise(resolve => setTimeout(resolve, 200))
        }        
    }

    get isOpen() {
        return !!this.session?.isOpen
    }

    start = async () => {
        if (!this.aconn) {
              this.aconn = new autobahn.Connection(this.connectionOptions)
              this.aconn.onopen = this.onOpen
              this.aconn.onclose = this.onClose
        }
        this.aconn.open()
        await this.firstConnection
    }

    stop = () => {
        this.aconn.close('wamp.close.normal', 'Connection closed by user')
    }

    call = async (method: string, args?: any[], kwargs?: any, options?: autobahn.ICallOptions): Promise<any> => {
        await this.sessionWait()
        const result = await this.session?.call(method, args, kwargs)
        return result
    }

    subscribe = async (topic: string, handler: SubscribeHandler) => {
        await this.sessionWait()
        const sub = await this.session?.subscribe(topic, handler)
        if (sub) this.subscriptions.push(sub)
    }

    publish = async (topic: string, args: any[], kwargs: any): Promise<IPublication | undefined> => {
        await this.sessionWait()
        const res = await this.session?.publish(topic, args, kwargs)
        return res
    }

    unsubscribe = async (topic: string, func?: (...args: any[]) => any) => {
        await this.sessionWait()
        if (func === undefined) {
            return this.unsubscribeTopic(topic)
        }

        for (let i = 0; i < this.subscriptions.length; i++) {
                
            const sub = this.subscriptions[i]
            if (sub.topic === topic && sub.handler === func) {
                try {
                    await this.session?.unsubscribe(sub)
                    this.subscriptions.splice(i, 1)
                    i--
                } catch (error) {
                    console.warn("Failed to unsubscribe", sub.topic, sub.handler, error)
                }
            }
        }
    }

    unsubscribeTopic = async (topic: string) => {
        const subList = this.subscriptions.filter(sub => sub.topic === topic)
        if(!subList.length) { return true }
        const promises = subList.filter(sub => sub.handler).map(sub => this.unsubscribe(sub.topic, sub.handler))
        return Promise.all(promises)
    }

    register = async (method: string, endpoint: RegisterEndpoint): Promise<IRegistration | undefined> => {
        await this.sessionWait()
        const registration = await this.session?.register(method, endpoint)
        if (registration) this.registrations.push(registration)
        return registration
    }

    unregister = async (registration: autobahn.Registration) => {
        await this.sessionWait()
        await this.session?.unregister(registration)
        if (registration) {
            const index = this.registrations.indexOf(registration);
            if (index !== -1) {
                this.registrations.splice(index, 1);
            }
        }
    }

    resubscribeAll = async () => {
        const subscriptionsCopy = [...this.subscriptions]
        this.subscriptions = []
        const promises = subscriptionsCopy.map(sub => this.subscribe(sub.topic, sub.handler))
        await Promise.all(promises)
    }

}

A connection can then be consumed by other modules of your app like this:

const conn = new WAMPConnection({... some config ...})
const myChat = new MyChat(conn)
conn.start()

Nice side effect here is, you can deal out the connection before the actual WAMP connection has been established.

inside of myChat you can then simply call

const result = await this.connection.call('my.topic', stuff)

and you don't need to care if the session momentarily is been set up or has been refreshed.

[oberstet]

Users would publish / subscribe / call / register on the connection, not on the session.

WAMP does have the concepts of a

1. "WAMP session"
2. "WAMP transport (instance)"

It does not have the concept of a "WAMP connection".

Such a concept would be a WAMP client library thing, not known to other WAMP peers or router(s) or the WAMP specification itself.

Now, assuming you want to have a concept of a "WAMP connection" - in AutobahnJS or any of the other Autobahns

What exactly should it mean, represent and how should it act? How should it related to WAMP sessions and WAMP transport instances?

I don't know answers that would make sense, in fact, I don't see the need nor the aim of a concept of "WAMP connections".

A procedure registered can only exist in the context of a WAMP session. Otherwise there is no Callee to call, and hence what should a "registration" mean?

Note that the definition of the types of WAMP procedure arguments of a procedure (in its abstract form) is of course interesting, and Crossbar.io and Autobahn support both "typed WAMP" (via WAMP Flatbuffers) as well as specifying schemata for that -- via respective meta APIs ..

[oberstet]

one more note rgd "communication of commitment": such an unconditional commitment can be looked at as a data offer by a data provider in XBR terms where the price for the data is zero, and the data offer of course then is available at run-time via XBR APIs.

offers are placed by data providers (callees) in a component running in Crossbar.io called the XBR market maker then

such call can be placed via Crossbar.io to the right place .. the market maker entry point for submission of data offers -- priced or not -- here

https://github.com/crossbario/crossbar/blob/0089c1ef6fbbb87fc7316088a91f1859fa84eeb0/crossbar/edge/worker/xbr/_marketmaker.py#L698

[oberstet]

Yes, and thats the 99% case IMHO.

ok, got you!

this is called "placing a data offer" in XBR terms, and it is done by an XBR data seller - whether the price is zero or not - by calling xbr.marketmaker.place_offer

it's implemented in Crossbar.io and Autobahn, e.g. here is how this is done from the client side .. in a simple Python data seller (== provider, == callee):

https://github.com/crossbario/autobahn-python/blob/f38f16ba28fa253dee951068cc729089c88d857d/autobahn/xbr/_seller.py#L319

besides, there is also lots of end-to-end encryption related code there .. sorry .. this is quite complex stuff .. the whole function I got my example above from actually is the Python client library side internals which will call into the XBR market maker automatically -- and periodically whenever new E2E keys are generated ... as in

Add a new (rotating) private encryption key for encrypting data on the given API.

[markope]

ok interesting, I didn't now that. But it is not exactly what I mean. So there is a way to communicate a commitment to crossbar.
If I'm offering such a commitment as a client, then I still have to make sure I have my registrations up at all time. The wamp router can not help me here. This is client side business. And that's what I want Wampy to handle for me inside my client code.

[oberstet]

I would maybe put it like: when a client makes a commitment and announces that via xbr.marketmaker.place_offer, that client is responsible for issuing WAMP procedure registrations required by the client to provide its committed services.

The code in the client that issues sending of WAMP procedure registrations can be in WAMP client library (such as AutobahnJS), or it can be at the user code level (such as JavaScript and some JS VM run-time).

Both are fine for me in general, and I am just trying to figure out what it is exactly we are looking for=)

[markope]

ok, I'll try again to describe the problem, that I'm trying to solve with my autobahn-js draft.

My browser app is a pretty big piece of js code with a lot of modules and dependencies. All kinds of components in my codebase want to publish some data or subscribe to smth. It is cumbersome to implement onOpen handlers in each of these components when it's always the same what I need to do:

1. listen to a "new-session" event.
2. resubscribe my subscriptions and registrations
3. use the new session for future calls and publishes.

And these steps (and others) are just boilerplate code, that should be done by Wampy, my WampSessionManager class.

Currently I'm using some code like above to help myself. I just find this so practical and also find this is such a common and generic requirement for the usability of all wamp client libraries. I should not need my wrapper for this. Autobahn-js should offer this out of the box IMO. And it wouldn't even be a breaking change, just add some more functions to the connection class.
I posted this here to discuss and hear the opinions of others.

[oberstet]

Autobahn-js should offer this out of the box IMO. And it wouldn't even be a breaking change, just add some more functions to the connection class.

yeah, gather feedback and comments, exchange ideas/details/views .. would be great=)

one thing to make it more concrete / tangible: come up with even a sketchy, incomplete PR! pls don't hesitate, a PR can be grown step by step, but at least it would e.g. to allow to define "some more functions" ...

[markope]

sure, but I think if I take on this endeavour, I'd like to do a complete rewrite of autobahn-js :)
(es6 + typescript)

[oberstet]

Draft version of a new autobahn-js-2 interface as I would like to use it
...
sure, but I think if I take on this endeavour

so you would like a new interface, but would like to avoid the work;) I mean, coming up with even some code bits which would allow to define and discuss "new interface". not even implement it.

sorry, for me it's just a question of sane and constructive approach to discussion of "the" issue. this issue. because I am not convinced there is even one;) I mean in the sense of defining the issue and goal is a necessary prerequisite of implementing a solution to it.

[markope]

I certainly don't expect anyone else to do it for me! I just want to take it slowly and gather some feedback.

To sum it up, my suggestion is:
Put pub/sub/call/reg functions on the connection, too.

(Maybe I should put this up in the title)
I find it super useful, but if people don't like it, I'll just not do it and live with my wrapper.

[oberstet]

Yeah, I get it, and I think gathering feedback, starting a conversation is super important. However, natural language has limits, and sooner or later inserting code snippets - such as e.g. w.r.t. interface change proposals - would be important .. because at least me I don't actually understand what is being proposed ..

To sum it up, my suggestion is:

I don't understand the suggestion

Put pub/sub/call/reg functions on the connection, too.

There is no such thing as a "connection" in WAMP, there only is "session" and "transport"

[markope]

There is no such thing as a "connection" in WAMP, there only is "session" and "transport"

there is a "connection" in autobahn-js. That's what I'm talking about.

[oberstet]

ahh, right, so your proposal is wrt AutobahnJS - not WAMP or WAMP client libraries in general - and it is about adding new functions to the Connection type there (which is a ABJS artifact)

autobahn-js/packages/autobahn/lib/connection.js

Line 20 in c831f35

var Connection = function (options) {

?

[markope]

exactly :)

[oberstet]

alright! let's try to explore that route.

Currently, unless and until there exists a WAMP transport, and an opened WAMP session running on that transport, Connection._session will be null, and the object of type Connection in that state is pretty useless.

Specifically, any future intent or desire by app code that it wants the Connection to have its session registered or subscribed any procedures or topics in the future, once the session is actually active/open, can not be already booked!

App code must wait until there actually is a transport and session ready before it can register & subscribe.

Of course it is true: even if there were a way to book a future intent or desire to register/subscribe (book within the client only of course), and then e.g. use such register/subscribe booking information to actually register/subscribe later, that is "realize" the previously only booked registrations/subscriptions), will then depend on an actual active/open session ... to realize.

When there is no WAMP session, a previously booked intent to register/subscribe cannot be realized. Simple as that.

Are we on the same page?

[markope]

yes

Of course it is true: even if there were a way to book a future intent or desire to register/subscribe (book within the client only of course), and then e.g. use such register/subscribe booking information to actually register/subscribe later, that is "realize" the previously only booked registrations/subscriptions), will then depend on an actual active/open session ... to realize.

That's my suggestion for reg/sub. I want to register/subscribe on the connection. This effectively means exactly what you describe: "I store a registration/subscription intent at the connection instance."

Similar for pub/call: I give the connection a pub/call intent. This should be immediately executed on the session if it exists, but we could allow for a grace period to send out later it the session comes online within that time.

[oberstet]

Rereading, yeah. Moreover, if we are on the same page, then I would restate your proposal like:

Extend AutobahnJS Connection class with

1. new functions to book a future intent or desire to register/subscribe a procedure/topic.
2. new function or option to automatically register/subscribe all such "booked future intents" once a session has been opened and is active - thus realize the (previously) future intents

In my eyes, such a thing would make sense, is sufficiently defined and could be implemented in ABJS.

And this holds in general, that is, one may say in general:

A WAMP client library should have a way for apps to book future intents of registration/subscription and allow for automatic registration/subscription based on this client-side booking information.

[markope]

True, this is a much clearer description of what I mean :)

[oberstet]

Great!

I would call it: "New ABJS / WAMP client library feature: Registration/subscription booking and auto-processing"

---

One could also imagine:

As an app developer, I want to register/subscribe - whether that is now resulting in a direct active registration/subscription, or later from a booking of the future intent, and realization of that intent once a session actually becomes open/active.

I think it is super important to spend a little time and some iterations and polishing on the user (== app dev) story.

Please comment, help improve, propose, ... we need this 1 sentence right .. this will be the root / anchor of any "New feature" XYZ issue ...

[oberstet]

Some 2nd level design questions:

The set of future intents of registrations/subscriptions must be tracked (booked) client-library side, and one natural option to hold such information could indeed be ABJS Connection objects

A different, much more radical design would factor the whole "intent thing" into a "virtual WAMP session state" or "would-be WAMP session" ... thus, a session could have a new state

With the latter approach, a call to subscribe or register on a session which is not open would no longer throw

autobahn-js/packages/autobahn/lib/session.js

Line 1347 in c831f35

if (!self.isOpen) {

but would submit and book a future intent transparently

a call to call or publish on such a session of course still throws.

we don't want to track intents for future calls or publications, do we? ;)

[markope]

A different, much more radical design would factor the whole "intent thing" into a "virtual WAMP session state" or "would-be WAMP session"

I think this is not a good idea because. If a session gets dropped and a new session is created it will have a different session id and it "physically" is a different session for the router. Then we would have to migrate registration intents to the new session. Sounds a bit over engineered to me.
Also I find the "Connection" is something the router needs not to now about. As such I would keep it a pure client side concern.

we don't want to track intents for future calls or publications, do we? ;)

Well, kind of yes! The main reason behind a "call intent" is to avoid notifying the whole user-code about the opening of a session with some custom callbacks or event mechanisms.

If the session is open, then the call-intent should be executed immediately on this session otherwise an error is thrown. This is not the same as directly calling on the session (!!), because the connection can swap the session without notifying user-code and the call-intent will still work.

[oberstet]

IOW: with intent booking, does it even make sense to book an intent rather than just execute in case the session currently is open / active? Can I book without execution (now)?

[markope]

The intent should still be kept in a RegistrationIntentList even if it is immediately executed to prepare for reregistrations.

Can I book without execution (now)?

IMO if I create a registration intent I really want it registered whenever as possible. This means registration intents should be always executed immediately. I think creating an "intent" that I don't want executed is not an "intent". (Picking words, but still)

Maybe for this reason it would be wise to choose different names for those functions on the connection to avoid confusion:

connection.createRegistrationIntent() instead of connection.register
...

hm, somehow I still like the shorter version on the right better. It's clearer more usable. The connection prefix already makes the "intent" clear enough maybe.

[oberstet]

Then we would have to migrate registration intents to the new session. Sounds a bit over engineered to me.
Also I find the "Connection" is something the router needs not to now about. As such I would keep it a pure client side concern.

I agree with both arguments.

My main issue is: if I want to be able to have Session.register/subscribe to always succeed, then I need to store the arguments of such a call as an "intent", not as a registration request

autobahn-js/packages/autobahn/lib/session.js

Line 1396 in c831f35

self._register_reqs[request] = [d, procedure, endpoint, options];

and if that storing is supposed to be not on the session object, then it still needs to be somewhere reachable from Session

autobahn-js/packages/autobahn/lib/session.js

Line 220 in c831f35

var Session = function (socket, defer, onchallenge, on_user_error, on_internal_error) {

long ago that I looked so closely at ABJS impl;) anyways, details, we'll get that sorted!

so overall, my gut feeling is, all good & fine, the outline should be a sustainable road to such a new feature ..

[oberstet]

createRegistrationIntent: I think this is one of the more important 2nd'ary design questions.

My thinking naturally would be: submitting an intent of registering a procedure should work "the same" as registering a procedure.

Thus, calling session.register in app code will succeed even when session.isOpen is false.

In such a case an intent will be stored internally. "Somewhere" (tbd).

Then, later, either automatically or manually triggered, that list of intents is executed, that is run on an open/active session.