Merge #9485: ZMQ example using python3 and asyncio

b471daf Adddress nits, use asyncio signal handling, create_task (Bob McElrath)
4bb7d1b Add python version checks and 3.4 example (Bob McElrath)
5406d51 Rewrite to not use Polling wrapper for asyncio, link to python2.7 example (Bob McElrath)
5ea5368 ZMQ example using python3 and asyncio (Bob McElrath)

1 files changed, 69 insertions, 28 deletions

diff --git a/contrib/zmq/zmq_sub.py b/contrib/zmq/zmq_sub.py
index 5707188f23..ea398a27ea 100755
--- a/contrib/zmq/zmq_sub.py
+++ b/contrib/zmq/zmq_sub.py
@@ -1,43 +1,84 @@
-#!/usr/bin/env python2
+#!/usr/bin/env python3
 # Copyright (c) 2014-2016 The Bitcoin Core developers
 # Distributed under the MIT software license, see the accompanying
 # file COPYING or http://www.opensource.org/licenses/mit-license.php.
 
+"""
+    ZMQ example using python3's asyncio
+
+    Bitcoin should be started with the command line arguments:
+        bitcoind -testnet -daemon \
+                -zmqpubhashblock=tcp://127.0.0.1:28332 \
+                -zmqpubrawtx=tcp://127.0.0.1:28332 \
+                -zmqpubhashtx=tcp://127.0.0.1:28332 \
+                -zmqpubhashblock=tcp://127.0.0.1:28332
+
+    We use the asyncio library here.  `self.handle()` installs itself as a
+    future at the end of the function.  Since it never returns with the event
+    loop having an empty stack of futures, this creates an infinite loop.  An
+    alternative is to wrap the contents of `handle` inside `while True`.
+
+    A blocking example using python 2.7 can be obtained from the git history:
+    https://github.com/bitcoin/bitcoin/blob/37a7fe9e440b83e2364d5498931253937abe9294/contrib/zmq/zmq_sub.py
+"""
+
 import binascii
+import asyncio
 import zmq
+import zmq.asyncio
+import signal
 import struct
+import sys
+
+if not (sys.version_info.major >= 3 and sys.version_info.minor >= 5):
+    print("This example only works with Python 3.5 and greater")
+    exit(1)
 
 port = 28332
 
-zmqContext = zmq.Context()
-zmqSubSocket = zmqContext.socket(zmq.SUB)
-zmqSubSocket.setsockopt(zmq.SUBSCRIBE, "hashblock")
-zmqSubSocket.setsockopt(zmq.SUBSCRIBE, "hashtx")
-zmqSubSocket.setsockopt(zmq.SUBSCRIBE, "rawblock")
-zmqSubSocket.setsockopt(zmq.SUBSCRIBE, "rawtx")
-zmqSubSocket.connect("tcp://127.0.0.1:%i" % port)
-
-try:
-    while True:
-        msg = zmqSubSocket.recv_multipart()
-        topic = str(msg[0])
+class ZMQHandler():
+    def __init__(self):
+        self.loop = zmq.asyncio.install()
+        self.zmqContext = zmq.asyncio.Context()
+
+        self.zmqSubSocket = self.zmqContext.socket(zmq.SUB)
+        self.zmqSubSocket.setsockopt_string(zmq.SUBSCRIBE, "hashblock")
+        self.zmqSubSocket.setsockopt_string(zmq.SUBSCRIBE, "hashtx")
+        self.zmqSubSocket.setsockopt_string(zmq.SUBSCRIBE, "rawblock")
+        self.zmqSubSocket.setsockopt_string(zmq.SUBSCRIBE, "rawtx")
+        self.zmqSubSocket.connect("tcp://127.0.0.1:%i" % port)
+
+    async def handle(self) :
+        msg = await self.zmqSubSocket.recv_multipart()
+        topic = msg[0]
         body = msg[1]
         sequence = "Unknown"
         if len(msg[-1]) == 4:
           msgSequence = struct.unpack('<I', msg[-1])[-1]
           sequence = str(msgSequence)
-        if topic == "hashblock":
-            print '- HASH BLOCK ('+sequence+') -'
-            print binascii.hexlify(body)
-        elif topic == "hashtx":
-            print '- HASH TX  ('+sequence+') -'
-            print binascii.hexlify(body)
-        elif topic == "rawblock":
-            print '- RAW BLOCK HEADER ('+sequence+') -'
-            print binascii.hexlify(body[:80])
-        elif topic == "rawtx":
-            print '- RAW TX ('+sequence+') -'
-            print binascii.hexlify(body)
-
-except KeyboardInterrupt:
-    zmqContext.destroy()
+        if topic == b"hashblock":
+            print('- HASH BLOCK ('+sequence+') -')
+            print(binascii.hexlify(body))
+        elif topic == b"hashtx":
+            print('- HASH TX  ('+sequence+') -')
+            print(binascii.hexlify(body))
+        elif topic == b"rawblock":
+            print('- RAW BLOCK HEADER ('+sequence+') -')
+            print(binascii.hexlify(body[:80]))
+        elif topic == b"rawtx":
+            print('- RAW TX ('+sequence+') -')
+            print(binascii.hexlify(body))
+        # schedule ourselves to receive the next message
+        asyncio.ensure_future(self.handle())
+
+    def start(self):
+        self.loop.add_signal_handler(signal.SIGINT, self.stop)
+        self.loop.create_task(self.handle())
+        self.loop.run_forever()
+
+    def stop(self):
+        self.loop.stop()
+        self.zmqContext.destroy()
+
+daemon = ZMQHandler()
+daemon.start()