(self)
| 103 | self.loop = loop |
| 104 | |
| 105 | def start(self): |
| 106 | prop = dict( |
| 107 | iserver=self.iserver, |
| 108 | loop=self.loop, |
| 109 | logger=self.logger, |
| 110 | policies=self._policies, |
| 111 | clients=self.clients |
| 112 | ) |
| 113 | protocol_factory = type('OPCUAProtocol', (OPCUAProtocol,), prop) |
| 114 | |
| 115 | coro = self.loop.create_server(protocol_factory, self.hostname, self.port) |
| 116 | self._server = self.loop.run_coro_and_wait(coro) |
| 117 | # get the port and the hostname from the created server socket |
| 118 | # only relevant for dynamic port asignment (when self.port == 0) |
| 119 | if self.port == 0 and len(self._server.sockets) == 1: |
| 120 | # will work for AF_INET and AF_INET6 socket names |
| 121 | # these are to only families supported by the create_server call |
| 122 | sockname = self._server.sockets[0].getsockname() |
| 123 | self.hostname = sockname[0] |
| 124 | self.port = sockname[1] |
| 125 | self.logger.warning('Listening on {0}:{1}'.format(self.hostname, self.port)) |
| 126 | |
| 127 | def stop(self): |
| 128 | self.logger.info("Closing asyncio socket server") |
nothing calls this directly
no test coverage detected