Learn RustChapter 10: Streaming replies
Lesson 10.2 · 18 min
Answers that arrive in pieces
Lesson 10.1 ended on a wall: a Box<dyn Replier> cannot hold a trait whose reply is an async fn. The way round it is to change what reply hands back. Instead of the answer, it returns at once with the receiving end of a pipe, and the answer comes down that pipe later, a word at a time. By the end, Chatterbox v1.1 prints the pretend model's answer as it is typed.
Adds to ChatterboxAnswers arrive through a channel; the pretend model types word by word.
Compiler errors you will meet E0373
Hand back a pipe, not an answer
At a busy café you do not stand at the till while your coffee is made. You get a buzzer straight away, sit down, and it tells you when something is ready. Chatterbox wants the same deal from a replier: something now, and the answer when it is ready.
In Rust that something is a channel: a pipe with two ends. Whatever goes in at one end comes out of the other, in the same order. So the new promise in the Replier trait is a plain, non-async method:
fn reply(&self, history: &[Message]) -> Pieces, wherePiecesis the receiving end of a channel whose items areResult<String, ChatError>: each one a piece of text, or what went wrong.
Because reply is an ordinary function again, Box<dyn Replier> works exactly as it did in lesson 8.4. The waiting has moved out of reply and into the pipe.
A channel has two ends
let (tx, rx) = mpsc::channel(8); makes one. It gives back a tuple (lesson 2.6) of its two ends. The short names are a habit: tx for "transmit" and rx for "receive".
mpscstands for multiple producer, single consumer: many senders may put things in, and exactly one receiver takes them out. It lives intokio::sync.txis the Sender.tx.send(piece).awaitputs one item in. The8is how many items may wait in the pipe; when it is full,sendpauses until there is room, so a fast writer cannot drown a slow reader.rxis the Receiver.rx.recv().awaitwaits for the next item and gives anOption:Some(piece), orNoneonce the pipe is empty and every Sender has gone.
A task: work that carries on by itself
Something has to put the words into the pipe after reply has already returned. That is a task: a piece of async work that the tokio runtime runs alongside the rest of your program. tokio::spawn(...) starts one and returns immediately, without waiting for it to finish.
What you give spawn is an async block: async { … } is code that can .await, wrapped up as a future (lesson 10.1) without being given a name. The word move in front of it means "this block takes ownership of every variable it uses". Why it is needed comes next. First, see what "returns immediately" means.
PredictWho speaks first?
use std::time::Duration;
use tokio::sync::mpsc;
fn reply(message: &str) -> mpsc::Receiver<String> {
let text = format!("you said: {message}");
let (tx, rx) = mpsc::channel(8);
tokio::spawn(async move {
for word in text.split(' ') {
tokio::time::sleep(Duration::from_millis(100)).await;
if tx.send(word.to_string()).await.is_err() {
return;
}
}
});
rx
}
#[tokio::main]
async fn main() {
let mut pieces = reply("hello there");
println!("reply has returned");
while let Some(word) = pieces.recv().await {
println!("got: {word}");
}
println!("the channel is closed");
}
Each word waits 100 milliseconds before it is sent. What is the FIRST line this program prints?
reply has returnedgot: yougot: you said: hello there
Reveal the answer
The program prints: reply has returned tokio::spawn only starts the task; it does not wait for it. So reply returns the Receiver at once, and main prints its line before any word is sent. Then recv().await waits for each word in turn, and when the task finishes, its Sender is dropped and the loop ends.
Why the task must own what it uses
Look at the order of events. reply returns, and its local variables text and tx are dropped at its closing brace. The task is still running, and it still needs both. If it had only borrowed them, it would be reading values that no longer exist, which lesson 4.4's rules forbid. Take move away and the compiler says so.
Compiler CoachThe compiler, translated
This is the exact output of the real compiler. Toggle the English, then pick a fix and see what actually happens.
use tokio::sync::mpsc;
fn reply(message: &str) -> mpsc::Receiver<String> {
let text = format!("you said: {message}");
let (tx, rx) = mpsc::channel(8);
tokio::spawn(async {
for word in text.split(' ') {
let _ = tx.send(word.to_string()).await;
}
});
rx
}
#[tokio::main]
async fn main() {
let mut pieces = reply("hello there");
while let Some(word) = pieces.recv().await {
println!("{word}");
}
}
error[E0373]: async block may outlive the current function, but it borrows `tx`, which is owned by the current function --> src/main.rs:6:18 | 6 | tokio::spawn(async { | ^^^^^ may outlive borrowed value `tx` 7 | for word in text.split(' ') { 8 | let _ = tx.send(word.to_string()).await; | -- `tx` is borrowed here | = note: async blocks are not executed immediately and must either take a reference or ownership of outside variables they use help: to force the async block to take ownership of `tx` (and any other referenced variables), use the `move` keyword | 6 | tokio::spawn(async move { | ++++
E0373 in the compiler error atlas →
async block may outlive the current function, but it borrows `tx`: the task can still be running afterreplyhas returned and droppedtx.- A borrow must end before its owner is dropped, and nothing promises that the task ends first.
- The help line says what to do: to force the async block to take ownership of `tx` (and any other referenced variables), write
async move. Here the other one istext.
Fixes that were compiled for real
- async move: the task owns text and tx — compiles. Ownership of both moves into the task, the way a value moves into a function (lesson 4.3). They are dropped when the task ends, and dropping tx is what closes the channel.
- Trap: move a borrowed message — still E0521. <code>message</code> is a <code>&str</code>, a borrow of text that belongs to the caller. Moving the borrow into the task still leaves the task pointing at text it does not own. That is why <code>reply</code> builds an owned <code>String</code> first.
- Trap: use text again after the move — still E0382. Once <code>text</code> has moved into the task, <code>reply</code> no longer owns it. The same rule as moving into a function.
- Clone what you still need here — compiles, with a cost. The task gets its own copy and <code>reply</code> keeps the original. Chatterbox's real model does this with its <code>url</code> and <code>key</code>, because a task cannot borrow from <code>self</code> either.
How an answer ends
There is no "the end" message. The channel itself says when an answer is complete: once every Sender has been dropped and the pipe is empty, recv() gives None. When the task finishes, it drops the tx it owns, so while let Some(piece) = pieces.recv().await stops on its own. In the program below, drop(tx) does the same thing by hand: it ends a value's life on the spot, as if it had reached the end of its block.
PredictAfter the last piece
use tokio::sync::mpsc;
#[tokio::main]
async fn main() {
let (tx, mut rx) = mpsc::channel(8);
tx.send("hello").await.unwrap();
drop(tx); // the only Sender is gone
println!("{:?}", rx.recv().await);
println!("{:?}", rx.recv().await);
}
What is the second line this program prints?
NoneSome("hello")Some("")
Reveal the answer
The program prints: None The first recv still gets Some("hello"): dropping the Sender does not throw away what is already in the pipe. The second finds the pipe empty and no Sender left, so nothing can ever arrive. Instead of waiting forever, it gives None.
It also works the other way round. If the Receiver is dropped, nobody is listening, and send returns an Err that hands the item back. That is why the pretend model's task checks every send: if tx.send(Ok(piece)).await.is_err() { return; }. If Chatterbox stops reading, the task stops typing.
PredictTalking to nobody
use tokio::sync::mpsc;
#[tokio::main]
async fn main() {
let (tx, rx) = mpsc::channel(8);
drop(rx); // the listener walks away
match tx.send(String::from("hello")).await {
Ok(()) => println!("sent"),
Err(_) => println!("nobody is listening"),
}
}
What does this program print?
sentnobody is listeninghello
Reveal the answer
The program prints: nobody is listening With the Receiver gone, nothing could ever take the item out, so send fails at once instead of putting it in a pipe nobody reads.
Chatterbox v1.1
The trait hands back Pieces
In replier.rs, pub type Pieces = mpsc::Receiver<Result<String, ChatError>>; gives that long type a short name, and reply returns it.
Echo answers all at once
all_at_once makes a channel with room for one, puts a whole answer (or an error) in it with try_send, and returns the Receiver. try_send sends without waiting. It fails if the pipe is full, or if the Receiver has already gone; here neither can happen, because the channel is brand new and all_at_once still holds the Receiver. .expect("…") is like unwrap() (lesson 5.4), with your own message if it ever fails. The Sender is dropped at the end of the function, so the channel is already closed: one piece, then None.
Pretend types
Pretend builds its sentence as before, then spawns a task that sends it a word at a time, with a 40 millisecond pause after each, so you can watch it arrive.
Model asks inside a task
Model moves its HTTP call into an async fn ask, using reqwest's async client: .send().await? and .json().await?. Its task owns clones of url, key and the request body, and sends the whole answer as one piece.
main waits for the keyboard
In main.rs, main becomes #[tokio::main] async fn main, and two use lines matter:
use std::io::{self, Write};brings inioitself (that is theself) andWrite, the trait that givesflush. A trait must be in scope before you can call its methods.use tokio::io::{AsyncBufReadExt, BufReader};: aBufReaderreads input in big blocks and hands it out a line at a time, andAsyncBufReadExtis the trait that adds.lines()andnext_lineto async readers.
BufReader::new(tokio::io::stdin()).lines() reads the keyboard; lines.next_line().await? gives Some(line), or None when input ends.
answer prints as pieces arrive
async fn answer asks the replier for its Pieces and prints each one with print!, which, unlike println!, adds no new line. The catch: the terminal is normally handed text only when a line ends. Until then it waits in a buffer, a holding area in memory. io::stdout().flush()? pushes out what is waiting, right now. answer also gathers the pieces into one String for the history, or None if a piece was an error.
Tests that can wait
The tests become #[tokio::test] async fn: like #[test], but each test gets a runtime of its own, so it can .await.
Going deeper Small print in v1.1
type makes a new name for an existing type, not a new type: a Pieces and a mpsc::Receiver<Result<String, ChatError>> are the same thing, and either name works anywhere.
In Model's task, let _ = tx.send(answer).await; ignores the result on purpose: if nobody is listening, there is no one to tell.
For your name, unwrap_or_default() takes the line, or empty text if input had already ended. The chat line and /again both call answer with replier.as_ref(), which borrows the boxed replier as a &dyn Replier.
You type
Sam hello! /remember pizza extra cheese /recall pizza /model pretend can you hear me? /stats /history /quit
Chatterbox prints (a real run)
Chatterbox v1.1 (replies by: echo) What should I call you? Hi Sam! I'm Chatterbox. Ask me anything. bot: you said: hello! bot: got it: pizza is extra cheese bot: pizza is extra cheese bot: replies now come from pretend bot: (pretend model) I read all 3 messages. You last said: can you hear me? bot: your top words: can (1), hear (1), hello! (1) 1. you: hello! 2. bot: you said: hello! 3. you: can you hear me? 4. bot: (pretend model) I read all 3 messages. You last said: can you hear me? bot: bye, Sam! messages so far: 2 saved 4 messages to chatterbox.json
Show the whole program (open it after you have tried the exercise below: it contains the answer)
src/main.rs39 new lines
// Chatterbox: a small chat program, built one lesson at a time.
mod command;
mod message;
mod replier;
mod store;
use command::{Command, parse};
use message::Message;
use replier::{Echo, Model, Pretend, Replier};
use std::collections::HashMap;
use std::io::{self, Write};
use tokio::io::{AsyncBufReadExt, BufReader};
/// What the bot is called. A constant: fixed for the whole program.
const BOT: &str = "Chatterbox";
/// The version as a pair of numbers: (major, minor).
const VERSION: (u32, u32) = (1, 1);
/// The longest message we accept, in bytes.
const MAX_LEN: usize = 200;
/// What /help prints.
const HELP: &str = "commands: /help, /again, /history, /count, /stats, /remember <name> <fact>, /recall <name>, /find <word>, /last <n>, /model <echo|pretend|real>, /say <text>, /shout <text>, /debug <text>, /quit (or /h, /q, /exit)";
/// The first line Chatterbox prints.
fn banner() -> String {
format!("{} v{}.{}", BOT, VERSION.0, VERSION.1)
}
/// The greeting, with a name filled in.
fn greeting(name: &str) -> String {
let mut text = format!("Hi {name}! I'm {BOT}.");
text.push_str(" Ask me anything.");
text
}
/// The last `n` items of any slice: messages, numbers, words, anything.
/// `<T>` says "for any type T"; the body never needs to know which one.
fn last_n<T>(items: &[T], n: usize) -> &[T] {
let start = items.len().saturating_sub(n);
&items[start..]
}
/// Print an answer as its pieces arrive, and give back the whole of it
/// (None if it went wrong: what went wrong has been printed instead).
async fn answer(replier: &dyn Replier, history: &[Message]) -> Result<Option<String>, io::Error> {
let mut pieces = replier.reply(history);
let mut whole = String::new();
print!("bot: ");
while let Some(piece) = pieces.recv().await {
match piece {
Ok(text) => {
print!("{text}");
io::stdout().flush()?; // show it now, not at the end of the line
whole.push_str(&text);
}
Err(e) => {
println!("(that went wrong: {e})");
return Ok(None);
}
}
}
println!();
Ok(Some(whole))
}
// #[tokio::main] starts tokio's runtime, the part that runs async code, and
// then runs this async main inside it.
#[tokio::main]
async fn main() -> Result<(), io::Error> {
let mut messages_seen: u32 = 0;
let quiet = false;
let mut last: Option<String> = None; // the last thing you said, for /again
let mut history: Vec<Message> = store::load(store::FILE)?; // last time's conversation, if any
let mut facts: HashMap<String, String> = HashMap::new(); // name -> fact, for /recall
// Box<dyn Replier>: "some Replier, decided while the program runs".
// With a key in the environment, start with the real model.
let mut replier: Box<dyn Replier> = match Model::from_env() {
Ok(model) => Box::new(model),
Err(_) => Box::new(Echo { max_len: MAX_LEN }),
};
println!("{} (replies by: {})", banner(), replier.name());
// The keyboard, read one line at a time without blocking the program.
let mut lines = BufReader::new(tokio::io::stdin()).lines();
println!("What should I call you?");
let name_line = lines.next_line().await?.unwrap_or_default();
let name = name_line.trim();
if !quiet {
println!("{}", greeting(name));
}
if !history.is_empty() {
println!(
"(I remember {} messages from last time. /history shows them.)",
history.len()
);
}
// next_line gives None when there is no more input: the keyboard was closed.
while let Some(line) = lines.next_line().await? {
let message = line.trim();
let cmd = match parse(message) {
Ok(cmd) => cmd,
Err(e) => {
println!("bot: {e}");
continue;
}
};
match cmd {
Command::Empty => {} // nothing typed: nothing to answer
Command::Quit => {
println!("bot: bye, {name}!");
break;
}
Command::Help => println!("bot: {HELP}"),
Command::Again => match &last {
Some(text) => {
answer(replier.as_ref(), &[Message::new("you", text)]).await?;
}
None => println!("bot: nothing to repeat yet"),
},
Command::History => {
if history.is_empty() {
println!("bot: nothing said yet");
} else {
for (i, m) in history.iter().enumerate() {
println!("{}. {}", i + 1, m.to_line());
}
}
}
Command::Count => {
let bytes: usize = history.iter().map(|m| m.text.len()).sum();
println!("bot: {} messages, {} bytes of text", history.len(), bytes);
}
Command::Stats => {
// One chain of steps: your messages -> their words -> a count per word.
let mut counts: HashMap<String, u32> = HashMap::new();
for word in history
.iter()
.filter(|m| m.author == "you")
.flat_map(|m| m.text.split_whitespace())
{
*counts.entry(word.to_lowercase()).or_insert(0) += 1;
}
let mut top: Vec<(String, u32)> = counts.into_iter().collect();
// most used first; ties in alphabetical order, so the answer never changes
top.sort_by(|a, b| b.1.cmp(&a.1).then(a.0.cmp(&b.0)));
let words: Vec<String> = top
.iter()
.take(3)
.map(|(word, n)| format!("{word} ({n})"))
.collect();
if words.is_empty() {
println!("bot: nothing said yet");
} else {
println!("bot: your top words: {}", words.join(", "));
}
}
Command::Remember(key, fact) => {
println!("bot: got it: {key} is {fact}");
facts.insert(key, fact);
}
Command::Recall(Some(key)) => match facts.get(&key) {
Some(fact) => println!("bot: {key} is {fact}"),
None => println!("bot: I don't know anything about {key}"),
},
Command::Recall(None) => {
if facts.is_empty() {
println!("bot: I don't remember anything yet. Try /remember");
} else {
// A HashMap has no order, so sort the names before printing them.
let mut names = Vec::new();
for name in facts.keys() {
names.push(name.as_str());
}
names.sort();
println!("bot: I remember: {}", names.join(", "));
}
}
Command::Find(word) => {
// The closure borrows `word` from here: a small function made on the spot.
let hits: Vec<&Message> = history
.iter()
.filter(|m| m.author == "you" && m.text.contains(&word))
.collect();
if hits.is_empty() {
println!("bot: you never said {word}");
} else {
for m in hits {
println!("{}", m.to_line());
}
}
}
Command::Last(n) => {
for m in last_n(&history, n) {
println!("{}", m.to_line());
}
}
Command::Model(choice) => {
let next: Result<Box<dyn Replier>, String> = match choice.as_str() {
"echo" => Ok(Box::new(Echo { max_len: MAX_LEN })),
"pretend" => Ok(Box::new(Pretend)),
"real" => match Model::from_env() {
Ok(model) => Ok(Box::new(model)),
Err(e) => Err(e.to_string()),
},
_ => Err(format!(
"I don't know a model called {choice}. Try echo, pretend or real"
)),
};
match next {
Ok(r) => {
replier = r;
println!("bot: replies now come from {}", replier.name());
}
Err(why) => println!("bot: {why}"),
}
}
Command::Say(text) => println!("bot: {text}"),
Command::Shout(text) => println!("bot: {}", Message::new("you", &text).shout()),
Command::Debug(text) => println!("bot: {:?}", parse(&text)),
Command::Chat(text) => {
messages_seen += 1;
history.push(Message::new("you", &text));
// the history holds its own copy, so `text` can move into `last`
last = Some(text);
if let Some(reply) = answer(replier.as_ref(), &history).await? {
history.push(Message::new("bot", &reply));
}
}
}
}
println!("messages so far: {messages_seen}");
store::save(store::FILE, &history)?;
let s = if history.len() == 1 { "" } else { "s" };
println!("saved {} message{s} to {}", history.len(), store::FILE);
Ok(())
}
// Tests: `cargo test` builds and runs them; the program itself never sees them.
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn last_n_takes_from_the_end() {
assert_eq!(last_n(&[1, 2, 3, 4], 2), &[3, 4]);
assert_eq!(last_n(&["a", "b"], 5), &["a", "b"]);
assert!(last_n(&[1, 2, 3], 0).is_empty());
}
}
src/command.rsunchanged
//! Commands: what a line of input means, and what can be wrong with it.
use std::fmt;
/// Everything a line of input can mean. One enum, so a `match` on it must
/// handle every kind of line: the compiler checks that nothing is forgotten.
#[derive(Debug)]
pub enum Command {
Empty,
Help,
Quit,
Again,
History,
Count,
Stats,
Remember(String, String),
Recall(Option<String>),
Find(String),
Last(usize),
Model(String),
Say(String),
Shout(String),
Debug(String),
Chat(String),
}
/// Everything that can be wrong with a line of input. parse() returns one of
/// these instead of a Command, and main prints it.
#[derive(Debug)]
pub enum CommandError {
Unknown(String),
Missing(String, String),
Usage(String),
NotACount(String),
}
/// How a CommandError reads when printed with {}.
impl fmt::Display for CommandError {
fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
match self {
CommandError::Unknown(word) => {
write!(f, "I don't know the command {word}. Try /help.")
}
CommandError::Missing(verb, example) => {
write!(f, "{verb} what? Try /{verb} {example}")
}
CommandError::Usage(form) => write!(f, "try {form}"),
CommandError::NotACount(text) => write!(f, "'{text}' is not a count. Try /last 3"),
}
}
}
/// The text after a command word, or the error that says it is missing.
fn need(line: &str, verb: &str, example: &str) -> Result<String, CommandError> {
match arg(line) {
Some(text) => Ok(String::from(text)),
None => Err(CommandError::Missing(
String::from(verb),
String::from(example),
)),
}
}
/// How many messages /last should show. "two" or "-1" is an error.
fn parse_count(text: &str) -> Result<usize, CommandError> {
match text.parse::<usize>() {
Ok(n) => Ok(n),
Err(_) => Err(CommandError::NotACount(String::from(text))),
}
}
/// Turn one line of input into a Command, or say what is wrong with it.
/// Each `?` means: if this step failed, return its error right here.
pub fn parse(line: &str) -> Result<Command, CommandError> {
let cmd = match first_word(line) {
"" => Command::Empty,
"/help" | "/h" => Command::Help,
"/quit" | "/q" | "/exit" => Command::Quit,
"/again" => Command::Again,
"/history" => Command::History,
"/count" => Command::Count,
"/stats" => Command::Stats,
"/remember" => {
let usage = CommandError::Usage(String::from("/remember <name> <fact>"));
let rest = match arg(line) {
Some(rest) => rest,
None => return Err(usage),
};
match arg(rest) {
Some(fact) => Command::Remember(String::from(first_word(rest)), String::from(fact)),
None => return Err(usage),
}
}
"/recall" => match arg(line) {
Some(name) => Command::Recall(Some(String::from(name))),
None => Command::Recall(None),
},
"/find" => Command::Find(need(line, "find", "hello")?),
"/last" => Command::Last(parse_count(&need(line, "last", "3")?)?),
"/model" => Command::Model(need(line, "model", "pretend")?),
"/debug" => Command::Debug(need(line, "debug", "/say hi")?),
"/say" => Command::Say(need(line, "say", "hello")?),
"/shout" => Command::Shout(need(line, "shout", "hello")?),
word => {
if is_command(word) {
return Err(CommandError::Unknown(String::from(word)));
}
Command::Chat(String::from(line))
}
};
Ok(cmd)
}
/// True when the text is a command: it starts with a slash.
fn is_command(text: &str) -> bool {
text.starts_with('/')
}
/// The first word of the text: a slice (&str) pointing into it, not a copy.
fn first_word(text: &str) -> &str {
match text.find(' ') {
Some(i) => &text[..i],
None => text,
}
}
/// The text after the command word, if there is any: Some("hello") for
/// "/say hello", None for a bare "/say". No more "" standing in for nothing.
fn arg(line: &str) -> Option<&str> {
match line.find(' ') {
Some(i) => {
let rest = line[i + 1..].trim();
if rest.is_empty() { None } else { Some(rest) }
}
None => None,
}
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn plain_text_is_chat() {
assert!(matches!(parse("hello there"), Ok(Command::Chat(t)) if t == "hello there"));
}
#[test]
fn help_has_a_short_form() {
assert!(matches!(parse("/help"), Ok(Command::Help)));
assert!(matches!(parse("/h"), Ok(Command::Help)));
}
#[test]
fn say_needs_text() {
assert_eq!(
parse("/say").unwrap_err().to_string(),
"say what? Try /say hello"
);
}
#[test]
fn last_needs_a_number() {
assert!(matches!(parse("/last 3"), Ok(Command::Last(3))));
assert_eq!(
parse("/last two").unwrap_err().to_string(),
"'two' is not a count. Try /last 3"
);
}
#[test]
fn unknown_commands_are_errors() {
assert_eq!(
parse("/dance").unwrap_err().to_string(),
"I don't know the command /dance. Try /help."
);
}
#[test]
fn remember_takes_a_name_and_a_fact() {
assert!(
matches!(parse("/remember pizza extra cheese"), Ok(Command::Remember(k, v)) if k == "pizza" && v == "extra cheese")
);
assert!(parse("/remember pizza").is_err());
}
}
src/message.rsunchanged
//! A Message: one line of the conversation.
use serde::{Deserialize, Serialize};
/// One line of the conversation: who said it, and what they said.
#[derive(Serialize, Deserialize)]
pub struct Message {
pub author: String,
pub text: String,
}
impl Message {
/// Make a Message from two pieces of borrowed text.
pub fn new(author: &str, text: &str) -> Message {
Message {
author: String::from(author),
text: String::from(text),
}
}
/// The message as one transcript line: "you: hello".
pub fn to_line(&self) -> String {
format!("{}: {}", self.author, self.text)
}
/// The text in capitals, for /shout.
pub fn shout(&self) -> String {
self.text.to_uppercase()
}
}
src/replier.rs108 new lines
//! Who answers: anything that can reply, behind one trait.
use crate::message::Message;
use serde::{Deserialize, Serialize};
use std::env;
use std::fmt;
use std::time::Duration;
use tokio::sync::mpsc;
/// What can go wrong while getting a reply.
#[derive(Debug)]
pub enum ChatError {
/// The request did not get a good answer (no network, bad key, server error).
Http(reqwest::Error),
/// The model answered, but with nothing in it.
Empty,
/// /model real was asked for, but a setting is missing.
NotSet(&'static str),
}
impl fmt::Display for ChatError {
fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
match self {
ChatError::Http(e) => write!(f, "the model call failed: {e}"),
ChatError::Empty => write!(
f,
"there was nothing to reply with (an empty conversation or an empty answer)"
),
ChatError::NotSet(name) => write!(f, "set {name} first (see lesson 9.2)"),
}
}
}
/// Lets `?` turn a reqwest error into a ChatError on its own.
impl From<reqwest::Error> for ChatError {
fn from(e: reqwest::Error) -> Self {
ChatError::Http(e)
}
}
/// The receiving end of an answer: its pieces arrive in order, as they are
/// ready, and the channel closes when the answer is complete.
pub type Pieces = mpsc::Receiver<Result<String, ChatError>>;
/// A promise with a name: any type that implements Replier can answer a
/// conversation. `reply` returns at once; the answer arrives on the channel.
pub trait Replier {
/// A short name, for "/model" and the banner.
fn name(&self) -> &str;
/// Start answering the conversation so far (the last message is the newest).
fn reply(&self, history: &[Message]) -> Pieces;
}
/// A channel that already holds a whole answer, or an error, and is closed.
fn all_at_once(answer: Result<String, ChatError>) -> Pieces {
let (tx, rx) = mpsc::channel(1);
// A new channel with room for one cannot be full, so this send always works.
tx.try_send(answer).expect("a new channel has room for one");
rx
}
/// The echo Chatterbox has had since chapter 3. It answers all at once.
pub struct Echo {
pub max_len: usize,
}
impl Replier for Echo {
fn name(&self) -> &str {
"echo"
}
fn reply(&self, history: &[Message]) -> Pieces {
let message = match history.last() {
Some(m) => &m.text,
None => return all_at_once(Err(ChatError::Empty)),
};
let text = if message.len() > self.max_len {
format!("that is too long (the limit is {} bytes)", self.max_len)
} else if message.ends_with('?') {
String::from("Good question! I can only echo. Try /model real to ask a real model.")
} else {
format!("you said: {message}")
};
all_at_once(Ok(text))
}
}
/// A stand-in for a real model. It types its answer a word at a time, the way
/// a real model does, so you can watch an answer arrive in pieces.
pub struct Pretend;
impl Replier for Pretend {
fn name(&self) -> &str {
"pretend"
}
fn reply(&self, history: &[Message]) -> Pieces {
let last = match history.last() {
Some(m) => &m.text,
None => return all_at_once(Err(ChatError::Empty)),
};
let text = format!(
"(pretend model) I read all {} message{}. You last said: {last}",
history.len(),
if history.len() == 1 { "" } else { "s" },
);
let (tx, rx) = mpsc::channel(8);
// The task runs on after this function has returned, so it cannot
// borrow anything from here: `move` hands `text` and `tx` over to it.
tokio::spawn(async move {
for (i, word) in text.split(' ').enumerate() {
let piece = if i == 0 {
word.to_string()
} else {
format!(" {word}")
};
if tx.send(Ok(piece)).await.is_err() {
return; // nobody is listening any more
}
tokio::time::sleep(Duration::from_millis(40)).await;
}
});
rx
}
}
/// One message as a chat API expects it: a role and some content.
#[derive(Serialize)]
struct ApiMessage {
role: String,
content: String,
}
/// The body we send: which model, and the whole conversation.
#[derive(Serialize)]
struct Request {
model: String,
messages: Vec<ApiMessage>,
}
/// The parts of the answer we read. serde ignores every other field.
#[derive(Deserialize)]
struct Response {
choices: Vec<Choice>,
}
#[derive(Deserialize)]
struct Choice {
message: ApiReply,
}
#[derive(Deserialize)]
struct ApiReply {
content: String,
}
/// A real model behind an OpenAI-style chat endpoint, configured from
/// environment variables so the key never sits in the code.
pub struct Model {
url: String,
key: String,
model: String,
}
impl Model {
/// Read the three settings, or say which one is missing.
pub fn from_env() -> Result<Model, ChatError> {
let key =
env::var("CHATTERBOX_API_KEY").map_err(|_| ChatError::NotSet("CHATTERBOX_API_KEY"))?;
let url = env::var("CHATTERBOX_URL").map_err(|_| ChatError::NotSet("CHATTERBOX_URL"))?;
let model =
env::var("CHATTERBOX_MODEL").map_err(|_| ChatError::NotSet("CHATTERBOX_MODEL"))?;
Ok(Model { url, key, model })
}
}
/// Send the conversation and wait for the whole answer.
async fn ask(url: &str, key: &str, body: &Request) -> Result<String, ChatError> {
let answer: Response = reqwest::Client::new()
.post(url)
.bearer_auth(key)
.json(body)
.send()
.await?
.error_for_status()?
.json()
.await?;
match answer.choices.into_iter().next() {
Some(choice) => Ok(choice.message.content),
None => Err(ChatError::Empty),
}
}
impl Replier for Model {
fn name(&self) -> &str {
"real"
}
fn reply(&self, history: &[Message]) -> Pieces {
// Our names ("you", "bot") become the API's ("user", "assistant").
let mut messages = Vec::new();
for m in history {
let role = if m.author == "you" {
"user"
} else {
"assistant"
};
messages.push(ApiMessage {
role: String::from(role),
content: m.text.clone(),
});
}
let body = Request {
model: self.model.clone(),
messages,
};
// Owned copies: the task may outlive `self`, so it cannot borrow from it.
let url = self.url.clone();
let key = self.key.clone();
let (tx, rx) = mpsc::channel(1);
tokio::spawn(async move {
let answer = ask(&url, &key, &body).await;
// If nobody is listening any more, there is no one to tell.
let _ = tx.send(answer).await;
});
rx
}
}
#[cfg(test)]
mod tests {
use super::*;
fn said(text: &str) -> Vec<Message> {
vec![Message::new("you", text)]
}
/// Wait for every piece of an answer and join them.
async fn whole(mut pieces: Pieces) -> Result<String, ChatError> {
let mut text = String::new();
while let Some(piece) = pieces.recv().await {
text.push_str(&piece?);
}
Ok(text)
}
#[tokio::test]
async fn echo_repeats_you() {
let answer = whole(Echo { max_len: 200 }.reply(&said("hi"))).await;
assert_eq!(answer.unwrap(), "you said: hi");
}
#[tokio::test]
async fn echo_refuses_long_messages() {
let long = "x".repeat(10);
let answer = whole(Echo { max_len: 5 }.reply(&said(&long))).await;
assert!(answer.unwrap().starts_with("that is too long"));
}
#[tokio::test]
async fn pretend_reads_the_whole_history() {
let history = vec![
Message::new("you", "a"),
Message::new("bot", "b"),
Message::new("you", "c"),
];
assert_eq!(
whole(Pretend.reply(&history)).await.unwrap(),
"(pretend model) I read all 3 messages. You last said: c"
);
}
#[tokio::test]
async fn pretend_answers_a_word_at_a_time() {
let mut pieces = Pretend.reply(&said("one two"));
let mut count = 0;
while let Some(piece) = pieces.recv().await {
piece.unwrap();
count += 1;
}
assert!(count > 1, "the answer came in {count} piece(s)");
}
#[tokio::test]
async fn nothing_to_reply_to_is_an_error() {
let answer = whole(Echo { max_len: 200 }.reply(&[])).await;
assert!(matches!(answer, Err(ChatError::Empty)));
}
}
src/store.rsunchanged
//! Saving the conversation to a file and reading it back, as JSON.
use crate::message::Message;
use std::fs;
use std::io;
/// Where the history is kept, next to wherever you run Chatterbox.
pub const FILE: &str = "chatterbox.json";
/// The saved history, or an empty one the very first time (no file yet).
pub fn load(path: &str) -> Result<Vec<Message>, io::Error> {
match fs::read_to_string(path) {
// `?` turns a JSON error into an io::Error for us (serde_json says how).
Ok(text) => Ok(serde_json::from_str(&text)?),
Err(e) if e.kind() == io::ErrorKind::NotFound => Ok(Vec::new()),
Err(e) => Err(e),
}
}
/// Write the whole history to the file as pretty-printed JSON.
pub fn save(path: &str, history: &[Message]) -> Result<(), io::Error> {
let text = serde_json::to_string_pretty(history)?;
fs::write(path, text)
}
Cargo.toml3 new lines
[package]
name = "chatterbox"
version = "1.1.0"
edition = "2024"
[dependencies]
reqwest = { version = "0.13", features = ["json"] }
serde = { version = "1", features = ["derive"] }
serde_json = "1"
tokio = { version = "1", features = ["macros", "rt-multi-thread", "io-std", "io-util", "sync", "time"] }
Download this version (zip)Unzip, then cargo run inside the folder.
The real model still works, through the same stand-in server as lesson 9.2. Its answer arrives as one piece, because Chatterbox never asked for anything else: the request does not ask for streaming, and .json().await waits for the whole body before it hands anything back. Asking with "stream": true, and reading the answer as the model writes it, is the job of the next lesson.
You type
Sam hello model and again /quit
Chatterbox prints (a real run)
Chatterbox v1.1 (replies by: real) What should I call you? Hi Sam! I'm Chatterbox. Ask me anything. bot: mock model saw 1 message; last: hello model bot: mock model saw 3 messages; last: and again bot: bye, Sam! messages so far: 2 saved 4 messages to chatterbox.json
Try itYour turn: the pretend model types
This is the whole of Chatterbox v1.1 with the second half of Pretend's reply taken out: the sentence text is built, and nothing sends it. Make a channel, spawn a task that sends text one word at a time (the first word as it is, every later one with a space in front), pausing 40 milliseconds after each, and return the Receiver straight away. Joined back together, the pieces must be exactly text.
This is the whole of Chatterbox v1.1 with the second half of Pretend's reply taken out: the sentence text is built, and nothing sends it. Make a channel, spawn a task that sends text one word at a time (the first word as it is, every later one with a space in front), pausing 40 milliseconds after each, and return the Receiver straight away. Joined back together, the pieces must be exactly text.
// Chatterbox: a small chat program, built one lesson at a time.
mod command {
//! Commands: what a line of input means, and what can be wrong with it.
use std::fmt;
/// Everything a line of input can mean. One enum, so a `match` on it must
/// handle every kind of line: the compiler checks that nothing is forgotten.
#[derive(Debug)]
pub enum Command {
Empty,
Help,
Quit,
Again,
History,
Count,
Stats,
Remember(String, String),
Recall(Option<String>),
Find(String),
Last(usize),
Model(String),
Say(String),
Shout(String),
Debug(String),
Chat(String),
}
/// Everything that can be wrong with a line of input. parse() returns one of
/// these instead of a Command, and main prints it.
#[derive(Debug)]
pub enum CommandError {
Unknown(String),
Missing(String, String),
Usage(String),
NotACount(String),
}
/// How a CommandError reads when printed with {}.
impl fmt::Display for CommandError {
fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
match self {
CommandError::Unknown(word) => {
write!(f, "I don't know the command {word}. Try /help.")
}
CommandError::Missing(verb, example) => {
write!(f, "{verb} what? Try /{verb} {example}")
}
CommandError::Usage(form) => write!(f, "try {form}"),
CommandError::NotACount(text) => write!(f, "'{text}' is not a count. Try /last 3"),
}
}
}
/// The text after a command word, or the error that says it is missing.
fn need(line: &str, verb: &str, example: &str) -> Result<String, CommandError> {
match arg(line) {
Some(text) => Ok(String::from(text)),
None => Err(CommandError::Missing(
String::from(verb),
String::from(example),
)),
}
}
/// How many messages /last should show. "two" or "-1" is an error.
fn parse_count(text: &str) -> Result<usize, CommandError> {
match text.parse::<usize>() {
Ok(n) => Ok(n),
Err(_) => Err(CommandError::NotACount(String::from(text))),
}
}
/// Turn one line of input into a Command, or say what is wrong with it.
/// Each `?` means: if this step failed, return its error right here.
pub fn parse(line: &str) -> Result<Command, CommandError> {
let cmd = match first_word(line) {
"" => Command::Empty,
"/help" | "/h" => Command::Help,
"/quit" | "/q" | "/exit" => Command::Quit,
"/again" => Command::Again,
"/history" => Command::History,
"/count" => Command::Count,
"/stats" => Command::Stats,
"/remember" => {
let usage = CommandError::Usage(String::from("/remember <name> <fact>"));
let rest = match arg(line) {
Some(rest) => rest,
None => return Err(usage),
};
match arg(rest) {
Some(fact) => Command::Remember(String::from(first_word(rest)), String::from(fact)),
None => return Err(usage),
}
}
"/recall" => match arg(line) {
Some(name) => Command::Recall(Some(String::from(name))),
None => Command::Recall(None),
},
"/find" => Command::Find(need(line, "find", "hello")?),
"/last" => Command::Last(parse_count(&need(line, "last", "3")?)?),
"/model" => Command::Model(need(line, "model", "pretend")?),
"/debug" => Command::Debug(need(line, "debug", "/say hi")?),
"/say" => Command::Say(need(line, "say", "hello")?),
"/shout" => Command::Shout(need(line, "shout", "hello")?),
word => {
if is_command(word) {
return Err(CommandError::Unknown(String::from(word)));
}
Command::Chat(String::from(line))
}
};
Ok(cmd)
}
/// True when the text is a command: it starts with a slash.
fn is_command(text: &str) -> bool {
text.starts_with('/')
}
/// The first word of the text: a slice (&str) pointing into it, not a copy.
fn first_word(text: &str) -> &str {
match text.find(' ') {
Some(i) => &text[..i],
None => text,
}
}
/// The text after the command word, if there is any: Some("hello") for
/// "/say hello", None for a bare "/say". No more "" standing in for nothing.
fn arg(line: &str) -> Option<&str> {
match line.find(' ') {
Some(i) => {
let rest = line[i + 1..].trim();
if rest.is_empty() { None } else { Some(rest) }
}
None => None,
}
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn plain_text_is_chat() {
assert!(matches!(parse("hello there"), Ok(Command::Chat(t)) if t == "hello there"));
}
#[test]
fn help_has_a_short_form() {
assert!(matches!(parse("/help"), Ok(Command::Help)));
assert!(matches!(parse("/h"), Ok(Command::Help)));
}
#[test]
fn say_needs_text() {
assert_eq!(
parse("/say").unwrap_err().to_string(),
"say what? Try /say hello"
);
}
#[test]
fn last_needs_a_number() {
assert!(matches!(parse("/last 3"), Ok(Command::Last(3))));
assert_eq!(
parse("/last two").unwrap_err().to_string(),
"'two' is not a count. Try /last 3"
);
}
#[test]
fn unknown_commands_are_errors() {
assert_eq!(
parse("/dance").unwrap_err().to_string(),
"I don't know the command /dance. Try /help."
);
}
#[test]
fn remember_takes_a_name_and_a_fact() {
assert!(
matches!(parse("/remember pizza extra cheese"), Ok(Command::Remember(k, v)) if k == "pizza" && v == "extra cheese")
);
assert!(parse("/remember pizza").is_err());
}
}
}
mod message {
//! A Message: one line of the conversation.
use serde::{Deserialize, Serialize};
/// One line of the conversation: who said it, and what they said.
#[derive(Serialize, Deserialize)]
pub struct Message {
pub author: String,
pub text: String,
}
impl Message {
/// Make a Message from two pieces of borrowed text.
pub fn new(author: &str, text: &str) -> Message {
Message {
author: String::from(author),
text: String::from(text),
}
}
/// The message as one transcript line: "you: hello".
pub fn to_line(&self) -> String {
format!("{}: {}", self.author, self.text)
}
/// The text in capitals, for /shout.
pub fn shout(&self) -> String {
self.text.to_uppercase()
}
}
}
mod replier {
//! Who answers: anything that can reply, behind one trait.
use crate::message::Message;
use serde::{Deserialize, Serialize};
use std::env;
use std::fmt;
use std::time::Duration;
use tokio::sync::mpsc;
/// What can go wrong while getting a reply.
#[derive(Debug)]
pub enum ChatError {
/// The request did not get a good answer (no network, bad key, server error).
Http(reqwest::Error),
/// The model answered, but with nothing in it.
Empty,
/// /model real was asked for, but a setting is missing.
NotSet(&'static str),
}
impl fmt::Display for ChatError {
fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
match self {
ChatError::Http(e) => write!(f, "the model call failed: {e}"),
ChatError::Empty => write!(
f,
"there was nothing to reply with (an empty conversation or an empty answer)"
),
ChatError::NotSet(name) => write!(f, "set {name} first (see lesson 9.2)"),
}
}
}
/// Lets `?` turn a reqwest error into a ChatError on its own.
impl From<reqwest::Error> for ChatError {
fn from(e: reqwest::Error) -> Self {
ChatError::Http(e)
}
}
/// The receiving end of an answer: its pieces arrive in order, as they are
/// ready, and the channel closes when the answer is complete.
pub type Pieces = mpsc::Receiver<Result<String, ChatError>>;
/// A promise with a name: any type that implements Replier can answer a
/// conversation. `reply` returns at once; the answer arrives on the channel.
pub trait Replier {
/// A short name, for "/model" and the banner.
fn name(&self) -> &str;
/// Start answering the conversation so far (the last message is the newest).
fn reply(&self, history: &[Message]) -> Pieces;
}
/// A channel that already holds a whole answer, or an error, and is closed.
fn all_at_once(answer: Result<String, ChatError>) -> Pieces {
let (tx, rx) = mpsc::channel(1);
// A new channel with room for one cannot be full, so this send always works.
tx.try_send(answer).expect("a new channel has room for one");
rx
}
/// The echo Chatterbox has had since chapter 3. It answers all at once.
pub struct Echo {
pub max_len: usize,
}
impl Replier for Echo {
fn name(&self) -> &str {
"echo"
}
fn reply(&self, history: &[Message]) -> Pieces {
let message = match history.last() {
Some(m) => &m.text,
None => return all_at_once(Err(ChatError::Empty)),
};
let text = if message.len() > self.max_len {
format!("that is too long (the limit is {} bytes)", self.max_len)
} else if message.ends_with('?') {
String::from("Good question! I can only echo. Try /model real to ask a real model.")
} else {
format!("you said: {message}")
};
all_at_once(Ok(text))
}
}
/// A stand-in for a real model. It types its answer a word at a time, the way
/// a real model does, so you can watch an answer arrive in pieces.
pub struct Pretend;
impl Replier for Pretend {
fn name(&self) -> &str {
"pretend"
}
fn reply(&self, history: &[Message]) -> Pieces {
let last = match history.last() {
Some(m) => &m.text,
None => return all_at_once(Err(ChatError::Empty)),
};
let text = format!(
"(pretend model) I read all {} message{}. You last said: {last}",
history.len(),
if history.len() == 1 { "" } else { "s" },
);
todo!()
}
}
/// One message as a chat API expects it: a role and some content.
#[derive(Serialize)]
struct ApiMessage {
role: String,
content: String,
}
/// The body we send: which model, and the whole conversation.
#[derive(Serialize)]
struct Request {
model: String,
messages: Vec<ApiMessage>,
}
/// The parts of the answer we read. serde ignores every other field.
#[derive(Deserialize)]
struct Response {
choices: Vec<Choice>,
}
#[derive(Deserialize)]
struct Choice {
message: ApiReply,
}
#[derive(Deserialize)]
struct ApiReply {
content: String,
}
/// A real model behind an OpenAI-style chat endpoint, configured from
/// environment variables so the key never sits in the code.
pub struct Model {
url: String,
key: String,
model: String,
}
impl Model {
/// Read the three settings, or say which one is missing.
pub fn from_env() -> Result<Model, ChatError> {
let key =
env::var("CHATTERBOX_API_KEY").map_err(|_| ChatError::NotSet("CHATTERBOX_API_KEY"))?;
let url = env::var("CHATTERBOX_URL").map_err(|_| ChatError::NotSet("CHATTERBOX_URL"))?;
let model =
env::var("CHATTERBOX_MODEL").map_err(|_| ChatError::NotSet("CHATTERBOX_MODEL"))?;
Ok(Model { url, key, model })
}
}
/// Send the conversation and wait for the whole answer.
async fn ask(url: &str, key: &str, body: &Request) -> Result<String, ChatError> {
let answer: Response = reqwest::Client::new()
.post(url)
.bearer_auth(key)
.json(body)
.send()
.await?
.error_for_status()?
.json()
.await?;
match answer.choices.into_iter().next() {
Some(choice) => Ok(choice.message.content),
None => Err(ChatError::Empty),
}
}
impl Replier for Model {
fn name(&self) -> &str {
"real"
}
fn reply(&self, history: &[Message]) -> Pieces {
// Our names ("you", "bot") become the API's ("user", "assistant").
let mut messages = Vec::new();
for m in history {
let role = if m.author == "you" {
"user"
} else {
"assistant"
};
messages.push(ApiMessage {
role: String::from(role),
content: m.text.clone(),
});
}
let body = Request {
model: self.model.clone(),
messages,
};
// Owned copies: the task may outlive `self`, so it cannot borrow from it.
let url = self.url.clone();
let key = self.key.clone();
let (tx, rx) = mpsc::channel(1);
tokio::spawn(async move {
let answer = ask(&url, &key, &body).await;
// If nobody is listening any more, there is no one to tell.
let _ = tx.send(answer).await;
});
rx
}
}
#[cfg(test)]
mod tests {
use super::*;
fn said(text: &str) -> Vec<Message> {
vec![Message::new("you", text)]
}
/// Wait for every piece of an answer and join them.
async fn whole(mut pieces: Pieces) -> Result<String, ChatError> {
let mut text = String::new();
while let Some(piece) = pieces.recv().await {
text.push_str(&piece?);
}
Ok(text)
}
#[tokio::test]
async fn echo_repeats_you() {
let answer = whole(Echo { max_len: 200 }.reply(&said("hi"))).await;
assert_eq!(answer.unwrap(), "you said: hi");
}
#[tokio::test]
async fn echo_refuses_long_messages() {
let long = "x".repeat(10);
let answer = whole(Echo { max_len: 5 }.reply(&said(&long))).await;
assert!(answer.unwrap().starts_with("that is too long"));
}
#[tokio::test]
async fn pretend_reads_the_whole_history() {
let history = vec![
Message::new("you", "a"),
Message::new("bot", "b"),
Message::new("you", "c"),
];
assert_eq!(
whole(Pretend.reply(&history)).await.unwrap(),
"(pretend model) I read all 3 messages. You last said: c"
);
}
#[tokio::test]
async fn pretend_answers_a_word_at_a_time() {
let mut pieces = Pretend.reply(&said("one two"));
let mut count = 0;
while let Some(piece) = pieces.recv().await {
piece.unwrap();
count += 1;
}
assert!(count > 1, "the answer came in {count} piece(s)");
}
#[tokio::test]
async fn nothing_to_reply_to_is_an_error() {
let answer = whole(Echo { max_len: 200 }.reply(&[])).await;
assert!(matches!(answer, Err(ChatError::Empty)));
}
}
}
mod store {
//! Saving the conversation to a file and reading it back, as JSON.
use crate::message::Message;
use std::fs;
use std::io;
/// Where the history is kept, next to wherever you run Chatterbox.
pub const FILE: &str = "chatterbox.json";
/// The saved history, or an empty one the very first time (no file yet).
pub fn load(path: &str) -> Result<Vec<Message>, io::Error> {
match fs::read_to_string(path) {
// `?` turns a JSON error into an io::Error for us (serde_json says how).
Ok(text) => Ok(serde_json::from_str(&text)?),
Err(e) if e.kind() == io::ErrorKind::NotFound => Ok(Vec::new()),
Err(e) => Err(e),
}
}
/// Write the whole history to the file as pretty-printed JSON.
pub fn save(path: &str, history: &[Message]) -> Result<(), io::Error> {
let text = serde_json::to_string_pretty(history)?;
fs::write(path, text)
}
}
use command::{Command, parse};
use message::Message;
use replier::{Echo, Model, Pretend, Replier};
use std::collections::HashMap;
use std::io::{self, Write};
use tokio::io::{AsyncBufReadExt, BufReader};
/// What the bot is called. A constant: fixed for the whole program.
const BOT: &str = "Chatterbox";
/// The version as a pair of numbers: (major, minor).
const VERSION: (u32, u32) = (1, 1);
/// The longest message we accept, in bytes.
const MAX_LEN: usize = 200;
/// What /help prints.
const HELP: &str = "commands: /help, /again, /history, /count, /stats, /remember <name> <fact>, /recall <name>, /find <word>, /last <n>, /model <echo|pretend|real>, /say <text>, /shout <text>, /debug <text>, /quit (or /h, /q, /exit)";
/// The first line Chatterbox prints.
fn banner() -> String {
format!("{} v{}.{}", BOT, VERSION.0, VERSION.1)
}
/// The greeting, with a name filled in.
fn greeting(name: &str) -> String {
let mut text = format!("Hi {name}! I'm {BOT}.");
text.push_str(" Ask me anything.");
text
}
/// The last `n` items of any slice: messages, numbers, words, anything.
/// `<T>` says "for any type T"; the body never needs to know which one.
fn last_n<T>(items: &[T], n: usize) -> &[T] {
let start = items.len().saturating_sub(n);
&items[start..]
}
/// Print an answer as its pieces arrive, and give back the whole of it
/// (None if it went wrong: what went wrong has been printed instead).
async fn answer(replier: &dyn Replier, history: &[Message]) -> Result<Option<String>, io::Error> {
let mut pieces = replier.reply(history);
let mut whole = String::new();
print!("bot: ");
while let Some(piece) = pieces.recv().await {
match piece {
Ok(text) => {
print!("{text}");
io::stdout().flush()?; // show it now, not at the end of the line
whole.push_str(&text);
}
Err(e) => {
println!("(that went wrong: {e})");
return Ok(None);
}
}
}
println!();
Ok(Some(whole))
}
// #[tokio::main] starts tokio's runtime, the part that runs async code, and
// then runs this async main inside it.
#[tokio::main]
async fn main() -> Result<(), io::Error> {
let mut messages_seen: u32 = 0;
let quiet = false;
let mut last: Option<String> = None; // the last thing you said, for /again
let mut history: Vec<Message> = store::load(store::FILE)?; // last time's conversation, if any
let mut facts: HashMap<String, String> = HashMap::new(); // name -> fact, for /recall
// Box<dyn Replier>: "some Replier, decided while the program runs".
// With a key in the environment, start with the real model.
let mut replier: Box<dyn Replier> = match Model::from_env() {
Ok(model) => Box::new(model),
Err(_) => Box::new(Echo { max_len: MAX_LEN }),
};
println!("{} (replies by: {})", banner(), replier.name());
// The keyboard, read one line at a time without blocking the program.
let mut lines = BufReader::new(tokio::io::stdin()).lines();
println!("What should I call you?");
let name_line = lines.next_line().await?.unwrap_or_default();
let name = name_line.trim();
if !quiet {
println!("{}", greeting(name));
}
if !history.is_empty() {
println!(
"(I remember {} messages from last time. /history shows them.)",
history.len()
);
}
// next_line gives None when there is no more input: the keyboard was closed.
while let Some(line) = lines.next_line().await? {
let message = line.trim();
let cmd = match parse(message) {
Ok(cmd) => cmd,
Err(e) => {
println!("bot: {e}");
continue;
}
};
match cmd {
Command::Empty => {} // nothing typed: nothing to answer
Command::Quit => {
println!("bot: bye, {name}!");
break;
}
Command::Help => println!("bot: {HELP}"),
Command::Again => match &last {
Some(text) => {
answer(replier.as_ref(), &[Message::new("you", text)]).await?;
}
None => println!("bot: nothing to repeat yet"),
},
Command::History => {
if history.is_empty() {
println!("bot: nothing said yet");
} else {
for (i, m) in history.iter().enumerate() {
println!("{}. {}", i + 1, m.to_line());
}
}
}
Command::Count => {
let bytes: usize = history.iter().map(|m| m.text.len()).sum();
println!("bot: {} messages, {} bytes of text", history.len(), bytes);
}
Command::Stats => {
// One chain of steps: your messages -> their words -> a count per word.
let mut counts: HashMap<String, u32> = HashMap::new();
for word in history
.iter()
.filter(|m| m.author == "you")
.flat_map(|m| m.text.split_whitespace())
{
*counts.entry(word.to_lowercase()).or_insert(0) += 1;
}
let mut top: Vec<(String, u32)> = counts.into_iter().collect();
// most used first; ties in alphabetical order, so the answer never changes
top.sort_by(|a, b| b.1.cmp(&a.1).then(a.0.cmp(&b.0)));
let words: Vec<String> = top
.iter()
.take(3)
.map(|(word, n)| format!("{word} ({n})"))
.collect();
if words.is_empty() {
println!("bot: nothing said yet");
} else {
println!("bot: your top words: {}", words.join(", "));
}
}
Command::Remember(key, fact) => {
println!("bot: got it: {key} is {fact}");
facts.insert(key, fact);
}
Command::Recall(Some(key)) => match facts.get(&key) {
Some(fact) => println!("bot: {key} is {fact}"),
None => println!("bot: I don't know anything about {key}"),
},
Command::Recall(None) => {
if facts.is_empty() {
println!("bot: I don't remember anything yet. Try /remember");
} else {
// A HashMap has no order, so sort the names before printing them.
let mut names = Vec::new();
for name in facts.keys() {
names.push(name.as_str());
}
names.sort();
println!("bot: I remember: {}", names.join(", "));
}
}
Command::Find(word) => {
// The closure borrows `word` from here: a small function made on the spot.
let hits: Vec<&Message> = history
.iter()
.filter(|m| m.author == "you" && m.text.contains(&word))
.collect();
if hits.is_empty() {
println!("bot: you never said {word}");
} else {
for m in hits {
println!("{}", m.to_line());
}
}
}
Command::Last(n) => {
for m in last_n(&history, n) {
println!("{}", m.to_line());
}
}
Command::Model(choice) => {
let next: Result<Box<dyn Replier>, String> = match choice.as_str() {
"echo" => Ok(Box::new(Echo { max_len: MAX_LEN })),
"pretend" => Ok(Box::new(Pretend)),
"real" => match Model::from_env() {
Ok(model) => Ok(Box::new(model)),
Err(e) => Err(e.to_string()),
},
_ => Err(format!(
"I don't know a model called {choice}. Try echo, pretend or real"
)),
};
match next {
Ok(r) => {
replier = r;
println!("bot: replies now come from {}", replier.name());
}
Err(why) => println!("bot: {why}"),
}
}
Command::Say(text) => println!("bot: {text}"),
Command::Shout(text) => println!("bot: {}", Message::new("you", &text).shout()),
Command::Debug(text) => println!("bot: {:?}", parse(&text)),
Command::Chat(text) => {
messages_seen += 1;
history.push(Message::new("you", &text));
// the history holds its own copy, so `text` can move into `last`
last = Some(text);
if let Some(reply) = answer(replier.as_ref(), &history).await? {
history.push(Message::new("bot", &reply));
}
}
}
}
println!("messages so far: {messages_seen}");
store::save(store::FILE, &history)?;
let s = if history.len() == 1 { "" } else { "s" };
println!("saved {} message{s} to {}", history.len(), store::FILE);
Ok(())
}
// Tests: `cargo test` builds and runs them; the program itself never sees them.
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn last_n_takes_from_the_end() {
assert_eq!(last_n(&[1, 2, 3, 4], 2), &[3, 4]);
assert_eq!(last_n(&["a", "b"], 5), &["a", "b"]);
assert!(last_n(&[1, 2, 3], 0).is_empty());
}
}
Hidden tests (4): an_empty_history_is_still_an_error pretend_reads_the_whole_history pretend_answers_a_word_at_a_time a_long_answer_fits_through_a_small_pipe
This exercise runs real Rust in your browser through the official Rust Playground, so it needs JavaScript.
If you have coded before
In Python's asyncio, tokio::spawn is close to asyncio.create_task, and an mpsc channel is close to an asyncio.Queue with a maxsize. The difference is the ending: a Queue has no idea when its writers are done, so you send a sentinel such as None yourself. A tokio channel closes when the last Sender is dropped, so ownership tells the reader that the answer is over.
What you have now
Chatterbox v1.1 prints each piece the moment it arrives, whichever replier it is. Ownership did two quiet jobs: move let the task outlive reply, and dropping the last Sender says an answer is finished.
Self-check
Say it out loud, then reveal
Self-checkRetrieval beats re-reading
What are the two ends of <code>mpsc::channel(8)</code>, and what does the 8 mean?Reveal
A Sender (
tx), which puts items in with send, and a Receiver (rx), which takes them out with recv. The 8 is how many items may wait in the pipe before send has to pause.Why does <code>tokio::spawn</code> need <code>async move</code> here?Reveal
The task can keep running after the function that started it has returned and dropped its variables. So the task must own what it uses, not borrow it; without
move the compiler stops with E0373.How does the reader of a channel know the answer is complete?Reveal
recv().await gives None once the pipe is empty and every Sender has been dropped. The task dropping its Sender when it finishes is what ends the answer.Why does <code>answer</code> call <code>io::stdout().flush()</code> after <code>print!</code>?Reveal
Text without a new line waits in a buffer. Flushing hands it to the terminal now, so each piece appears as it arrives instead of all at the end.
That is lesson 10.2. Tick it off so the course remembers where you are.
Go deeper
Questions people ask
What is the difference between a tokio task and a thread?
Both are work that runs alongside the rest of the program. A thread is given to you by the operating system and is fairly heavy. A tokio task is much lighter: the tokio runtime runs many tasks on a few threads, switching between them each time one is waiting at an await.
What does mpsc stand for?
Multiple producer, single consumer. A channel can have many Senders, made by cloning the first one, but only one Receiver.
Why does reply return a channel instead of being an async fn?
A trait with an async fn cannot be used as Box dyn Replier, which Chatterbox needs to swap repliers while it runs. A plain function that returns a Receiver at once can, and the answer still arrives later, down the channel.
What happens if I forget to flush after print?
Nothing goes wrong, but the pieces sit in the output buffer until a new line is printed, so the whole answer appears at once at the end and the streaming effect is lost.