如何从faust应用程序向websocket发送数据

我正在使用Kafka和罗比尼奥的《浮士德》处理来自Kafka的数据。我已经成功地做了计算，结果我需要被打印到控制台我的浮士德工人正在运行。
现在我想找到一种方法，不仅可以在控制台中获得结果，而且可以在html页面中看到结果。我已经看了websockets图书馆，但我不能让它与浮士德一起工作。我得到的错误是 Crashed reason=RuntimeError('This event loop is already running') 我认为这是因为代码是针对每个正在处理的消息执行的。
非常感谢您的帮助
这是我正在使用的代码：

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']

被此asyncio错误消息弄糊涂是正常的：）
你不能打电话 loop.run_until_complete 从一个 async def 功能。
您需要做的是在后台启动websocket服务器。这应该很容易，而且它正在使用 asyncio.ensure_future ，但您也希望在应用程序退出时，websocket服务器正常关闭。
因此，faust使用“服务”，您可以为websocket服务器定义服务：

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')

# [...]

如何从faust应用程序向websocket发送数据

1条答案

相关问题

热门标签

最新问答