Channel messaging
- Rust course
- Advanced
- Lesson 27/81
- 6 min
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.
Practice 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.
Starter code
use std::io::{self, BufRead, BufReader, Write};
use std::net::{TcpListener, TcpStream};
use std::sync::mpsc::{self, Sender};
use std::thread;
fn main() {
// buraya yaz
}Sample 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(())
}Expected output
[0] selam [1] nasılsın
Questions and answers
Review the lesson questions together with their answers. To test yourself, solve the lesson interactively.
1 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 Correct answer
- At the first message
Explanation: If the accept loop keeps its own tx copy, it must drop it too when the loop ends.
2 Hand over the sender
let tx = tx.____();
thread::spawn(move || istemci_isle(akis, id, tx));let tx = tx.clone();
thread::spawn(move || istemci_isle(akis, id, tx));Explanation: Each client handler gets its own sender copy.
3 Component and role
| Term | Meaning |
|---|---|
Sender<String> |
Handler to center |
Receiver |
Central reader |
enumerate() |
Client id |
lines() biter |
Client left |