Ratchet

Ratchet 是一个快速、健壮、轻量级且完全异步的 RFC6455 (WebSocket 协议) 实现。还包括 RFC7692 (WebSocket 压缩扩展) 的可选实现。

特性

• 使用 ratchet_ext 实现自己的扩展。
• 使用 ratchet_deflate 进行按消息的 deflate 或通过 deflate 功能启用。
• 使用 split 功能分割 WebSocket。

测试

Ratchet 完全经过测试，并且通过了客户端和服务器模式下的所有 Autobahn 测试。

示例

客户端

#[tokio::main]
async fn main() -> Result<(), Error> {
  let stream = TcpStream::connect("127.0.0.1:9001").await?;
  
  let upgraded = subscribe(
    WebSocketConfig::default(),
    stream,
    "ws://127.0.0.1/hello".try_into_request()?,
  )
  .await?;

  let UpgradedClient{ socket, subprotocol }=upgraded;
  let mut buf = BytesMut::new();

  loop {
    match websocket.read(&mut buf).await? {
      Message::Text => {
        websocket.write(&mut buf, PayloadType::Text).await?;
        buf.clear();
      }
      Message::Binary => {
        websocket.write(&mut buf, PayloadType::Binary).await?;
        buf.clear();
      }
      Message::Ping | Message::Pong => {
        // Ping messages are transparently handled by Ratchet
      }
      Message::Close(_) => break Ok(()),
    }
  }
}

服务器

#[tokio::main]
async fn main() -> Result<(), Error> {
    let listener = TcpListener::bind("127.0.0.1:9001").await?;
    let mut incoming = TcpListenerStream::new(listener);

    while let Some(socket) = incoming.next().await {
        let socket = socket?;

        // An upgrader contains information about what the peer has requested.
        let mut upgrader = ratchet::accept_with(
            socket,
            WebSocketConfig::default(),
            NoExtProvider,
            ProtocolRegistry::default(),
        )
        .await?;

        // You could opt to reject the connection
        // upgrader.reject(WebSocketResponse::new(404)?).await?;
        // continue;
      
        // Or you could reject the connection with headers
        // upgrader.reject(WebSocketResponse::with_headers(404, headers)?).await;
        // continue;

        let UpgradedServer {
            request,
            mut websocket,
            subprotocol,
        } = upgrader.upgrade().await?;
        
        let mut buf = BytesMut::new();

        loop {
            match websocket.read(&mut buf).await? {
                Message::Text => {
                    websocket.write(&mut buf, PayloadType::Text).await?;
                    buf.clear();
                }
                Message::Binary => {
                    websocket.write(&mut buf, PayloadType::Binary).await?;
                    buf.clear();
                }
                Message::Ping | Message::Pong => {
                  // Ping messages are transparently handled by Ratchet
                }
                Message::Close(_) => break,
            }
        }
    }
    
    Ok(())
}

Deflate

  let mut websocket = ratchet::accept_with(
      socket,
      WebSocketConfig::default(),
      DeflateProvider,
      ProtocolRegistry::default(),
  )
  .await?;

Split

// A split operation will only fail if the WebSocket is already closed.
let (mut sender, mut receiver) = websocket.split()?;
    
loop {
    match receiver.read(&mut buf).await? {
        Message::Text => {
            sender.write(&mut buf, PayloadType::Text).await?;
            buf.clear();
        }
        Message::Binary => {
            sender.write(&mut buf, PayloadType::Binary).await?;
            buf.clear();
        }
        Message::Ping | Message::Pong => {}
        Message::Close(_) => break Ok(()),
    }
}

计划中的特性

• futures-rs Sink 和 Stream 实现。
• tokio AsyncRead 和 AsyncWrite 实现。

许可证

Ratchet 根据 Apache License 2.0 许可。