fastwebsockets_stream/lib.rs
1//! # fastwebsockets-stream
2//!
3//! `fastwebsockets-stream` provides an adapter that exposes a `fastwebsockets::WebSocket`
4//! as an `AsyncRead` / `AsyncWrite` byte stream compatible with `tokio` and
5//! utilities such as `tokio_util::codec::Framed`.
6//!
7//! ## Overview
8//!
9//! The adapter type is [`WebSocketStream`], which wraps a `fastwebsockets::WebSocket<S>`
10//! and presents websocket application payloads as a continuous byte stream.
11//! This is useful when you want to reuse existing codecs (length-delimited,
12//! line-based, protobuf, etc.) or any code that operates on `AsyncRead` /
13//! `AsyncWrite` without reimplementing websocket framing logic.
14//!
15//! The adapter supports both text and binary payloads (controlled by
16//! [`PayloadType`]) and will validate that incoming data frames match the
17//! configured payload type. Control frames (Ping/Pong) are handled automatically
18//! by the underlying `fastwebsockets::WebSocket` (auto-pong) and `Close` frames
19//! are translated to EOF for `AsyncRead` consumers.
20//!
21//! ## Key types
22//!
23//! * [`WebSocketStream<S>`] — the main adapter implementing `tokio::io::AsyncRead`
24//! and `tokio::io::AsyncWrite`.
25//! * [`PayloadType`] — selects whether the stream works with Text or Binary
26//! application frames.
27//!
28//! ## Examples
29//!
30//! The example below demonstrates a minimal server and client that speak a
31//! simple binary protocol in the same process: it creates an actual TCP
32//! listener, upgrades the connection to WebSocket using `fastwebsockets` +
33//! `hyper`, and then uses `WebSocketStream` as an `AsyncRead`/`AsyncWrite`
34//! stream on both ends.
35//!
36//! ```rust
37//! use fastwebsockets::{WebSocketError, handshake, upgrade};
38//! use fastwebsockets_stream::{PayloadType, WebSocketStream};
39//! use http_body_util::Empty;
40//! use hyper::body::Bytes;
41//! use hyper::body::Incoming;
42//! use hyper::header::{CONNECTION, UPGRADE};
43//! use hyper::server::conn::http1;
44//! use hyper::service::service_fn;
45//! use hyper::{Request, Response};
46//! use hyper_util::rt::TokioIo;
47//! use std::future::Future;
48//! use std::net::Ipv4Addr;
49//! use tokio::io::{AsyncReadExt, AsyncWriteExt};
50//! use tokio::net::{TcpListener, TcpStream};
51//!
52//! struct SpawnExecutor;
53//!
54//! impl<F> hyper::rt::Executor<F> for SpawnExecutor
55//! where
56//! F: Future + Send + 'static,
57//! F::Output: Send + 'static,
58//! {
59//! fn execute(&self, fut: F) {
60//! tokio::task::spawn(fut);
61//! }
62//! }
63//!
64//! // Server-side connection handler: upgrades the request, then echoes a
65//! // single binary message back to the client over `WebSocketStream`.
66//! async fn handle(mut request: Request<Incoming>) -> Result<Response<Empty<Bytes>>, WebSocketError> {
67//! assert!(upgrade::is_upgrade_request(&request));
68//! let (response, ws_fut) = upgrade::upgrade(&mut request)?;
69//!
70//! tokio::spawn(async move {
71//! let ws = ws_fut.await.unwrap();
72//! let mut ws_stream = WebSocketStream::new(ws, PayloadType::Binary);
73//!
74//! let mut buf = [0u8; 6];
75//! ws_stream.read_exact(&mut buf).await.unwrap();
76//! ws_stream.write_all(&buf).await.unwrap();
77//! ws_stream.shutdown().await.unwrap();
78//! });
79//!
80//! Ok(response)
81//! }
82//!
83//! # #[tokio::main]
84//! # async fn main() {
85//! // Bind an ephemeral local port and remember the address we actually got.
86//! let listener = TcpListener::bind((Ipv4Addr::LOCALHOST, 0u16)).await.unwrap();
87//! let addr = listener.local_addr().unwrap();
88//!
89//! // Accept and upgrade incoming connections in the background.
90//! tokio::spawn(async move {
91//! loop {
92//! let (stream, _) = listener.accept().await.unwrap();
93//! let io = TokioIo::new(stream);
94//! tokio::spawn(async move {
95//! let _ = http1::Builder::new()
96//! .serve_connection(io, service_fn(handle))
97//! .with_upgrades()
98//! .await;
99//! });
100//! }
101//! });
102//!
103//! // Client: connect to the port the server actually bound above.
104//! let stream = TcpStream::connect(addr).await.unwrap();
105//! let request = Request::builder()
106//! .method("GET")
107//! .uri(format!("ws://{addr}"))
108//! .header("Host", addr.to_string())
109//! .header(UPGRADE, "websocket")
110//! .header(CONNECTION, "upgrade")
111//! .header("Sec-WebSocket-Key", handshake::generate_key())
112//! .header("Sec-WebSocket-Version", "13")
113//! .body(Empty::<Bytes>::new())
114//! .unwrap();
115//!
116//! let (ws, _response) = handshake::client(&SpawnExecutor, request, stream)
117//! .await
118//! .unwrap();
119//! let mut ws_stream = WebSocketStream::new(ws, PayloadType::Binary);
120//!
121//! let n = ws_stream.write(b"Hello!").await.unwrap();
122//! assert_eq!(n, 6);
123//!
124//! let mut buf = [0u8; 6];
125//! ws_stream.read_exact(&mut buf).await.unwrap();
126//! assert_eq!(&buf, b"Hello!");
127//! # }
128//! ```
129//!
130//! ## Notes and caveats
131//!
132//! * Each call to `AsyncWrite::poll_write` will produce a single WebSocket data
133//! frame containing exactly the bytes provided in that call. If you need to
134//! stream a large logical message across multiple websocket frames, you
135//! should implement framing at the codec layer above the `WebSocketStream`.
136//! * If the peer sends a data frame with an opcode that doesn't match the
137//! configured `PayloadType` (e.g. the stream is configured `Binary` but the
138//! peer sends `Text`), reads will return an error.
139//! * `into_inner` on the adapter will return the inner `WebSocket` only if no
140//! read/write future currently owns it (i.e. there is no in-flight read or
141//! write). If an operation is in-progress, `into_inner` will return `None`.
142//!
143//! ## See Also
144//! - [`fastwebsockets`](https://docs.rs/fastwebsockets)
145//! - [`tokio::io::AsyncRead`](https://docs.rs/tokio/latest/tokio/io/trait.AsyncRead.html)
146//! - [`tokio::io::AsyncWrite`](https://docs.rs/tokio/latest/tokio/io/trait.AsyncWrite.html)
147//!
148//! ## License
149//!
150//! Licensed under the [MIT License](https://opensource.org/licenses/MIT).
151//!
152//! See the `LICENSE` file in the repository root for details.
153
154mod stream;
155
156pub use stream::PayloadType;
157pub use stream::WebSocketStream;