Rewrite to not use Polling wrapper for asyncio, link to python2.7 example

2 files changed, 51 insertions, 81 deletions

diff --git a/contrib/zmq/zmq_sub.py b/contrib/zmq/zmq_sub.py
index 6f57aa47e9..e462198bfe 100755
--- a/contrib/zmq/zmq_sub.py
+++ b/contrib/zmq/zmq_sub.py
@@ -1,43 +1,65 @@
-#!/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.
 
+# 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 array
 import binascii
-import zmq
+import asyncio, zmq, zmq.asyncio
+import signal
 import struct
 
 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):
+        asyncio.ensure_future(self.handle())
+        self.loop.run_forever()
+
+    def stop(self):
+        self.loop.stop()
+        self.zmqContext.destroy()
+
+daemon = ZMQHandler()
+def signal_handler(num, frame):
+    daemon.stop()
+    exit(0)
+signal.signal(signal.SIGINT, signal_handler)
+daemon.start()
diff --git a/contrib/zmq/zmq_sub3.py b/contrib/zmq/zmq_sub3.py
deleted file mode 100755
index 79cb5b484c..0000000000
--- a/contrib/zmq/zmq_sub3.py
+++ /dev/null
@@ -1,52 +0,0 @@
-#!/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.
-
-import array
-import binascii
-import asyncio, zmq, zmq.asyncio
-import struct
-
-port = 28332
-
-zmqContext = zmq.asyncio.Context()
-
-async def recv_and_process():
-    zmqSubSocket = zmqContext.socket(zmq.SUB)
-    zmqSubSocket.setsockopt_string(zmq.SUBSCRIBE, "hashblock")
-    zmqSubSocket.setsockopt_string(zmq.SUBSCRIBE, "hashtx")
-    zmqSubSocket.setsockopt_string(zmq.SUBSCRIBE, "rawblock")
-    zmqSubSocket.setsockopt_string(zmq.SUBSCRIBE, "rawtx")
-    zmqSubSocket.connect("tcp://127.0.0.1:%i" % port)
-
-    poller = zmq.asyncio.Poller()
-    poller.register(zmqSubSocket, zmq.POLLIN)
-    while True:
-        s = await poller.poll()
-        msg = await s[0][0].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 == 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))
-
-try:
-    loop = zmq.asyncio.ZMQEventLoop()
-    asyncio.set_event_loop(loop)
-    loop.run_until_complete(setup())
-except KeyboardInterrupt:
-    zmqContext.destroy()