use std::collections::HashSet;
use std::hash::Hash;
pub struct DedupByKey<I, K, F> {
inner: I,
key_fn: F,
seen: HashSet<K>,
}
impl<I, K, F> Iterator for DedupByKey<I, K, F>
where
I: Iterator,
K: Eq + Hash,
F: FnMut(&I::Item) -> K,
{
type Item = I::Item;
fn next(&mut self) -> Option<Self::Item> {
loop {
let item = self.inner.next()?;
let key = (self.key_fn)(&item);
if self.seen.insert(key) {
return Some(item);
}
}
}
}
pub trait DedupByKeyExt: Iterator + Sized {
fn dedup_by_key<K, F>(self, key_fn: F) -> DedupByKey<Self, K, F>
where
K: Eq + Hash,
F: FnMut(&Self::Item) -> K,
{
DedupByKey {
inner: self,
key_fn,
seen: HashSet::new(),
}
}
}
impl<I: Iterator> DedupByKeyExt for I {}
use std::collections::HashSet;
use std::hash::Hash;
use std::pin::Pin;
use std::task::{Context, Poll};
use futures::stream::Stream;
pub struct DedupStream<S, K, F> {
inner: Pin<Box<S>>,
key_fn: F,
seen: HashSet<K>,
}
impl<S, K, F> DedupStream<S, K, F> {
pub fn new(inner: S, key_fn: F) -> Self {
DedupStream {
inner: Box::pin(inner),
key_fn,
seen: HashSet::new(),
}
}
}
impl<S, K, F> Stream for DedupStream<S, K, F>
where
S: Stream,
K: Eq + Hash + Unpin,
F: FnMut(&S::Item) -> K + Unpin,
{
type Item = S::Item;
fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<S::Item>> {
let this = self.get_mut();
loop {
match this.inner.as_mut().poll_next(cx) {
Poll::Ready(Some(item)) => {
let key = (this.key_fn)(&item);
if this.seen.insert(key) {
return Poll::Ready(Some(item));
}
}
Poll::Ready(None) => return Poll::Ready(None),
Poll::Pending => return Poll::Pending,
}
}
}
}
mod dedup_by_key;
mod dedup_stream;
use dedup_by_key::DedupByKeyExt;
use dedup_stream::DedupStream;
use futures::stream::{self, StreamExt};
#[derive(Clone, Debug)]
struct Event {
user_id: u32,
action: &'static str,
}
#[tokio::main]
async fn main() {
let events = vec![
Event { user_id: 1, action: "login" },
Event { user_id: 2, action: "login" },
Event { user_id: 1, action: "click" },
Event { user_id: 3, action: "login" },
Event { user_id: 2, action: "logout" },
];
let unique: Vec<_> = events
.iter()
.cloned()
.dedup_by_key(|e| e.user_id)
.collect();
println!("sync first-wins: {:?}", unique);
let deduped = DedupStream::new(stream::iter(events), |e: &Event| e.user_id);
let collected: Vec<_> = deduped.collect().await;
println!("async first-wins: {:?}", collected);
}
This snippet shows two complementary ways to deduplicate a stream of records by a key while keeping the original arrival order: a synchronous Iterator adapter and an asynchronous Stream adapter built on futures. The core idea in both is a first-wins policy — the first record seen for a given key passes through, and every later record with the same key is dropped. Order is preserved because filtering happens inline as items flow past, never by sorting or buffering the whole sequence.
In dedup_by_key.rs, DedupByKey wraps any inner Iterator and owns a HashSet<K> of keys already emitted. The key_fn closure extracts a hashable, ownable key from each borrowed item. On every next(), it pulls from the inner iterator and calls seen.insert(key); because HashSet::insert returns true only when the value was newly added, that boolean doubles as the keep/skip decision. The surrounding loop skips duplicates without recursion, so a long run of repeats costs constant stack space. The DedupByKeyExt trait then bolts a dedup_by_key method onto every iterator, matching the ergonomics of the standard .filter() or .map() combinators.
The async side in dedup_stream.rs mirrors that design against futures::Stream. DedupStream is pinned and delegates to the inner stream via poll_next, using Poll::Ready(Some(item)) to apply the same seen.insert guard and Poll::Pending to propagate backpressure untouched. Wrapping the inner stream in Pin<Box<S>> keeps the adapter usable with unnamed async stream types. The #[pin_project] alternative is avoided here to keep the pinning explicit and dependency-light.
main.rs exercises both paths: a plain vector run and a tokio-driven async run over stream::iter, printing that only the first event per user survives. The trade-off is memory — the HashSet grows with the number of distinct keys, which is unbounded for an infinite stream, so this fits bounded key spaces or windowed use. For last-wins semantics a reversed pass or a HashMap staging buffer would be needed instead, sacrificing streaming. This pattern is the right reach whenever duplicates must be collapsed cheaply without reordering, such as event ingestion or log compaction.
Related snips
export interface RetryOptions {
retries: number;
baseMs: number;
maxMs: number;
signal?: AbortSignal;
onRetry?: (attempt: number, delay: number, err: unknown) => void;Exponential backoff with jitter for retries
struct Config<'a> {
name: &'a str,
value: &'a str,
}
fn parse_config(line: &str) -> Config {Lifetime annotations for flexible borrowing in structs
use crossbeam::channel::unbounded;
use std::thread;
fn main() {
let (tx, rx) = unbounded();Crossbeam for advanced concurrent data structures
// Creating a Promise
const myPromise = new Promise((resolve, reject) => {
const success = true;
setTimeout(() => {
if (success) {Promises and async/await patterns for asynchronous JavaScript
use tracing::{info, instrument};
#[instrument]
fn process_request(user_id: u64) {
info!(user_id, "Processing request");
// Work happens heretracing for structured logging and distributed tracing
use std::sync::mpsc;
use std::thread;
fn main() {
let (tx, rx) = mpsc::channel();Channels (mpsc) for message passing between threads
Share this code
Here's the card — post it anywhere.
