Add a shutdown() function for the server

src/socketio/asyncio_server.py

@@ -387,6 +387,15 @@ async def disconnect(self, sid, namespace=None, ignore_queue=False):
             await self.manager.disconnect(sid, namespace=namespace,
                                           ignore_queue=True)
 
+    async def shutdown(self):
+        """Stop Socket.IO background tasks.
+
+        This method stops all background activity initiated by the Socket.IO
+        server. It must be called before shutting down the web server.
+        """
+        self.logger.info('Socket.IO is shutting down')
+        await self.eio.shutdown()
+
     async def handle_request(self, *args, **kwargs):
         """Handle an HTTP request from the client.
 

src/socketio/server.py

@@ -570,6 +570,15 @@ def disconnect(self, sid, namespace=None, ignore_queue=False):
             self.manager.disconnect(sid, namespace=namespace,
                                     ignore_queue=True)
 
+    def shutdown(self):
+        """Stop Socket.IO background tasks.
+
+        This method stops all background activity initiated by the Socket.IO
+        server. It must be called before shutting down the web server.
+        """
+        self.logger.info('Socket.IO is shutting down')
+        self.eio.shutdown()
+
     def transport(self, sid):
         """Return the name of the transport used by the client.
 

tests/asyncio/test_asyncio_server.py

@@ -981,6 +981,12 @@ def test_async_handlers(self, eio):
             None,
         )
 
+    def test_shutdown(self, eio):
+        s = asyncio_server.AsyncServer()
+        s.eio.shutdown = AsyncMock()
+        _run(s.shutdown())
+        s.eio.shutdown.mock.assert_called_once_with()
+
     def test_start_background_task(self, eio):
         s = asyncio_server.AsyncServer()
         s.start_background_task('foo', 'bar', baz='baz')

tests/common/test_server.py

@@ -917,6 +917,11 @@ def test_async_handlers(self, eio):
             None,
         )
 
+    def test_shutdown(self, eio):
+        s = server.Server()
+        s.shutdown()
+        s.eio.shutdown.assert_called_once_with()
+
     def test_start_background_task(self, eio):
         s = server.Server()
         s.start_background_task('foo', 'bar', baz='baz')