lambda-channel 0.1.3

A crossbeam-like channel concurrently executes lambda functions.
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
59
60
61
62
63

use std::thread;
use std::time::Duration;

use crossbeam_channel::{unbounded, Receiver, Sender};

fn spawn_thread<T: Send + 'static, U: Send + 'static, V: Clone + Send + 'static>(
    input_channel: Receiver<T>,
    output_channel: Sender<U>,
    shared_resource: V,
    function: fn(&V, T) -> U,
) {
    thread::spawn(move || loop {
        output_channel
            .send(function(&shared_resource, input_channel.recv().unwrap()))
            .unwrap();
    });
}

pub fn new_simplified_lamda_channel<
    T: Send + 'static,
    U: Send + 'static,
    V: Clone + Send + 'static,
>(
    threads: usize,
    shared_resource: V,
    function: fn(&V, T) -> U,
) -> (Sender<T>, Receiver<U>) {
    let (out_tx, rx) = unbounded();
    let (tx, in_rx) = unbounded();

    for _ in 0..threads - 1 {
        spawn_thread(
            in_rx.clone(),
            out_tx.clone(),
            shared_resource.clone(),
            function,
        );
    }

    spawn_thread(in_rx, out_tx, shared_resource, function);

    (tx, rx)
}

fn fast_lambda(_: &Option<()>, x: u32) -> u32 {
    let y = 2 * x;
    println!("Hello World! 2 * {} = {}", x, y);
    y
}

fn main() {
    let (tx, rx) = new_simplified_lamda_channel(2, None, fast_lambda);

    for i in 0..5 {
        println!("Sending {}", i);
        tx.send(i).unwrap();
        thread::sleep(Duration::from_millis(1));
    }

    for _ in 0..5 {
        println!("Receiving {}", rx.recv().unwrap());
    }
}