| 11 | |
| 12 | |
| 13 | class SocketIoRunner(object): |
| 14 | def __init__(self, url): |
| 15 | self.host, port_str = url.split(':') |
| 16 | self.port = int(port_str) |
| 17 | self.server = None |
| 18 | |
| 19 | # create the thread object |
| 20 | self.thread = threading.Thread(target=self._start_listening_blocking) |
| 21 | |
| 22 | # wrap Flask application with socketio's middleware |
| 23 | self.app = socketio.Middleware(sio, app) |
| 24 | |
| 25 | def start_listening_async(self): |
| 26 | wsgi.is_accepting = True |
| 27 | self.thread.start() |
| 28 | |
| 29 | def stop_listening(self): |
| 30 | wsgi.is_accepting = False |
| 31 | |
| 32 | def _start_listening_blocking(self): |
| 33 | # deploy as an eventlet WSGI server |
| 34 | listener = eventlet.listen((self.host, self.port)) |
| 35 | self.server = wsgi.server(listener, self.app, log_output=False, debug=False) |
no outgoing calls
no test coverage detected