New websockets

examples/websocket.py

@@ -1,13 +1,14 @@
 from sanic import Sanic
-from sanic.response import file
+from sanic.response import redirect
 
 app = Sanic(__name__)
 
 
-@app.route('/')
-async def index(request):
-    return await file('websocket.html')
+app.static('index.html', "websocket.html")
 
+@app.route('/')
+def index(request):
+    return redirect("index.html")
 
 @app.websocket('/feed')
 async def feed(request, ws):

sanic/app.py

@@ -74,9 +74,10 @@
 from sanic.server import AsyncioServer, HttpProtocol
 from sanic.server import Signal as ServerSignal
 from sanic.server import serve, serve_multiple, serve_single
+from sanic.server.protocols.websocket_protocol import WebSocketProtocol
+from sanic.server.websockets.impl import ConnectionClosed
 from sanic.signals import Signal, SignalRouter
 from sanic.touchup import TouchUp, TouchUpMeta
-from sanic.websocket import ConnectionClosed, WebSocketProtocol
 
 
 class Sanic(BaseSanic, metaclass=TouchUpMeta):
@@ -871,39 +872,31 @@ async def handle_request(self, request: Request):  # no cov
     async def _websocket_handler(
         self, handler, request, *args, subprotocols=None, **kwargs
     ):
-        request.app = self
-        if not getattr(handler, "__blueprintname__", False):
-            request._name = handler.__name__
-        else:
-            request._name = (
-                getattr(handler, "__blueprintname__", "") + handler.__name__
-            )
-
-            pass
-
         if self.asgi:
             ws = request.transport.get_websocket_connection()
             await ws.accept(subprotocols)
         else:
             protocol = request.transport.get_protocol()
-            protocol.app = self
-
             ws = await protocol.websocket_handshake(request, subprotocols)
 
         # schedule the application handler
         # its future is kept in self.websocket_tasks in case it
         # needs to be cancelled due to the server being stopped
         fut = ensure_future(handler(request, ws, *args, **kwargs))
         self.websocket_tasks.add(fut)
+        cancelled = False
         try:
             await fut
         except Exception as e:
             self.error_handler.log(request, e)
         except (CancelledError, ConnectionClosed):
-            pass
+            cancelled = True
         finally:
             self.websocket_tasks.remove(fut)
-            await ws.close()
+            if cancelled:
+                ws.end_connection(1000)
+            else:
+                await ws.close()
 
     # -------------------------------------------------------------------- #
     # Testing

sanic/asgi.py

@@ -10,7 +10,7 @@
 from sanic.models.asgi import ASGIReceive, ASGIScope, ASGISend, MockTransport
 from sanic.request import Request
 from sanic.server import ConnInfo
-from sanic.websocket import WebSocketConnection
+from sanic.server.websockets.connection import WebSocketConnection
 
 
 class Lifespan:

sanic/config.py

@@ -35,12 +35,9 @@
     "REQUEST_MAX_SIZE": 100000000,  # 100 megabytes
     "REQUEST_TIMEOUT": 60,  # 60 seconds
     "RESPONSE_TIMEOUT": 60,  # 60 seconds
-    "WEBSOCKET_MAX_QUEUE": 32,
     "WEBSOCKET_MAX_SIZE": 2 ** 20,  # 1 megabyte
     "WEBSOCKET_PING_INTERVAL": 20,
     "WEBSOCKET_PING_TIMEOUT": 20,
-    "WEBSOCKET_READ_LIMIT": 2 ** 16,
-    "WEBSOCKET_WRITE_LIMIT": 2 ** 16,
 }
 
 
@@ -62,12 +59,9 @@ class Config(dict):
     REQUEST_MAX_SIZE: int
     REQUEST_TIMEOUT: int
     RESPONSE_TIMEOUT: int
-    WEBSOCKET_MAX_QUEUE: int
     WEBSOCKET_MAX_SIZE: int
     WEBSOCKET_PING_INTERVAL: int
     WEBSOCKET_PING_TIMEOUT: int
-    WEBSOCKET_READ_LIMIT: int
-    WEBSOCKET_WRITE_LIMIT: int
 
     def __init__(
         self,

sanic/mixins/routes.py

@@ -121,8 +121,11 @@ def decorator(handler):
                         "Expected either string or Iterable of host strings, "
                         "not %s" % host
                     )
-
-            if isinstance(subprotocols, (list, tuple, set)):
+            if isinstance(subprotocols, list):
+                # Ordered subprotocols, maintain order
+                subprotocols = tuple(subprotocols)
+            elif isinstance(subprotocols, set):
+                # subprotocol is unordered, keep it unordered
                 subprotocols = frozenset(subprotocols)
 
             route = FutureRoute(

sanic/models/asgi.py

@@ -3,7 +3,7 @@
 from typing import Any, Awaitable, Callable, MutableMapping, Optional, Union
 
 from sanic.exceptions import InvalidUsage
-from sanic.websocket import WebSocketConnection
+from sanic.server.websockets.connection import WebSocketConnection
 
 
 ASGIScope = MutableMapping[str, Any]

sanic/server/protocols/websocket_protocol.py

@@ -0,0 +1,165 @@
+from typing import TYPE_CHECKING, Optional, Sequence
+
+from websockets.connection import CLOSED, CLOSING, OPEN
+from websockets.server import ServerConnection
+
+from sanic.exceptions import ServerError
+from sanic.log import error_logger
+from sanic.server import HttpProtocol
+
+from ..websockets.impl import WebsocketImplProtocol
+
+
+if TYPE_CHECKING:
+    from websockets import http11
+
+
+class WebSocketProtocol(HttpProtocol):
+
+    websocket: Optional[WebsocketImplProtocol]
+    websocket_timeout: float
+    websocket_max_size = Optional[int]
+    websocket_ping_interval = Optional[float]
+    websocket_ping_timeout = Optional[float]
+
+    def __init__(
+        self,
+        *args,
+        websocket_timeout: float = 10.0,
+        websocket_max_size: Optional[int] = None,
+        websocket_max_queue: Optional[int] = None,  # max_queue is deprecated
+        websocket_read_limit: Optional[int] = None,  # read_limit is deprecated
+        websocket_write_limit: Optional[int] = None,  # write_limit deprecated
+        websocket_ping_interval: Optional[float] = 20.0,
+        websocket_ping_timeout: Optional[float] = 20.0,
+        **kwargs,
+    ):
+        super().__init__(*args, **kwargs)
+        self.websocket = None
+        self.websocket_timeout = websocket_timeout
+        self.websocket_max_size = websocket_max_size
+        if websocket_max_queue is not None and websocket_max_queue > 0:
+            # TODO: Reminder remove this warning in v22.3
+            error_logger.warning(
+                DeprecationWarning(
+                    "Websocket no longer uses queueing, so websocket_max_queue"
+                    " is no longer required."
+                )
+            )
+        if websocket_read_limit is not None and websocket_read_limit > 0:
+            # TODO: Reminder remove this warning in v22.3
+            error_logger.warning(
+                DeprecationWarning(
+                    "Websocket no longer uses read buffers, so "
+                    "websocket_read_limit is not required."
+                )
+            )
+        if websocket_write_limit is not None and websocket_write_limit > 0:
+            # TODO: Reminder remove this warning in v22.3
+            error_logger.warning(
+                DeprecationWarning(
+                    "Websocket no longer uses write buffers, so "
+                    "websocket_write_limit is not required."
+                )
+            )
+        self.websocket_ping_interval = websocket_ping_interval
+        self.websocket_ping_timeout = websocket_ping_timeout
+
+    def connection_lost(self, exc):
+        if self.websocket is not None:
+            self.websocket.connection_lost(exc)
+        super().connection_lost(exc)
+
+    def data_received(self, data):
+        if self.websocket is not None:
+            self.websocket.data_received(data)
+        else:
+            # Pass it to HttpProtocol handler first
+            # That will (hopefully) upgrade it to a websocket.
+            super().data_received(data)
+
+    def eof_received(self) -> Optional[bool]:
+        if self.websocket is not None:
+            return self.websocket.eof_received()
+        else:
+            return False
+
+    def close(self, timeout: Optional[float] = None):
+        # Called by HttpProtocol at the end of connection_task
+        # If we've upgraded to websocket, we do our own closing
+        if self.websocket is not None:
+            # Note, we don't want to use websocket.close()
+            # That is used for user's application code to send a
+            # websocket close packet. This is different.
+            self.websocket.end_connection(1001)
+        else:
+            super().close()
+
+    def close_if_idle(self):
+        # Called by Sanic Server when shutting down
+        # If we've upgraded to websocket, shut it down
+        if self.websocket is not None:
+            if self.websocket.connection.state in (CLOSING, CLOSED):
+                return True
+            elif self.websocket.loop is not None:
+                self.websocket.loop.create_task(self.websocket.close(1001))
+            else:
+                self.websocket.end_connection(1001)
+        else:
+            return super().close_if_idle()
+
+    async def websocket_handshake(
+        self, request, subprotocols=Optional[Sequence[str]]
+    ):
+        # let the websockets package do the handshake with the client
+        try:
+            if subprotocols is not None:
+                # subprotocols can be a set or frozenset,
+                # but ServerConnection needs a list
+                subprotocols = list(subprotocols)
+            ws_conn = ServerConnection(
+                max_size=self.websocket_max_size,
+                subprotocols=subprotocols,
+                state=OPEN,
+                logger=error_logger,
+            )
+            resp: "http11.Response" = ws_conn.accept(request)
+        except Exception:
+            msg = (
+                "Failed to open a WebSocket connection.\n"
+                "See server log for more information.\n"
+            )
+            raise ServerError(msg, status_code=500)
+        if 100 <= resp.status_code <= 299:
+            rbody = "".join(
+                [
+                    "HTTP/1.1 ",
+                    str(resp.status_code),
+                    " ",
+                    resp.reason_phrase,
+                    "\r\n",
+                ]
+            )
+            rbody += "".join(f"{k}: {v}\r\n" for k, v in resp.headers.items())
+            if resp.body is not None:
+                rbody += f"\r\n{resp.body}\r\n\r\n"
+            else:
+                rbody += "\r\n"
+            await super().send(rbody.encode())
+        else:
+            raise ServerError(resp.body, resp.status_code)
+
+        self.websocket = WebsocketImplProtocol(
+            ws_conn,
+            ping_interval=self.websocket_ping_interval,
+            ping_timeout=self.websocket_ping_timeout,
+            close_timeout=self.websocket_timeout,
+        )
+        loop = (
+            request.transport.loop
+            if hasattr(request, "transport")
+            and hasattr(request.transport, "loop")
+            else None
+        )
+        await self.websocket.connection_made(self, loop=loop)
+        return self.websocket

sanic/server/runners.py

@@ -175,15 +175,11 @@ def serve(
 
         # Force close non-idle connection after waiting for
         # graceful_shutdown_timeout
-        coros = []
         for conn in connections:
             if hasattr(conn, "websocket") and conn.websocket:
-                coros.append(conn.websocket.close_connection())
+                conn.websocket.fail_connection(code=1001)
             else:
                 conn.abort()
-
-        _shutdown = asyncio.gather(*coros)
-        loop.run_until_complete(_shutdown)
         loop.run_until_complete(app._server_event("shutdown", "after"))
 
         remove_unix_socket(unix)
@@ -278,9 +274,6 @@ def _build_protocol_kwargs(
     if hasattr(protocol, "websocket_handshake"):
         return {
             "websocket_max_size": config.WEBSOCKET_MAX_SIZE,
-            "websocket_max_queue": config.WEBSOCKET_MAX_QUEUE,
-            "websocket_read_limit": config.WEBSOCKET_READ_LIMIT,
-            "websocket_write_limit": config.WEBSOCKET_WRITE_LIMIT,
             "websocket_ping_timeout": config.WEBSOCKET_PING_TIMEOUT,
             "websocket_ping_interval": config.WEBSOCKET_PING_INTERVAL,
         }