| 38 | return hash256(byte_str)[::-1] |
| 39 | |
| 40 | class ZMQSubscriber: |
| 41 | def __init__(self, socket, topic): |
| 42 | self.sequence = None # no sequence number received yet |
| 43 | self.socket = socket |
| 44 | self.topic = topic |
| 45 | |
| 46 | self.socket.setsockopt(zmq.SUBSCRIBE, self.topic) |
| 47 | |
| 48 | # Receive message from publisher and verify that topic and sequence match |
| 49 | def _receive_from_publisher_and_check(self): |
| 50 | topic, body, seq = self.socket.recv_multipart() |
| 51 | # Topic should match the subscriber topic. |
| 52 | assert_equal(topic, self.topic) |
| 53 | # Sequence should be incremental. |
| 54 | received_seq = struct.unpack('<I', seq)[-1] |
| 55 | if self.sequence is None: |
| 56 | self.sequence = received_seq |
| 57 | else: |
| 58 | assert_equal(received_seq, self.sequence) |
| 59 | self.sequence += 1 |
| 60 | return body |
| 61 | |
| 62 | def receive(self): |
| 63 | return self._receive_from_publisher_and_check() |
| 64 | |
| 65 | def receive_sequence(self): |
| 66 | body = self._receive_from_publisher_and_check() |
| 67 | hash = body[:32].hex() |
| 68 | label = chr(body[32]) |
| 69 | mempool_sequence = None if len(body) != 32+1+8 else struct.unpack("<Q", body[32+1:])[0] |
| 70 | if mempool_sequence is not None: |
| 71 | assert label == "A" or label == "R" |
| 72 | else: |
| 73 | assert label == "D" or label == "C" |
| 74 | return (hash, label, mempool_sequence) |
| 75 | |
| 76 | |
| 77 | class ZMQTestSetupBlock: |