How to send data to a Websocket from a faust application

I am working on an usecase at the moment using Kafka and robinhood's faust to process the data coming from Kafka. I have succeeded doing the computation and the results I need are being printed to the console my faust worker is running in.

Now I want to find a way to get my results not only in console but visible in a HTML page. I have taken a look at the websockets library but I can't get it to work in conjunction with faust. The error I get is Crashed reason=RuntimeError('This event loop is already running') I think this is caused because the code is executed for every message that is being processed.

Any help is highly appreciated

This is the code I am using:

    import faust, datetime, websockets, asyncio

app = faust.App(
    'UseCase',
    broker='kafka://localhost:29092',
)

usecase_topic = app.topic('usecase',partitions=8)

usecase_table = app.Table('usecase', default=int)

checkfailure = {}

@app.agent(usecase_topic)
async def process_record(records):
    async for record in records:
        #count records for each Sensor
        print(record)
        sensor = record['ext_id']
        usecase_table[sensor] += 1
        #print(f'Records for Sensor {sensor}: {usecase_table[sensor]}')

        #write current timestamp of record and previous timestamp for each sensor to usecase_table dict
        currtime_id = record['ext_id']+'c'
        prevtime_id = record['ext_id']+'p'
        usecase_table[currtime_id] = datetime.datetime.strptime(record['tag_tsp'], "%Y%m%d%H%M%S.%f")

        #print current time
        print(f'Current time for Sensor {sensor}: {usecase_table[currtime_id]}')


        #calculate and print timestamp delta; if no previous value is given print message
        if usecase_table[prevtime_id] == 0:
            print(f'no previous timestamp for sensor {sensor}')
        else:
            usecase_table[prevtime_id] = datetime.datetime.strptime(usecase_table[prevtime_id], "%Y%m%d%H%M%S.%f")
            print(f'previous time for Sensor {sensor}: {usecase_table[prevtime_id]}')
            tsdelta = usecase_table[currtime_id] - usecase_table[prevtime_id]
            tsdelta_id = record['ext_id']+'t'
            usecase_table[tsdelta_id] = str(tsdelta)
            print(f'Sensor: {sensor} timestamp delta: {usecase_table[tsdelta_id]}')

        #calculate value delta
        currvalue_id = record['ext_id']+'cv'
        prevvalue_id = record['ext_id']+'pv'
        usecase_table[currvalue_id] = record['tag_value_int']

        print(f'current value for Sensor {sensor}: {usecase_table[currvalue_id]}')

        if usecase_table[prevvalue_id] == 0:
            print(f'no previous record for sensor {sensor}')
        else:
            print(f'previous value for Sensor {sensor}: {usecase_table[prevvalue_id]}')
            vdelta = usecase_table[currvalue_id] - usecase_table[prevvalue_id]
            vdelta_id = record['ext_id']+'v'
            usecase_table[vdelta_id] = vdelta
            print(f'Sensor: {sensor} value delta:{usecase_table[vdelta_id]}')

        #calculate cycle time
        if usecase_table[prevtime_id] != 0 and usecase_table[prevvalue_id] != 0 and usecase_table[vdelta_id] != 0:
            cycletime = tsdelta / usecase_table[vdelta_id]
            cyclemsg = f'Sensor {sensor}; Cycletime {cycletime}'
            print(cyclemsg)

        #add timestamp to checkfailure dict
        checkfailure[sensor] = datetime.datetime.strptime(record['tag_tsp'], "%Y%m%d%H%M%S.%f")
        #check if newest timestamp for a sensor is older than 10 secs
        for key in checkfailure:
            if datetime.datetime.now() - checkfailure[key] >= datetime.timedelta(seconds=10):
                failuremsg = f'Error: Sensor {key}'
                print(failuremsg)

        #send results to websocket
        async def send_result(websocket,path):
            results = cyclemsg + failuremsg
            await websockets.send(results)
        start_server = websockets.serve(send_result, '127.0.0.1', 5678)
        asyncio.get_event_loop().run_until_complete(start_server)

        #set previous value and timestamp to current
        usecase_table[prevtime_id] = record['tag_tsp']
        usecase_table[prevvalue_id] = record['tag_value_int']

Solution

It is normal to be confused by this asyncio error message :)

You cannot call loop.run_until_complete from an async def function.

What you need to do is start the websocket server in the background. That should be easy, and it is using asyncio.ensure_future, but you also want your websocket server to gracefully shut down when your application exits.

For this reason Faust uses "services", and you can define a service for your websocket server:

import faust
import websockets
from mode import Service
from websockets.exceptions import ConnectionClosed
from websockets.server import WebSocketServerProtocol


class App(faust.App):

   def on_init(self):
       self.websockets = Websockets(self)

   async def on_start(self):
       await self.add_runtime_dependency(self.websockets)



class Websockets(Service):

    def __init__(self, app, bind: str = 'localhost', port: int = 9999, **kwargs):
        self.app = app
        self.bind = bind
        self.port = port
        super().__init__(**kwargs)

    async def on_message(self, ws, message):
        ...

    async def on_messages(self,
                          ws: WebSocketServerProtocol,
                          path: str) -> None:
        try:
            async for message in ws:
                await self.on_message(ws, message)
        except ConnectionClosed:
            await self.on_close(ws)
        except asyncio.CancelledError:
            pass

    async def on_close(self, ws):
        # called when websocket socket is closed.
        ...

    @Service.task
    def _background_server(self):
         await websockets.serve(self.on_messages, self.bind, self.port)

app = App('UseCase')
# [...]