# polyoxide-ws
The WebSocket kit the polyoxide socket clients share, one copy of each block
they would otherwise each carry:
- `ensure_crypto_provider` installs the process-default rustls provider a
`tokio-tungstenite` connect needs.
- `Backoff` is the reconnect delay schedule, and `Timing` holds a supervised
connection's ping, staleness, backoff, connect and rotation timings, with
`timing_setters!` for a builder and `next_wait` for the wait until the next
deadline.
- `connect` opens a connection within a timeout (`DEFAULT_CONNECT_TIMEOUT`,
10 s), `connect_over` upgrades a TCP connection the caller opened itself,
`with_timeout` bounds a connect of the caller's own, and
`retrying_can_fix` says whether a reconnect can help after one failed, by a
status rule the caller passes in.
- The transport seam decides how a connection is dialled: `SocketTransport` is
the dial, `BoxSocket` the connection it returns, and `Capabilities` says
what the transport's sockets can do (send pings, see the server's pings,
carry handshake headers). `TungsteniteTransport` is the native dial and
`default_socket_transport` the target's choice. `transport_setters!` gives a
builder its `runtime` and `socket_transport` setters, `Connection::with_runtime`
and the `Supervisor`'s `runtime` and `socket_transport` setters put a
connection's and a supervisor's clocks, timers and spawns on the injected
runtime and its dial on the injected transport, and a transport that cannot
do what a caller asks fails as `Kind::Unsupported`.
- `Connection` is the bare tier: one connection as a `Stream` of its frames,
in `Mode::Data` (text and binary) or `Mode::All` (every frame). It answers
the server's close with the RFC 6455 reply, says how the server closed, and
correlates requests with their answers through a caller's builder and
matcher (`Correlated`), with optional pacing and an answer timeout.
- `Supervisor` runs one supervised connection for a venue's `Protocol`: pings
on the wall clock, even while the consumer's queue is full, staleness,
reconnect with backoff and a replay of the membership, rotation, outage
markers, and membership changes through a `MembershipHandle`, all as one
`Supervised` stream. A venue keeps only what is its own: its frames, its
ping, what counts as liveness, how it recovers and whether it announces an
outage.
- `WsError` is the one socket error, read through its `Kind`.
`polyoxide_venue::impl_ws_classification!` classifies it in the venue crate
that wraps it.
- `tungstenite` is re-exported, so a venue names `Message` and `CloseFrame`
through this crate.
The `test-server` feature adds a doc-hidden scripted local server and the
helpers the offline suites share. It is for tests, not consumers.
It depends on neither `polyoxide-core` nor `polyoxide-venue`, and builds no
HTTP or signing stack, so a credential-free feed built on it stays light. It
logs under the `polyoxide_ws` target and never installs a subscriber.
On `wasm32` the kit builds no TLS or TCP stack. `WebSocketTransport`, the browser's
`WebSocket`, is the default socket transport there; `connect_over`,
`ensure_crypto_provider`, the scripted test server and `TungsteniteTransport` are built for
native targets only, and the `test-server` feature turns on nothing.
More information about this crate can be found in the [crate documentation](https://docs.rs/polyoxide-ws/).
## Installation
```toml
[dependencies]
polyoxide-ws = "0.47"
```
## Usage
A request on the bare tier, then the frames that follow it:
```rust,no_run
use std::time::Duration;
use futures_util::StreamExt;
use polyoxide_ws::{connect, tungstenite::Message, Connection, Correlated, Mode, DEFAULT_CONNECT_TIMEOUT};
# async fn run() -> Result<(), polyoxide_ws::WsError> {
let socket = connect("wss://example.com/ws", DEFAULT_CONNECT_TIMEOUT).await?;
let mut connection = Connection::new(socket, Mode::Data)
.min_request_interval(Duration::from_millis(200))
.answer_timeout(Duration::from_secs(10));
connection
.request(
|id| Message::Text(format!(r#"{{"id":{id},"op":"subscribe"}}"#).into()),
|id, frame| match frame {
Message::Text(text) if text.contains(&format!(r#""id":{id}"#)) => Correlated::Answer(()),
// Anything else is kept, and yielded below in the order it came.
_ => Correlated::Push,
},
)
.await?;
while let Some(frame) = connection.next().await {
println!("{:?}", frame?);
}
if let Some(closed) = connection.closed() {
eprintln!("{closed}");
}
# Ok(())
# }
```
A supervised builder takes the shared timings, and its reconnect loop waits
out the schedule they describe:
```rust
use std::time::Duration;
use polyoxide_ws::Timing;
#[derive(Default)]
pub struct FeedBuilder {
timing: Timing,
}
impl FeedBuilder {
polyoxide_ws::timing_setters! {
/// How often to ping, whatever the traffic.
ping_interval;
/// Reconnect delay bounds.
backoff;
}
}
let timing = FeedBuilder::default()
.backoff(Duration::from_millis(100), Duration::from_millis(300))
.timing;
let mut backoff = timing.backoff();
// A connection that delivered something starts the schedule over.
backoff.after_connection_ended(true);
assert_eq!(backoff.take(), Duration::from_millis(100));
```