Browse Source

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)
0.15
Wladimir J. van der Laan 8 years ago
parent
commit
61a640ea97
No known key found for this signature in database
GPG Key ID: 74810B012346C9A6
  1. 97
      contrib/zmq/zmq_sub.py
  2. 89
      contrib/zmq/zmq_sub3.4.py

97
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 # Copyright (c) 2014-2016 The Bitcoin Core developers
# Distributed under the MIT software license, see the accompanying # Distributed under the MIT software license, see the accompanying
# file COPYING or http://www.opensource.org/licenses/mit-license.php. # 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 binascii
import asyncio
import zmq import zmq
import zmq.asyncio
import signal
import struct 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 port = 28332
zmqContext = zmq.Context() class ZMQHandler():
zmqSubSocket = zmqContext.socket(zmq.SUB) def __init__(self):
zmqSubSocket.setsockopt(zmq.SUBSCRIBE, "hashblock") self.loop = zmq.asyncio.install()
zmqSubSocket.setsockopt(zmq.SUBSCRIBE, "hashtx") self.zmqContext = zmq.asyncio.Context()
zmqSubSocket.setsockopt(zmq.SUBSCRIBE, "rawblock")
zmqSubSocket.setsockopt(zmq.SUBSCRIBE, "rawtx") self.zmqSubSocket = self.zmqContext.socket(zmq.SUB)
zmqSubSocket.connect("tcp://127.0.0.1:%i" % port) self.zmqSubSocket.setsockopt_string(zmq.SUBSCRIBE, "hashblock")
self.zmqSubSocket.setsockopt_string(zmq.SUBSCRIBE, "hashtx")
try: self.zmqSubSocket.setsockopt_string(zmq.SUBSCRIBE, "rawblock")
while True: self.zmqSubSocket.setsockopt_string(zmq.SUBSCRIBE, "rawtx")
msg = zmqSubSocket.recv_multipart() self.zmqSubSocket.connect("tcp://127.0.0.1:%i" % port)
topic = str(msg[0])
async def handle(self) :
msg = await self.zmqSubSocket.recv_multipart()
topic = msg[0]
body = msg[1] body = msg[1]
sequence = "Unknown" sequence = "Unknown"
if len(msg[-1]) == 4: if len(msg[-1]) == 4:
msgSequence = struct.unpack('<I', msg[-1])[-1] msgSequence = struct.unpack('<I', msg[-1])[-1]
sequence = str(msgSequence) sequence = str(msgSequence)
if topic == "hashblock": if topic == b"hashblock":
print '- HASH BLOCK ('+sequence+') -' print('- HASH BLOCK ('+sequence+') -')
print binascii.hexlify(body) print(binascii.hexlify(body))
elif topic == "hashtx": elif topic == b"hashtx":
print '- HASH TX ('+sequence+') -' print('- HASH TX ('+sequence+') -')
print binascii.hexlify(body) print(binascii.hexlify(body))
elif topic == "rawblock": elif topic == b"rawblock":
print '- RAW BLOCK HEADER ('+sequence+') -' print('- RAW BLOCK HEADER ('+sequence+') -')
print binascii.hexlify(body[:80]) print(binascii.hexlify(body[:80]))
elif topic == "rawtx": elif topic == b"rawtx":
print '- RAW TX ('+sequence+') -' print('- RAW TX ('+sequence+') -')
print binascii.hexlify(body) print(binascii.hexlify(body))
# schedule ourselves to receive the next message
except KeyboardInterrupt: asyncio.ensure_future(self.handle())
zmqContext.destroy()
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()

89
contrib/zmq/zmq_sub3.4.py

@ -0,0 +1,89 @@
#!/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`.
The `@asyncio.coroutine` decorator and the `yield from` syntax found here
was introduced in python 3.4 and has been deprecated in favor of the `async`
and `await` keywords respectively.
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 >= 4):
print("This example only works with Python 3.4 and greater")
exit(1)
port = 28332
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)
@asyncio.coroutine
def handle(self) :
msg = yield from 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 == 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()
Loading…
Cancel
Save