contrib/zmq/zmq_sub3.4.py

   1 #!/usr/bin/env python3
   2 # Copyright (c) 2014-2016 The Bitcoin Core 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 """
   7     ZMQ example using python3's asyncio
   8
   9     Bitcoin should be started with the command line arguments:
  10         bitcoind -testnet -daemon \
  11                 -zmqpubhashblock=tcp://127.0.0.1:28332 \
  12                 -zmqpubrawtx=tcp://127.0.0.1:28332 \
  13                 -zmqpubhashtx=tcp://127.0.0.1:28332 \
  14                 -zmqpubhashblock=tcp://127.0.0.1:28332
  15
  16     We use the asyncio library here.  `self.handle()` installs itself as a
  17     future at the end of the function.  Since it never returns with the event
  18     loop having an empty stack of futures, this creates an infinite loop.  An
  19     alternative is to wrap the contents of `handle` inside `while True`.
  20
  21     The `@asyncio.coroutine` decorator and the `yield from` syntax found here
  22     was introduced in python 3.4 and has been deprecated in favor of the `async`
  23     and `await` keywords respectively.
  24
  25     A blocking example using python 2.7 can be obtained from the git history:
  26     https://github.com/bitcoin/bitcoin/blob/37a7fe9e440b83e2364d5498931253937abe9294/contrib/zmq/zmq_sub.py
  27 """
  28
  29 import binascii
  30 import asyncio
  31 import zmq
  32 import zmq.asyncio
  33 import signal
  34 import struct
  35 import sys
  36
  37 if not (sys.version_info.major >= 3 and sys.version_info.minor >= 4):
  38     print("This example only works with Python 3.4 and greater")
  39     exit(1)
  40
  41 port = 28332
  42
  43 class ZMQHandler():
  44     def __init__(self):
  45         self.loop = zmq.asyncio.install()
  46         self.zmqContext = zmq.asyncio.Context()
  47
  48         self.zmqSubSocket = self.zmqContext.socket(zmq.SUB)
  49         self.zmqSubSocket.setsockopt_string(zmq.SUBSCRIBE, "hashblock")
  50         self.zmqSubSocket.setsockopt_string(zmq.SUBSCRIBE, "hashtx")
  51         self.zmqSubSocket.setsockopt_string(zmq.SUBSCRIBE, "rawblock")
  52         self.zmqSubSocket.setsockopt_string(zmq.SUBSCRIBE, "rawtx")
  53         self.zmqSubSocket.connect("tcp://127.0.0.1:%i" % port)
  54
  55     @asyncio.coroutine
  56     def handle(self) :
  57         msg = yield from self.zmqSubSocket.recv_multipart()
  58         topic = msg[0]
  59         body = msg[1]
  60         sequence = "Unknown"
  61         if len(msg[-1]) == 4:
  62           msgSequence = struct.unpack('<I', msg[-1])[-1]
  63           sequence = str(msgSequence)
  64         if topic == b"hashblock":
  65             print('- HASH BLOCK ('+sequence+') -')
  66             print(binascii.hexlify(body))
  67         elif topic == b"hashtx":
  68             print('- HASH TX  ('+sequence+') -')
  69             print(binascii.hexlify(body))
  70         elif topic == b"rawblock":
  71             print('- RAW BLOCK HEADER ('+sequence+') -')
  72             print(binascii.hexlify(body[:80]))
  73         elif topic == b"rawtx":
  74             print('- RAW TX ('+sequence+') -')
  75             print(binascii.hexlify(body))
  76         # schedule ourselves to receive the next message
  77         asyncio.ensure_future(self.handle())
  78
  79     def start(self):
  80         self.loop.add_signal_handler(signal.SIGINT, self.stop)
  81         self.loop.create_task(self.handle())
  82         self.loop.run_forever()
  83
  84     def stop(self):
  85         self.loop.stop()
  86         self.zmqContext.destroy()
  87
  88 daemon = ZMQHandler()
  89 daemon.start()