Using the Decorator approach with AutobahnWS, how to publish messages independent from subscription callbacks and it's Session-Reference?

Question

When working with Autobahn and WAMP before I have been using the Subclassing-Approach but stumbled over decorator / functions approach which I really prefer over subclassing.

However. I have a function that is being called from an external hardware (via callback) and this function needs to publish to Crossbar.io Router whenever it is being called.

This is how I've done this, keeping a reference of the Session right after the on_join -> async def joined(session, details) was called.

from autobahn.asyncio.component import Component
from autobahn.asyncio.component import run

global_session = None

comp = Component(
    transports=u"ws://localhost:8080/ws",
    realm=u"realm1",
)

def callback_from_hardware(msg):
    if global_session is None:
        return
    global_session.publish(u'com.someapp.somechannel', msg)

@comp.on_join
async def joined(session, details):
    global global_session
    global_session = session
    print("session ready")

if __name__ == "__main__":
    run([comp])

This approach of keeping a reference after component has joined connection feels however a bit "odd". Is there a different approach to this? Can this done on some other way.

If not than it feels a bit more "right" with subclassing and having all the application depended code within that subclass (but however keeping everything of my app within one subclass also feels odd).

Answer 1

There are multiple approaches you could take here, though the simplest IMO would be to use Crossbar's http bridge. So whenever an event callback is received from your hardware, you can just make a http POST request to Crossbar and your message will get delivered

More details about http bridge https://crossbar.io/docs/HTTP-Bridge-Publisher/

Answer 2

I would recommend to use asynchronous queue instead of shared session:

import asyncio

from autobahn.asyncio.component import Component
from autobahn.asyncio.component import run

queue = asyncio.queues.Queue()

comp = Component(
    transports=u"ws://localhost:8080/ws",
    realm=u"realm1",
)


def callback_from_hardware(msg):
    queue.put_nowait((u'com.someapp.somechannel', msg,))


@comp.on_join
async def joined(session, details):
    print("session ready")

    while True:
        topic, message, = await queue.get()
        print("Publishing: topic: `%s`, message: `%s`" % (topic, message))
        session.publish(topic, message)


if __name__ == "__main__":
    callback_from_hardware("dassdasdasd")
    run([comp])