Struct ResponseSender

pub struct ResponseSender { /* private fields */ }

Encode Message into Bytes and send it to ResponseStream.

Implementations

impl ResponseSender

pub fn downgrade(&self) -> ResponseWeakSender

downgrade Self to a weak sender.

pub fn send( &self, msg: Message, ) -> impl Future<Output = Result<(), ProtocolError>>

encode Message and add to ResponseStream.

pub fn send_error( &self, err: Error, ) -> impl Future<Output = Result<(), ProtocolError>>

add io::Error to ResponseStream.

the error should be used as a signal to the TCP connection associated with ResponseStream to close immediately.

Examples

use std::{future::poll_fn, pin::Pin, time::Duration};

use futures_core::Stream;
use http_ws::{CloseCode, Message, RequestStream, ResponseSender, ResponseStream};
use tokio::{io::AsyncWriteExt, time::timeout, net::TcpStream};

// thread1:
// read and write websocket message.
async fn sender<S, T, E>(tx: ResponseSender, mut rx: Pin<&mut RequestStream<S>>)
where
    S: Stream<Item = Result<T, E>>,
    T: AsRef<[u8]>,
{
    // send close message to client
    tx.send(Message::Close(Some(CloseCode::Away.into()))).await.unwrap();

    // the client failed to respond to close message in 5 seconds time window.
    if let Err(_) = timeout(Duration::from_secs(5), poll_fn(|cx| rx.as_mut().poll_next(cx))).await {
        // send io error to thread2
        tx.send_error(std::io::ErrorKind::UnexpectedEof.into()).await.unwrap();
    }
}

// thread2:
// receive websocket message from thread1 and transfer it on tcp connection.
async fn io_write(conn: &mut TcpStream, mut rx: Pin<&mut ResponseStream>) {
    // the first message is the "go away" close message in Ok branch.
    let msg = poll_fn(|cx| rx.as_mut().poll_next(cx)).await.unwrap().unwrap();

    // send msg to client
    conn.write_all(&msg).await.unwrap();

    // the second message is the io::Error in Err branch.
    let err = poll_fn(|cx| rx.as_mut().poll_next(cx)).await.unwrap().unwrap_err();

    // at this point we should close the tcp connection by either graceful close or
    // just return immediately and drop the TcpStream.
    let _ = conn.shutdown().await;
}

// thread3:
// receive message from tcp connection and send it to thread1:
async fn io_read(conn: &mut TcpStream) {
    // this part is ignored as it has no relation to the send_error api.
}

pub fn text( &self, txt: impl Into<String>, ) -> impl Future<Output = Result<(), ProtocolError>>

encode Message::Text variant and add to ResponseStream.

pub fn binary( &self, bin: impl Into<Bytes>, ) -> impl Future<Output = Result<(), ProtocolError>>

encode Message::Binary variant and add to ResponseStream.

pub async fn close( self, reason: Option<impl Into<CloseReason>>, ) -> Result<(), ProtocolError>

encode Message::Close variant and add to ResponseStream. take ownership of Self as after close message no more message can be sent to client.

Trait Implementations

impl Debug for ResponseSender

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

Formats the value using the given formatter.