diff --git a/docs/feature.md b/docs/feature.md index aad3be1..2326a0b 100644 --- a/docs/feature.md +++ b/docs/feature.md @@ -22,7 +22,7 @@ | Тема | Зазор | Комментарий | |------|--------|-------------| -| **WebSockets** | Нет | RSGI/текущий стек заточен под HTTP-запрос/ответ. Нужна отдельная ветка согласно Granian/RSGI для WS и мост в Python. | +| **WebSockets** | Частично (ASGI spike) | Есть `@app.websocket(...)` в ASGI bridge (см. [websocket.md](websocket.md)); RSGI/Rust-native WS path пока не реализован. | | **SSE / длинный стрим ответа** | Частично | Есть `send_sse` (см. [sse.md](sse.md)); инкрементальный стрим зависит от `response_stream` в серверном протоколе, иначе fallback в буферизованный ответ. | | **HTTP/2 push, trailers** | Не в фокусе | Обычно на стороне сервера; фреймворк редко экспонирует. | | **ASGI: lifespan, `websocket`, background** | Частично | Мост ASGI — **только `http`**; нет полноценного ASGI-приложения с lifecycle из спеки. | diff --git a/docs/index.md b/docs/index.md index 431a13f..7d035fb 100644 --- a/docs/index.md +++ b/docs/index.md @@ -46,6 +46,7 @@ Granian still invokes a Python `App` object; the “win” is doing routing, bod | [CSRF](csrf.md) | Double-submit, `apply_csrf`, `csrf_layer` + CORS | | [JWT](jwt.md) | `require_jwt`, HS* / RSA / EC PEM, `decode_jwt_hs` (HS* tests) | | [SSE](sse.md) | `send_sse`, event framing, streaming caveats | +| [WebSocket](websocket.md) | ASGI spike API and current limitations | | [HTTP/2 with Granian](http2.md) | Transport guarantees vs server/proxy responsibilities | | [Dependencies](dependencies.md) | `Depends`, `dependencies=[...]`, `freeze` | | [OpenAPI](openapi.md) | `openapi.json` route, title, `openapi_json()` | diff --git a/docs/websocket.md b/docs/websocket.md new file mode 100644 index 0000000..5aef718 --- /dev/null +++ b/docs/websocket.md @@ -0,0 +1,49 @@ +# WebSocket (ASGI spike) + +[← Documentation index](index.md) + +Current WebSocket support in OxyRoute is an **ASGI bridge spike**, not full RSGI-native support yet. + +## Current API + +Use `@app.websocket(path)` for exact-path handlers on `App.__call__` (ASGI entry): + +```python +from oxyroute import App + +app = App() + + +@app.websocket("/ws") +async def ws(sock): + await sock.accept() + text = await sock.receive_text() + await sock.send_text(f"echo:{text}") + await sock.close() +``` + +The handler receives a small `WebSocket` helper with: + +- `accept(subprotocol=None)` +- `receive()` / `receive_text()` +- `send_text(text)` / `send_bytes(data)` +- `close(code=1000)` + +## Scope and limitations + +- Works on the optional ASGI entry (`app(scope, receive, send)`). +- **Not** wired into Rust request routing (`run_rsgi`) yet. +- Path matching is currently exact string match (no path-params router for WS yet). +- No first-class dependency/JWT/middleware chain for WS handlers in this spike. + +## Design split (what lives where) + +- **Python/ASGI now:** websocket handshake + frame loop helper and handler dispatch. +- **Rust/RSGI later:** unified route table, WS protocol lifecycle in native path, shared middleware/auth story. + +This keeps a practical testable path now while preserving room for a proper RSGI-native implementation. + +## See also + +- [ASGI bridge](asgi.md) +- [Feature gaps](feature.md) diff --git a/oxyroute/app.py b/oxyroute/app.py index b171a49..5e30b7c 100644 --- a/oxyroute/app.py +++ b/oxyroute/app.py @@ -1,5 +1,6 @@ from __future__ import annotations +import asyncio import inspect import json from collections.abc import Callable, Mapping @@ -47,6 +48,7 @@ def __init__(self, title: str = "OxyRoute", *, include_openapi: bool = True) -> self._app = _oxyroute.App(include_openapi=include_openapi) self._app.set_openapi_title(title) self.title = title + self._websocket_routes: dict[str, Callable[..., Any]] = {} # Per-process mutable bag for ``__rsgi_init__`` / factory setup (DB pool, clients, …). self.state: SimpleNamespace = SimpleNamespace() self._asgi3: Callable[..., Any] = build_asgi_caller(self) @@ -297,6 +299,35 @@ def options( jwt_cookie=jwt_cookie, ) + def websocket(self, path: str) -> Callable[[F], F]: + """ + Register an ASGI websocket handler for exact `path`. + + This route family is available on the optional ASGI bridge (`App.__call__`), not + on the RSGI request path. + """ + + def wrap(handler: F) -> F: + self._websocket_routes[path] = handler + return handler + + return wrap + + async def _handle_asgi_websocket(self, scope: dict[str, Any], receive: Any, send: Any) -> None: + path = str(scope.get("path", "/") or "/") + handler = self._websocket_routes.get(path) + if handler is None: + await send({"type": "websocket.close", "code": 1000}) + return + from .asgi import WebSocket + + ws = WebSocket(receive, send) + out = handler(ws) + if inspect.isawaitable(out): + await out + return + await asyncio.get_running_loop().run_in_executor(None, lambda: out) + def _route( self, method: str, diff --git a/oxyroute/asgi.py b/oxyroute/asgi.py index cbd3377..811d19f 100644 --- a/oxyroute/asgi.py +++ b/oxyroute/asgi.py @@ -12,6 +12,68 @@ from typing import Any +class WebSocket: + """Small ASGI websocket helper used by the optional ASGI bridge.""" + + __slots__ = ("_accepted", "_closed", "_connected", "_receive", "_send") + + def __init__(self, receive: Any, send: Any) -> None: + self._receive = receive + self._send = send + self._accepted = False + self._closed = False + self._connected = False + + async def accept(self, subprotocol: str | None = None) -> None: + if self._accepted: + return + if not self._connected: + first = await self._receive() + t = first.get("type", "") + if t == "websocket.disconnect": + self._closed = True + raise RuntimeError("websocket disconnected before accept") + if t != "websocket.connect": + raise RuntimeError(f"unexpected websocket event before accept: {t}") + self._connected = True + msg: dict[str, Any] = {"type": "websocket.accept"} + if subprotocol is not None: + msg["subprotocol"] = subprotocol + await self._send(msg) + self._accepted = True + + async def receive(self) -> dict[str, Any]: + return await self._receive() + + async def receive_text(self) -> str: + msg = await self._receive() + t = msg.get("type", "") + if t == "websocket.disconnect": + raise RuntimeError("websocket disconnected") + if t != "websocket.receive": + raise RuntimeError(f"unexpected websocket event: {t}") + text = msg.get("text") + if text is None: + raise RuntimeError("expected text websocket frame") + return str(text) + + async def send_text(self, text: str) -> None: + if self._closed: + return + await self._send({"type": "websocket.send", "text": text}) + + async def send_bytes(self, data: bytes) -> None: + if self._closed: + return + await self._send({"type": "websocket.send", "bytes": data}) + + async def close(self, code: int = 1000) -> None: + if self._closed: + return + await self._send({"type": "websocket.close", "code": int(code)}) + self._closed = True + + def _run_handle_rsgi_blocking( app_rsgi: Callable[[Any, Any], Any], rscope: Any, @@ -194,11 +256,19 @@ async def _go() -> None: async def asgi_to_rsgi( app_rsgi: Callable[[Any, Any], Any], + app_ws: Callable[[dict[str, Any], Any, Any], Any] | None, scope: dict[str, Any], receive: Any, send: Any, ) -> None: - if scope.get("type") != "http": + st = scope.get("type") + if st == "websocket": + if app_ws is None: + await send({"type": "websocket.close", "code": 1000}) + return + await app_ws(scope, receive, send) + return + if st != "http": return body = b"" while True: @@ -249,7 +319,14 @@ def _rsgi(s: Any, p: Any) -> Any: inner = getattr(c, "_app", c) return inner.handle_rsgi(s, p) + async def _ws(scope: dict[str, Any], receive: Any, send: Any) -> None: + h = getattr(framework_app, "_handle_asgi_websocket", None) + if h is None or not callable(h): + await send({"type": "websocket.close", "code": 1000}) + return + await h(scope, receive, send) + async def asgi3(scope: dict[str, Any], receive: Any, send: Any) -> None: - await asgi_to_rsgi(_rsgi, scope, receive, send) + await asgi_to_rsgi(_rsgi, _ws, scope, receive, send) return asgi3 diff --git a/tests/test_websocket_asgi.py b/tests/test_websocket_asgi.py new file mode 100644 index 0000000..d617e89 --- /dev/null +++ b/tests/test_websocket_asgi.py @@ -0,0 +1,54 @@ +"""ASGI websocket spike for issue #52.""" + +from __future__ import annotations + +import asyncio +from collections import deque +from typing import Any + +from oxyroute import App + + +def test_asgi_websocket_echo_flow() -> None: + app = App() + + @app.websocket("/ws") + async def ws(sock: Any) -> None: + await sock.accept() + text = await sock.receive_text() + await sock.send_text(f"echo:{text}") + await sock.close(code=1000) + + incoming: deque[dict[str, Any]] = deque( + [ + {"type": "websocket.connect"}, + {"type": "websocket.receive", "text": "ping"}, + ] + ) + sent: list[dict[str, Any]] = [] + + async def receive() -> dict[str, Any]: + if incoming: + return incoming.popleft() + return {"type": "websocket.disconnect", "code": 1000} + + async def send(message: dict[str, Any]) -> None: + sent.append(message) + + scope = { + "type": "websocket", + "path": "/ws", + "headers": [], + "query_string": b"", + "scheme": "ws", + } + + async def _run() -> None: + await app(scope, receive, send) + + asyncio.run(_run()) + + assert sent[0]["type"] == "websocket.accept" + assert sent[1]["type"] == "websocket.send" + assert sent[1]["text"] == "echo:ping" + assert sent[2]["type"] == "websocket.close"