Async Listen

该包包含使用 async-std 在 Rust 中编写生产级服务器的各种辅助工具。

文档 | Github | Crate

功能

• 接受循环中的错误处理
• 限制传入连接的数量

许可证

许可协议为以下之一：

• Apache 许可协议 2.0，(./LICENSE-APACHE 或 https://apache.ac.cn/licenses/LICENSE-2.0)
• MIT 许可证 (./LICENSE-MIT 或 http://opensource.org/licenses/MIT)，任选其一。

贡献

除非您明确声明，否则您有意提交以包含在作品中的任何贡献，根据 Apache-2.0 许可协议定义，应按上述方式双许可，不得附加任何额外条款或条件。

lib.rs:

Async Listen

该包包含使用 async-std 在 Rust 中编写生产级服务器的各种辅助工具。

文档 | Github | Crate

实用工具

• ListenExt -- 为接受套接字流提供扩展 trait，为流提供有用的组合器
• error_hint -- 显示用户提示，说明如何解决 最重要的错误

低级实用工具

• is_transient_error -- 确定从 accept() 返回的错误是否可以忽略

示例

以下是一个相当详细的示例，演示了以下内容：

• 背压（同时连接数的限制）
• 错误处理
• Tcp 和 Unix 套接字的统一

use std::env::args;
use std::error::Error;
use std::fs::remove_file;
use std::io;
use std::time::Duration;

use async_std::task;
use async_std::net::TcpListener;
use async_std::prelude::*;

use async_listen::{ListenExt, ByteStream, backpressure, error_hint};


fn main() -> Result<(), Box<dyn Error>> {
    let (_, bp) = backpressure::new(10);
    #[cfg(unix)] {
        use async_std::os::unix::net::UnixListener;

        if args().any(|x| x == "--unix") {
            remove_file("./example.sock").ok();
            return task::block_on(async {
                let listener = UnixListener::bind("./example.sock").await?;
                eprintln!("Accepting connections on ./example.sock");
                let mut incoming = listener.incoming()
                    .log_warnings(log_accept_error)
                    .handle_errors(Duration::from_millis(500))
                    .backpressure_wrapper(bp);
                while let Some(stream) = incoming.next().await {
                    task::spawn(connection_loop(stream));
                }
                Ok(())
            });
        }
    }
    task::block_on(async {
        let listener = TcpListener::bind("localhost:8080").await?;
        eprintln!("Accepting connections on localhost:8080");
        let mut incoming = listener.incoming()
            .log_warnings(log_accept_error)
            .handle_errors(Duration::from_millis(500))
            .backpressure_wrapper(bp);
        while let Some(stream) = incoming.next().await {
            task::spawn(async {
                if let Err(e) = connection_loop(stream).await {
                    eprintln!("Error: {}", e);
                }
            });
        }
        Ok(())
    })
}

async fn connection_loop(mut stream: ByteStream) -> Result<(), io::Error> {
    println!("Connected from {}", stream.peer_addr()?);
    task::sleep(Duration::from_secs(5)).await;
    stream.write_all("hello\n".as_bytes()).await?;
    Ok(())
}

fn log_accept_error(e: &io::Error) {
    eprintln!("Accept error: {}. Sleeping 0.5s. {}", e, error_hint(&e));
}