Skip to content

Instantly share code, notes, and snippets.

@taotetek
Created January 29, 2011 16:12
Show Gist options
  • Star 2 You must be signed in to star a gist
  • Fork 2 You must be signed in to fork a gist
  • Save taotetek/801957 to your computer and use it in GitHub Desktop.
Save taotetek/801957 to your computer and use it in GitHub Desktop.
Sending timed signals to multiple processes using twisted and zeromq
import zmq
from twisted.internet.task import LoopingCall
from twisted.internet import reactor
from datetime import datetime
from time import sleep
from multiprocessing import Process
class PulseTransmitter(object):
def __init__(self):
self.context = zmq.Context()
self.publisher = self.context.socket(zmq.PUB)
self.publisher.bind("tcp://*:5558")
self.counter = 0
def transmit(self):
if self.counter < 10:
self.publisher.send("PULSE")
print("PULSE snt at %s" % datetime.now())
self.counter = self.counter + 1
else:
self.publisher.send("TERM")
print("TERM snt at %s" % datetime.now())
reactor.stop()
def listener(pnum):
context = zmq.Context()
pulse_receiver = context.socket(zmq.SUB)
pulse_receiver.connect("tcp://*:5558")
pulse_receiver.setsockopt(zmq.SUBSCRIBE,"")
poller = zmq.Poller()
poller.register(pulse_receiver, zmq.POLLIN)
while True:
socks = dict(poller.poll())
if socks.get(pulse_receiver) == zmq.POLLIN:
message = pulse_receiver.recv()
if message == "PULSE":
print("PULSE rcv at %s by %i" % (datetime.now(), pnum))
elif message == "TERM":
print("TERM rcv at %s by %i" % (datetime.now(), pnum))
break
if __name__ == "__main__":
listeners = range(10)
for pnum in range(len(listeners)):
listeners[pnum] = Process(target=listener, args=(pnum,))
listeners[pnum].start()
sender = PulseTransmitter()
pulser = LoopingCall(sender.transmit)
pulser.start(1)
reactor.run()
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment