Is it possible to read from a python websocket and write to kafka topic?

Question

I have a requirement to read data from a public topic over websocket and write to kafka topic. I tried this with below snippet, but the websocket client is getting closed upon receiving the first message.

Websocket client

ws_client = await WsClient.create(URL, client, msg_handler)
topic = '/my-topic'
wait ws_client.subscribe(topic)
while True:
    await asyncio.sleep(60, loop=loop)

Message handler - Here I am trying to produce the message to a kafka topic

async def msg_handler(msg):
        if msg:
            try:
                f = producer.send(kf_topic, value=msg)
                record_metadata = f.get(timeout=10)
                print(record_metadata)
            except KafkaError as er:
                print(er)

Debug log

DEBUG:websockets.protocol:client - state = OPEN
DEBUG:websockets.protocol:client - event = data_received(<60 bytes>)
DEBUG:websockets.protocol:client < Frame(fin=True, opcode=, data=b'{"id":"dc043deb9c08465d9e20fa9355a6b2f2","type":"welcome"}', rsv1=False, rsv2=False, rsv3=False)
DEBUG:websockets.protocol:client > Frame(fin=True, opcode=, data=b'{"type": "subscribe", "topic": "/my-topic", "response": true, "id": "1631228981047"}', rsv1=False, rsv2=False, rsv3=False)
DEBUG:websockets.protocol:client - event = data_received(<37 bytes>)
DEBUG:websockets.protocol:client < Frame(fin=True, opcode=, data=b'{"id":"1631228981047","type":"ack"}', rsv1=False, rsv2=False, rsv3=False)
DEBUG:websockets.protocol:client - event = data_received(<353 bytes>)
DEBUG:websockets.protocol:client < Frame(fin=True, opcode=, data=b'{"type":"message","topic":"/my-topic","data":{"name":"abc"}}', rsv1=False, rsv2=False, rsv3=False)
DEBUG:websockets.protocol:client - event = data_received(<348 bytes>)
DEBUG:websockets.protocol:client < Frame(fin=True, opcode=, data=b'{"type":"message","topic":"/my-topic","data":{"name":"abc"}}', rsv1=False, rsv2=False, rsv3=False)
DEBUG:websockets.protocol:client - state = CLOSING
DEBUG:websockets.protocol:client > Frame(fin=True, opcode=, data=b'\x03\xe8', rsv1=False, rsv2=False, rsv3=False)
DEBUG:websockets.protocol:client - event = data_received(<4 bytes>)
DEBUG:websockets.protocol:client < Frame(fin=True, opcode=, data=b'\x03\xe8', rsv1=False, rsv2=False, rsv3=False)
DEBUG:websockets.protocol:client - event = connection_lost(None)
DEBUG:websockets.protocol:client - state = CLOSED
DEBUG:websockets.protocol:client x code = 1000, reason = [no reason]
DEBUG:websockets.protocol:client x closing TCP connection

Is it possible to read from a python websocket and write to kafka topic?

0 Answers