Create a Tokio MPSC Channel for Concurrent Scraper Results

Mục tiêu: Refactor a concurrent Rust web scraper by introducing a Tokio MPSC (multi-producer, single-consumer) channel. This task involves creating the channel to replace the previous JoinHandle collection logic, setting the stage for processing results as they become available.

You’ve successfully built a scraper that can launch multiple tasks and gather their results. The method you used—collecting JoinHandles and awaiting them in a second loop—is a perfectly valid pattern often called “structured concurrency.” It works well, but it has one primary characteristic: you must wait for all the tasks to be launched before you can start processing any of the results.

What if you wanted to start processing data the moment the first scraper finishes, without waiting for the slowest one? This is where a more flexible and powerful communication primitive comes into play: channels.

A Better Way to Communicate: Introducing Channels

In concurrent programming, a channel is a conduit used to send data from one task to another. Think of it like a conveyor belt: one or more “producer” tasks can place items onto the belt, and a single “consumer” task can pick items off the end. This pattern elegantly decouples the producers from the consumer. The scrapers (producers) don’t need to know anything about the main function (consumer), other than how to send their finished product down the channel.

The tokio crate provides its own asynchronous, multi-producer, single-consumer (MPSC) channels, which are perfect for our use case. * Multi-Producer: We can give a “sender” end of the channel to each of our spawned scraping tasks. * Single-Consumer: Our main function will hold the one and only “receiver” end, collecting all the results in one central place. * Asynchronous: Sending or receiving on these channels are async operations. If a producer tries to send data into a full channel, it will pause gracefully (.await) without blocking the thread, waiting for the consumer to make space.

Our first task is to create this channel in our main function. We will use the tokio::sync::mpsc::channel() function.

This function takes one argument: the channel’s buffer capacity. This is the number of messages that can be held in the channel waiting for the receiver to pick them up. It acts as a buffer against mismatched speeds between producers and consumers. If the buffer fills up, any producer trying to send another message will be paused until there’s room. This mechanism, known as backpressure, is crucial for preventing a swarm of fast producers from overwhelming a slower consumer and exhausting system memory. We’ll start with a reasonable capacity of 100.

The channel() function returns a tuple containing two values: the sender and the receiver, often abbreviated as tx (for transmit) and rx (for receive).

Let’s modify src/main.rs to create this channel. We will also remove the old logic for collecting and processing JoinHandles, as the channel will completely replace that mechanism.

// In src/main.rs

// Add the new `use` statement for mpsc channels
use tokio::sync::mpsc;
// ... (Product struct and scrape_page function remain unchanged) ...

#[tokio::main]
async fn main() {
    let urls_to_scrape = vec![
        "http://books.toscrape.com/catalogue/category/books/travel_2/index.html",
        "http://books.toscrape.com/catalogue/category/books/mystery_3/index.html",
        "http://books.toscrape.com/catalogue/category/books/historical-fiction_4/index.html",
    ];

    println!("Starting the scraper with MPSC channel...");

    // highlight-start
    // Create a multi-producer, single-consumer channel.
    // The channel will transport our `Vec<Product>` results. Rust's type inference
    // will figure out the exact type later when we first use the channel.
    // `100` is the buffer capacity of the channel.
    let (tx, mut rx) = mpsc::channel(100);
    // We declare `rx` as mutable because receiving from it will change its state.
    // highlight-end

    let mut tasks = vec![];

    for url in urls_to_scrape {
        let url_owned = url.to_string();

        // This is the old spawn logic. We will modify it in the next steps
        // to use the channel's sender (`tx`).
        let task = tokio::spawn(async move {
            println!("Spawning task for URL: {}", url_owned);
            scrape_page(&url_owned).await
        });
        tasks.push(task);
    }

    // The old logic for awaiting JoinHandles is now removed.
    // We will replace it with a loop that receives from `rx`.

    println!("\n--- All scraping tasks launched! ---\n");
}

Code Breakdown

use tokio::sync::mpsc;: We bring the mpsc module into scope so we can use mpsc::channel().
let (tx, mut rx) = mpsc::channel(100);: This single line is the core of our task.
- It calls the channel function, establishing our communication conduit with a buffer size of 100.
- It uses destructuring to unpack the returned tuple into two new variables: tx (the sender) and rx (the receiver).
- We mark rx as mut because the recv() method we will use later modifies the receiver’s internal state as it consumes messages.

Right now, if you compile this code using cargo check, the compiler will correctly warn you that tx and rx are unused variables. This is expected! We have successfully built the communication channel; in the next tasks, we will plumb it into our concurrent tasks and our main result-processing loop.

What’s Next?

You have the “sender” half of the channel, tx. However, you have multiple producer tasks, and each one will need its own handle to the sending side. In the next task, you will learn how to clone the sender (tx) inside your for loop, preparing a unique copy to be moved into each spawned task.

Clone a Tokio MPSC Sender for Concurrent Tasks

Mục tiêu: Implement the multi-producer pattern by cloning the sender (tx) of a Tokio MPSC channel inside a loop, allowing multiple spawned tasks to send data back to a single receiver while respecting Rust’s ownership rules.

Excellent! You’ve just set up the communication backbone for our scraper by creating a Tokio MPSC channel. You now have the two ends of this powerful conduit: tx, the sender, and rx, the receiver. This channel will allow our concurrent tasks to send their results back to the main function in a safe, efficient, and decoupled manner.

However, we face a classic challenge in Rust’s concurrent programming landscape, a challenge rooted in its greatest strength: the ownership system. We have multiple scraper tasks to spawn, but we only have one sender, tx. If we were to give this sender to the first task we spawn, it would be gone, “moved” and consumed, leaving us with nothing to give to the second, third, and subsequent tasks.

The Multi-Producer Solution: Cloning the Sender

This is precisely what the “Multi-Producer” in MPSC is designed to solve. The mpsc::Sender type is not a single-use object; it’s a clonable “handle” to the sending side of the channel.

When you call .clone() on a Sender, you are not creating a new, separate channel. Instead, you are creating a new, lightweight pointer or reference to the same underlying channel. Each cloned sender is a distinct value in terms of ownership, but they all feed into the single, shared channel that our one receiver (rx) is listening to.

Think of it like this: you have a single mailbox (rx). To allow multiple people to send you letters, you don’t give them the mailbox itself. You give each person a copy of your mailing address (tx.clone()). All those addresses lead to the same mailbox.

Our strategy will be to perform this clone operation inside our for loop. For each URL, we will create a fresh clone of the sender. This new, cloned sender is what we will then give to the new task we spawn for that URL.

Let’s update our main function to implement this cloning step.

// In src/main.rs

use tokio::sync::mpsc;
use scraper::{Html, Selector};

// ... (Product struct remains unchanged) ...

// ... (scrape_page function remains unchanged) ...

#[tokio::main]
async fn main() {
    let urls_to_scrape = vec![
        "http://books.toscrape.com/catalogue/category/books/travel_2/index.html",
        "http://books.toscrape.com/catalogue/category/books/mystery_3/index.html",
        "http://books.toscrape.com/catalogue/category/books/historical-fiction_4/index.html",
    ];

    println!("Starting the scraper with MPSC channel...");

    let (tx, mut rx) = mpsc::channel(100);

    let mut tasks = vec![];

    for url in urls_to_scrape {
        // highlight-start
        // For each URL, we clone the sender part of the channel.
        // This is crucial because each spawned task needs its own sender handle
        // to send results back. The `move` keyword in `tokio::spawn` will take
        // ownership of this clone. The original `tx` remains in the `main` function,
        // ready to be cloned again for the next iteration of the loop.
        let tx_clone = tx.clone();
        // highlight-end

        let url_owned = url.to_string();

        let task = tokio::spawn(async move {
            println!("Spawning task for URL: {}", url_owned);
            // In the next task, we will modify this block to use `tx_clone`.
            scrape_page(&url_owned).await
        });
        tasks.push(task);
    }

    println!("\n--- All scraping tasks launched! ---\n");
}

By adding let tx_clone = tx.clone();, you have successfully prepared the necessary tool for each of your concurrent workers. Each iteration of the loop now produces a unique, ownable tx_clone that points to our central channel. You have satisfied the ownership checker while enabling a true multi-producer setup.

What’s Next?

You now have a unique tx_clone ready for each task. The next logical and final step in wiring up our producers is to move this cloned sender into the spawned task. You will modify the async move block to take ownership of tx_clone and, after the scraping is complete, use it to send the resulting Vec<Product> down the channel to the waiting receiver.

Transfer Ownership to Tokio Tasks with `async move`

Mục tiêu: Modify the tokio::spawn call to use an async move block. This transfers ownership of the cloned channel sender and the URL into the new asynchronous task, allowing it to send the scraping result back through the MPSC channel.

In the previous task, you expertly set up the multi-producer side of our communication channel by cloning the sender (tx) inside the for loop. You created a tx_clone for each URL, preparing a unique “mailing address” for each of our concurrent workers.

Now, it’s time to hand that address to the worker and give it its instructions. This task is all about connecting the cloned sender to the spawned task, a process enabled by one of Rust’s most powerful and safety-critical features: the move keyword.

From Borrowing to Owning: The `async move` Block

When we use tokio::spawn, we are sending a self-contained unit of work to the Tokio runtime. The runtime makes no guarantees about when or for how long this task will run. It could finish in milliseconds, or it could run for seconds. It could even outlive the for loop that created it.

Because of this uncertainty, Rust’s borrow checker enforces a strict but vital rule: the spawned task must be 'static. This means it cannot contain any borrowed references to data that might disappear while the task is still running. The task must own all the data it needs to complete its job.

This is where async move becomes our most important tool. By default, closures in Rust try to borrow the variables they use from their surrounding environment. The move keyword fundamentally changes this behavior. It forces the closure (our async block) to take ownership of any captured variables.

In our case, the async move block will now do two things:

It will take ownership of url_owned, the String containing the URL for that specific task.
It will take ownership of tx_clone, the cloned channel sender for that specific task.

This transfer of ownership is the “move” we are performing. It bundles the instructions (scrape_page), the target (url_owned), and the communication channel (tx_clone) into a single, independent package that can be safely executed by the runtime.

Modifying the Task to Use Its New Tools

To demonstrate this move in a practical way, we’ll also perform the next logical action: modifying the task’s logic. Instead of returning the result of scrape_page, the task will now use its owned sender (tx_clone) to send the result back through the channel. This clearly shows the tx_clone has been moved and is now being used by its new owner.

We will send the entire Result from scrape_page down the channel. This is a robust pattern that allows the receiver in our main function to handle both successful scrapes and specific scraping errors from a single source.

Let’s update the tokio::spawn call in your main function.

// In src/main.rs

use tokio::sync::mpsc;
use scraper::{Html, Selector};

// ... (Product struct and scrape_page function remain unchanged) ...

#[tokio::main]
async fn main() {
    let urls_to_scrape = vec![
        "http://books.toscrape.com/catalogue/category/books/travel_2/index.html",
        "http://books.toscrape.com/catalogue/category/books/mystery_3/index.html",
        "http://books.toscrape.com/catalogue/category/books/historical-fiction_4/index.html",
    ];

    println!("Starting the scraper with MPSC channel...");

    // The channel will now transport the entire `Result` of our scraping operation.
    // Rust's type inference can often figure this out, but being explicit can be clearer.
    let (tx, mut rx) = mpsc::channel::<Result<Vec<Product>, Box<dyn std::error::Error>>>(100);

    // We no longer need to collect JoinHandles for their return values,
    // but we might keep them to ensure tasks complete, or to check for panics.
    // For now, let's simplify and remove the `tasks` vector.
    // let mut tasks = vec![]; // This is no longer needed for our channel-based approach.

    for url in urls_to_scrape {
        let tx_clone = tx.clone();
        let url_owned = url.to_string();

        // highlight-start
        // Spawn a new asynchronous task.
        tokio::spawn(async move {
            // The `move` keyword transfers ownership of `tx_clone` and `url_owned`
            // from the `for` loop's scope into this new task.

            // Perform the scraping operation.
            let scrape_result = scrape_page(&url_owned).await;

            // After scraping, send the result (whether Ok or Err) down the channel.
            // The `.send()` method is asynchronous. It will wait if the channel
            // buffer is full (this is called "backpressure").
            // We use `if let Err` to handle the case where the channel might have
            // been closed by the receiver, which would make sending impossible.
            if let Err(e) = tx_clone.send(scrape_result).await {
                eprintln!("Failed to send result for url {}: {}", url_owned, e);
            }
        });
        // We no longer push the handle, as we will receive results from the channel.
        // tasks.push(task);
        // highlight-end
    }

    println!("\n--- All scraping tasks launched! ---\n");
}

Deconstructing the New `async move` Block

Let’s break down the updated tokio::spawn block:

tokio::spawn(async move { ... }): This is the core. The move keyword is the instruction that tells Rust to transfer ownership of tx_clone and url_owned into this task. The task is now completely self-sufficient.
let scrape_result = scrape_page(&url_owned).await;: The task performs its primary function just as before, producing a Result<Vec<Product>, ...>.
tx_clone.send(scrape_result).await: This is the communication step. The task uses its owned sender to send the scrape_result back to the main function.
- The .send() method takes ownership of the data being sent.
- It is an async operation. If the channel’s buffer is full, it will pause the task gracefully until there is space, preventing memory overflows.
- The .send() call itself returns a Result. It will return an Err if the receiver (rx) has been dropped, which means no one is listening on the other end of the channel. Our if let Err(e) handles this unlikely but possible scenario by printing an error.
No Return Value: Notice that the async block no longer returns a value. Its job is not to return data to a JoinHandle, but to send data through the channel. This simplifies our logic significantly. As a result, we’ve also removed the tasks vector, as we no longer need to collect JoinHandles to get their return values.

You have now successfully wired up the “producer” side of your channel. Each spawned task is a true, independent worker that knows its job and how to report back.

What’s Next?

The producers are ready to send data, but the main function isn’t yet listening. Also, there’s a subtle but critical detail we must address. If we start listening now, our receiving loop will never end, because the original tx in the main function is still alive, keeping the channel open. In the next task, you will explicitly drop the original tx after the loop. This is the signal to the receiver that no more producers will ever be created, and the channel can be closed once all the clones are gone.

Implement Asynchronous Sending in Tokio Tasks

Mục tiêu: Refactor the spawned asynchronous tasks to send their results back to the main function using a cloned MPSC channel sender. This involves calling the .send().await method on the sender within each task, transitioning from a result-pulling model to a result-pushing model.

Building on our last step, we’ve successfully equipped each spawned task with a unique, cloned sender (tx_clone). This sender is the task’s personal communication line back to our main function. The ownership of this sender has been safely transferred into the async move block, making each task a completely self-contained unit of work.

Now, we must instruct the task on how to use this new tool. The entire purpose of this refactoring is to shift from a “pull” model (where main awaits JoinHandles to pull results) to a “push” model, where the tasks proactively push their results back as soon as they’re ready. This is accomplished using the sender’s .send() method.

From Returning to Sending: A Paradigm Shift

Previously, our async block ended with the line scrape_page(&url_owned).await. This made the Result from scrape_page the return value of the task, which we would later retrieve from its JoinHandle.

We will now change this logic. The task will no longer return a value. Instead, it will:

Execute scrape_page and await its result.
Take that result (whether it was a success Ok(Vec<Product>) or a failure Err(...)).
Use its owned sender, tx_clone, to send this result down the channel.

This fundamentally decouples the task’s completion from the collection of its result. The task’s only job is to do the work and report back via the channel.

The Anatomy of an Asynchronous Send

The line tx_clone.send(scrape_result).await is packed with important concurrent programming concepts:

tx_clone.send(scrape_result): This method takes the data you want to send (scrape_result) and places it into the channel’s buffer. A key detail is that .send() takes ownership of the data. Once you’ve sent the scrape_result, the task can no longer use it. This is a powerful part of Rust’s safety guarantee, preventing you from accidentally modifying data that’s already “in transit.”
.await: Why is sending an asynchronous operation? The answer is backpressure. Recall that we created our channel with a buffer capacity of 100. If our scrapers are very fast and the main function is slow to process the results, this buffer can fill up. If a task tries to send a result into a full buffer, .await will gracefully pause that specific task. It will sit dormant, consuming no CPU, until the main function receives a message and makes space in the buffer. The Tokio runtime will then automatically wake the task up to complete its send. This is a crucial mechanism for building stable, resilient systems that don’t crash from out-of-control memory usage.
Error Handling the Send: The .send() method itself returns a Result. It will return an Err if, and only if, the receiving end of the channel (rx) has been dropped. This signifies that there’s no one left to listen for messages, so the send operation fails. While this is unlikely in our current code, it’s a best practice to handle this possibility. The if let Err(...) pattern is a clean and idiomatic way to check for this failure and log an error if it occurs.

Let’s modify the tokio::spawn block in main to implement this sending logic. We will also remove the tasks vector, as we no longer need to collect JoinHandles to get their return values.

// In src/main.rs

use tokio::sync::mpsc;
use scraper::{Html, Selector};

// ... (Product struct and scrape_page function remain unchanged) ...

#[tokio::main]
async fn main() {
    let urls_to_scrape = vec![
        "http://books.toscrape.com/catalogue/category/books/travel_2/index.html",
        "http://books.toscrape.com/catalogue/category/books/mystery_3/index.html",
        "http://books.toscrape.com/catalogue/category/books/historical-fiction_4/index.html",
    ];

    println!("Starting the scraper with MPSC channel...");

    // The channel will transport the entire `Result` of our scraping operation.
    let (tx, mut rx) = mpsc::channel::<Result<Vec<Product>, Box<dyn std::error::Error>>>(100);

    // We've removed the `tasks` vector as it's no longer needed for collecting results.

    for url in urls_to_scrape {
        let tx_clone = tx.clone();
        let url_owned = url.to_string();

        // highlight-start
        tokio::spawn(async move {
            // 1. Perform the scraping. The result is either Ok(data) or Err(error).
            let scrape_result = scrape_page(&url_owned).await;

            // 2. Use the owned sender to send the result back through the channel.
            //    This is an async operation that will wait if the channel buffer is full.
            if let Err(e) = tx_clone.send(scrape_result).await {
                // This error occurs if the receiver has been dropped.
                eprintln!("Failed to send result for url {}: {}", url_owned, e);
            }
        });
        // highlight-end
    }

    println!("\n--- All scraping tasks launched! ---\n");
}

You have now successfully implemented the “producer” half of the channel pattern. Each of your concurrent scrapers knows its job, has its target URL, and now has a direct line to send its findings back to headquarters the moment its work is done.

What’s Next?

Our producers are diligently sending results, but our main function isn’t yet listening. There’s also a subtle but critical detail remaining. The original sender, tx, is still alive in the main function. This keeps the channel open indefinitely from the receiver’s perspective. In the next task, we will explicitly drop the original tx after our spawning loop. This is the crucial signal that no new senders will be created, allowing the receiver to know when all the work is truly done.

Preventing Deadlock by Dropping the MPSC Sender

Mục tiêu: Solve a common concurrency deadlock by explicitly dropping the original MPSC channel sender (tx) after spawning all producer tasks. This signals to the receiver that no more messages will be sent, allowing the channel to close gracefully once all tasks are complete.

You have brilliantly wired up all your producers. Each concurrent scraping task is now a self-sufficient worker, equipped with its target URL and a direct communication line—the cloned channel sender—to report its findings back to main. The for loop that launched these tasks has finished, and your army of scrapers is now working in parallel.

Before we can start listening for their reports, we must address a subtle but critical piece of channel management. It’s the final signal that separates the “setup” phase from the “listening” phase.

The Last Sender and the Open Channel Problem

Think about the lifecycle of our channel. How does the receiver (rx) know when all the work is done and it can stop listening? The rule is simple: a channel is considered “closed” only when every single Sender handle associated with it has been dropped.

Let’s take inventory of our Senders:

For each URL, we created a tx_clone which was moved into a spawned task. These Senders will be dropped automatically when their respective tasks finish.
The original tx variable still exists and is owned by the main function.

Herein lies the problem. Even after all three of our scraping tasks have completed their work and dropped their cloned senders, the original tx in main is still alive. From the receiver’s perspective, the channel is still open because there is still one valid Sender in existence. If we were to start listening for messages now, our receiver loop would successfully receive the three results and then… it would wait. Forever. It would be stuck waiting for another message from the main function’s tx, a message that will never be sent. This is a classic concurrency bug known as a deadlock.

The Solution: Explicitly Dropping the Original Sender

To solve this, we must explicitly signal to the channel that no new producers will be created. We do this by dropping the original tx in the main function. The perfect place to do this is immediately after the for loop that was responsible for creating all the producers.

Once we drop the original tx, the fate of the channel is now tied exclusively to the lifecycle of our spawned tasks. The channel will now close automatically at the exact moment the very last scraping task finishes its work and its tx_clone is dropped. This is the precise behavior we want.

We use the standard library function std::mem::drop(). This function takes ownership of a value and immediately runs its “destructor,” causing it to go out of scope.

Let’s add this crucial line to your main function in src/main.rs.

// In src/main.rs

use tokio::sync::mpsc;
use scraper::{Html, Selector};

// ... (Product struct and scrape_page function remain unchanged) ...

#[tokio::main]
async fn main() {
    let urls_to_scrape = vec![
        "http://books.toscrape.com/catalogue/category/books/travel_2/index.html",
        "http://books.toscrape.com/catalogue/category/books/mystery_3/index.html",
        "http://books.toscrape.com/catalogue/category/books/historical-fiction_4/index.html",
    ];

    println!("Starting the scraper with MPSC channel...");

    let (tx, mut rx) = mpsc::channel::<Result<Vec<Product>, Box<dyn std::error::Error>>>(100);

    for url in urls_to_scrape {
        let tx_clone = tx.clone(); // We need the original `tx` here to clone it.
        let url_owned = url.to_string();

        tokio::spawn(async move {
            let scrape_result = scrape_page(&url_owned).await;
            if let Err(e) = tx_clone.send(scrape_result).await {
                eprintln!("Failed to send result for url {}: {}", url_owned, e);
            }
        });
    }

    // highlight-start
    // After the loop, we explicitly drop the original sender `tx`.
    // This is the critical signal to the receiver `rx`. It tells the receiver
    // that no more new senders will be created. The channel will close
    // once all the cloned senders (held by the spawned tasks) are dropped.
    drop(tx);
    // highlight-end

    println!("\n--- All scraping tasks launched! Now waiting for results... ---\n");
}

By adding this single, elegant line, drop(tx);, you have made your channel logic complete and robust. You have now perfectly orchestrated the entire lifecycle of the producer side of the channel, ensuring a clean and predictable shutdown.

What’s Next?

The stage is now perfectly set. The producers are working, and you have guaranteed that the channel will close when they are finished. In the next task, you will finally implement the consumer side of the pattern. You will write a loop in main that listens on the receiver (rx), gracefully handling incoming results until the channel closes, at which point the loop will terminate automatically.

Implement the MPSC Channel Consumer Loop

Mục tiêu: In the main function, create a while let loop to asynchronously receive scraping results from the MPSC channel’s receiver. The loop should process each result as it arrives and terminate gracefully when all sender tasks are complete and the channel is closed.

You have perfectly orchestrated the producer side of our concurrent system. By cloning the sender for each task and, most importantly, dropping the original sender after the loop, you’ve created a robust system where an army of scrapers can work independently and signal when the entire operation is complete. The stage is now perfectly set for the star of the show: the consumer.

Our main function will now take on its final role in this step: to be the single, attentive consumer of all the results our scrapers produce. It will listen patiently, processing results as they arrive, and it will know exactly when to stop.

The Art of Listening: Receiving from a Channel

To receive a message, we use the recv() method on our Receiver handle, rx. This method is the counterpart to the send() method we used in the tasks.

rx.recv() is asynchronous: Just like sending can pause if the channel is full, receiving will pause if the channel is empty. When you call rx.recv().await, if there are no messages waiting in the buffer, your main function will yield control back to the Tokio runtime. It will sleep efficiently, consuming no CPU, until a scraper task sends a new message. The runtime will then wake main back up to process it.
rx.recv() returns an Option<T>: This is the most elegant part of the channel’s design. The return value is not the message itself, but an Option wrapping the message. This Option has two states that perfectly map to the state of our work:
1. Some(message): A message was successfully received from the channel. The message is the Result<Vec<Product>, ...> that our scraper task sent.
2. None: This is the crucial signal. The receiver gets None if and only if the channel has been closed. The channel closes when all Sender handles (our original tx and all the tx_clones) have been dropped. Because we so carefully drop(tx) after the loop, this None value is the definitive signal that all scraping tasks have finished their work.

The Perfect Tool for the Job: The `while let` Loop

How do we create a loop that continues as long as it receives Some(message) and stops automatically when it receives None? Rust has a beautiful and idiomatic control flow pattern for exactly this scenario: the while let loop.

The structure while let Some(variable) = expression { ... } does the following:

It evaluates the expression (in our case, rx.recv().await).
It checks if the result matches the pattern Some(variable).
If it matches, it binds the inner value to variable and executes the loop body.
If it does not match (i.e., it gets None), the loop terminates immediately.

This is far cleaner and safer than a manual loop with a match and a break statement. It perfectly expresses our intent: “process messages until there are no more.”

Let’s implement this consumer loop in main. For now, we will simply process each incoming message by printing a confirmation, setting the stage for aggregating the data in the final task of this step.

// In src/main.rs

use tokio::sync::mpsc;
use scraper::{Html, Selector};

// ... (Product struct and scrape_page function remain unchanged) ...

#[tokio::main]
async fn main() {
    let urls_to_scrape = vec![
        "http://books.toscrape.com/catalogue/category/books/travel_2/index.html",
        "http://books.toscrape.com/catalogue/category/books/mystery_3/index.html",
        "http://books.toscrape.com/catalogue/category/books/historical-fiction_4/index.html",
    ];

    println!("Starting the scraper with MPSC channel...");

    let (tx, mut rx) = mpsc::channel::<Result<Vec<Product>, Box<dyn std::error::Error>>>(100);

    for url in urls_to_scrape {
        let tx_clone = tx.clone();
        let url_owned = url.to_string();

        tokio::spawn(async move {
            let scrape_result = scrape_page(&url_owned).await;
            if let Err(e) = tx_clone.send(scrape_result).await {
                eprintln!("Failed to send result for url {}: {}", url_owned, e);
            }
        });
    }

    drop(tx);

    println!("\n--- All scraping tasks launched! Now waiting for results... ---\n");

// highlight-start
    // This is our consumer loop.
    // `while let Some(result)` will loop as long as `rx.recv()` returns `Some` message.
    // When the channel closes (because all senders are dropped), `rx.recv()` will
    // return `None`, and the loop will terminate gracefully.
    while let Some(scrape_result) = rx.recv().await {
        // Inside the loop, `scrape_result` is the `Result<Vec<Product>, ...>`
        // that was sent by one of the scraper tasks.
        match scrape_result {
            Ok(products) => {
                // The scrape was successful. For now, just confirm receipt.
                println!("Received a batch of {} products.", products.len());
            }
            Err(e) => {
                // A specific scraper task encountered an error.
                eprintln!("A scraper task reported an error: {}", e);
            }
        }
    }

    println!("\n--- All results received. The channel is closed. ---");
// highlight-end
}

Run your program now with cargo run. You will see the “tasks launched” message, followed by the “Received a batch…” messages as they come in from the concurrent scrapers. Finally, once all tasks are done, the loop will end, and you’ll see the “channel is closed” message. You have successfully built a real-time data processing pipeline!

What’s Next?

You are now successfully receiving the batches of scraped products as they are completed. This is a huge step! However, right now, you are just printing a confirmation message and then discarding the data. In the final task of this step, you will take these individual Vec<Product> batches and collect them into a single, master Vec<Product>, preparing the complete dataset to be written to a file.

Rust: Aggregate MPSC Channel Data with `Vec::extend`

Mục tiêu: Collect multiple batches of scraped data received from an MPSC channel into a single master vector using the efficient Vec::extend method to complete the data aggregation pipeline.

You have reached the final and most gratifying task of this step. In the previous task, you masterfully constructed the consumer loop using while let. Your main function now acts as a central processing unit, receiving batches of scraped data in real-time as your concurrent workers complete their jobs. You are successfully printing a confirmation for each batch, but the valuable products vector is then discarded at the end of each loop iteration.

Now, we will complete the pipeline. We will take these individual streams of data and merge them into a single, comprehensive dataset. This task is about aggregation: collecting all the Vec<Product> batches into one final, flattened Vec<Product>.

From Batches to a Master List

The goal is to have a single, unified vector at the end of our program that contains every product scraped from every successfully processed page. The most straightforward way to achieve this is to:

Create a new, empty, and mutable vector before our receiving loop begins. This will be our master list.
Inside the loop, for each successful batch of products we receive, we will add all of its contents to our master list.

The Best Tool for Aggregation: `Vec::extend()`

You might be tempted to loop through the incoming products vector and .push() each product one by one into the master list. While this would work, Rust provides a more efficient, expressive, and idiomatic tool for this exact job: the .extend() method.

When you call master_list.extend(incoming_list), you are telling Rust to append all the elements from incoming_list onto the end of master_list. This is often more efficient than a manual loop because the vector can perform a single, larger memory allocation if needed, rather than potentially reallocating multiple times inside a loop.

Furthermore, extend is powerful because it works with any iterator. Under the hood, when you pass a Vec<T> to extend, it is converted into an iterator that moves each element out of the original vector. This means the data is transferred with maximum efficiency, without unnecessary copying. After the extend operation, the incoming_list will be empty, as ownership of all its elements has been passed to master_list.

Let’s integrate this aggregation logic into our consumer loop. We will create the all_products vector and then use .extend() to populate it. Finally, after the loop finishes, we’ll print a summary of our complete dataset.

Here are the final changes for src/main.rs:

// In src/main.rs

use tokio::sync::mpsc;
use scraper::{Html, Selector};

// ... (Product struct and scrape_page function remain unchanged) ...

#[tokio::main]
async fn main() {
    let urls_to_scrape = vec![
        "http://books.toscrape.com/catalogue/category/books/travel_2/index.html",
        "http://books.toscrape.com/catalogue/category/books/mystery_3/index.html",
        "http://books.toscrape.com/catalogue/category/books/historical-fiction_4/index.html",
    ];

    println!("Starting the scraper with MPSC channel...");

    let (tx, mut rx) = mpsc::channel::<Result<Vec<Product>, Box<dyn std::error::Error>>>(100);

    for url in urls_to_scrape {
        let tx_clone = tx.clone();
        let url_owned = url.to_string();

        tokio::spawn(async move {
            let scrape_result = scrape_page(&url_owned).await;
            if let Err(e) = tx_clone.send(scrape_result).await {
                eprintln!("Failed to send result for url {}: {}", url_owned, e);
            }
        });
    }

    drop(tx);

    println!("\n--- All scraping tasks launched! Now waiting for results... ---\n");

    // highlight-start
    // Create a master vector to hold all products from all scraped pages.
    let mut all_products: Vec<Product> = Vec::new();
    // highlight-end

    while let Some(scrape_result) = rx.recv().await {
        match scrape_result {
            // highlight-start
            Ok(products) => {
                println!("Received a batch of {} products.", products.len());
                // Extend our master list with the products from this batch.
                // `.extend()` is an efficient method that moves all elements from the
                // `products` vector into `all_products`.
                all_products.extend(products);
            }
            // highlight-end
            Err(e) => {
                eprintln!("A scraper task reported an error: {}", e);
            }
        }
    }

    // highlight-start
    println!("\n--- All results received. The channel is closed. ---");
    println!("Total products scraped: {}", all_products.len());
    println!("--- Sample of all scraped products: ---");

    // Print the first 5 products for a clean preview of our final, aggregated data.
    for product in all_products.iter().take(5) {
        println!("{:#?}", product);
    }
    // highlight-end
}

Congratulations! You have now completed a major architectural refactoring of your application. You have moved from a simple “spawn-and-wait” model to a sophisticated, real-time data pipeline using MPSC channels. Your main function now successfully launches a fleet of scrapers, receives their results concurrently, and aggregates them into a final, complete dataset.

Run your program with cargo run. You will see the real-time “Received a batch…” messages, followed by a final summary showing the total number of products collected from all pages, and a sample of the final data.

What’s Next?

You now have all the scraped data neatly collected in memory in the all_products vector. The next logical and final step in our core project is to persist this data. In the next step, you will use the csv crate to take this vector of Product structs and write it cleanly into a products.csv file, producing a tangible, usable output from your scraper.