ZeroMQ-Pushpin: received message with invalid format

Pushpin
1 
payload = {
    "items": [
        {
            "channel": "0001010",
            "formats": {
                "http-stream": {
                    "content-bin": base64.b64encode(b"hello world").decode("ascii")
                }
            },
        }
    ]
}

I have the above which works well when used with http /publish, please ignore the b"hello world" part, it’s just a placeholder for the data I am streaming.

import base64

import tnetstring
import zmq

context = zmq.Context()
socket = context.socket(zmq.PUSH)
socket.connect("tcp://localhost:5560")


def encode_values(obj):
    if isinstance(obj, dict):
        return {
            k.encode("utf-8") if isinstance(k, str) else k: encode_values(v)
            for k, v in obj.items()
        }
    elif isinstance(obj, list):
        return [encode_values(item) for item in obj]
    elif isinstance(obj, str):
        return obj.encode("utf-8")
    else:
        return obj


payload = {
    "items": [
        {
            "channel": "0001010",
            "formats": {
                "http-stream": {
                    "content-bin": base64.b64encode(b"hello world").decode("ascii")
                }
            },
        }
    ]
}

tnet_message = tnetstring.dumps(encode_values(payload))


socket.send(tnet_message)
print("Sent message:", tnet_message)


socket.close()
context.term()

The above is the code but I get the below error

[DEBUG] 2025-05-25 05:02:49.046 [handler] IN pull: { "items": [ { "formats": { "http-stream": { "content-bin": "aGVsbG8gd29ybGQ=" } }, "channel": "0001010" } ] }
[WARN] 2025-05-25 05:02:49.046 [handler] IN pull: received message with invalid format: publish item object does not contain 'channel', skipping

It means there’s something I am not doing well. Please, I need help with it, thank you.

2

For sending items via ZeroMQ PUSH or PUB, they should be sent without the {"items": [ ]} wrapping.

3

Yes, that works. Thanks.

I noticed it doesn’t allow content-bin. What I’m streaming only works if I used the content-bin line exactly the way I wrote it!

I tried the content type but that didn’t work.

Is there another way to make it work?

5

Another difference with sending messages over ZeroMQ is the use of tnetstring format instead of JSON, which supports binary data. So there is no content-bin field, only content, which can contain arbitrary data without base64-encoding.

6

I had to put on a thinking cap after you wrote this. I was determined to make it work with tnetstring and it did.

Here is something I did differently though

    def off_take(self, content):
        if self.channel:
            try:
                msg = tnetstring.dumps(
                    {
                        b"channel": self.channel.encode("utf-8"),
                        b"formats": {b"http-stream": {b"content": content}},
                    }
                )
                socket.send(msg, zmq.DONTWAIT)
            except (Exception,) as exc:
                bl.error(exc)

Every other thing I did failed until this pattern worked. content is left as-is! I am leaving this for anyone that has such issues later on.

Once more @jkarneges thanks for coming through with your help despite your busy schedule. You rock!