server/client that can republish

[k2ev]

I would like to set a server that can subscribe to an external websocket (ws1), receive data and then republish that data after curating to internal clients connecting to this server over websockets (ws2).

Can this be achieved using this package? And if it can be, can you please point to a simple example. I really appreciate your help,
https://stackoverflow.com/questions/72040359/python-websocket-pub-sub-with-re-publish-broadcast

[orweis]

Hi @k2ev :)

Yes, this can (easily) be achieved with this package.

You can have as many sockets as you'd like , but you don't actually have to have two sockets here - you can have one socket and split the roles based on topics.

I don't have a specific example for you that is exactly what you're asking for- but you can look at this test which is pretty similar:
https://github.com/permitio/fastapi_websocket_pubsub/blob/master/tests/multiprocess_test.py

[nneskildsf]

Hi @k2ev

I think I might have been trying to solve the same problem that you describe here.

In my case I have a number of backends that I would like to register with an aggregator and dump their data.
The aggregator should also allow users to subscribe to data from backends as well as allow users to call methods on the backends.

flowchart LR
    subgraph S3[ ]
    A[Aggregator]
    end

    subgraph S1[Users]
    X[User 1]
    Y[User 2]
    Z[User 3]
    end

    subgraph S2[Backends]
    U[Backend 1]
    V[Backend 2]
    W[Backend 3]
    end
    
    X  <-->|WebSockets| A <-->|WebSockets| U
    Y  <-->|WebSockets| A <-->|WebSockets| V
    Z  <-->|WebSockets| A <-->|WebSockets| W
    
    style S3 fill:#FFF, stroke-width:0px

Loading

Figure 1. Diagram of the problem.

My solution is given in the code below. Start off running python aggregator.py and then launch backends and users.

Br Eskild

aggregator.py

'''
User 1 <--> |^^^^^^^^^^^^| <--> Backend 1
User 2 <--> | Aggregator | <--> Backend 2
User 3 <--> |____________| <--> Backend 3
'''

import asyncio
import uvicorn
from fastapi import FastAPI
from fastapi.routing import APIRouter
from fastapi_websocket_pubsub import PubSubEndpoint

app =  FastAPI()

rpc_channels = {}

# Methods to expose to the clients
from fastapi_websocket_pubsub.rpc_event_methods import RpcEventServerMethods
class AggregatorServer(RpcEventServerMethods):
    async def call_backend(self, backend_id, fn, *args, **kwargs):
        '''
        This endpoint forwards function calls from a User
        to a specific Backend.
        '''
        global rpc_channels
        if backend_id in rpc_channels:
            ch = rpc_channels[backend_id]
            async def f():
                await getattr(ch.other, fn)(*args, **kwargs)
            asyncio.create_task(f())
            return True
        return False
    async def register_backend(self, backend_id):
        '''
        This endpoint is called by Backends
        that want to register themselves.
        This is done by letting the Aggregator
        subscribe to data from the Backend and
        passing it on to Users.
        '''
        print(f'Registering backend {backend_id}')
        global rpc_channels
        rpc_channels[backend_id] = self.channel
        
        async def forward_data_to_users(topic, data):
            data = {'backend_id': backend_id, 'data': data}
            await psendpoint.publish(['U', f'U{backend_id:d}'], data=data)
        await psendpoint.subscribe([f'E{backend_id:d}'], forward_data_to_users)

#PubSub stuff
router = APIRouter()
psendpoint = PubSubEndpoint(methods_class=AggregatorServer)
psendpoint.register_route(router, '/pubsub')
app.include_router(router)

# Start the server itself
uvicorn.run(app, host='localhost', port=8000)

backend.py

import asyncio
import random
from fastapi_websocket_pubsub import PubSubClient
from fastapi_websocket_pubsub.rpc_event_methods import RpcEventClientMethods

PORT = 8000
backend_id = int(random.random()*1000)

class BackendServer(RpcEventClientMethods):
    async def action(self, *args, **kwargs):
        print('Backend was served an action:', kwargs)

async def main():
    client = PubSubClient([], methods_class=BackendServer)
    
    client.start_client(f"ws://localhost:{PORT}/pubsub")
    await client.wait_until_ready()
    await client._rpc_channel.other.register_backend(backend_id=backend_id)
    while 1:
        await asyncio.sleep(1)
        live_data = int(random.random()*1000)
        await client.publish([f'E{backend_id:d}'], data={'live_data': live_data,})
        print(f'>> {backend_id:04d} {live_data:04d}')
    await client.wait_until_done()

asyncio.run(main())

user.py

import asyncio
from fastapi_websocket_pubsub import PubSubClient

PORT = 8000

async def main():
    client = PubSubClient([])

    async def send_action_to_backend(backend_id):
        response = await client._rpc_channel.other.call_backend(backend_id=backend_id, fn='action', a=5, b=4)
        if response is False:
            print(f'It seems that backend {backend} does not exist.')

    async def on_data(topic, data):
        backend_id = data['backend_id']
        live_data = data['data']['live_data']
        print(f'<< {backend_id:04d} {live_data:04d}')
        asyncio.create_task(send_action_to_backend(int(backend_id)))

    client.subscribe('U', on_data)
    client.start_client(f"ws://localhost:{PORT}/pubsub")
    await client.wait_until_ready()
    await client.wait_until_done()

asyncio.run(main())