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
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211

/*
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 tokio::sync::{broadcast, mpsc::channel};
use tokio_util::sync::CancellationToken;

use crate::{handle, Error, Task, BUFFER_CAPACITY};

// here I use a mod just to keep clean and ordered the file :)
mod modes {
    use tokio::sync::{broadcast, mpsc::Sender};

    /// This is the Worker’s mode that lets build a worker that is not able to
    /// communicate with the controlled task.
    pub struct Isolated {}

    /// Worker's mode that has a communication channel between the worker and its
    /// task. In this channel the messages, with type Message, can be send from the
    /// worker to the task using the function [`super::Worker<Mode>::post_message()`].
    pub struct OneWay<Message> {
        // used to send messages toward Task
        pub(super) sender_to_tsk: Sender<Message>,
    }

    /// This mode is used when a bidirectional channel is needed between worker
    /// and its task. These messages can have diffent types.
    pub struct TwoWay<Message, TaskMessage> {
        // used to send messages toward Task
        pub(super) sender_to_tsk: Sender<Message>,
        // used by interested tasks to subscribe to messages sent by this worker
        // controlled task.
        pub(super) broadcast_from_tsk: broadcast::Sender<TaskMessage>,
    }
}

pub use modes::*;

// // // // // // // // // // // // // // // // // // // // // // // // // // //

/// As the Web Workers API, [`Worker`] makes it possible to spawn a new
/// [`Task`] and, depending of used `Mode`, having some functions that lets
/// its user to interact with the spawned task.
///
/// The type parameter `Mode` is used to assign the wanted features to the [`Worker`].
/// Not always we want a worker that can interact with the task that it spawned,
/// excluding the possibility to terminate its activity using the function [`terminate`]
/// that is always available.
///
/// A brief description of available modes can be found in [`crate::worker`].
///
/// A [`Worker`] can be generated only using the function [`spawn`].
///
/// [`Worker`]: Worker<Mode>
/// [`terminate`]: Worker<Mode>::terminate
/// [`spawn`]: Worker<Mode>::spawn

pub struct Worker<Mode> {
    // used to terminate Task
    termination_token: CancellationToken,
    // mode is used to differenziate the Worker's behaviour.
    mode: Mode,
}

impl<Mode> Worker<Mode> {
    /// Terminates this worker and the related task.
    pub fn terminate(self) {
        self.termination_token.cancel();
    }
}

// // // // // // // // // // // // // // // // // // // // // // // // // // //

impl Worker<Isolated> {
    /// Creates an isolated worker that can only terminate the spawned task.
    pub fn spawn<T>(task: T) -> Worker<Isolated>
    where
        T: Task<Handle = handle::Worker<handle::Isolated>>,
        <T as Task>::Output: Send + 'static,
    {
        // This token is used to terminate the worker and its controlled task.
        let token = CancellationToken::new();

        // Worker's handle that will be used by the Task to communicate with
        // this worker and to terminate both.
        let wkh = handle::Worker::isolated(token.clone());

        // The Task is spawned here
        tokio::spawn(task.spawn(wkh));

        Worker {
            termination_token: token,
            mode: Isolated {},
        }
    }
}

// // // // // // // // // // // // // // // // // // // // // // // // // // //

impl<Message> Worker<OneWay<Message>> {
    /// Creates a worker that is able to send messages to its controlled `task`.
    pub fn spawn<T>(task: T) -> Worker<OneWay<Message>>
    where
        T: Task<Handle = handle::Worker<handle::OneWay<Message>>>,
        <T as Task>::Output: Send + 'static,
    {
        // This token is used to terminate the worker and its controlled task.
        let token = CancellationToken::new();

        // the channel used by Worker to communicate with its Task.
        let (send_to_task, recv_from_wk) = channel::<Message>(BUFFER_CAPACITY);

        // Worker's handle that will be used by the Task to communicate with
        // this worker and to terminate both.
        let wkh = handle::Worker::one_way(token.clone(), recv_from_wk);

        // The Task is spawned here
        tokio::spawn(task.spawn(wkh));

        Worker {
            termination_token: token,
            mode: OneWay {
                sender_to_tsk: send_to_task,
            },
        }
    }

    /// Send message `msg` to the spawned task.
    pub async fn post_message(&self, msg: Message) -> Result<(), Error> {
        self.mode
            .sender_to_tsk
            .send(msg)
            .await
            .map_err(|e| Error::from(&e))
    }
}

// // // // // // // // // // // // // // // // // // // // // // // // // // //

impl<Message, TaskMessage: Clone> Worker<TwoWay<Message, TaskMessage>> {
    /// Creates a worker that is able to communicate in a bidirectional way with
    /// the `task` that is spowned. The back channel is a broadcast one so many
    /// subscriber tasks will be able to subscribe, with the function [`Self::on_message()`],
    /// to the events sent by this worker's controlled task.
    pub fn spawn<T>(task: T) -> Worker<TwoWay<Message, TaskMessage>>
    where
        T: Task<Handle = handle::Worker<handle::TwoWay<Message, TaskMessage>>>,
        <T as Task>::Output: Send + 'static,
    {
        // This token is used to terminate the worker and its controlled task.
        let token = CancellationToken::new();

        // the channel used by Worker to communicate with its Task.
        let (send_to_task, recv_from_wk) = channel::<Message>(BUFFER_CAPACITY);

        // the broadcast channel used by the Task to communicate with this Worker.
        let (broadcast_to_wk, _) = broadcast::channel::<TaskMessage>(BUFFER_CAPACITY);

        // Worker's handle that will be used by the Task to communicate with
        // this worker and to terminate both.
        let wkh = handle::Worker::two_way(token.clone(), recv_from_wk, broadcast_to_wk.to_owned());

        // The Task is spawned here
        tokio::spawn(task.spawn(wkh));

        Worker {
            termination_token: token,
            mode: TwoWay {
                sender_to_tsk: send_to_task,
                broadcast_from_tsk: broadcast_to_wk,
            },
        }
    }

    /// Send message `msg` to the spawned task.
    pub async fn post_message(&self, msg: Message) -> Result<(), Error> {
        self.mode
            .sender_to_tsk
            .send(msg)
            .await
            .map_err(|e| Error::from(&e))
    }

    /// Let `task` to subscribe to event messages that will be sent by this
    /// two-way worker's task. Every subscription will receive independently
    /// the sent events. The OnEvent handle is able to `terminate` itself and
    /// the subscriber task, but not the two-way worker or task.
    pub fn on_message<T>(&self, task: T) -> Worker<Isolated>
    where
        T: Task<Handle = handle::Worker<handle::OnEvent<TaskMessage>>>,
        <T as Task>::Output: Send + 'static,
    {
        // This token is used to terminate the worker and its controlled task.
        let token = CancellationToken::new();

        // OnEvent worker's handle that will be used by the Task to receive
        // events sent by this two-way task.
        let wkh = handle::Worker::on_event(token.clone(), self.mode.broadcast_from_tsk.subscribe());

        // The Task is spawned here
        tokio::spawn(task.spawn(wkh));

        Worker {
            termination_token: token,
            mode: Isolated {},
        }
    }
}