From cc0504c44d0d366b5663cdb081bd5c710474f6e3 Mon Sep 17 00:00:00 2001 From: yumoqing Date: Sat, 13 Nov 2021 20:59:06 +0800 Subject: [PATCH] bugfix --- appPublic/zmq_topic.py | 119 +++++++++++++++++++++++++++++++++++++++++ doc/.zmq_topic.md.swp | Bin 0 -> 12288 bytes doc/zmq_topic.md | 60 +++++++++++++++++++++ 3 files changed, 179 insertions(+) create mode 100644 appPublic/zmq_topic.py create mode 100644 doc/.zmq_topic.md.swp create mode 100644 doc/zmq_topic.md diff --git a/appPublic/zmq_topic.py b/appPublic/zmq_topic.py new file mode 100644 index 0000000..a7c18b2 --- /dev/null +++ b/appPublic/zmq_topic.py @@ -0,0 +1,119 @@ +import sys +import zmq +import time +from zmq import Context +from appPublic.jsonConfig import getConfig + +class TopicServer: + def __init__(self, address='127.0.0.1', pub_port='5566', sub_port='5567'): + # get ZeroMQ version + print("Current libzmq version is %s" % zmq.zmq_version()) + print("Current pyzmq version is %s" % zmq.pyzmq_version()) + + self.context = Context.instance() + # 2 sockets, because we can only bind once to a socket (as opposed to connect) + self.pub_port = "tcp://{}:{}".format(address, pub_port) + self.sub_port = "tcp://{}:{}".format(address, sub_port) + + self.xpub_xsub_proxy() + + # N publishers to 1 sub; proxy 1 sub to 1 pub; publish to M subscribers + def xpub_xsub_proxy(self): + print("Init proxy") + + # Socket subscribing to publishers + frontend_pubs = self.context.socket(zmq.XSUB) + frontend_pubs.bind(self.pub_port) + + # Socket publishing to subscribers + backend_subs = self.context.socket(zmq.XPUB) + backend_subs.bind(self.sub_port) + + print("Try: Proxy... CONNECT!") + zmq.proxy(frontend_pubs, backend_subs) + print("CONNECT successful!") + """ + while True: + time.sleep(1) + """ + +class ConfiguredTopicServer(TopicServer): + """ + in config file has a topicserver key + { + "topicserver":{ + "address":"11.11.1.11", + "pub_port":1234, + "sub_server":1235 + } + } + """ + def __init__(self): + config = getConfig() + params = config.topicserver + if not params: + raise MissTopicServerConfig + super(ConfiguredTopicServer, self).__init__(**params) + +class TopicPublisher: + def __init__(self, topic='en', address='127.0.0.1', port='5566'): + # get ZeroMQ version + print("Current libzmq version is %s" % zmq.zmq_version()) + print("Current pyzmq version is %s" % zmq.pyzmq_version()) + + self.topic = topic + self._topic = topic.encode('utf-8') + self.context = Context.instance() + self.url = "tcp://{}:{}".format(address, port) + self.pub = self.context.socket(zmq.PUB) + self.pub.connect(self.url) + + def send(self, message): + self.pub.send_multipart([self._topic, message.encode('utf-8')]) + +class ConfiguredTopicPublisher(TopicPublisher): + def __init__(self, topic=''): + config = getConfig() + params = config.topicserver + if not params: + raise MissTopicServerConfig + super(ConfiguredTopicPublisher, self).__init__(topic=topic, + address = params.address, + port=params.pub_port) + +class TopicSubscriber: + def __init__(self, topic='', address='127.0.0.1', port='5567', callback=None): + # get ZeroMQ version + print("Current libzmq version is %s" % zmq.zmq_version()) + print("Current pyzmq version is %s" % zmq.pyzmq_version()) + + self.callback = callback + self.topic = topic + self._topic = self.topic.encode('utf-8') + self.context = Context.instance() + self.url = "tcp://{}:{}".format(address, port) + self.sub = self.context.socket(zmq.SUB) + self.sub.connect(self.url) + # subscribe to topic 'en' or 'jp' + self.sub.setsockopt(zmq.SUBSCRIBE, self._topic) + + + def run(self): + # keep listening to all published message, filtered on topic + print("Sub {}: Going to wait for messages!".format(self.topic)) + while True: + msg_received = self.sub.recv_multipart() + print("sub {}: {}".format(self.topic, msg_received)) + if self.callback: + self.callback(msg_received) + +class ConfiguredTopicSubscriber(TopicSubscriber): + def __init__(self, topic=''): + config = getConfig() + params = config.topicserver + if not params: + raise MissTopicServerConfig + super(ConfiguredTopicSubscriber, self).__init__(topic=topic, + address=params.address, + port=params.sub_port) + diff --git a/doc/.zmq_topic.md.swp b/doc/.zmq_topic.md.swp new file mode 100644 index 0000000000000000000000000000000000000000..73d3c6480540070c5ac9ed96d4955d3b6ca40853 GIT binary patch literal 12288 zcmeI2&2Jk;7>B1l04{Ao?WLS}iIMCGcVnB6q*i@Okf;!%Qe1?f)n>grwpZ*|cXs21 z&;vcR2M)c#9dY2?1OEVL;NC0r52&ExOQ`+Lcz5f#ttjfDqIx%a9PjMRJ2THaV=3_r z+aH{HkJhZN!1baK)9EIAgzbzxZMUY=8~00XDz}*Z><~18jf| zumLu}2A&cF4!(82{hSbg)-WM_|F8cC@Y@SQ+&n78FMz-~Z}1cN5qtrz zf-B%Numb|n2d{#k4-0Vv`~a?lufbQ~ORxuG@F8e}CGaeG1{?&xKQF|u;Ct{bxCSnR z&%qh61YQCsz(0qCxCL&2>);!36m%~2G{@_ zU;}L6sWecplc?7Pp!GN!`Qtp5-asG0+{}l{P5ogYQ~mToWeY9LK~6Q9tG};gq4uUs zx96%Bw_96VLew&;GFue`kWIC#EGHdrf84pSENV4PJ~~)HtV{vFx3X z_NvBJ8t4bog9 z(K><&kd&FJ7Uv@*rcC+T@;&#ZpZ{dO>Qof@(9X)sXf literal 0 HcmV?d00001 diff --git a/doc/zmq_topic.md b/doc/zmq_topic.md new file mode 100644 index 0000000..814e87b --- /dev/null +++ b/doc/zmq_topic.md @@ -0,0 +1,60 @@ +# zmq_topic +zmq_topic implements a topic message machine, it need a TopicServer with created using a server address, publish_port and subscrible_port, and one or more publisher with can publish topiced message and one or more topic message subscriber which comsumes topiced message. + +If the server gone, subscriber will not received message, and when the server restart, subscriber will received message again, but the message will miss send by publisher when the server is down. + +## Usage examples + +topic server +``` +from appPublic.zmq_topic import TopicServer + +x = TopicServer(address='127.0.0.1', pub_port=5678, sub_port=5679) +``` + +publisher +``` +import time +from appPublic.zmq_topic import TopicPublisher +def test_publish(topic): + p = TopicPublisher(topic=topic, address='127.0.0.1', port=5678) + cnt = 1 + while True: + p.send(f'message {cnt}') + time.sleep(1) + cnt += 1 + +if __name__ == '__main__': + import sys + if len(sys.argv) < 2: + print(f'Usage:\n{sys.argv[0]} topic') + sys.exit(1) + test_publish(sys.argv[1]) + +``` +subscriber +``` +from appPublic.zmq_topic import TopicSubscriber +def test_subscribe(topic): + def print_message(msg): + print(msg) + + s = TopicSubscriber(topic=topic, address='127.0.0.1', + port=5679, callback=print_message) + s.run() + +if __name__ == '__main__': + import sys + if len(sys.argv) < 2: + print(f'Usage:\n{sys.argv[0]} topic') + sys.exit(1) + test_subscribe(sys.argv[1]) + +``` +## TopicServer +## ConfiguredTopicServer +## TopicPublisher +## ConfiguredTopicPublisher +## TopicSubscribler +## ConfiguredTopicSubscribler +