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

use crate::*;
use futures::Future;
use std::{sync::Arc};

/// [spawn] with [Config::default].
pub fn spawn_default<T, R, Fun, Fut>(fun: Fun) -> (Child<R>, Address<T>)
where
    Fun: FnOnce(Inbox<T>) -> Fut + Send + 'static,
    Fut: Future<Output = R> + Send + 'static,
    R: Send + 'static,
    T: Send + 'static,
{
    spawn(Config::default(), fun)
}

/// [spawn_pooled] with [Config::default].
pub fn spawn_default_pooled<T, R, I, Fun, Fut>(
    iter: impl IntoIterator<Item = I>,
    fun: Fun,
) -> (ChildPool<R>, Address<T>)
where
    Fun: FnOnce(I, Inbox<T>) -> Fut + Send + 'static + Clone,
    Fut: Future<Output = R> + Send + 'static,
    R: Send + 'static,
    T: Send + 'static,
    I: Send + 'static,
{
    spawn_pooled(iter, Config::default(), fun)
}

/// Spawn a new `Actor` with a single `Process`. This will return a [Child] and
/// and [Address]. The `Process` is spawned with a single [Inbox].
/// 
/// # Example
/// ```no_run
///# use tiny_actor::*;
///# #[tokio::main]
///# async fn main() {
/// let (child, address) = 
///     spawn(Config::default(), |mut inbox: Inbox<u32>| async move {
///         loop {
///             let msg = inbox.recv().await;
///             println!("Received message: {msg:?}");
///         }
///     });
///# }
/// ```
pub fn spawn<T, R, Fun, Fut>(config: Config, fun: Fun) -> (Child<R>, Address<T>)
where
    Fun: FnOnce(Inbox<T>) -> Fut + Send + 'static,
    Fut: Future<Output = R> + Send + 'static,
    R: Send + 'static,
    T: Send + 'static,
{
    let (inbox, address, channel) = setup_channel(config.capacity);

    let handle = tokio::task::spawn(async move { fun(inbox).await });

    let child = Child::new(channel, handle, config.link);

    (child, address)
}

/// Spawn a new `Actor` with a multiple `Process`es. This will return a [ChildPool] and
/// and [Address]. The `Process`es are spawned with [Inbox]es.
/// 
/// The amount of `Process`es that are spawned is equal to the length of the iterator.
/// Every process get's access to a single item within the iterator as it's first argument.
/// 
/// # Example
/// ```no_run
///# use tiny_actor::*;
///# #[tokio::main]
///# async fn main() {
/// let (child, address) = 
///     spawn_pooled(0..5, Config::default(), |i, mut inbox: Inbox<u32>| async move {
///         loop {
///             let msg = inbox.recv().await;
///             println!("Received message on actor {i}: {msg:?}");
///         }
///     });
///# }
/// ```
pub fn spawn_pooled<T, R, I, Fun, Fut>(
    iter: impl IntoIterator<Item = I>,
    config: Config,
    fun: Fun,
) -> (ChildPool<R>, Address<T>)
where
    Fun: FnOnce(I, Inbox<T>) -> Fut + Send + 'static + Clone,
    Fut: Future<Output = R> + Send + 'static,
    R: Send + 'static,
    T: Send + 'static,
    I: Send + 'static,
{
    let iterator = iter.into_iter();
    let mut handles = Vec::with_capacity(iterator.size_hint().0);

    let (inbox, address, channel) = setup_channel(config.capacity);

    for i in iterator {
        let fun = fun.clone();
        let inbox = inbox.clone_inbox();
        let handle = tokio::task::spawn(async move { fun(i, inbox).await });
        handles.push(handle);
    }

    let children = ChildPool::new(channel, handles, config.link);

    (children, address)
}

fn setup_channel<T>(capacity: Capacity) -> (Inbox<T>, Address<T>, Arc<Channel<T>>) {
    let channel = Arc::new(Channel::new(1, 1, capacity));
    let inbox = Inbox::from_channel(channel.clone());
    let address = Address::from_channel(channel.clone());
    (inbox, address, channel)
}