ptest.py

#!/usr/bin/env python3
import sys

import pulsar
import argparse


# pulsar_host = "http://10.217.0.171:9080"
pulsar_host = "http://131.84.1.232"

if __name__ == '__main__':
    parser = argparse.ArgumentParser(description="Test pulsar")
    parser.add_argument("--pulsar-host", default=pulsar_host, help="Pulsar server")
    parser.add_argument("--pub", action="store_true", help="publish (else we subscribe)")
    parser.add_argument("--topic", default="test_topic", help="topic")
    opts = parser.parse_args()

    client = pulsar.Client(opts.pulsar_host)

    if opts.pub:
        producer = client.create_producer(opts.topic)

        for i in range(10):
            producer.send(('Hello-%d' % i).encode('utf-8'))
    else:
        consumer = client.subscribe(opts.topic, sys.argv[0])
        while True:
            msg = consumer.receive()
            print("Received message '{}' id='{}'".format(msg.data(), msg.message_id()))
            consumer.acknowledge(msg)

    client.close()