Skip to content

Commit 5406d51

Browse files
author
Bob McElrath
committed
Rewrite to not use Polling wrapper for asyncio, link to python2.7 example
1 parent 5ea5368 commit 5406d51

File tree

2 files changed

+51
-81
lines changed

2 files changed

+51
-81
lines changed

contrib/zmq/zmq_sub.py

Lines changed: 51 additions & 29 deletions
Original file line numberDiff line numberDiff line change
@@ -1,43 +1,65 @@
1-
#!/usr/bin/env python2
1+
#!/usr/bin/env python3
22
# Copyright (c) 2014-2016 The Bitcoin Core developers
33
# Distributed under the MIT software license, see the accompanying
44
# file COPYING or http://www.opensource.org/licenses/mit-license.php.
55

6+
# A blocking example using python 2.7 can be obtained from the git history:
7+
# https://github.com/bitcoin/bitcoin/blob/37a7fe9e440b83e2364d5498931253937abe9294/contrib/zmq/zmq_sub.py
8+
9+
import array
610
import binascii
7-
import zmq
11+
import asyncio, zmq, zmq.asyncio
12+
import signal
813
import struct
914

1015
port = 28332
1116

12-
zmqContext = zmq.Context()
13-
zmqSubSocket = zmqContext.socket(zmq.SUB)
14-
zmqSubSocket.setsockopt(zmq.SUBSCRIBE, "hashblock")
15-
zmqSubSocket.setsockopt(zmq.SUBSCRIBE, "hashtx")
16-
zmqSubSocket.setsockopt(zmq.SUBSCRIBE, "rawblock")
17-
zmqSubSocket.setsockopt(zmq.SUBSCRIBE, "rawtx")
18-
zmqSubSocket.connect("tcp://127.0.0.1:%i" % port)
19-
20-
try:
21-
while True:
22-
msg = zmqSubSocket.recv_multipart()
23-
topic = str(msg[0])
17+
class ZMQHandler():
18+
def __init__(self):
19+
self.loop = zmq.asyncio.install()
20+
self.zmqContext = zmq.asyncio.Context()
21+
22+
self.zmqSubSocket = self.zmqContext.socket(zmq.SUB)
23+
self.zmqSubSocket.setsockopt_string(zmq.SUBSCRIBE, "hashblock")
24+
self.zmqSubSocket.setsockopt_string(zmq.SUBSCRIBE, "hashtx")
25+
self.zmqSubSocket.setsockopt_string(zmq.SUBSCRIBE, "rawblock")
26+
self.zmqSubSocket.setsockopt_string(zmq.SUBSCRIBE, "rawtx")
27+
self.zmqSubSocket.connect("tcp://127.0.0.1:%i" % port)
28+
29+
async def handle(self) :
30+
msg = await self.zmqSubSocket.recv_multipart()
31+
topic = msg[0]
2432
body = msg[1]
2533
sequence = "Unknown";
2634
if len(msg[-1]) == 4:
2735
msgSequence = struct.unpack('<I', msg[-1])[-1]
2836
sequence = str(msgSequence)
29-
if topic == "hashblock":
30-
print '- HASH BLOCK ('+sequence+') -'
31-
print binascii.hexlify(body)
32-
elif topic == "hashtx":
33-
print '- HASH TX ('+sequence+') -'
34-
print binascii.hexlify(body)
35-
elif topic == "rawblock":
36-
print '- RAW BLOCK HEADER ('+sequence+') -'
37-
print binascii.hexlify(body[:80])
38-
elif topic == "rawtx":
39-
print '- RAW TX ('+sequence+') -'
40-
print binascii.hexlify(body)
41-
42-
except KeyboardInterrupt:
43-
zmqContext.destroy()
37+
if topic == b"hashblock":
38+
print('- HASH BLOCK ('+sequence+') -')
39+
print(binascii.hexlify(body))
40+
elif topic == b"hashtx":
41+
print('- HASH TX ('+sequence+') -')
42+
print(binascii.hexlify(body))
43+
elif topic == b"rawblock":
44+
print('- RAW BLOCK HEADER ('+sequence+') -')
45+
print(binascii.hexlify(body[:80]))
46+
elif topic == b"rawtx":
47+
print('- RAW TX ('+sequence+') -')
48+
print(binascii.hexlify(body))
49+
# schedule ourselves to receive the next message
50+
asyncio.ensure_future(self.handle())
51+
52+
def start(self):
53+
asyncio.ensure_future(self.handle())
54+
self.loop.run_forever()
55+
56+
def stop(self):
57+
self.loop.stop()
58+
self.zmqContext.destroy()
59+
60+
daemon = ZMQHandler()
61+
def signal_handler(num, frame):
62+
daemon.stop()
63+
exit(0)
64+
signal.signal(signal.SIGINT, signal_handler)
65+
daemon.start()

contrib/zmq/zmq_sub3.py

Lines changed: 0 additions & 52 deletions
This file was deleted.

0 commit comments

Comments
 (0)