fastapi/sub.py

137 lines
5.3 KiB
Python
Raw Normal View History

2023-03-10 10:46:21 +00:00
#!/usr/bin/env python3
# Copyright (c) 2014-2018 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
Litecoind should be started with the command line arguments:
litecoind -testnet -daemon \
-zmqpubrawtx=tcp://127.0.0.1:28332 \
-zmqpubrawblock=tcp://127.0.0.1:28332 \
-zmqpubhashtx=tcp://127.0.0.1:28332 \
-zmqpubhashblock=tcp://127.0.0.1:28332 \
-zmqpubsequence=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
import time, requests, json
if (sys.version_info.major, sys.version_info.minor) < (3, 5):
print("This example only works with Python 3.5 and greater")
sys.exit(1)
class RPCHost(object):
def __init__(self, url):
self._session = requests.Session()
self._url = url
self._headers = {'content-type': 'application/json'}
def call(self, rpcMethod, *params):
payload = json.dumps({"method": rpcMethod, "params": list(params), "jsonrpc": "2.0"})
tries = 5
hadConnectionFailures = False
while True:
try:
response = self._session.post(self._url, headers=self._headers, data=payload)
except requests.exceptions.ConnectionError:
tries -= 1
if tries == 0:
raise Exception('Failed to connect for remote procedure call.')
hadFailedConnections = True
print("Couldn't connect for remote procedure call, will sleep for five seconds and then try again ({} more tries)".format(tries))
time.sleep(10)
else:
if hadConnectionFailures:
print('Connected for remote procedure call after retry.')
break
if not response.status_code in (200, 500):
raise Exception('RPC connection failure: ' + str(response.status_code) + ' ' + response.reason)
responseJSON = response.json()
if 'error' in responseJSON and responseJSON['error'] != None:
raise Exception('Error in RPC call: ' + str(responseJSON['error']))
return responseJSON['result']
ZMQ_PORT = 28442
SERVER_IP = '172.16.0.10'
SERVER_PORT = 8442
SERVER_USER = 'litecoinrpc'
SERVER_PASS = 'uhalalala'
serverURL = "http://%s:%s@%s:%s" % (SERVER_USER, SERVER_PASS, SERVER_IP, SERVER_PORT)
rpc = RPCHost(serverURL)
class ZMQHandler():
def __init__(self):
self.loop = asyncio.get_event_loop()
self.zmqContext = zmq.asyncio.Context()
self.zmqSubSocket = self.zmqContext.socket(zmq.SUB)
self.zmqSubSocket.setsockopt(zmq.RCVHWM, 0)
#self.zmqSubSocket.setsockopt_string(zmq.SUBSCRIBE, "")
#self.zmqSubSocket.setsockopt_string(zmq.SUBSCRIBE, "hashtx")
#self.zmqSubSocket.setsockopt_string(zmq.SUBSCRIBE, "rawblock")
#self.zmqSubSocket.setsockopt_string(zmq.SUBSCRIBE, "rawtx")
#self.zmqSubSocket.setsockopt_string(zmq.SUBSCRIBE, "sequence")
self.zmqSubSocket.connect("tcp://%s:%i" % (SERVER_IP, ZMQ_PORT))
async def handle(self) :
topic, body, seq = await self.zmqSubSocket.recv_multipart()
sequence = "Unknown"
if len(seq) == 4:
sequence = str(struct.unpack('<I', seq)[-1])
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+') -')
#decode = rpc.call('decoderawtransaction', binascii.hexlify(body).decode('utf-8'))
#data = json.dumps(decode)
#print(data)
#print(binascii.hexlify(body))
elif topic == b"sequence":
hash = binascii.hexlify(body[:32])
label = chr(body[32])
mempool_sequence = None if len(body) != 32+1+8 else struct.unpack("<Q", body[32+1:])[0]
print('- SEQUENCE ('+sequence+') -')
print(hash, label, mempool_sequence)
else:
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()