Publishing OpenCV VideoCapture to NATS server

[pcrete]

Hi, I am using NATS to perform video streaming between microservices. It works great on NATS video subscriber.

However, I'm facing a problem with reading video frames from webcam and publishing them to nats-server by using OpenCV, cv2.VideoCapture.

I guess, the problem is because of I/O blocking from cv2.VideoCapture.read (I've tried with cv2.imread too, and it caused the same problem.)

And when I move these 3 lines outside the loop, the streaming works fine.

        frame = cap.read()[1]
        frame_encoded = cv2.imencode('.jpg', frame)[1]
        frame_bytes = frame_encoded.tostring()

Can anyone suggest me a solution to solve this NATS video streaming?
or what could be best practices for doing video streaming between microservices by using NATS.io?

Here is my sample code:
publisher.py

import cv2
import asyncio
from nats.aio.client import Client as NATS

async def run(loop):
    nc = NATS()
    await nc.connect(loop=loop)

    cap = cv2.VideoCapture(0)
    while True:
        frame = cap.read()[1]
        frame_encoded = cv2.imencode('.jpg', frame)[1]
        frame_bytes = frame_encoded.tostring()
        
        await nc.publish('foo', frame_bytes)

if __name__ == '__main__':
    loop = asyncio.get_event_loop()
    loop.run_until_complete(run(loop))
    loop.run_forever()

My subscriber just simply listens and prints random output to check if it's receiving something.
subscriber.py

import uuid
import asyncio
from nats.aio.client import Client as NATS

async def run(loop):
    nc = NATS()
    await nc.connect(loop=loop)

    async def cb(msg):
        print(uuid.uuid4())

    await nc.subscribe('foo', cb=cb)

if __name__ == '__main__':
    loop = asyncio.get_event_loop()
    loop.run_until_complete(run(loop))
    loop.run_forever()

[takanoshota]

lt' probably due to Delimiters of nats protocol(https://github.com/nats-io/docs/blob/master/nats_protocol/nats-protocol.md).

Codes below works for me.

note

If you want to send over 1MB message, you need to change default value of
nats server by editing nats-server.conf like as max_payload: 100000000(it means 100MB).

publisher.py

import cv2
import asyncio
from nats.aio.client import Client as NATS
import base64

async def run(loop):
    nc = NATS()
    await nc.connect(loop=loop)
    cap = cv2.VideoCapture(0)
    while True:
        frame = cap.read()[1]
        frame = cv2.resize(frame, dsize=(640,360))
        cv2.imwrite('before.png',frame)
        frame_base64 = base64.b64encode(frame)
        await nc.publish('foo', frame_base64)

if __name__ == '__main__':
    loop = asyncio.get_event_loop()
    loop.run_until_complete(run(loop))
    loop.run_forever()

subscriber.py

import asyncio
from nats.aio.client import Client as NATS
import base64
import numpy as np
import cv2

async def run(loop):
    nc = NATS()
    await nc.connect(loop=loop)

    async def receiver(msg):
        data = msg.data.decode()
        frame_encoded = base64.b64decode(data)
        frame_encoded = np.frombuffer(frame_encoded, dtype=np.uint8)
        frame_encoded =frame_encoded.reshape(360,640,3)
        await cv2.imwrite('after.png',frame_encoded)

    await nc.subscribe("foo", cb=receiver)

if __name__ == '__main__':
    loop = asyncio.get_event_loop()
    loop.run_until_complete(run(loop))
    loop.run_forever()

[wallyqs]

This use case would now be better suited with OrderedConsumers from JetStream: https://github.com/nats-io/nats.py/blob/a1a9270/tests/test_js.py#L523-L590