|
|
|
@ -80,34 +80,43 @@ class ZMQTest (BitcoinTestFramework):
|
|
|
|
|
self.log.debug("Destroying ZMQ context")
|
|
|
|
|
self.ctx.destroy(linger=None)
|
|
|
|
|
|
|
|
|
|
# Restart node with the specified zmq notifications enabled, subscribe to
|
|
|
|
|
# all of them and return the corresponding ZMQSubscriber objects.
|
|
|
|
|
def setup_zmq_test(self, services, recv_timeout=60, connect_nodes=False):
|
|
|
|
|
subscribers = []
|
|
|
|
|
for topic, address in services:
|
|
|
|
|
socket = self.ctx.socket(zmq.SUB)
|
|
|
|
|
socket.set(zmq.RCVTIMEO, recv_timeout*1000)
|
|
|
|
|
subscribers.append(ZMQSubscriber(socket, topic.encode()))
|
|
|
|
|
|
|
|
|
|
self.restart_node(0, ["-zmqpub%s=%s" % (topic, address) for topic, address in services])
|
|
|
|
|
|
|
|
|
|
if connect_nodes:
|
|
|
|
|
self.connect_nodes(0, 1)
|
|
|
|
|
|
|
|
|
|
for i, sub in enumerate(subscribers):
|
|
|
|
|
sub.socket.connect(services[i][1])
|
|
|
|
|
|
|
|
|
|
# Relax so that the subscribers are ready before publishing zmq messages
|
|
|
|
|
sleep(0.2)
|
|
|
|
|
|
|
|
|
|
return subscribers
|
|
|
|
|
|
|
|
|
|
def test_basic(self):
|
|
|
|
|
|
|
|
|
|
# Invalid zmq arguments don't take down the node, see #17185.
|
|
|
|
|
self.restart_node(0, ["-zmqpubrawtx=foo", "-zmqpubhashtx=bar"])
|
|
|
|
|
|
|
|
|
|
address = 'tcp://127.0.0.1:28332'
|
|
|
|
|
sockets = []
|
|
|
|
|
subs = []
|
|
|
|
|
services = [b"hashblock", b"hashtx", b"rawblock", b"rawtx"]
|
|
|
|
|
for service in services:
|
|
|
|
|
sockets.append(self.ctx.socket(zmq.SUB))
|
|
|
|
|
sockets[-1].set(zmq.RCVTIMEO, 60000)
|
|
|
|
|
subs.append(ZMQSubscriber(sockets[-1], service))
|
|
|
|
|
|
|
|
|
|
# Subscribe to all available topics.
|
|
|
|
|
subs = self.setup_zmq_test(
|
|
|
|
|
[(topic, address) for topic in ["hashblock", "hashtx", "rawblock", "rawtx"]],
|
|
|
|
|
connect_nodes=True)
|
|
|
|
|
|
|
|
|
|
hashblock = subs[0]
|
|
|
|
|
hashtx = subs[1]
|
|
|
|
|
rawblock = subs[2]
|
|
|
|
|
rawtx = subs[3]
|
|
|
|
|
|
|
|
|
|
self.restart_node(0, ["-zmqpub%s=%s" % (sub.topic.decode(), address) for sub in [hashblock, hashtx, rawblock, rawtx]])
|
|
|
|
|
self.connect_nodes(0, 1)
|
|
|
|
|
for socket in sockets:
|
|
|
|
|
socket.connect(address)
|
|
|
|
|
|
|
|
|
|
# Relax so that the subscriber is ready before publishing zmq messages
|
|
|
|
|
sleep(0.2)
|
|
|
|
|
|
|
|
|
|
num_blocks = 5
|
|
|
|
|
self.log.info("Generate %(n)d blocks (and %(n)d coinbase txes)" % {"n": num_blocks})
|
|
|
|
|
genhashes = self.nodes[0].generatetoaddress(num_blocks, ADDRESS_BCRT1_UNSPENDABLE)
|
|
|
|
@ -174,25 +183,10 @@ class ZMQTest (BitcoinTestFramework):
|
|
|
|
|
|
|
|
|
|
address = 'tcp://127.0.0.1:28333'
|
|
|
|
|
|
|
|
|
|
services = [b"hashblock", b"hashtx"]
|
|
|
|
|
sockets = []
|
|
|
|
|
subs = []
|
|
|
|
|
for service in services:
|
|
|
|
|
sockets.append(self.ctx.socket(zmq.SUB))
|
|
|
|
|
# 2 second timeout to check end of notifications
|
|
|
|
|
sockets[-1].set(zmq.RCVTIMEO, 2000)
|
|
|
|
|
subs.append(ZMQSubscriber(sockets[-1], service))
|
|
|
|
|
|
|
|
|
|
# Subscribe to all available topics.
|
|
|
|
|
hashblock = subs[0]
|
|
|
|
|
hashtx = subs[1]
|
|
|
|
|
|
|
|
|
|
# Should only notify the tip if a reorg occurs
|
|
|
|
|
self.restart_node(0, ["-zmqpub%s=%s" % (sub.topic.decode(), address) for sub in [hashblock, hashtx]])
|
|
|
|
|
for socket in sockets:
|
|
|
|
|
socket.connect(address)
|
|
|
|
|
# Relax so that the subscriber is ready before publishing zmq messages
|
|
|
|
|
sleep(0.2)
|
|
|
|
|
hashblock, hashtx = self.setup_zmq_test(
|
|
|
|
|
[(topic, address) for topic in ["hashblock", "hashtx"]],
|
|
|
|
|
recv_timeout=2) # 2 second timeout to check end of notifications
|
|
|
|
|
|
|
|
|
|
# Generate 1 block in nodes[0] with 1 mempool tx and receive all notifications
|
|
|
|
|
payment_txid = self.nodes[0].sendtoaddress(self.nodes[0].getnewaddress(), 1.0)
|
|
|
|
@ -240,15 +234,7 @@ class ZMQTest (BitcoinTestFramework):
|
|
|
|
|
<32-byte hash>A<8-byte LE uint> : Transactionhash added mempool
|
|
|
|
|
"""
|
|
|
|
|
self.log.info("Testing 'sequence' publisher")
|
|
|
|
|
address = 'tcp://127.0.0.1:28333'
|
|
|
|
|
socket = self.ctx.socket(zmq.SUB)
|
|
|
|
|
socket.set(zmq.RCVTIMEO, 60000)
|
|
|
|
|
seq = ZMQSubscriber(socket, b'sequence')
|
|
|
|
|
|
|
|
|
|
self.restart_node(0, ['-zmqpub%s=%s' % (seq.topic.decode(), address)])
|
|
|
|
|
socket.connect(address)
|
|
|
|
|
# Relax so that the subscriber is ready before publishing zmq messages
|
|
|
|
|
sleep(0.2)
|
|
|
|
|
[seq] = self.setup_zmq_test([("sequence", "tcp://127.0.0.1:28333")])
|
|
|
|
|
|
|
|
|
|
# Mempool sequence number starts at 1
|
|
|
|
|
seq_num = 1
|
|
|
|
@ -399,16 +385,7 @@ class ZMQTest (BitcoinTestFramework):
|
|
|
|
|
return
|
|
|
|
|
|
|
|
|
|
self.log.info("Testing 'mempool sync' usage of sequence notifier")
|
|
|
|
|
address = 'tcp://127.0.0.1:28333'
|
|
|
|
|
socket = self.ctx.socket(zmq.SUB)
|
|
|
|
|
socket.set(zmq.RCVTIMEO, 60000)
|
|
|
|
|
seq = ZMQSubscriber(socket, b'sequence')
|
|
|
|
|
|
|
|
|
|
self.restart_node(0, ['-zmqpub%s=%s' % (seq.topic.decode(), address)])
|
|
|
|
|
self.connect_nodes(0, 1)
|
|
|
|
|
socket.connect(address)
|
|
|
|
|
# Relax so that the subscriber is ready before publishing zmq messages
|
|
|
|
|
sleep(0.2)
|
|
|
|
|
[seq] = self.setup_zmq_test([("sequence", "tcp://127.0.0.1:28333")], connect_nodes=True)
|
|
|
|
|
|
|
|
|
|
# In-memory counter, should always start at 1
|
|
|
|
|
next_mempool_seq = self.nodes[0].getrawmempool(mempool_sequence=True)["mempool_sequence"]
|
|
|
|
@ -508,26 +485,17 @@ class ZMQTest (BitcoinTestFramework):
|
|
|
|
|
|
|
|
|
|
def test_multiple_interfaces(self):
|
|
|
|
|
# Set up two subscribers with different addresses
|
|
|
|
|
subscribers = []
|
|
|
|
|
for i in range(2):
|
|
|
|
|
address = 'tcp://127.0.0.1:%d' % (28334 + i)
|
|
|
|
|
socket = self.ctx.socket(zmq.SUB)
|
|
|
|
|
socket.set(zmq.RCVTIMEO, 60000)
|
|
|
|
|
hashblock = ZMQSubscriber(socket, b"hashblock")
|
|
|
|
|
socket.connect(address)
|
|
|
|
|
subscribers.append({'address': address, 'hashblock': hashblock})
|
|
|
|
|
|
|
|
|
|
self.restart_node(0, ['-zmqpub%s=%s' % (subscriber['hashblock'].topic.decode(), subscriber['address']) for subscriber in subscribers])
|
|
|
|
|
|
|
|
|
|
# Relax so that the subscriber is ready before publishing zmq messages
|
|
|
|
|
sleep(0.2)
|
|
|
|
|
subscribers = self.setup_zmq_test([
|
|
|
|
|
("hashblock", "tcp://127.0.0.1:28334"),
|
|
|
|
|
("hashblock", "tcp://127.0.0.1:28335"),
|
|
|
|
|
])
|
|
|
|
|
|
|
|
|
|
# Generate 1 block in nodes[0] and receive all notifications
|
|
|
|
|
self.nodes[0].generatetoaddress(1, ADDRESS_BCRT1_UNSPENDABLE)
|
|
|
|
|
|
|
|
|
|
# Should receive the same block hash on both subscribers
|
|
|
|
|
assert_equal(self.nodes[0].getbestblockhash(), subscribers[0]['hashblock'].receive().hex())
|
|
|
|
|
assert_equal(self.nodes[0].getbestblockhash(), subscribers[1]['hashblock'].receive().hex())
|
|
|
|
|
assert_equal(self.nodes[0].getbestblockhash(), subscribers[0].receive().hex())
|
|
|
|
|
assert_equal(self.nodes[0].getbestblockhash(), subscribers[1].receive().hex())
|
|
|
|
|
|
|
|
|
|
if __name__ == '__main__':
|
|
|
|
|
ZMQTest().main()
|
|
|
|
|