New websockets

sanic/app.py

@@ -76,7 +76,8 @@
 from sanic.server import serve, serve_multiple, serve_single
 from sanic.signals import Signal, SignalRouter
 from sanic.touchup import TouchUp, TouchUpMeta
-from sanic.websocket import ConnectionClosed, WebSocketProtocol
+from sanic.server.protocols.websocket_protocol import WebSocketProtocol
+from sanic.server.websockets.impl import ConnectionClosed
 
 
 class Sanic(BaseSanic, metaclass=TouchUpMeta):
@@ -871,23 +872,11 @@ 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

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,7 +35,6 @@
     "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,
@@ -62,7 +61,6 @@ 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

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)
+            if 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,113 @@
+from typing import (
+    Optional,
+    Union,
+    Sequence,
+    TYPE_CHECKING
+)
+
+from httptools import HttpParserUpgrade  # type: ignore
+from websockets.server import ServerConnection
+from websockets.connection import OPEN, CLOSING, CLOSED
+
+from sanic.exceptions import SanicException
+from sanic.server import HttpProtocol
+from sanic.log import error_logger
+from ..websockets.impl import WebsocketImplProtocol
+
+if TYPE_CHECKING:
+    from websockets import http11
+
+class WebSocketProtocol(HttpProtocol):
+    def __init__(
+        self,
+        *args,
+        websocket_timeout=10,
+        websocket_max_size=None,
+        websocket_max_queue=None,
+        websocket_read_limit=2 ** 16,
+        websocket_write_limit=2 ** 16,
+        websocket_ping_interval=20,
+        websocket_ping_timeout=20,
+        **kwargs,
+    ):
+        super().__init__(*args, **kwargs)
+        self.websocket = None  # type: Union[None, WebsocketImplProtocol]
+        # self.app = None
+        self.websocket_timeout = websocket_timeout
+        self.websocket_max_size = websocket_max_size
+        if websocket_max_queue is not None and int(websocket_max_queue) > 0:
+            error_logger.warning(DeprecationWarning("websocket_max_queue is no longer used. No websocket message queueing is implemented."))
+        self.websocket_read_limit = websocket_read_limit
+        self.websocket_write_limit = websocket_write_limit
+        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:
+            self.websocket.fail_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
+            else:
+                return self.websocket.fail_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
+        headers = {"Upgrade": "websocket", "Connection": "Upgrade"}
+        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 as exc:
+            msg = (
+                    "Failed to open a WebSocket connection.\n"
+                    "See server log for more information.\n"
+                )
+            raise SanicException(msg, status_code=500)
+        if 100 <= resp.status_code <= 299:
+            rbytes = b"".join([b"HTTP/1.1 ", b'%d' % resp.status_code, b" ", resp.reason_phrase.encode("utf-8"), b"\r\n"])
+            for k, v in resp.headers.items():
+                rbytes += k.encode("utf-8") + b": " + v.encode("utf-8") + b"\r\n"
+            if resp.body:
+                rbytes += b"\r\n" + resp.body + b"\r\n"
+            rbytes += b"\r\n"
+            await super().send(rbytes)
+        else:
+            raise SanicException(resp.body, resp.status_code)
+
+        self.websocket = WebsocketImplProtocol(ws_conn, ping_interval=self.websocket_ping_interval, ping_timeout=self.websocket_ping_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

@@ -178,7 +178,7 @@ def serve(
         coros = []
         for conn in connections:
             if hasattr(conn, "websocket") and conn.websocket:
-                coros.append(conn.websocket.close_connection())
+                coros.append(conn.websocket.close(code=1001))
             else:
                 conn.abort()
 
@@ -278,7 +278,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,

sanic/server/websockets/connection.py

@@ -0,0 +1,70 @@
+from typing import Optional, List, Callable, Awaitable, Union, Dict, MutableMapping, Any
+
+ASIMessage = MutableMapping[str, Any]
+
+class WebSocketConnection:
+    """
+    This is for ASGI Connections.
+    It provides an interface similar to WebsocketProtocol, but
+    sends/receives over an ASGI connection.
+    """
+    # TODO
+    # - Implement ping/pong
+
+    def __init__(
+        self,
+        send: Callable[[ASIMessage], Awaitable[None]],
+        receive: Callable[[], Awaitable[ASIMessage]],
+        subprotocols: Optional[List[str]] = None,
+    ) -> None:
+        self._send = send
+        self._receive = receive
+        self._subprotocols = subprotocols or []
+
+    async def send(self, data: Union[str, bytes], *args, **kwargs) -> None:
+        message: Dict[str, Union[str, bytes]] = {"type": "websocket.send"}
+
+        if isinstance(data, bytes):
+            message.update({"bytes": data})
+        else:
+            message.update({"text": str(data)})
+
+        await self._send(message)
+
+    async def recv(self, *args, **kwargs) -> Optional[str]:
+        message = await self._receive()
+
+        if message["type"] == "websocket.receive":
+            return message["text"]
+        elif message["type"] == "websocket.disconnect":
+            pass
+
+        return None
+
+    receive = recv
+
+    async def accept(self, subprotocols: Optional[List[str]] = None) -> None:
+        subprotocol = None
+        if subprotocols:
+            for subp in subprotocols:
+                if subp in self.subprotocols:
+                    subprotocol = subp
+                    break
+
+        await self._send(
+            {
+                "type": "websocket.accept",
+                "subprotocol": subprotocol,
+            }
+        )
+
+    async def close(self, code: int = 1000, reason: str = "") -> None:
+        pass
+
+    @property
+    def subprotocols(self):
+        return self._subprotocols
+
+    @subprotocols.setter
+    def subprotocols(self, subprotocols: Optional[List[str]] = None):
+        self._subprotocols = subprotocols or []