sockudo-ws

use futures_util::{SinkExt, StreamExt};
use sockudo_ws::{Config, Message, WebSocketStream};
use tokio::net::TcpStream;

async fn handle(stream: TcpStream) {
    // After WebSocket handshake...
    let mut ws = WebSocketStream::server(stream, Config::default());

    while let Some(msg) = ws.next().await {
        match msg.unwrap() {
            Message::Text(text) => {
                ws.send(Message::text(text)).await.unwrap();
            }
            Message::Binary(data) => {
                ws.send(Message::Binary(data)).await.unwrap();
            }
            Message::Close(_) => break,
            _ => {}
        }
    }
}

use sockudo_ws::{Config, Message, WebSocketStream};
use tokio::sync::mpsc;

async fn handle(stream: TcpStream) {
    let ws = WebSocketStream::server(stream, Config::default());
    
    // Split into independent read/write halves
    // Reader and writer can operate 100% concurrently!
    let (mut reader, mut writer) = ws.split();

    // Writer task - NEVER blocks reader
    let (tx, mut rx) = mpsc::channel::<Message>(32);
    tokio::spawn(async move {
        while let Some(msg) = rx.recv().await {
            writer.send(msg).await.unwrap();
        }
    });

    // Reader loop - NEVER blocks writer
    while let Some(msg) = reader.next().await {
        match msg.unwrap() {
            Message::Text(text) => {
                tx.send(Message::text(text)).await.unwrap();
            }
            Message::Close(_) => break,
            _ => {}
        }
    }
}

use sockudo_ws::{Config, Http1};
use sockudo_ws::client::WebSocketClient;
use tokio::net::TcpStream;

async fn connect() -> sockudo_ws::Result<()> {
    let stream = TcpStream::connect("example.com:80").await?;
    let headers = vec![
        ("Authorization".to_string(), "Bearer token".to_string()),
        ("User-Agent".to_string(), "my-client".to_string()),
    ];
    let client = WebSocketClient::<Http1>::new(Config::default());
    let (_websocket, _handshake) = client
        .connect_with_headers(
            stream,
            "example.com",
            "/ws",
            None,
            Some(&headers),
        )
        .await?;
    Ok(())
}

use sockudo_ws::compio::{accept_async, connect_async, net::{TcpListener, TcpStream}};
use sockudo_ws::{Config, Message};

#[sockudo_ws::compio::main]
async fn main() -> sockudo_ws::Result<()> {
    let listener = TcpListener::bind("127.0.0.1:9001").await?;

    let server = sockudo_ws::compio::runtime::spawn(async move {
        let (stream, _) = listener.accept().await.unwrap();
        let (mut ws, _) = accept_async(stream, Config::default()).await.unwrap();

        while let Some(msg) = ws.next().await {
            let msg = msg.unwrap();
            if matches!(msg, Message::Close(_)) {
                break;
            }
            ws.send(msg).await.unwrap();
        }
    });

    let stream = TcpStream::connect("127.0.0.1:9001").await?;
    let (mut ws, _) = connect_async(
        stream,
        "127.0.0.1:9001",
        "/",
        None,
        Config::default(),
    )
    .await?;
    ws.send_text("hello").await?;
    let _ = ws.next().await;

    server.await.unwrap();
    Ok(())
}

use sockudo_ws::compio::{
    connect_http2, serve_http2,
    net::{TcpListener, TcpStream},
};
use sockudo_ws::Config;

#[sockudo_ws::compio::main]
async fn main() -> sockudo_ws::Result<()> {
    let listener = TcpListener::bind("127.0.0.1:9002").await?;
    let addr = listener.local_addr()?;

    let server = sockudo_ws::compio::runtime::spawn(async move {
        let (stream, _) = listener.accept().await.unwrap();
        serve_http2(stream, Config::default(), |mut ws, _req| async move {
            if let Some(Ok(msg)) = ws.next().await {
                ws.send(msg).await.unwrap();
            }
        })
        .await
        .unwrap();
    });

    let stream = TcpStream::connect(addr).await?;
    let mut ws = connect_http2(
        stream,
        &format!("https://localhost:{}/chat", addr.port()),
        None,
        Config::default(),
    )
    .await?;

    ws.send_text("hello over h2").await?;
    let _ = ws.next().await;

    server.await.unwrap();
    Ok(())
}

use sockudo_ws::compio::{CompioHttp3Server, connect_http3_multiplexed};
use sockudo_ws::Config;

async fn http3_example(
    server_addr: std::net::SocketAddr,
    server_tls: rustls::ServerConfig,
    client_tls: rustls::ClientConfig,
) -> sockudo_ws::Result<()> {
    let server = CompioHttp3Server::bind(server_addr, server_tls, Config::default()).await?;
    let addr = server.local_addr()?;

    let server_task = sockudo_ws::compio::runtime::spawn(async move {
        server
            .serve(|mut ws, _req| async move {
                if let Some(Ok(msg)) = ws.next().await {
                    ws.send(msg).await.unwrap();
                }
            })
            .await
            .unwrap();
    });

    let mut mux = connect_http3_multiplexed(addr, "localhost", client_tls, Config::default()).await?;
    let mut chat = mux.open_websocket("/chat", None).await?;
    let mut notifications = mux.open_websocket("/notifications", None).await?;

    chat.send_text("hello h3").await?;
    notifications.send_text("ping").await?;

    mux.close();
    server_task.await.unwrap();
    Ok(())
}

use sockudo_ws::pubsub::PubSub;
use sockudo_ws::Message;
use tokio::sync::mpsc;

// Create pub/sub system
let pubsub = PubSub::new();

// Create a subscriber with a message channel
let (tx, mut rx) = mpsc::unbounded_channel();
let subscriber_id = pubsub.create_subscriber(tx);

// Subscribe to topics
pubsub.subscribe(subscriber_id, "chat/general");
pubsub.subscribe(subscriber_id, "notifications");

// Publish to all subscribers
let msg = Message::text("Hello everyone!");
pubsub.publish("chat/general", msg);

// Publish excluding a specific subscriber (useful for echo prevention)
let msg = Message::text("Broadcast from user");
pubsub.publish_excluding(subscriber_id, "chat/general", msg);

// Unsubscribe from a topic
pubsub.unsubscribe(subscriber_id, "chat/general");

// Remove subscriber when connection closes
pubsub.remove_subscriber(subscriber_id);

use sockudo_ws::pubsub::PubSub;

let pubsub = PubSub::new();

// Generate Pusher-style socket ID (format: "1234567890.9876543210")
let socket_id = PubSub::generate_socket_id();

// Create subscriber with custom socket ID
let (tx, rx) = mpsc::unbounded_channel();
let subscriber_id = pubsub.create_subscriber_with_id(&socket_id, tx);

// Subscribe/publish using socket ID
pubsub.subscribe_by_socket_id(&socket_id, "private-channel");
pubsub.publish_excluding_socket_id(&socket_id, "chat", Message::text("Hello"));

// Lookup subscriber by socket ID
if let Some(id) = pubsub.get_subscriber_by_socket_id(&socket_id) {
    println!("Found subscriber: {:?}", id);
}

// Get statistics
let topic_count = pubsub.topic_count();
let subscriber_count = pubsub.subscriber_count();
let messages_published = pubsub.messages_published();
let subscribers_in_topic = pubsub.topic_subscriber_count("chat/general");

use axum::{Router, body::Body, extract::Request, http::{Response, StatusCode, header}, routing::get};
use futures_util::{SinkExt, StreamExt};
use hyper_util::rt::TokioIo;
use sockudo_ws::{Config, Message, WebSocketStream, handshake::generate_accept_key};

async fn ws_handler(req: Request) -> Response<Body> {
    let key = req.headers().get("sec-websocket-key").unwrap().to_str().unwrap();
    let accept_key = generate_accept_key(key);

    tokio::spawn(async move {
        if let Ok(upgraded) = hyper::upgrade::on(req).await {
            let mut ws = WebSocketStream::server(TokioIo::new(upgraded), Config::default());
            
            while let Some(Ok(msg)) = ws.next().await {
                match msg {
                    Message::Text(text) => { ws.send(Message::text(text)).await.ok(); }
                    Message::Binary(data) => { ws.send(Message::Binary(data)).await.ok(); }
                    Message::Close(_) => break,
                    _ => {}
                }
            }
        }
    });

    Response::builder()
        .status(StatusCode::SWITCHING_PROTOCOLS)
        .header(header::UPGRADE, "websocket")
        .header(header::CONNECTION, "Upgrade")
        .header("Sec-WebSocket-Accept", accept_key)
        .body(Body::empty())
        .unwrap()
}

use sockudo_ws::{WebSocketServer, Http2, Config, Message};
use futures_util::{SinkExt, StreamExt};

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let listener = tokio::net::TcpListener::bind("127.0.0.1:8080").await?;
    
    let config = Config::builder()
        .http2_max_streams(100)
        .build();
    
    let server = WebSocketServer::<Http2>::new(config);

    loop {
        let (stream, _) = listener.accept().await?;
        
        // In production: wrap with TLS first
        // let tls_stream = tls_acceptor.accept(stream).await?;
        
        let server = server.clone();
        tokio::spawn(async move {
            server.serve(stream, |mut ws, req| async move {
                println!("HTTP/2 WebSocket at: {}", req.path);
                
                // Same API as HTTP/1.1!
                while let Some(msg) = ws.next().await {
                    if let Ok(msg) = msg {
                        ws.send(msg).await.ok();
                    }
                }
            }).await.ok();
        });
    }
}

use sockudo_ws::{WebSocketClient, Http2, Config, Message};

let client = WebSocketClient::<Http2>::new(Config::default());
let mut ws = client.connect(tls_stream, "wss://example.com/ws", None).await?;

ws.send(Message::text("Hello!")).await?;

use sockudo_ws::{WebSocketClient, Http2, Config};

let client = WebSocketClient::<Http2>::new(Config::default());
let mut conn = client.connect_multiplexed(tls_stream).await?;

// Open multiple WebSocket streams on the same connection
let mut ws1 = conn.open_websocket("wss://example.com/chat", None).await?;
let mut ws2 = conn.open_websocket("wss://example.com/notifications", None).await?;

use sockudo_ws::{WebSocketServer, Http3, Config, Message};
use futures_util::{SinkExt, StreamExt};

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    // Load TLS certificates (required for QUIC)
    let tls_config = load_server_tls_config()?;
    
    let ws_config = Config::builder()
        .http3_idle_timeout(30_000)
        .build();

    let server = WebSocketServer::<Http3>::bind(
        "0.0.0.0:4433".parse()?,
        tls_config,
        ws_config,
    ).await?;

    println!("HTTP/3 server listening on {}", server.local_addr()?);

    server.serve(|mut ws, req| async move {
        println!("HTTP/3 WebSocket at: {}", req.path);
        
        // Same API as HTTP/1.1 and HTTP/2!
        while let Some(msg) = ws.next().await {
            if let Ok(msg) = msg {
                ws.send(msg).await.ok();
            }
        }
    }).await?;

    Ok(())
}

use sockudo_ws::io_uring::UringStream;
use sockudo_ws::{Config, WebSocketStream};

#[tokio_uring::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let listener = tokio_uring::net::TcpListener::bind("127.0.0.1:8080".parse()?)?;

    loop {
        let (tcp_stream, _) = listener.accept().await?;
        
        // Wrap in UringStream for io_uring I/O
        let uring_stream = UringStream::new(tcp_stream);
        
        tokio_uring::spawn(async move {
            let mut ws = WebSocketStream::server(uring_stream, Config::default());
            
            while let Some(msg) = ws.next().await {
                if let Ok(msg) = msg {
                    ws.send(msg).await.ok();
                }
            }
        });
    }
}

use sockudo_ws::io_uring::UringStream;
use sockudo_ws::{WebSocketServer, Http2, Config};

#[tokio_uring::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let listener = tokio_uring::net::TcpListener::bind("127.0.0.1:8080".parse()?)?;
    let server = WebSocketServer::<Http2>::new(Config::default());

    loop {
        let (tcp_stream, _) = listener.accept().await?;
        
        // 1. Wrap TCP in UringStream for io_uring I/O
        let uring_stream = UringStream::new(tcp_stream);
        
        // 2. Add TLS (required for HTTP/2)
        // let tls_stream = tls_acceptor.accept(uring_stream).await?;
        
        // 3. Run HTTP/2 WebSocket over io_uring transport
        let server = server.clone();
        tokio_uring::spawn(async move {
            server.serve(uring_stream, |mut ws, req| async move {
                // WebSocket over HTTP/2 over io_uring!
                while let Some(msg) = ws.next().await {
                    if let Ok(msg) = msg {
                        ws.send(msg).await.ok();
                    }
                }
            }).await.ok();
        });
    }
}

// HTTP/1.1 (default)
let ws = WebSocketStream::server(tcp_stream, config);

// HTTP/2
let ws = WebSocketStream::server(h2_stream, config);

// HTTP/3
let ws = WebSocketStream::server(h3_stream, config);

// io_uring
let ws = WebSocketStream::server(uring_stream, config);

// Same message loop for all!
while let Some(msg) = ws.next().await {
    ws.send(msg?).await?;
}

use sockudo_ws::{Config, Compression};

let config = Config::builder()
    .compression(Compression::Shared)      // SHARED_COMPRESSOR
    .max_payload_length(16 * 1024)         // 16KB max message
    .ping_interval(30)                     // Ping after 30s inbound inactivity
    .pong_timeout(10)                      // Matching Pong deadline
    .pong_timeout_close(4201, "Pong reply not received in time")
    .idle_timeout(0)                       // Independent hard idle limit disabled
    .close_timeout(5)                      // Bounded Close flush/shutdown
    .max_backpressure(1024 * 1024)         // 1MB backpressure limit
    .build();

// Or use uWebSockets-style defaults
let config = Config::uws_defaults();

let config = Config::builder()
    .http2_window_size(1024 * 1024)        // 1MB stream window
    .http2_connection_window_size(2 * 1024 * 1024)  // 2MB connection window
    .http2_max_streams(100)                // Max concurrent streams
    .build();

let config = Config::builder()
    .http3_idle_timeout(30_000)            // 30 second idle timeout
    .build();

// Create server-side stream
let ws = WebSocketStream::server(tcp_stream, config);

// Create client-side stream
let ws = WebSocketStream::client(tcp_stream, config);

// Send messages
ws.send(Message::text("hello")).await?;
ws.send(Message::binary(bytes)).await?;

// Receive messages
while let Some(msg) = ws.next().await {
    // handle msg
}

// Close connection
ws.close(1000, "goodbye").await?;

// Backpressure handling
if ws.is_backpressured() {
    // Write buffer is full, consider slowing down
}

let (reader, writer) = ws.split();

// SplitReader - owns decoding and reports shared terminal state
reader.next().await  // Receive message
reader.is_closed()   // Check if closed (non-blocking)

// SplitWriter - bounded command handle to the connection writer driver
writer.send(msg).await?;
writer.send_text("hello").await?;
writer.send_binary(bytes).await?;
writer.close(1000, "bye").await?;
writer.is_closed()   // Check if closed (non-blocking)
writer.flush().await?;  // Flush accepted application writes

pub enum Message {
    Text(Bytes),      // Zero-copy, UTF-8 validated
    Binary(Bytes),
    Ping(Bytes),
    Pong(Bytes),
    Close(Option<CloseReason>),
}

// Create messages
let text_msg = Message::text("hello");           // From &str
let text_msg = Message::text(String::from("hello")); // From String
let binary_msg = Message::binary(vec![1, 2, 3]); // From Vec<u8>

// Access text content
if let Message::Text(bytes) = msg {
    let text: &str = msg.as_text().unwrap(); // Returns Option<&str>
}