| 70 | self.stop_event.set() |
| 71 | |
| 72 | def _start_thread(self, started_event): |
| 73 | self.loop = asyncio.new_event_loop() |
| 74 | asyncio.set_event_loop(self.loop) |
| 75 | |
| 76 | self.connectivity = asyncio.Event() |
| 77 | self.connectivity.set() |
| 78 | self.connectivity_loss = asyncio.Event() |
| 79 | self.stop_event = asyncio.Event() |
| 80 | |
| 81 | if self.listening_port is None: |
| 82 | self.listening_port = cluster.find_available_port() |
| 83 | |
| 84 | self.sock = socket.socket() |
| 85 | self.sock.bind((self.listening_addr, self.listening_port)) |
| 86 | self.sock.listen(50) |
| 87 | self.sock.setblocking(False) |
| 88 | |
| 89 | try: |
| 90 | self.loop.run_until_complete(self._main(started_event)) |
| 91 | finally: |
| 92 | self.loop.close() |
| 93 | |
| 94 | async def _main(self, started_event): |
| 95 | self.listen_task = asyncio.ensure_future(self.listen()) |