Skip to content

Instantly share code, notes, and snippets.

@ichux
Last active November 2, 2025 17:24
Show Gist options
  • Select an option

  • Save ichux/3cf100e2c4f81b3202b9181e815192b9 to your computer and use it in GitHub Desktop.

Select an option

Save ichux/3cf100e2c4f81b3202b9181e815192b9 to your computer and use it in GitHub Desktop.
import msgpack
class Event:
@staticmethod
def pack(payload):
return msgpack.packb(payload, use_bin_type=True)
@staticmethod
def unpack(blob):
return msgpack.unpackb(blob, raw=False, strict_map_key=False)
# >>> Event.unpack(Event.pack({1: 2}))
# {1: 2}
# >>> Event.unpack(Event.pack({'1': '2'}))
# {'1': '2'}
# server.py
import zmq
import msgpack
ctx = zmq.Context()
socket = ctx.socket(zmq.REP)
socket.bind("tcp://*:5555")
while True:
req = msgpack.unpackb(socket.recv(), raw=False)
print(f"Received: {req}")
socket.send(msgpack.packb({'ok': True, 'echo': req}, use_bin_type=True))
# client.py
import zmq
import msgpack
ctx = zmq.Context()
socket = ctx.socket(zmq.REQ)
socket.connect("tcp://127.0.0.1:5555")
socket.send(msgpack.packb({'hello': 'world'}, use_bin_type=True))
reply = msgpack.unpackb(socket.recv(), raw=False)
print(reply)
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment