Dedup piled-up mpsc channel messages

Question

When using a channel, let's say at a given moment we have the following messages in the queue for a given receiver:

abc
blah
abc
something
something
blah
something

Usually messages are processed like this:

let (sender, receiver) = mpsc::channel::<Msg>();
for msg in receiver {
   // Process a message...
}

What is the easiest way to de-duplicate messages that have already piled up?

Using the above example, I'd like to only process 3 (not 7) messages:

abc
blah
something

Note: After a message abc has been processed, it's fine to receive (and process) another message abc again in the future. The goal is to only de-duplicate messages that have already piled-up in the queue.

Answer 1

Solution 1

Here is an example solution:

loop {
    let deduped_msgs : HashSet<Msg> = HashSet::from_iter(receiver.try_iter());
    for msg in deduped_msgs {
        // Process a message...                
    }
    thread::sleep(time::Duration::from_millis(10));
}

The above uses the non-blocking try_iter method. I've added a short sleep() to avoid excessive pressure on the CPU.

Solution 2

Here is a similar solution, as per Chayim's suggestions, that avoids the need for periodic checks every X milliseconds:

let mut deduped_msgs = HashSet::new();
loop {
    deduped_msgs.insert(receiver.recv().unwrap());
    deduped_msgs.extend(receiver.try_iter());               
    for msg in deduped_msgs.drain() {
        // Process a message... 
    }
}

Order of processing

If the order in which messages are processed matters, HashSet (alone) might not be the best option.

• To process messages in the order they arrived (only taking into account the first occurrence of each message), one could use IndexSet for example.
• In some cases the natural order of messages might make more sense and then BTreeSet could be the simplest option.
• To apply some sort of a custom priority ordering, one could, for example, utilize a vector and sort it via sort_unstable_by or sort_unstable_by_key (probably simpler to first dedup them with a HashSet, but another option might be to dedup an already sorted vector).