Add wrapper to create threads compatible with the selected async mode

engineio/server.py

@@ -308,6 +308,12 @@ def _get_socket(self, sid):
             raise KeyError('Session is disconnected')
         return s
 
+    def _start_background_task(self, target, *args, **kwargs):
+        """Start a background task using the appropriate async model."""
+        return getattr(self.async['threading'],
+                       self.async['thread_class'])(target=target, args=args,
+                                                   kwargs=kwargs)
+
     def _ok(self, packets=None, headers=None, b64=False):
         """Generate a successful HTTP response."""
         if packets is not None:

engineio/socket.py

@@ -150,8 +150,7 @@ def writer():
                 except:
                     break
 
-        writer_task = getattr(self.server.async['threading'],
-                              self.server.async['thread_class'])(target=writer)
+        writer_task = self.server._start_background_task(writer)
         writer_task.start()
 
         self.server.logger.info(

tests/test_server.py

@@ -619,3 +619,16 @@ def loads(*args, **kwargs):
 
         # restore the default JSON module
         packet.Packet.json = json
+
+    def test_background_tasks(self):
+        flag = {}
+
+        def bg_task():
+            flag['task'] = True
+
+        s = server.Server()
+        task = s._start_background_task(bg_task)
+        task.start()
+        task.join()
+        self.assertIn('task', flag)
+        self.assertTrue(flag['task'])

tests/test_socket.py

@@ -28,6 +28,11 @@ def _get_mock_server(self):
                              'queue': queue,
                              'queue_class': 'Queue',
                              'websocket': None}
+
+        def bg_task(target, *args, **kwargs):
+            return threading.Thread(target=target, args=args, kwargs=kwargs)
+
+        mock_server._start_background_task = bg_task
         return mock_server
 
     def test_create(self):