sod-tungstenite

sod::Service implementations to interact with tungstenite websockets.

Service Impls

All Services are Retryable and are able to be blocking or non-blocking.

• WsSession is a MutService that wraps a tungstenite::WebSocket, accepting WsSessionEvent to send or receive messages. WsSession::into_split can split a WsSession into a WsReader, WsWriter, and WsFlusher.
• WsReader is a Service that wraps a Mutex<tungstenite::WebSocket>, accepting a () as input and producing tungstenite::Message as output.
• WsWriter is a Service that wraps a Mutex<tungstenite::WebSocket>, accepting a tungstenite::Message as input.
• WsFlusher is a Service that wraps a Mutex<tungstenite::WebSocket>, accepting a () as input.
• WsServer is a Service that that listens on a TCP port, accepting a () as input and producing a WsSession as output.

Features

• native-tls to enable Native TLS
• __rustls-tls to enable Rustls TLS

Blocking Example

use sod::{idle::backoff, MaybeProcessService, MutService, RetryService, Service, ServiceChain};
use sod_tungstenite::{UninitializedWsSession, WsServer, WsSession, WsSessionEvent};
use std::{sync::atomic::Ordering, thread::spawn};
use tungstenite::{http::StatusCode, Message};
use url::Url;

// server session logic to add `"pong: "` in front of text payload
struct PongService;
impl Service for PongService {
    type Input = Message;
    type Output = Option<Message>;
    type Error = ();
    fn process(&self, input: Message) -> Result<Self::Output, Self::Error> {
        match input {
            Message::Text(text) => Ok(Some(Message::Text(format!("pong: {text}")))),
            _ => Ok(None),
        }
    }
}

// wires session logic and spawns in new thread
struct SessionSpawner;
impl Service for SessionSpawner {
    type Input = UninitializedWsSession;
    type Output = ();
    type Error = ();
    fn process(&self, input: UninitializedWsSession) -> Result<Self::Output, Self::Error> {
        spawn(|| {
            let (r, w, f) = input.handshake().unwrap().into_split();
            let chain = ServiceChain::start(r)
                .next(PongService)
                .next(MaybeProcessService::new(w))
                .next(MaybeProcessService::new(f))
                .end();
            sod::thread::spawn_loop(chain, |err| {
                println!("Session: {err:?}");
                Err(err) // stop thread on error
            });
        });
        Ok(())
    }
}

// start a blocking server that creates blocking sessions
let server = WsServer::bind("127.0.0.1:48490").unwrap();

// spawn a thread to start accepting new server sessions
let handle = sod::thread::spawn_loop(
    ServiceChain::start(server).next(SessionSpawner).end(),
    |err| {
        println!("Server: {err:?}");
        Err(err) // stop thread on error
    },
);

// connect a client to the server
let (mut client, _) =
    WsSession::connect(Url::parse("ws://127.0.0.1:48490/socket").unwrap()).unwrap();

// client writes `"hello world"` payload
client
    .process(WsSessionEvent::WriteMessage(Message::Text(
        "hello world!".to_owned(),
    )))
    .unwrap();

// client receives `"pong: hello world"` payload
println!(
    "Received: {:?}",
    client.process(WsSessionEvent::ReadMessage).unwrap()
);

// join until server crashes
handle.join().unwrap();

Non-Blocking Example

use sod::{idle::backoff, MaybeProcessService, MutService, RetryService, Service, ServiceChain};
use sod_tungstenite::{UninitializedWsSession, WsServer, WsSession, WsSessionEvent};
use std::{sync::atomic::Ordering, thread::spawn};
use tungstenite::{http::StatusCode, Message};
use url::Url;

// server session logic to add `"pong: "` in front of text payload
struct PongService;
impl Service for PongService {
    type Input = Message;
    type Output = Option<Message>;
    type Error = ();
    fn process(&self, input: Message) -> Result<Self::Output, Self::Error> {
        match input {
            Message::Text(text) => Ok(Some(Message::Text(format!("pong: {text}")))),
            _ => Ok(None),
        }
    }
}

// wires session logic and spawns in new thread
struct SessionSpawner;
impl Service for SessionSpawner {
    type Input = UninitializedWsSession;
    type Output = ();
    type Error = ();
    fn process(&self, input: UninitializedWsSession) -> Result<Self::Output, Self::Error> {
        spawn(|| {
            let (r, w, f) = input.handshake().unwrap().into_split();
            let chain = ServiceChain::start(RetryService::new(r, backoff))
                .next(PongService)
                .next(MaybeProcessService::new(RetryService::new(w, backoff)))
                .next(MaybeProcessService::new(f))
                .end();
            sod::thread::spawn_loop(chain, |err| {
                println!("Session: {err:?}");
                Err(err) // stop thread on error
            });
        });
        Ok(())
    }
}

// start a non-blocking server that creates non-blocking sessions
let server = WsServer::bind("127.0.0.1:48490")
    .unwrap()
    .with_nonblocking_sessions(true)
    .with_nonblocking_server(true)
    .unwrap();

// spawn a thread to start accepting new server sessions
let handle = sod::thread::spawn_loop(
    ServiceChain::start(RetryService::new(server, backoff))
        .next(SessionSpawner)
        .end(),
    |err| {
        println!("Server: {err:?}");
        Err(err) // stop thread on error
    },
);

// connect a client to the server
let (mut client, response) =
    WsSession::connect(Url::parse("ws://127.0.0.1:48490/socket").unwrap()).unwrap();
assert_eq!(response.status(), StatusCode::SWITCHING_PROTOCOLS);

// client writes `"hello world"` payload
client
    .process(WsSessionEvent::WriteMessage(Message::Text(
        "hello world!".to_owned(),
    )))
    .unwrap();

// client receives `"pong: hello world"` payload
assert_eq!(
    client.process(WsSessionEvent::ReadMessage).unwrap(),
    Some(Message::Text("pong: hello world!".to_owned()))
);

// stop the server
sod::idle::KEEP_RUNNING.store(false, Ordering::Release);
handle.join().unwrap();