async-smux

crates.io

为 smol/async-std 和任何与 futures 兼容的异步运行时提供轻量级的异步 smux (Simple MUltipleXing) 库。

async-smux 消费一个实现 AsyncRead + AsyncWrite + Unpin + Send 的结构体，例如 TcpStream 和 TlsStream，以创建一个 Mux<T> 结构体。然后您可以在 Mux<T> 上创建多个 MuxStream<T>（最多 4294967295 个），这也实现了 AsyncRead + AsyncWrite + Unpin + Send。

基准测试

以下是我本地机器上的简单基准测试结果，与原始版本 smux（用 go 编写）进行比较。

运行 cargo bench 以自行测试。有关更多详细信息，请检查 /benches 目录。

惰性

此库不会生成任何线程或任务。它只生成一些 future。因此，它与运行时完全独立。

Mux 和 MuxStream 完全是惰性的，如果您不调用 poll()，它们将不会执行任何操作。

任何轮询操作，包括 .read()、.write()、accept() 和 connect()，都将使 Mux 和 MuxStream 运行。

规范

VERSION(1B) | CMD(1B) | LENGTH(2B) | STREAMID(4B) | DATA(LENGTH)

VERSION: 1

CMD:
    SYN(0)
    FIN(1)
    PSH(2)
    NOP(3)

STREAMID: Randomly chosen number

示例

use async_smux::{Mux, MuxConfig};
use async_std::net::{TcpListener, TcpStream};
use async_std::prelude::*;

async fn echo_server() {
    let listener = TcpListener::bind("0.0.0.0:12345").await.unwrap();
    let (stream, _) = listener.accept().await.unwrap();
    let mux = Mux::new(stream, MuxConfig::default());
    loop {
        let mut mux_stream = mux.accept().await.unwrap();
        let mut buf = [0u8; 1024];
        let size = mux_stream.read(&mut buf).await.unwrap();
        mux_stream.write(&buf[..size]).await.unwrap();
    }
}

fn main() {
    async_std::task::spawn(echo_server());
    async_std::task::block_on(async {
        smol::Timer::after(std::time::Duration::from_secs(1)).await;
        let stream = TcpStream::connect("127.0.0.1:12345").await.unwrap();
        let mux = Mux::new(stream, MuxConfig::default());
        for i in 0..100 {
            let mut mux_stream = mux.connect().await.unwrap();
            let mut buf = [0u8; 1024];
            mux_stream.write(b"hello").await.unwrap();
            let size = mux_stream.read(&mut buf).await.unwrap();
            let reply = String::from_utf8(buf[..size].to_vec()).unwrap();
            println!("{}: {}", i, reply);
        }
    });
}