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
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
use crate::{
    spawn::{spawn_task, task_name},
    Bus, Lifeline,
};
use log::{debug, error};
use std::future::Future;
use std::{any::TypeId, fmt::Debug};

/// Takes channels from the [Bus](./trait.Bus.html), and spawns a tree of tasks.  Returns one or more [Lifeline](./struct.Lifeline.html) values.  
/// When the [Lifeline](./struct.Lifeline.html) is dropped, the task tree is immediately cancelled.
///
/// - Simple implementations can return the [Lifeline](./struct.Lifeline.html) value, a handle returned by [Task::task](./trait.Task.html#method.task).
/// - Implementations which have fallible spawns can return `anyhow::Result<Lifeline>`.
/// - Implementations which spawn multiple tasks can store lifelines for each task in self, and return `anyhow::Result<Self>`.
///
/// ## Example
/// ```
/// use lifeline::prelude::*;
/// use tokio::sync::mpsc;
///
/// lifeline_bus!(pub struct ExampleBus);
///
/// #[derive(Debug, Clone)]
/// struct ExampleMessage {}
///
/// impl Message<ExampleBus> for ExampleMessage {
///     type Channel = mpsc::Sender<Self>;
/// }    
///
/// struct ExampleService {
///     _run: Lifeline   
/// }
///
/// impl Service for ExampleService {
///     type Bus = ExampleBus;
///     type Lifeline = anyhow::Result<Self>;
///
///     fn spawn(bus: &ExampleBus) -> anyhow::Result<Self> {
///         let mut rx = bus.rx::<ExampleMessage>()?;
///
///         let _run = Self::task("run", async move {
///             while let Some(msg) = rx.recv().await {
///                 log::info!("got message: {:?}", msg);
///             }
///         });
///
///         Ok(Self { _run })
///     }
/// }
///
/// async fn run() {
///     let bus = ExampleBus::default();
///     let _service = ExampleService::spawn(&bus);
/// }
/// ```
pub trait Service: Task {
    /// The bus, which must be provided to spawn the task
    type Bus: Bus;

    /// The service lifeline.  When dropped, all spawned tasks are immediately cancelled.
    type Lifeline;

    /// Spawns the service with all sub-tasks, and returns a lifeline value.  When the lifeline is dropped, all spawned tasks are immediately cancelled.
    ///
    /// Implementations should synchronously take channels from the bus, and then use them asynchronously.  This makes errors occur as early and predictably as possible.
    fn spawn(bus: &Self::Bus) -> Self::Lifeline;
}

/// Constructs the bus, spawns the service, and returns both.
pub trait DefaultService: Service {
    fn spawn_default() -> (Self::Bus, Self::Lifeline);
}

impl<T> DefaultService for T
where
    T: Service,
{
    fn spawn_default() -> (Self::Bus, Self::Lifeline) {
        let bus = Self::Bus::default();
        let lifeline = Self::spawn(&bus);

        (bus, lifeline)
    }
}

/// Carries messages between **two** bus instances. A variant of the [Service](./trait.Service.html).
///
/// Bus types form a tree, with a 'root application' bus, and multiple busses focused on particular domains.  This structure provides isolation,
/// and predictable failures when [Services](./trait.Service.html) spawn.
/// ```text
/// - MainBus
///   | ListenerBus
///   |  | ConnectionBus
///   | DomainSpecificBus
///   |  | ...
/// ```
///
/// This trait can be implemented to carry messages between the root and the leaf of the tree.
///
/// ## Example
/// ```
/// use lifeline::prelude::*;
/// use tokio::sync::mpsc;
/// lifeline_bus!(pub struct MainBus);
/// lifeline_bus!(pub struct LeafBus);
///
/// #[derive(Debug, Clone)]
/// struct LeafShutdown {}
///
/// #[derive(Debug, Clone)]
/// struct MainShutdown {}
///
/// impl Message<LeafBus> for LeafShutdown {
///     type Channel = mpsc::Sender<Self>;   
/// }
///
/// impl Message<MainBus> for MainShutdown {
///     type Channel = mpsc::Sender<Self>;   
/// }
///
/// pub struct LeafMainCarrier {
///    _forward_shutdown: Lifeline
/// }
///
/// impl CarryFrom<MainBus> for LeafBus {
///     type Lifeline = anyhow::Result<LeafMainCarrier>;
///     fn carry_from(&self, from: &MainBus) -> Self::Lifeline {
///         let mut rx = self.rx::<LeafShutdown>()?;
///         let mut tx = from.tx::<MainShutdown>()?;
///
///         let _forward_shutdown = Self::try_task("forward_shutdown", async move {
///             if let Some(msg) = rx.recv().await {
///                 tx.send(MainShutdown{}).await?;
///             }
///
///             Ok(())
///         });
///
///         Ok(LeafMainCarrier { _forward_shutdown })
///     }
/// }
/// ```
pub trait CarryFrom<FromBus: Bus>: Bus + Task + Sized {
    /// The carrier lifeline.  When dropped, all spawned tasks are immediately cancelled.
    type Lifeline;

    /// Spawns the carrier service, returning the lifeline value.
    fn carry_from(&self, from: &FromBus) -> Self::Lifeline;
}

/// The receprocial of the [CarryFrom](./trait.CarryFrom.html) trait.  Implemented for all types on which [CarryFrom](./trait.CarryFrom.html) is implemented.
pub trait CarryInto<IntoBus: Bus>: Bus + Task + Sized {
    /// The carrier lifeline.  When dropped, all spawned tasks are immediately cancelled.
    type Lifeline;

    /// Spawns the carrier service, returning the lifeline value.
    fn carry_into(&self, into: &IntoBus) -> Self::Lifeline;
}

impl<F, I> CarryInto<I> for F
where
    I: CarryFrom<F>,
    F: Bus,
    I: Bus,
{
    type Lifeline = <I as CarryFrom<F>>::Lifeline;

    fn carry_into(&self, into: &I) -> Self::Lifeline {
        into.carry_from(self)
    }
}

/// Constructs two bus types, and spawns the carrier between them.
/// Returns both busses, and the carrier's lifeline.
pub trait DefaultCarrier<FromBus: Bus>: CarryFrom<FromBus> {
    fn carry_default() -> (Self, FromBus, Self::Lifeline) {
        let into = Self::default();
        let from = FromBus::default();
        let lifeline = into.carry_from(&from);

        (into, from, lifeline)
    }
}

/// Provides the [Self::task](./trait.Task.html#method.task) and [Self::try_task](./trait.Task.html#method.try_task) associated methods for all types.
///
/// Lifeline supports the following task executors (using feature flags), and will use the first enabled flag:
/// - `tokio-executor`
/// - `async-std-executor`
///
/// Fallible tasks can be invoked with [Self::try_task](./trait.Task.html#method.try_task).  Lifeline will log OK/ERR status when the task finishes.
///
/// # Example
/// ```
/// use lifeline::prelude::*;
/// use tokio::sync::mpsc;
///
/// lifeline_bus!(pub struct ExampleBus);
///
/// #[derive(Debug, Clone)]
/// struct ExampleMessage {}
///
/// impl Message<ExampleBus> for ExampleMessage {
///     type Channel = mpsc::Sender<Self>;
/// }    
///
/// struct ExampleService {
///     _run: Lifeline   
/// }
///
/// impl Service for ExampleService {
///     type Bus = ExampleBus;
///     type Lifeline = anyhow::Result<Self>;
///
///     fn spawn(bus: &ExampleBus) -> anyhow::Result<Self> {
///         let mut rx = bus.rx::<ExampleMessage>()?;
///
///         let _run = Self::task("run", async move {
///             while let Some(msg) = rx.recv().await {
///                 log::info!("got message: {:?}", msg);
///             }
///         });
///
///         Ok(Self { _run })
///     }
/// }
/// ```
pub trait Task {
    /// Spawns an infallible task using the provided executor, wrapping it in a [Lifeline](./struct.Lifeline.html) handle.
    /// The task will run until it finishes, or until the [Lifeline](./struct.Lifeline.html) is droped.
    fn task<Out>(name: &str, fut: impl Future<Output = Out> + Send + 'static) -> Lifeline
    where
        Out: Debug + Send + 'static,
        Self: Sized,
    {
        let service_name = task_name::<Self>(name);
        spawn_task(service_name, fut)
    }

    /// Spawns an fallible task using the provided executor, wrapping it in a [Lifeline](./struct.Lifeline.html) handle.
    /// The task will run until it finishes, or until the [Lifeline](./struct.Lifeline.html) is droped.
    ///
    /// If the task finishes, lifeline will log an 'OK' or 'ERR' message with the return value.
    fn try_task<Out>(
        name: &str,
        fut: impl Future<Output = anyhow::Result<Out>> + Send + 'static,
    ) -> Lifeline
    where
        Out: Debug + 'static,
        Self: Sized,
    {
        let service_name = task_name::<Self>(name);
        spawn_task(service_name.clone(), async move {
            match fut.await {
                Ok(val) => {
                    if TypeId::of::<Out>() != TypeId::of::<()>() {
                        debug!("OK {}: {:?}", service_name, val);
                    } else {
                        debug!("OK {}", service_name);
                    }
                }
                Err(err) => {
                    error!("ERR: {}: {}", service_name, err);
                }
            }
        })
    }
}

impl<T> Task for T {}