如何在Rust+Tokio中实现无缓冲超时读取？（用于Telnet服务器处理）

需求与Python实现示例 #

想要编写一个简易Telnet服务器，无需用户按下回车即可捕获其按键输入。该需求可通过Python的telnetlib3库实现，示例代码如下：

import asyncio

import telnetlib3
from telnetlib3 import TelnetReaderUnicode, TelnetWriterUnicode


async def shell(reader: TelnetReaderUnicode, writer: TelnetWriterUnicode):
    # 处理输入的函数
    async def on_input(inp: str):
        match inp:
            case '\x1b[C' | 'd':  # 右键
                writer.write('Right pressed\r\n')
            case '\x1b[D' | 'a':  # 左键
                writer.write('Left pressed\r\n')
            case '\x1b' | 'q' | '\x03':  # Esc、q或Ctrl+C
                writer.write('\r\nBye!\r\n')
                writer.close()
                return True
            case _:
                print(f'Unknown input: {repr(inp)}')
    
    # 监听输入
    while True:
        inp: str = await reader.read(3)
        if inp and await on_input(inp):
            return

if __name__ == '__main__':
    # 创建事件循环，启动服务器并等待关闭
    loop = asyncio.new_event_loop()
    asyncio.set_event_loop(loop)
    coro = telnetlib3.create_server(port=2323, shell=shell)
    server = loop.run_until_complete(coro)
    loop.run_until_complete(server.wait_closed())

Rust+Tokio实现的问题 #

使用Rust结合Tokio搭建了类似的TCP服务器，但无论设置多小的缓冲区（例如设为3），只有当Telnet客户端按下回车时，服务器才能收到输入。请问是否可以直接使用无缓冲IO来捕获无换行的客户端按键输入？

当前Rust代码如下：

use std::io::Result;
use tokio::{
    io::{AsyncReadExt, AsyncWriteExt},
    net::TcpListener,
};

#[tokio::main]
async fn main() -> Result<()> {
    const ADDR: &str = "127.0.0.1:2323";
    let listener = TcpListener::bind(ADDR).await?;

    println!("Server running on {ADDR}");

    while let Ok((mut socket, _addr)) = listener.accept().await {
        // 生成任务处理连接
        tokio::spawn(async move {
            let (mut reader, mut writer) = socket.split();

            // 设置终端为原始模式以接收转义序列
            writer.write_all(b"\x1B[?25l").await.unwrap(); // 隐藏光标
            writer.write_all(b"\x1B[?1049h").await.unwrap(); // 保存光标位置并切换到备用屏幕
            writer.flush().await.unwrap();

            // 发送欢迎消息
            writer.write_all(b"Meow~\n").await.unwrap();

            // 读取客户端输入直到连接关闭
            let mut buf = [0; 3];
            loop {
                // 打印转义后的字符用于调试
                let n = reader.read(&mut buf).await.unwrap();
                for i in 0..n {
                    for e in std::ascii::escape_default(buf[i]) {
                        print!("{}", e as char);
                    }
                }

                writer
                    .write_all(
                        format!(
                            "\nReceived {n} bytes: {}",
                            String::from_utf8_lossy(&buf[..n])
                        )
                        .as_ref(),
                    )
                    .await
                    .unwrap();

                if n == 0 {
                    break;
                }
            }
        });
    }

    Ok(())
}

解决方案 #

问题的核心并非缓冲区大小，而是Telnet客户端默认启用行缓冲，需通过Telnet协议的选项协商让客户端切换为字符模式，逐个发送按键输入，无需等待回车。

关键步骤： #

1. 发送Telnet协议选项，请求客户端关闭行缓冲：
   • 发送WILL SUPPRESS GO AHEAD（选项3）：禁用"前进抑制"，允许字符逐个发送
   • 发送WILL ECHO（选项1）：请求服务器回显输入（可选，根据需求调整）
2. 处理客户端返回的Telnet控制报文，过滤掉协议控制码，只保留实际输入字符。

修改后的Rust代码： #

use std::io::Result;
use tokio::{
    io::{AsyncReadExt, AsyncWriteExt},
    net::TcpListener,
};

#[tokio::main]
async fn main() -> Result<()> {
    const ADDR: &str = "127.0.0.1:2323";
    let listener = TcpListener::bind(ADDR).await?;

    println!("Server running on {ADDR}");

    while let Ok((mut socket, _addr)) = listener.accept().await {
        tokio::spawn(async move {
            let (mut reader, mut writer) = socket.split();

            // 发送Telnet选项协商，开启字符模式
            // WILL SUPPRESS GO AHEAD (选项3)
            writer.write_all(&[0xFF, 0xFB, 0x03]).await.unwrap();
            // WILL ECHO (选项1)
            writer.write_all(&[0xFF, 0xFB, 0x01]).await.unwrap();
            writer.flush().await.unwrap();

            // 设置终端为原始模式
            writer.write_all(b"\x1B[?25l").await.unwrap();
            writer.write_all(b"\x1B[?1049h").await.unwrap();
            writer.flush().await.unwrap();

            writer.write_all(b"Meow~ Press any key (q to quit)\r\n").await.unwrap();
            writer.flush().await.unwrap();

            let mut buf = [0; 1];
            loop {
                let n = reader.read(&mut buf).await.unwrap();
                if n == 0 {
                    break;
                }

                let byte = buf[0];
                // 处理Telnet控制码
                match byte {
                    0xFF => {
                        // 读取后续的Telnet命令字节
                        let mut cmd_buf = [0; 2];
                        if reader.read_exact(&mut cmd_buf).await.is_err() {
                            break;
                        }
                        // 跳过协议响应，可根据需求扩展处理逻辑
                        continue;
                    }
                    b'q' | 0x1B | 0x03 => { // q、Esc、Ctrl+C
                        writer.write_all(b"\r\nBye!\r\n").await.unwrap();
                        writer.flush().await.unwrap();
                        break;
                    }
                    _ => {
                        // 处理普通按键输入
                        let msg = format!("\rReceived key: {:?}", byte as char);
                        writer.write_all(msg.as_bytes()).await.unwrap();
                        writer.flush().await.unwrap();
                    }
                }
            }

            // 恢复终端原始状态
            writer.write_all(b"\x1B[?1049l").await.unwrap();
            writer.write_all(b"\x1B[?25h").await.unwrap();
            writer.flush().await.unwrap();
        });
    }

    Ok(())
}

代码说明： #

• 通过Telnet选项协商，强制客户端关闭行缓冲，改为逐个字符发送输入
• 过滤Telnet协议的控制码（如0xFF开头的命令序列），避免干扰实际输入处理
• 简化缓冲区为1字节，确保每次读取单个按键
• 添加终端恢复逻辑，避免客户端终端状态异常

内容的提问来源于stack exchange，提问作者Hykilpikonna