spacepaste

Paste details

reply | raw

  1.  
  2. #!/usr/bin/env python
  3. """
  4. ZeroMQ Testing
  6. Python 2.6.7
  7. zeromq-2.1.10.tar.gz
  8. pyzmq-2.1.10.zip
  9. """
  11. import pickle
  12. import time
  13. import traceback as tb
  14. import sys
  15. from common import getLogger
  16. log = getLogger()
  18. try:
  19. import zmq
  20. except:
  21. sys.stderr.write("Unable to load zmq python extensions.")
  23. try:
  24. try:
  25. try:
  26. from multiprocessing import Process as Thread
  27. except:
  28. from processing import Process as Thread
  29. except:
  30. from threading import Thread
  31. except:
  32. sys.stderr.write("Unable to load multiprocessing")
  33. sys.exit(-1)
  36. def publish(url):
  37. zmq_context = zmq.Context()
  38. pub_socket = zmq_context.socket(zmq.PUB)
  39. pub_socket.bind(url)
  40. for i in range(0,1000):
  41. pub_socket.send(pickle.dumps(i))
  43. def recv(socket):
  44. ts = time.time()
  45. while time.time() - ts < 2:
  46. try:
  47. return pickle.loads(socket.recv(zmq.NOBLOCK))
  48. except zmq.ZMQError, e:
  49. if e.errno == zmq.EAGAIN:
  50. time.sleep(0.5)
  51. continue
  52. else:
  53. log.error("".join(tb.format_exception(*sys.exc_info())))
  55. def subscribe(url):
  56. zmq_context = zmq.Context()
  57. sub_socket = zmq_context.socket(zmq.SUB)
  58. sub_socket.setsockopt(zmq.SUBSCRIBE, "")
  59. sub_socket.connect(url)
  60. total = 0
  61. while True:
  62. msg = recv(sub_socket)
  63. if msg is None:
  64. log.debug("Exiting")
  65. break
  66. total += 1
  67. log.debug("%d msgs received" % (total))
  69. url = "tcp://*:7000"
  70. max_subs = 10
  72. log.debug("Creating subscribers")
  73. subs = list()
  74. for i in range(0,max_subs):
  75. sub = Thread(target=subscribe, args=(url,))
  76. sub.start()
  77. subs.append(sub)
  79. time.sleep(1)
  80. log.debug("Publishing to %s" % (url,))
  81. publish(url)
  83. log.debug("Joining subscribers")
  84. while len(subs) > 0:
  85. joined = None
  86. for s in subs:
  87. s.join(0.1)
  88. if not s.is_alive():
  89. joined = s
  90. if joined is not None:
  91. subs.remove(joined)
  94. """
  95. $ reset; ./test.py;
  96. 6800 Creating subscribers
  97. 6800 Publishing to tcp://*:7000
  98. 6800 Joining subscribers
  99. 6804 Exiting
  100. 6804 0 msgs received
  101. 6807 Exiting
  102. 6807 0 msgs received
  103. 6806 Exiting
  104. 6806 0 msgs received
  105. 6809 Exiting
  106. 6803 Exiting
  107. 6803 0 msgs received
  108. 6809 0 msgs received
  109. 6808 Exiting
  110. 6810 Exiting
  111. 6810 0 msgs received
  112. 6808 0 msgs received
  113. 6801 Exiting
  114. 6801 408 msgs received
  115. 6802 Exiting
  116. 6802 509 msgs received
  117. 6805 Exiting
  118. 6805 877 msgs received
  119. """
  120.  
Powered by spacepaste. Paste removal.