Getting Started
Installation
muxws has no required runtime dependencies in either language. Transports, codecs and web framework integrations are optional extras or optional peer dependencies.
# Python. The extras are transports and codecs, and each one is optional:
# muxws[starlette] - the FastAPI / Starlette acceptor
# muxws[websockets] - the `websockets` dialer and acceptor
# muxws[msgpack] - the msgpack codec
# The quick start below runs a FastAPI server under uvicorn, so it asks for those two as well.
pip install "muxws[starlette,websockets]" fastapi uvicorn# TypeScript. `ws` and `@msgpack/msgpack` are optional peer dependencies: `ws` is needed only for a
# Node dialer or acceptor (a browser has WebSocket already), `@msgpack/msgpack` only for the msgpack
# codec. `tsx` is here to run the .ts file below without a build step.
npm install muxws ws
npm install --save-dev tsx
# The client below is an ES module (it uses top-level `await`), and the package.json created by npm
# does not declare `"type": "module"`. Without this line `npx tsx` rejects the file with
# "Top-level await is currently not supported with the cjs output format".
npm pkg set type=moduleQuick Start
Three files, one socket, two call shapes: a unary request that receives exactly one answer, and a streaming response that arrives in several payloads. The server is Python; the client is Python or TypeScript. Both clients print the same output, because both use the same wire protocol.
1. The server
The route passes the socket to accept(); after that, the route contains no transport-specific code. accept() performs the WebSocket upgrade itself, because it selects the muxws.v1.<codec> subprotocol.
Each peer has one on_stream handler, and every stream the other end opens is delivered to it. stream.reply() sends one payload and ends the stream; stream.send() any number of times followed by stream.end() is the streaming shape.
"""Quick-start acceptor: FastAPI, one WebSocket route, one muxws peer.
Run it with `python quickstart_server.py`. It listens on 127.0.0.1:8000 unless the `MUXWS_PORT`
environment variable says otherwise - the documentation test starts it on a free port that way.
The route does three things: it hands the socket to `accept()`, registers the peer's `on_stream`
handler, and runs the read loop. Every stream the dialer opens on this connection arrives at that
handler.
"""
import os
from fastapi import FastAPI, WebSocket
from muxws import accept, Stream
#: 127.0.0.1: the quick start runs locally and is not a deployment.
HOST = "127.0.0.1"
PORT = int(os.environ.get("MUXWS_PORT", "8000"))
app = FastAPI()
async def on_stream(payload: object, stream: Stream) -> None:
"""The incoming-stream handler. `payload` is the value the dialer opened the stream with."""
if isinstance(payload, dict) and payload.get("say") == "hello":
# Unary: one payload and the end of the stream in a single call.
await stream.reply({"greeting": "hello, muxws"})
return
if isinstance(payload, dict) and isinstance(payload.get("count"), int):
# Streaming response: any number of payloads, then an end. `end` is a flag on the last frame,
# not a separate frame, so three chunks take three frames.
total = payload["count"]
for n in range(1, total + 1):
await stream.send({"chunk": n, "of": total})
await stream.end()
return
# Any other stream, including the reconnect hello from the last section, is acknowledged by
# returning: muxws ends a stream its handler left open.
@app.websocket("/ws")
async def muxws_endpoint(websocket: WebSocket) -> None:
"""`accept()` performs the WebSocket upgrade, because it selects the subprotocol."""
peer = await accept(websocket)
peer.on_stream(on_stream)
await peer.serve()
if __name__ == "__main__":
import uvicorn
uvicorn.run(app, host=HOST, port=PORT, log_level="warning")Run it:
python quickstart_server.pyIt listens on 127.0.0.1:8000. Set MUXWS_PORT to use another port.
2. The Python client
connect() returns a peer that is already serving. request() is the unary shape: one payload out, exactly one payload back. open() is the general shape. It is synchronous and returns the Stream in the same call that allocates the stream id; end=True means this side sends nothing more.
"""Quick-start dialer: the `websockets` transport, two call shapes on one socket.
Start `quickstart_server.py` first, then run `python quickstart_client.py`. The URL is
`ws://127.0.0.1:8000/ws` unless the `MUXWS_URL` environment variable says otherwise.
"""
import asyncio
import os
from muxws import connect
URL = os.environ.get("MUXWS_URL", "ws://127.0.0.1:8000/ws")
async def main() -> None:
peer = await connect(URL)
try:
# Unary. `request()` sends one payload, waits for exactly one back, and raises if the remote
# sends a second; `await stream` does not raise in that case.
answer = await peer.request({"say": "hello"})
print(f"greeting: {answer['greeting']}")
# Streaming response. `open()` is synchronous: it returns the Stream in the same turn in which
# it allocates the id. `end=True` ends this side's sending, so the acceptor can start
# streaming back immediately.
stream = peer.open({"count": 3}, end=True)
async for chunk in stream:
print(f"chunk {chunk['chunk']} of {chunk['of']}")
finally:
await peer.close()
if __name__ == "__main__":
asyncio.run(main())Run it in a second terminal:
python quickstart_client.py3. The TypeScript client
The same two calls, against the same server. There are two imports: muxws is the package, and importing it registers the built-in json codec; muxws/node is the subpath that provides the ws socket adapter, and it is the only part of the package that imports ws. A browser dialer imports connect from muxws instead; the rest of the code is unchanged.
Units: every duration in the TypeScript API is milliseconds as an integer; in the Python API it is seconds as a float. Neither call below takes a duration; the rule applies to request(..., { timeoutMs }) and result({ timeoutMs }).
/**
* Quick-start dialer in TypeScript, on Node, against the same `quickstart_server.py`.
*
* Run it with `npx tsx quickstart-client.ts`. The URL is `ws://127.0.0.1:8000/ws` unless the
* `MUXWS_URL` environment variable says otherwise.
*
* Importing `muxws` registers the built-in `json` codec. `muxws/node` provides `connect` over the
* `ws` package; it is the only module in the package that imports `ws`. A browser dialer imports
* `connect` from `muxws` instead; the rest of the code is the same.
*/
import 'muxws';
import { connect } from 'muxws/node';
const url = process.env.MUXWS_URL ?? 'ws://127.0.0.1:8000/ws';
interface Greeting {
greeting: string;
}
interface Chunk {
chunk: number;
of: number;
}
const peer = await connect(url);
try {
// Unary. `request()` sends one payload, waits for exactly one back, and rejects if the remote
// sends a second. `await stream` does not reject in that case.
const answer = await peer.request<Greeting>({ say: 'hello' });
console.log(`greeting: ${answer.greeting}`);
// Streaming response. `open()` is synchronous: it returns the Stream in the same turn that
// allocates the id. `end: true` closes the sending side, so the acceptor can start streaming back
// immediately.
const stream = peer.open<Chunk>({ count: 3 }, { end: true });
for await (const chunk of stream) {
console.log(`chunk ${chunk.chunk} of ${chunk.of}`);
}
} finally {
await peer.close();
}Run it in a second terminal:
npx tsx quickstart-client.tsSet MUXWS_URL to point either client somewhere other than ws://127.0.0.1:8000/ws.
4. What you see
Either client prints exactly this. The documentation test compares the output of both clients with this block byte for byte:
greeting: hello, muxws
chunk 1 of 3
chunk 2 of 3
chunk 3 of 3The first line arrived on one stream and the other three on a second stream, over the same socket, without a second connection or an application-defined correlation id. This client completes the request before it opens the second stream, but this is not required. If both streams were open at once, the writer would interleave their frames, because a stream holds at most one unsent fragment at a time and the writer serves streams round-robin. For the same reason a 1 MB export does not delay a 200-byte progress update on another stream.
Next: server push
muxws has no separate push API. Peer is a single symmetric type: the acceptor calls peer.open() in the same way as the dialer, on the existing socket, and the dialer receives the stream in its own on_stream handler. Correlation and cancellation work the same way in both directions.
The acceptor side. peer is in scope in the route, and the handler uses it:
"""Server push: the acceptor opens a stream of its own.
Run it with `python push_server.py`; it listens on 127.0.0.1:8001 unless `MUXWS_PORT` says
otherwise, and `push_client.py` (or `push-client.ts`) is the other half.
There is no separate push API. `peer.open()` here is the same call the dialer makes in
`quickstart_client.py`, on the same socket, with the same correlation and cancellation: either peer
can open a stream.
"""
import os
from fastapi import FastAPI, WebSocket
from muxws import accept, Stream
HOST = "127.0.0.1"
PORT = int(os.environ.get("MUXWS_PORT", "8001"))
app = FastAPI()
@app.websocket("/ws")
async def muxws_endpoint(websocket: WebSocket) -> None:
peer = await accept(websocket)
async def on_stream(payload: object, _stream: Stream) -> None:
"""Handle the dialer's one-shot notify by pushing a stream; the notify itself gets no answer."""
topic = payload["subscribe"] if isinstance(payload, dict) else "ticks"
# Opened by the acceptor. The dialer receives it in its own on_stream handler, in the same way
# this peer received the dialer's stream.
ticks = peer.open({"topic": topic})
for n in (1, 2, 3):
await ticks.send({"tick": n})
await ticks.end()
peer.on_stream(on_stream)
await peer.serve()
if __name__ == "__main__":
import uvicorn
uvicorn.run(app, host=HOST, port=PORT, log_level="warning")The dialer side. The handler is passed to connect() rather than registered after connect() returns. The acceptor may open a stream as soon as the connection is accepted, and a stream that arrives before a handler is registered is answered with reset(REFUSED):
"""Server push, dialer side: a dialer with its own `on_stream` handler.
Start `push_server.py` first, then run `python push_client.py`. The URL is `ws://127.0.0.1:8001/ws`
unless `MUXWS_URL` says otherwise.
The handler is passed to `connect()` instead of being registered after it returns: the acceptor may
push a stream as soon as the connection is up, and a stream that arrives before a handler is
registered is answered with `reset(REFUSED)`.
"""
import asyncio
import os
from muxws import connect, Stream
URL = os.environ.get("MUXWS_URL", "ws://127.0.0.1:8001/ws")
async def main() -> None:
finished = asyncio.Event()
async def on_push(payload: object, stream: Stream) -> None:
print(f"push: {payload['topic']}")
async for tick in stream:
print(f"tick {tick['tick']}")
finished.set()
peer = await connect(URL, on_stream=on_push)
try:
# One-shot: `notify()` sends a payload, ends the stream, and returns no handle; no reply is
# expected.
await peer.notify({"subscribe": "ticks"})
# 10.0 seconds, so that a stuck example fails instead of hanging.
await asyncio.wait_for(finished.wait(), timeout=10.0)
finally:
await peer.close()
if __name__ == "__main__":
asyncio.run(main())Or the same dialer in TypeScript:
/**
* Server push, dialer half, in TypeScript on Node - the other end of `push_server.py`.
*
* Run it with `npx tsx push-client.ts`. The URL is `ws://127.0.0.1:8001/ws` unless `MUXWS_URL` says
* otherwise.
*
* `onStream` is passed to `connect()` rather than registered after it returns, because the acceptor
* may push a stream as soon as the connection is established. A stream that arrives before a handler is
* registered is reset with `REFUSED`.
*/
import 'muxws';
import { connect } from 'muxws/node';
const url = process.env.MUXWS_URL ?? 'ws://127.0.0.1:8001/ws';
interface Topic {
topic: string;
}
interface Tick {
tick: number;
}
let finished: () => void = () => undefined;
const pushDone = new Promise<void>((resolve) => {
finished = resolve;
});
const peer = await connect(url, {
onStream: async (payload, stream) => {
console.log(`push: ${(payload as Topic).topic}`);
for await (const tick of stream) {
console.log(`tick ${(tick as Tick).tick}`);
}
finished();
},
});
try {
// One-shot: `notify()` sends a payload, ends the stream and returns nothing to await; no response
// is expected.
await peer.notify({ subscribe: 'ticks' });
// 10000 milliseconds, so a stuck example fails instead of hanging.
const deadline = new Promise<never>((_resolve, reject) => {
setTimeout(() => {
reject(new Error('the push never arrived within 10000 milliseconds'));
}, 10_000).unref();
});
await Promise.race([pushDone, deadline]);
} finally {
await peer.close();
}Start python push_server.py (127.0.0.1:8001) and run either client. Both print:
push: ticks
tick 1
tick 2
tick 3Next: reconnecting after a disconnect
Reconnection is a dialer-only helper, enabled by passing two more arguments to connect(): a hello and a Reconnect.
"""Surviving a disconnect: the same dialer with `hello=` and `reconnect=Reconnect()`.
Start `quickstart_server.py` first, then run `python reconnect_client.py`. The client sends one
request and exits, so `on_reconnect` does not fire unless the connection drops; it shows where a
long-running program is notified of a reconnection.
"""
import asyncio
import os
from muxws import connect, Peer, Reconnect
URL = os.environ.get("MUXWS_URL", "ws://127.0.0.1:8000/ws")
def on_reconnect(reconnection: int, _peer: Peer) -> None:
"""Called once per re-established connection: socket open, subprotocol accepted, hello answered.
The number counts reconnections of this peer, starting at 1; it is not the number of dial
attempts.
"""
print(f"reconnected (reconnection {reconnection})")
async def main() -> None:
peer = await connect(
URL,
# Captured once, by value, and replayed unchanged on every later connection. It is an
# ordinary stream: the acceptor receives it in its on_stream handler and acknowledges it by
# returning. Do not put credentials in it; authenticate at the upgrade.
hello={"client": "quickstart"},
# initial_delay 0.25 seconds, factor 2.0, max_delay 30.0 seconds, jitter 0.3, unlimited
# attempts. Every duration in the Python implementation is in seconds, as a float.
reconnect=Reconnect(),
on_reconnect=on_reconnect,
)
try:
answer = await peer.request({"say": "hello"})
print(f"greeting: {answer['greeting']}")
finally:
await peer.close()
if __name__ == "__main__":
asyncio.run(main())When an established connection is lost, the dialer redials on a jittered exponential schedule (initial_delay 0.25 seconds, factor 2.0, capped at max_delay 30.0 seconds, jitter ±30 %) and replays the hello verbatim on every new socket. The acceptor receives the hello as an ordinary stream in its on_stream handler and acknowledges it by returning, so the dialer's identity is re-established without reconnect-specific code on the acceptor.
No stream survives a reconnect. Every stream that was open when the socket closed has failed, nothing is buffered while the peer has no socket, stream ids start again at 1, and peer.id changes, so the log shows a reconnected peer as a new connection. A reconnect restores a socket and the identity established by the hello, and nothing else.
If the first connect() fails, it raises regardless of reconnect=, so a wrong URL is reported immediately instead of being retried indefinitely.
Run it against the same quickstart_server.py. It connects, sends one request and exits, so on a working connection the on_reconnect line is never printed; the file contains it to show where a reconnection is reported:
greeting: hello, muxws