This commit is contained in:
yumoqing 2021-11-13 20:59:06 +08:00
parent 75369977ff
commit cc0504c44d
3 changed files with 179 additions and 0 deletions

119
appPublic/zmq_topic.py Normal file
View File

@ -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)

BIN
doc/.zmq_topic.md.swp Normal file

Binary file not shown.

60
doc/zmq_topic.md Normal file
View File

@ -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