Rust
Channel messaging
Collect and dispatch messages with mpsc.
Messages through a channel
In a chat, client threads must forward incoming messages to a central place. Each client handler is given a copy of a Sender; the central thread reads from the single Receiver. This way messages gather in a single order without locking shared state:
fn istemci_isle(akis: TcpStream, id: usize, tx: Sender<String>) -> io::Result<()> {
for s in BufReader::new(akis).lines() {
tx.send(format!("[{id}] {}", s?)).ok();
}
Ok(())
}
When the client closes the connection, lines() ends, the thread finishes and its sender is dropped.
Wiring up the accept loop
The accepting thread gives each connection an increasing id and a copy of the sender:
let (tx, rx) = mpsc::channel();
thread::spawn(move || {
for (id, akis) in dinleyici.incoming().take(2).enumerate() {
let tx = tx.clone();
let akis = akis.unwrap();
thread::spawn(move || istemci_isle(akis, id, tx));
}
// tx is dropped here
});
A real server has no take; a shutdown signal is sent through a separate channel.
Exercise Forward to the center
Build istemci_isle and the accept loop above. Let the main thread connect two clients in turn: one writes selam, the other nasılsın, each closing its connection. Then collect all messages with rx.iter(), sort them and print each one.
Show example solution
use std::io::{self, BufRead, BufReader, Write};
use std::net::{TcpListener, TcpStream};
use std::sync::mpsc::{self, Sender};
use std::thread;
fn istemci_isle(akis: TcpStream, id: usize, tx: Sender<String>) -> io::Result<()> {
for s in BufReader::new(akis).lines() {
tx.send(format!("[{id}] {}", s?)).ok();
}
Ok(())
}
fn main() -> io::Result<()> {
let dinleyici = TcpListener::bind("127.0.0.1:0")?;
let adres = dinleyici.local_addr()?;
let (tx, rx) = mpsc::channel();
thread::spawn(move || {
for (id, akis) in dinleyici.incoming().take(2).enumerate() {
let tx = tx.clone();
let akis = akis.unwrap();
thread::spawn(move || istemci_isle(akis, id, tx));
}
});
for mesaj in ["selam", "nasılsın"] {
let mut akis = TcpStream::connect(adres)?;
writeln!(akis, "{mesaj}")?;
}
let mut hepsi: Vec<String> = rx.iter().collect();
hepsi.sort();
for m in hepsi {
println!("{}", m);
}
Ok(())
}
Test yourself
-
When can the central receiver say "all messages have arrived"?
- After a second
- Never
- When all senders (each handler's and the accept loop's) are dropped
- At the first message
Show answer
When all senders (each handler's and the accept loop's) are dropped — If the accept loop keeps its own tx copy, it must drop it too when the loop ends.
-
Hand over the sender
let tx = tx.____();
thread::spawn(move || istemci_isle(akis, id, tx));
Show answer
let tx = tx.clone();
thread::spawn(move || istemci_isle(akis, id, tx));
-
Component and role
- Sender<String>
- Handler to center
- Receiver
- Central reader
- enumerate()
- Client id
- lines() biter
- Client left
Next lesson: Broadcast
