opifex 0.3.0

Opifex is a Latin word meaning artisan or manufacturer and referring to a worker who created something.
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
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127

/*
SPDX-License-Identifier: GPL-3.0-only

Copyright (C) 2024  Attilio Donà attilio.dona@gmail.com
Copyright (C) 2024  Claudio Carraro carraro.claudio@gmail.com
*/

use std::time::Duration;
use tokio::time::sleep;

use opifex::{
    handle,
    worker::{TwoWay, Worker},
    Task,
};

#[derive(Clone, Debug)]
pub struct Sum {
    a: i32,
    b: i32,
}

#[derive(Clone, Debug)]
pub struct Result {
    sum: i32,
}

impl From<Sum> for Result {
    fn from(value: Sum) -> Self {
        let Sum { a, b } = value;
        Result {
            sum: a.saturating_add(b),
        }
    }
}

pub struct Adder {}

impl Task for Adder {
    type Handle = handle::Worker<handle::TwoWay<Sum, Result>>;
    type Output = usize;

    fn spawn(
        &self,
        wk_hnd: Self::Handle,
    ) -> impl std::future::Future<Output = Self::Output> + Send + 'static {
        let (mut rx, hnd) = wk_hnd.receiver();

        async move {
            let mut count: usize = 0;

            loop {
                tokio::select! {
                    Some(sum) = rx.recv() => {
                        count += 1;
                        if let Err(e) = hnd.post_message(Result::from(sum)).await {
                            println!("Oops! Sending message reports: {e}");
                        }
                    }
                    () = hnd.terminated() => {
                        println!("Worker is terminated. Bye from adder task!");
                        break;
                    }
                }
            }

            count
        }
    }
}

pub struct Response;

impl Task for Response {
    type Handle = handle::Worker<handle::OnEvent<Result>>;
    type Output = usize;

    fn spawn(
        &self,
        wk_hnd: Self::Handle,
    ) -> impl std::future::Future<Output = Self::Output> + Send + 'static {
        let (mut rx, hnd) = wk_hnd.receiver();

        async move {
            let mut count: usize = 0;

            loop {
                tokio::select! {
                    Ok(res) = rx.recv() => {
                        count += 1;
                        let Result { sum } = res;
                        println!("{count} Result is {sum:?}");
                    }
                    () = hnd.terminated() => {
                        println!("Worker is terminated. Bye from result task!");
                        break;
                    }
                }
            }

            count
        }
    }
}

#[tokio::main]
async fn main() {
    let adder_worker = Worker::<TwoWay<Sum, Result>>::spawn(Adder {});
    let response_worker = adder_worker.on_message(Response {});

    if let Err(e) = adder_worker.post_message(Sum { a: 24, b: 28 }).await {
        eprintln!("Oops! sending a message to adder reports: {e}");
    }

    if let Err(e) = adder_worker.post_message(Sum { a: 123, b: 45 }).await {
        eprintln!("Oops! sending a message to adder reports: {e}");
    }

    // put to sleep this thread just to let worker's tasks to do the work!
    sleep(Duration::from_secs(2)).await;

    response_worker.terminate();
    adder_worker.terminate();

    // waiting to let terminations happens...
    sleep(Duration::from_secs(1)).await;
}