Prevent pubsub managers from ever crashing

Fixes #1262

src/socketio/async_pubsub_manager.py

@@ -230,11 +230,13 @@ class AsyncPubSubManager(AsyncManager):
                                     await self._handle_close_room(data)
                         except asyncio.CancelledError:
                             raise  # let the outer try/except handle it
-                        except:
+                        except Exception:
                             self.server.logger.exception(
-                                'Unknown error in pubsub listening task')
+                                'Handler error in pubsub listening thread')
+                self.server.logger.error('pubsub listen() exited unexpectedly')
+                break  # loop should never exit except in unit tests!
             except asyncio.CancelledError:  # pragma: no cover
                 break
-            except:  # pragma: no cover
+            except Exception:  # pragma: no cover
-                import traceback
+                self.server.logger.exception('Unexpected Error in pubsub '
-                traceback.print_exc()
+                                             'listening thread')

src/socketio/pubsub_manager.py

@@ -188,38 +188,45 @@ class PubSubManager(Manager):
                            namespace=message.get('namespace'))
 
     def _thread(self):
-        for message in self._listen():
+        while True:
-            data = None
+            try:
-            if isinstance(message, dict):
+                for message in self._listen():
-                data = message
+                    data = None
-            else:
+                    if isinstance(message, dict):
-                if isinstance(message, bytes):  # pragma: no cover
+                        data = message
-                    try:
+                    else:
-                        data = pickle.loads(message)
+                        if isinstance(message, bytes):  # pragma: no cover
-                    except:
+                            try:
-                        pass
+                                data = pickle.loads(message)
-                if data is None:
+                            except:
-                    try:
+                                pass
-                        data = json.loads(message)
+                        if data is None:
-                    except:
+                            try:
-                        pass
+                                data = json.loads(message)
-            if data and 'method' in data:
+                            except:
-                self._get_logger().debug('pubsub message: {}'.format(
+                                pass
-                    data['method']))
+                    if data and 'method' in data:
-                try:
+                        self._get_logger().debug('pubsub message: {}'.format(
-                    if data['method'] == 'callback':
+                            data['method']))
-                        self._handle_callback(data)
+                        try:
-                    elif data.get('host_id') != self.host_id:
+                            if data['method'] == 'callback':
-                        if data['method'] == 'emit':
+                                self._handle_callback(data)
-                            self._handle_emit(data)
+                            elif data.get('host_id') != self.host_id:
-                        elif data['method'] == 'disconnect':
+                                if data['method'] == 'emit':
-                            self._handle_disconnect(data)
+                                    self._handle_emit(data)
-                        elif data['method'] == 'enter_room':
+                                elif data['method'] == 'disconnect':
-                            self._handle_enter_room(data)
+                                    self._handle_disconnect(data)
-                        elif data['method'] == 'leave_room':
+                                elif data['method'] == 'enter_room':
-                            self._handle_leave_room(data)
+                                    self._handle_enter_room(data)
-                        elif data['method'] == 'close_room':
+                                elif data['method'] == 'leave_room':
-                            self._handle_close_room(data)
+                                    self._handle_leave_room(data)
-                except:
+                                elif data['method'] == 'close_room':
-                    self.server.logger.exception(
+                                    self._handle_close_room(data)
-                        'Unknown error in pubsub listening thread')
+                        except Exception:
+                            self.server.logger.exception(
+                                'Handler error in pubsub listening thread')
+                self.server.logger.error('pubsub listen() exited unexpectedly')
+                break  # loop should never exit except in unit tests!
+            except Exception:  # pragma: no cover
+                self.server.logger.exception('Unexpected Error in pubsub '
+                                             'listening thread')

tests/async/test_pubsub_manager.py

@@ -541,7 +541,6 @@ class TestAsyncPubSubManager(unittest.TestCase):
                    'host_id': host_id}
             yield pickle.dumps({'method': 'close_room', 'value': 'baz',
                                 'host_id': host_id})
-            raise asyncio.CancelledError()  # force the thread to exit
 
         self.pm._listen = messages
         _run(self.pm._thread())