Struct AxumSink 

#[repr(transparent)]
pub struct AxumSink(pub SplitSink<WebSocket, Message>);

An axum websocket sink wrapper, that implements futures Sink in a way, that makes it compatible with y-sync protocol, so that it can be used by y-sync crate [BroadcastGroup].

Examples

use std::net::SocketAddr;
use std::str::FromStr;
use std::sync::Arc;
use tokio::sync::Mutex;
use tokio::task::JoinHandle;
use futures_util::stream::StreamExt;
use axum::{
    Router,
    routing::get,
    extract::ws::{WebSocket, WebSocketUpgrade},
    extract::State,
    response::IntoResponse,
};
use yrs_axum::broadcast::BroadcastGroup;
use yrs_axum::ws::{AxumSink, AxumStream};

async fn start_server(
    addr: &str,
    bcast: Arc<BroadcastGroup>,
) -> Result<JoinHandle<()>, Box<dyn std::error::Error>> {
    let addr = SocketAddr::from_str(addr)?;
    let listener = tokio::net::TcpListener::bind(addr).await?;
     
    let app = Router::new()
        .route("/my-room", get(ws_handler))
        .with_state(bcast);

    Ok(tokio::spawn(async move {
        axum::serve(listener, app.into_make_service())
            .await
            .unwrap();
    }))
}

async fn ws_handler(
    ws: WebSocketUpgrade,
    State(bcast): State<Arc<BroadcastGroup>>,
) -> impl IntoResponse {
    ws.on_upgrade(move |socket| peer(socket, bcast))
}

async fn peer(ws: WebSocket, bcast: Arc<BroadcastGroup>) {
    let (sink, stream) = ws.split();
    // convert axum web socket into compatible sink/stream
    let sink = Arc::new(Mutex::new(AxumSink(sink)));
    let stream = AxumStream(stream);
    // subscribe to broadcast group
    let sub = bcast.subscribe(sink, stream);
    // wait for subscribed connection to close itself
    match sub.completed().await {
        Ok(_) => println!("broadcasting for channel finished successfully"),
        Err(e) => eprintln!("broadcasting for channel finished abruptly: {}", e),
    }
}

Tuple Fields

0: SplitSink<WebSocket, Message>

Trait Implementations

impl Debug for AxumSink

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

Formats the value using the given formatter.

impl From<SplitSink<WebSocket, Message>> for AxumSink

fn from(sink: SplitSink<WebSocket, Message>) -> Self

Converts to this type from the input type.

impl Into<SplitSink<WebSocket, Message>> for AxumSink

fn into(self) -> SplitSink<WebSocket, Message>

Converts this type into the (usually inferred) input type.

impl Sink<Vec<u8>> for AxumSink

type Error = Error

The type of value produced by the sink when an error occurs.

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

Attempts to prepare the Sink to receive a value.

fn start_send(self: Pin<&mut Self>, item: Vec<u8>) -> Result<(), Self::Error>

Begin the process of sending a value to the sink. Each call to this function must be preceded by a successful call to poll_ready which returned Poll::Ready(Ok(())).

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

Flush any remaining output from this sink.

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

Flush any remaining output and close this sink, if necessary.