You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.

zmq_sub.py 3.0KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384
  1. #!/usr/bin/env python3
  2. # Copyright (c) 2014-2016 The Starwels developers
  3. # Distributed under the MIT software license, see the accompanying
  4. # file COPYING or http://www.opensource.org/licenses/mit-license.php.
  5. """
  6. ZMQ example using python3's asyncio
  7. Starwels should be started with the command line arguments:
  8. starwelsd -ai -daemon \
  9. -zmqpubhashblock=tcp://127.0.0.1:28352 \
  10. -zmqpubrawtx=tcp://127.0.0.1:28352 \
  11. -zmqpubhashtx=tcp://127.0.0.1:28352 \
  12. -zmqpubhashblock=tcp://127.0.0.1:28352
  13. We use the asyncio library here. `self.handle()` installs itself as a
  14. future at the end of the function. Since it never returns with the event
  15. loop having an empty stack of futures, this creates an infinite loop. An
  16. alternative is to wrap the contents of `handle` inside `while True`.
  17. A blocking example using python 2.7 can be obtained from the git history:
  18. https://github.com/starwels/starwels/blob/37a7fe9e440b83e2364d5498931253937abe9294/contrib/zmq/zmq_sub.py
  19. """
  20. import binascii
  21. import asyncio
  22. import zmq
  23. import zmq.asyncio
  24. import signal
  25. import struct
  26. import sys
  27. if not (sys.version_info.major >= 3 and sys.version_info.minor >= 5):
  28. print("This example only works with Python 3.5 and greater")
  29. exit(1)
  30. port = 28352
  31. class ZMQHandler():
  32. def __init__(self):
  33. self.loop = zmq.asyncio.install()
  34. self.zmqContext = zmq.asyncio.Context()
  35. self.zmqSubSocket = self.zmqContext.socket(zmq.SUB)
  36. self.zmqSubSocket.setsockopt_string(zmq.SUBSCRIBE, "hashblock")
  37. self.zmqSubSocket.setsockopt_string(zmq.SUBSCRIBE, "hashtx")
  38. self.zmqSubSocket.setsockopt_string(zmq.SUBSCRIBE, "rawblock")
  39. self.zmqSubSocket.setsockopt_string(zmq.SUBSCRIBE, "rawtx")
  40. self.zmqSubSocket.connect("tcp://127.0.0.1:%i" % port)
  41. async def handle(self) :
  42. msg = await self.zmqSubSocket.recv_multipart()
  43. topic = msg[0]
  44. body = msg[1]
  45. sequence = "Unknown"
  46. if len(msg[-1]) == 4:
  47. msgSequence = struct.unpack('<I', msg[-1])[-1]
  48. sequence = str(msgSequence)
  49. if topic == b"hashblock":
  50. print('- HASH BLOCK ('+sequence+') -')
  51. print(binascii.hexlify(body))
  52. elif topic == b"hashtx":
  53. print('- HASH TX ('+sequence+') -')
  54. print(binascii.hexlify(body))
  55. elif topic == b"rawblock":
  56. print('- RAW BLOCK HEADER ('+sequence+') -')
  57. print(binascii.hexlify(body[:80]))
  58. elif topic == b"rawtx":
  59. print('- RAW TX ('+sequence+') -')
  60. print(binascii.hexlify(body))
  61. # schedule ourselves to receive the next message
  62. asyncio.ensure_future(self.handle())
  63. def start(self):
  64. self.loop.add_signal_handler(signal.SIGINT, self.stop)
  65. self.loop.create_task(self.handle())
  66. self.loop.run_forever()
  67. def stop(self):
  68. self.loop.stop()
  69. self.zmqContext.destroy()
  70. daemon = ZMQHandler()
  71. daemon.start()