Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion docs/feature.md
Original file line number Diff line number Diff line change
Expand Up @@ -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 из спеки. |
Expand Down
1 change: 1 addition & 0 deletions docs/index.md
Original file line number Diff line number Diff line change
Expand Up @@ -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()` |
Expand Down
49 changes: 49 additions & 0 deletions docs/websocket.md
Original file line number Diff line number Diff line change
@@ -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)
31 changes: 31 additions & 0 deletions oxyroute/app.py
Original file line number Diff line number Diff line change
@@ -1,5 +1,6 @@
from __future__ import annotations

import asyncio
import inspect
import json
from collections.abc import Callable, Mapping
Expand Down Expand Up @@ -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)
Expand Down Expand Up @@ -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,
Expand Down
81 changes: 79 additions & 2 deletions oxyroute/asgi.py
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand Down Expand Up @@ -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:
Expand Down Expand Up @@ -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
54 changes: 54 additions & 0 deletions tests/test_websocket_asgi.py
Original file line number Diff line number Diff line change
@@ -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"
Loading