ZeroMQ-based python message broker server/client
You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.

56 lines
1.8 KiB

def listen_native(host, port, channels):
Subscribe to the given server/channels using the msgbus client
from contextlib import closing
from msgbus.client import MsgbusSubClient
with closing(MsgbusSubClient(host, port)) as client:
if channels:
for channel in channels:
while True:
yield "{} {}".format(*client.recv())
def listen_zmq(host, port, channels):
Example subscribing to the given server/channels using a raw zeromq socket
import zmq
with zmq.Context() as ctx:
forward_socket = ctx.socket(zmq.SUB)
forward_socket.connect("tcp://{}:{}".format(host, port))
if channels:
for channel in channels:
forward_socket.setsockopt(zmq.SUBSCRIBE, channel.encode("utf-8"))
forward_socket.setsockopt(zmq.SUBSCRIBE, b'')
while True:
transport = forward_socket.recv()
yield transport.decode("utf-8")
def main():
import argparse
parser = argparse.ArgumentParser(description="dump all messages from msgbus")
parser.add_argument("-i", "--host", default="", help="host to connect to")
parser.add_argument("-p", "--port", type=int, help="port to connect to")
parser.add_argument("-c", "--channel", nargs="+", help="dump only channels")
parser.add_argument("--type", default="native", choices=["native", "raw"], help="client type")
args = parser.parse_args()
client = None
if args.type == "native":
client = listen_native(, args.port,
elif args.type == "raw":
client = listen_zmq(, args.port,
for line in client:
if __name__ == '__main__':