From cce638a8adf4e045ca5505afea4bda57753c31dd Mon Sep 17 00:00:00 2001 From: Micah Anderson Date: Mon, 11 Aug 2014 16:33:29 -0400 Subject: initial import of debian package --- examples/poll/pair.py | 56 ++++++++++++++++++++++++++++++++++++++ examples/poll/pubsub.py | 57 +++++++++++++++++++++++++++++++++++++++ examples/poll/reqrep.py | 71 +++++++++++++++++++++++++++++++++++++++++++++++++ 3 files changed, 184 insertions(+) create mode 100644 examples/poll/pair.py create mode 100644 examples/poll/pubsub.py create mode 100644 examples/poll/reqrep.py (limited to 'examples/poll') diff --git a/examples/poll/pair.py b/examples/poll/pair.py new file mode 100644 index 0000000..81c8b3a --- /dev/null +++ b/examples/poll/pair.py @@ -0,0 +1,56 @@ +"""A thorough test of polling PAIR sockets.""" + +#----------------------------------------------------------------------------- +# Copyright (c) 2010 Brian Granger +# +# Distributed under the terms of the New BSD License. The full license is in +# the file COPYING.BSD, distributed as part of this software. +#----------------------------------------------------------------------------- + +import time +import zmq + +print "Running polling tests for PAIR sockets..." + +addr = 'tcp://127.0.0.1:5555' +ctx = zmq.Context() +s1 = ctx.socket(zmq.PAIR) +s2 = ctx.socket(zmq.PAIR) + +s1.bind(addr) +s2.connect(addr) + +# Sleep to allow sockets to connect. +time.sleep(1.0) + +poller = zmq.Poller() +poller.register(s1, zmq.POLLIN|zmq.POLLOUT) +poller.register(s2, zmq.POLLIN|zmq.POLLOUT) + +# Now make sure that both are send ready. +socks = dict(poller.poll()) +assert socks[s1] == zmq.POLLOUT +assert socks[s2] == zmq.POLLOUT + +# Now do a send on both, wait and test for zmq.POLLOUT|zmq.POLLIN +s1.send('msg1') +s2.send('msg2') +time.sleep(1.0) +socks = dict(poller.poll()) +assert socks[s1] == zmq.POLLOUT|zmq.POLLIN +assert socks[s2] == zmq.POLLOUT|zmq.POLLIN + +# Make sure that both are in POLLOUT after recv. +s1.recv() +s2.recv() +socks = dict(poller.poll()) +assert socks[s1] == zmq.POLLOUT +assert socks[s2] == zmq.POLLOUT + +poller.unregister(s1) +poller.unregister(s2) + +# Wait for everything to finish. +time.sleep(1.0) + +print "Finished." \ No newline at end of file diff --git a/examples/poll/pubsub.py b/examples/poll/pubsub.py new file mode 100644 index 0000000..a590fa9 --- /dev/null +++ b/examples/poll/pubsub.py @@ -0,0 +1,57 @@ +"""A thorough test of polling PUB/SUB sockets.""" + +#----------------------------------------------------------------------------- +# Copyright (c) 2010 Brian Granger +# +# Distributed under the terms of the New BSD License. The full license is in +# the file COPYING.BSD, distributed as part of this software. +#----------------------------------------------------------------------------- + +import time +import zmq + +print "Running polling tets for PUB/SUB sockets..." + +addr = 'tcp://127.0.0.1:5555' +ctx = zmq.Context() +s1 = ctx.socket(zmq.PUB) +s2 = ctx.socket(zmq.SUB) +s2.setsockopt(zmq.SUBSCRIBE, '') + +s1.bind(addr) +s2.connect(addr) + +# Sleep to allow sockets to connect. +time.sleep(1.0) + +poller = zmq.Poller() +poller.register(s1, zmq.POLLIN|zmq.POLLOUT) +poller.register(s2, zmq.POLLIN|zmq.POLLOUT) + +# Now make sure that both are send ready. +socks = dict(poller.poll()) +assert socks[s1] == zmq.POLLOUT +assert not socks.has_key(s2) + +# Make sure that s1 stays in POLLOUT after a send. +s1.send('msg1') +socks = dict(poller.poll()) +assert socks[s1] == zmq.POLLOUT + +# Make sure that s2 is POLLIN after waiting. +time.sleep(0.5) +socks = dict(poller.poll()) +assert socks[s2] == zmq.POLLIN + +# Make sure that s2 goes into 0 after recv. +s2.recv() +socks = dict(poller.poll()) +assert not socks.has_key(s2) + +poller.unregister(s1) +poller.unregister(s2) + +# Wait for everything to finish. +time.sleep(1.0) + +print "Finished." diff --git a/examples/poll/reqrep.py b/examples/poll/reqrep.py new file mode 100644 index 0000000..ef4436c --- /dev/null +++ b/examples/poll/reqrep.py @@ -0,0 +1,71 @@ +"""A thorough test of polling REQ/REP sockets.""" + +#----------------------------------------------------------------------------- +# Copyright (c) 2010 Brian Granger +# +# Distributed under the terms of the New BSD License. The full license is in +# the file COPYING.BSD, distributed as part of this software. +#----------------------------------------------------------------------------- + +import time +import zmq + +print "Running polling tests for REQ/REP sockets..." + +addr = 'tcp://127.0.0.1:5555' +ctx = zmq.Context() +s1 = ctx.socket(zmq.REP) +s2 = ctx.socket(zmq.REQ) + +s1.bind(addr) +s2.connect(addr) + +# Sleep to allow sockets to connect. +time.sleep(1.0) + +poller = zmq.Poller() +poller.register(s1, zmq.POLLIN|zmq.POLLOUT) +poller.register(s2, zmq.POLLIN|zmq.POLLOUT) + +# Make sure that s1 is in state 0 and s2 is in POLLOUT +socks = dict(poller.poll()) +assert not socks.has_key(s1) +assert socks[s2] == zmq.POLLOUT + +# Make sure that s2 goes immediately into state 0 after send. +s2.send('msg1') +socks = dict(poller.poll()) +assert not socks.has_key(s2) + +# Make sure that s1 goes into POLLIN state after a time.sleep(). +time.sleep(0.5) +socks = dict(poller.poll()) +assert socks[s1] == zmq.POLLIN + +# Make sure that s1 goes into POLLOUT after recv. +s1.recv() +socks = dict(poller.poll()) +assert socks[s1] == zmq.POLLOUT + +# Make sure s1 goes into state 0 after send. +s1.send('msg2') +socks = dict(poller.poll()) +assert not socks.has_key(s1) + +# Wait and then see that s2 is in POLLIN. +time.sleep(0.5) +socks = dict(poller.poll()) +assert socks[s2] == zmq.POLLIN + +# Make sure that s2 is in POLLOUT after recv. +s2.recv() +socks = dict(poller.poll()) +assert socks[s2] == zmq.POLLOUT + +poller.unregister(s1) +poller.unregister(s2) + +# Wait for everything to finish. +time.sleep(1.0) + +print "Finished." -- cgit v1.2.3