mio — Metal I/O: Non-Blocking Networking in Rust from First Principles

use std::io::{self, Read};

fn read_all(stream: &mut mio::net::TcpStream, buf: &mut Vec<u8>) -> io::Result<bool> {
    let mut tmp = [0u8; 4096];
    loop {
        match stream.read(&mut tmp) {
            Ok(0) => return Ok(false),  // connection closed gracefully
            Ok(n) => buf.extend_from_slice(&tmp[..n]), // got n bytes, keep reading
            Err(ref e) if e.kind() == io::ErrorKind::WouldBlock => {
                // No more data right now — we've drained the socket buffer
                // STOP reading. mio will fire the event again when more data arrives.
                return Ok(true);
            }
            Err(ref e) if e.kind() == io::ErrorKind::Interrupted => {
                // Signal interrupted the syscall — retry immediately
                continue;
            }
            Err(e) => return Err(e),
        }
    }
}

use mio::net::{TcpListener, TcpStream};
use mio::{Events, Interest, Poll, Token};
use std::collections::HashMap;
use std::io::{self, Read, Write};
use std::net::SocketAddr;

const SERVER: Token = Token(0);

fn main() -> io::Result<()> {
    let mut poll = Poll::new()?;
    let mut events = Events::with_capacity(128);

    let addr: SocketAddr = "127.0.0.1:9000".parse().unwrap();
    let mut server = TcpListener::bind(addr)?;
    poll.registry().register(&mut server, SERVER, Interest::READABLE)?;

    let mut connections: HashMap<Token, TcpStream> = HashMap::new();
    let mut next_id: usize = 1;

    println!("Echo server on {}", addr);
    println!("Test with: nc 127.0.0.1 9000");

    loop {
        // Block until the OS has at least one ready event
        poll.poll(&mut events, None)?;

        for event in events.iter() {
            match event.token() {
                SERVER => {
                    // New connection(s) waiting — accept all of them
                    loop {
                        match server.accept() {
                            Ok((mut stream, addr)) => {
                                let token = Token(next_id);
                                next_id += 1;
                                poll.registry().register(
                                    &mut stream,
                                    token,
                                    Interest::READABLE,
                                )?;
                                connections.insert(token, stream);
                                println!("[{:?}] connected from {}", token, addr);
                            }
                            Err(ref e) if e.kind() == io::ErrorKind::WouldBlock => break,
                            Err(e) => return Err(e),
                        }
                    }
                }

                token => {
                    // Data ready on an existing client
                    let done = if let Some(stream) = connections.get_mut(&token) {
                        let mut buf = Vec::with_capacity(4096);
                        let mut tmp = [0u8; 1024];
                        let alive = loop {
                            match stream.read(&mut tmp) {
                                Ok(0) => break false, // EOF
                                Ok(n) => buf.extend_from_slice(&tmp[..n]),
                                Err(ref e) if e.kind() == io::ErrorKind::WouldBlock => {
                                    break true
                                }
                                Err(ref e) if e.kind() == io::ErrorKind::Interrupted => {
                                    continue
                                }
                                Err(e) => return Err(e),
                            }
                        };

                        if !buf.is_empty() {
                            stream.write_all(&buf)?;
                            println!(
                                "[{:?}] echoed {} bytes",
                                token,
                                buf.len()
                            );
                        }
                        !alive
                    } else {
                        false
                    };

                    if done {
                        // Deregister the socket before dropping it
                        if let Some(mut stream) = connections.remove(&token) {
                            poll.registry().deregister(&mut stream)?;
                            println!("[{:?}] disconnected", token);
                        }
                    }
                }
            }
        }
    }
}

# Terminal 1: run the server
cargo run --release

# Terminal 2: connect with netcat and type anything
nc 127.0.0.1 9000
hello, mio!        ← you type this
hello, mio!        ← server echoes it back

# Terminal 3: open a second simultaneous client
nc 127.0.0.1 9000
another client     ← both clients handled by the SAME single thread