Struct WebSocketStream 

pub struct WebSocketStream<S> { /* private fields */ }

An AsyncRead / AsyncWrite adapter over a fastwebsockets::WebSocket.

WebSocketStream<S> wraps a WebSocket<S> and exposes a byte-stream view (implementing tokio::io::AsyncRead and tokio::io::AsyncWrite) so that websocket application payloads can be used with existing I/O and codec infrastructure such as tokio_util::codec::Framed.

Behavior

• Incoming WebSocket data frames (Text or Binary depending on the stream’s PayloadType) are presented as a continuous byte stream. Each data frame’s payload is returned in-order; if a read buffer provided by the caller is smaller than a frame payload, the remainder is buffered internally and served on subsequent reads.
• Control frames (Ping/Pong) are handled by the underlying WebSocket (auto-pong) or ignored by this adapter. A Close frame marks EOF and subsequent reads return Ok(()) with zero bytes (standard EOF semantics).
• Writes produce single complete WebSocket data frames of the configured PayloadType. Each poll_write call sends one WebSocket data frame with the provided bytes as payload. The number of bytes reported as written is the length of buf supplied to poll_write.

Notes on threading and ownership

The adapter temporarily takes ownership of the inner WebSocket when it needs to perform an asynchronous read or write operation. To achieve this without requiring WebSocket itself to be Sync/Send across await points we spawn a boxed future that owns the websocket and returns it when the operation completes. This is implemented internally using ReadState and WriteState.

Example

use tokio::io::{AsyncReadExt, AsyncWriteExt};
use tokio::net::TcpStream;
use fastwebsockets::WebSocket;
use fastwebsockets_stream::{WebSocketStream, PayloadType};

// Wrap the websocket and apply a line-based codec:
async fn example<S>(_ws: WebSocket<S>)
    where S: tokio::io::AsyncRead + tokio::io::AsyncWrite + Unpin + Send + 'static {
    // This example is illustrative: constructing a real `WebSocket` requires
    // an underlying transport (e.g. a `TcpStream`) and the fastwebsockets
    // connection/handshake. Assume `ws` is a valid WebSocket<TcpStream>.

    let ws: WebSocket<S> = unimplemented!();
    let mut ws_stream = WebSocketStream::new(ws, PayloadType::Binary);

    // Write bytes -> sends a Binary frame
    let _n = ws_stream.write(b"hello").await;

    // Read bytes
    let mut buf = vec![0_u8; 1024];
    let _ = ws_stream.read(&mut buf).await;

    // Shutdown (sends Close)
    let _ = ws_stream.shutdown().await;
}

Another common usage is to use tokio_util::codec::Framed to apply a codec on top of WebSocketStream (for example a length-delimited or line-based codec). Example:

use tokio_util::codec::{Framed, LinesCodec};
use fastwebsockets::WebSocket;
use fastwebsockets_stream::{WebSocketStream, PayloadType};

// Wrap the websocket and apply a line-based codec:
async fn example<S>(_ws: WebSocket<S>)
    where S: tokio::io::AsyncRead + tokio::io::AsyncWrite + Unpin + Send + 'static {
    let ws: WebSocket<S> = unimplemented!();
    let stream = WebSocketStream::new(ws, PayloadType::Text);
    let mut framed = Framed::new(stream, LinesCodec::new());

    // Now you can use framed.read() / framed.send() to work with String frames.
}

Implementations

impl<S> WebSocketStream<S>

where S: AsyncRead + AsyncWrite + Unpin + Send + 'static,

pub fn new(websocket: WebSocket<S>, payload_type: PayloadType) -> Self

Create a new WebSocketStream wrapping the provided WebSocket.

This will enable automatic Pong replies and automatic Close handling on the wrapped WebSocket and initialize internal buffers and state.

payload_type selects whether this stream should read/write Text or Binary data. If the peer sends data frames with an opcode that does not match payload_type, reads will return an error.

pub fn into_inner(self) -> Option<WebSocket<S>>

Consume the adapter and attempt to return the inner WebSocket.

This returns Some(WebSocket<S>) if the websocket currently resides in the adapter. If there is an outstanding future that currently owns the websocket (i.e. a read or write in progress) this method will return None because the adapter cannot recover the websocket until that future completes.

pub fn is_closed(&self) -> bool

Returns true if we’ve observed a Close frame from the peer and the stream reached EOF.

Trait Implementations

impl<S> AsyncRead for WebSocketStream<S>

where S: AsyncRead + AsyncWrite + Unpin + Send + 'static,

fn poll_read( self: Pin<&mut Self>, cx: &mut Context<'_>, buf: &mut ReadBuf<'_>, ) -> Poll<Result<()>>

Attempts to read from the AsyncRead into buf.

impl<S> AsyncWrite for WebSocketStream<S>

where S: AsyncRead + AsyncWrite + Unpin + Send + 'static,

fn poll_write( self: Pin<&mut Self>, cx: &mut Context<'_>, buf: &[u8], ) -> Poll<Result<usize>>

Attempt to write bytes from buf into the object.

fn poll_flush(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<()>>

Attempts to flush the object, ensuring that any buffered data reach their destination.

fn poll_shutdown(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<()>>

Initiates or attempts to shut down this writer, returning success when the I/O connection has completely shut down.

fn poll_write_vectored( self: Pin<&mut Self>, cx: &mut Context<'_>, bufs: &[IoSlice<'_>], ) -> Poll<Result<usize, Error>>

Like poll_write, except that it writes from a slice of buffers.

fn is_write_vectored(&self) -> bool

Determines if this writer has an efficient poll_write_vectored implementation.

impl<S> Debug for WebSocketStream<S>

fn fmt(&self, f: &mut Formatter<'_>) -> Result

Formats the value using the given formatter.