| 129 | # Restart node with the specified zmq notifications enabled, subscribe to |
| 130 | # all of them and return the corresponding ZMQSubscriber objects. |
| 131 | def setup_zmq_test(self, services, *, recv_timeout=60, sync_blocks=True, ipv6=False): |
| 132 | subscribers = [] |
| 133 | for topic, address in services: |
| 134 | socket = self.ctx.socket(zmq.SUB) |
| 135 | if ipv6: |
| 136 | socket.setsockopt(zmq.IPV6, 1) |
| 137 | subscribers.append(ZMQSubscriber(socket, topic.encode())) |
| 138 | |
| 139 | self.restart_node(0, [f"-zmqpub{topic}={address}" for topic, address in services] + |
| 140 | self.extra_args[0]) |
| 141 | |
| 142 | for i, sub in enumerate(subscribers): |
| 143 | sub.socket.connect(services[i][1]) |
| 144 | |
| 145 | # Ensure that all zmq publisher notification interfaces are ready by |
| 146 | # running the following "sync up" procedure: |
| 147 | # 1. Generate a block on the node |
| 148 | # 2. Try to receive the corresponding notification on all subscribers |
| 149 | # 3. If all subscribers get the message within the timeout (1 second), |
| 150 | # we are done, otherwise repeat starting from step 1 |
| 151 | for sub in subscribers: |
| 152 | sub.socket.set(zmq.RCVTIMEO, 1000) |
| 153 | while True: |
| 154 | test_block = ZMQTestSetupBlock(self, self.nodes[0]) |
| 155 | recv_failed = False |
| 156 | for sub in subscribers: |
| 157 | try: |
| 158 | while not test_block.caused_notification(sub.receive().hex()): |
| 159 | self.log.debug("Ignoring sync-up notification for previously generated block.") |
| 160 | except zmq.error.Again: |
| 161 | self.log.debug("Didn't receive sync-up notification, trying again.") |
| 162 | recv_failed = True |
| 163 | if not recv_failed: |
| 164 | self.log.debug("ZMQ sync-up completed, all subscribers are ready.") |
| 165 | break |
| 166 | |
| 167 | # set subscriber's desired timeout for the test |
| 168 | for sub in subscribers: |
| 169 | sub.socket.set(zmq.RCVTIMEO, recv_timeout*1000) |
| 170 | |
| 171 | self.connect_nodes(0, 1) |
| 172 | if sync_blocks: |
| 173 | self.sync_blocks() |
| 174 | |
| 175 | return subscribers |
| 176 | |
| 177 | def test_basic(self): |
| 178 | |