carpool 0.1.0

Deduplicate and batch concurrent async requests through a collection window
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58

// Walk the queue through its observable behavior:
// filling, a leaving caller, both slicing paths, delivery.

use std::time::Duration;

use carpool::queue::{Pending, Queue};
use tokio::sync::oneshot::{self, Receiver};
use tokio::time::timeout;

fn deliver(batch: Vec<Pending<&str, String>>) {
    for (input, tx) in batch {
        let _ = tx.send(input.to_uppercase());
    }
}

#[tokio::main(flavor = "current_thread")]
async fn main() {
    let queue: Queue<&str, String> = Queue::default();

    // Filling: three callers enqueue and keep their receivers.
    let mut receivers: Vec<Receiver<String>> = Vec::new();
    for input in ["alpha", "beta", "gamma"] {
        let (tx, rx) = oneshot::channel();
        queue.push(input, tx);
        receivers.push(rx);
    }
    println!("pushed 3 requests: len = {}", queue.len());

    // One caller leaves; the raw len does not notice.
    drop(receivers.remove(1));
    println!("\"beta\" left: len = {}", queue.len());

    // Threshold slice: the scan buries the dead entry, and two live are not three.
    match queue.take_if(3) {
        Some(batch) => deliver(batch),
        None => println!("take_if(3) refused: len = {}", queue.len()),
    }

    // Two live entries meet a threshold of two; delivery stays with us.
    let batch = queue.take_if(2).expect("two live entries");
    println!("take_if(2) sliced a batch of {}", batch.len());
    deliver(batch);
    for rx in receivers.drain(..) {
        println!("caller got {:?}", rx.await);
    }

    // Timer slice: the threshold is out of reach, the deadline fires instead.
    let (tx, rx) = oneshot::channel();
    queue.push("delta", tx);
    if timeout(Duration::from_millis(50), queue.reached(3))
        .await
        .is_err()
    {
        println!("reached(3) timed out with len = {}", queue.len());
    }
    deliver(queue.take(usize::MAX));
    println!("timer batch delivered: {:?}", rx.await);
}