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

//! Run commands and keep track of subscriptions.
use crate::{subscription, Command, Executor, Subscription};

use futures::{channel::mpsc, Sink};
use std::marker::PhantomData;

/// A batteries-included runtime of commands and subscriptions.
///
/// If you have an [`Executor`], a [`Runtime`] can be leveraged to run any
/// [`Command`] or [`Subscription`] and get notified of the results!
///
/// [`Runtime`]: struct.Runtime.html
/// [`Executor`]: executor/trait.Executor.html
/// [`Command`]: struct.Command.html
/// [`Subscription`]: subscription/struct.Subscription.html
#[derive(Debug)]
pub struct Runtime<Hasher, Event, Executor, Sender, Message> {
    executor: Executor,
    sender: Sender,
    subscriptions: subscription::Tracker<Hasher, Event>,
    _message: PhantomData<Message>,
}

impl<Hasher, Event, Executor, Sender, Message>
    Runtime<Hasher, Event, Executor, Sender, Message>
where
    Hasher: std::hash::Hasher + Default,
    Event: Send + Clone + 'static,
    Executor: self::Executor,
    Sender:
        Sink<Message, Error = mpsc::SendError> + Unpin + Send + Clone + 'static,
    Message: Send + 'static,
{
    /// Creates a new empty [`Runtime`].
    ///
    /// You need to provide:
    /// - an [`Executor`] to spawn futures
    /// - a `Sender` implementing `Sink` to receive the results
    ///
    /// [`Runtime`]: struct.Runtime.html
    pub fn new(executor: Executor, sender: Sender) -> Self {
        Self {
            executor,
            sender,
            subscriptions: subscription::Tracker::new(),
            _message: PhantomData,
        }
    }

    /// Runs the given closure inside the [`Executor`] of the [`Runtime`].
    ///
    /// See [`Executor::enter`] to learn more.
    ///
    /// [`Executor`]: executor/trait.Executor.html
    /// [`Runtime`]: struct.Runtime.html
    /// [`Executor::enter`]: executor/trait.Executor.html#method.enter
    pub fn enter<R>(&self, f: impl FnOnce() -> R) -> R {
        self.executor.enter(f)
    }

    /// Spawns a [`Command`] in the [`Runtime`].
    ///
    /// The resulting `Message` will be forwarded to the `Sender` of the
    /// [`Runtime`].
    ///
    /// [`Command`]: struct.Command.html
    /// [`Runtime`]: struct.Runtime.html
    pub fn spawn(&mut self, command: Command<Message>) {
        use futures::{FutureExt, SinkExt};

        let futures = command.futures();

        for future in futures {
            let mut sender = self.sender.clone();

            let future = future.then(|message| async move {
                let _ = sender.send(message).await;

                ()
            });

            self.executor.spawn(future);
        }
    }

    /// Tracks a [`Subscription`] in the [`Runtime`].
    ///
    /// It will spawn new streams or close old ones as necessary! See
    /// [`Tracker::update`] to learn more about this!
    ///
    /// [`Subscription`]: subscription/struct.Subscription.html
    /// [`Runtime`]: struct.Runtime.html
    /// [`Tracker::update`]: subscription/struct.Tracker.html#method.update
    pub fn track(
        &mut self,
        subscription: Subscription<Hasher, Event, Message>,
    ) {
        let Runtime {
            executor,
            subscriptions,
            sender,
            ..
        } = self;

        let futures = executor
            .enter(|| subscriptions.update(subscription, sender.clone()));

        for future in futures {
            executor.spawn(future);
        }
    }

    /// Broadcasts an event to all the subscriptions currently alive in the
    /// [`Runtime`].
    ///
    /// See [`Tracker::broadcast`] to learn more.
    ///
    /// [`Runtime`]: struct.Runtime.html
    /// [`Tracker::broadcast`]:
    /// subscription/struct.Tracker.html#method.broadcast
    pub fn broadcast(&mut self, event: Event) {
        self.subscriptions.broadcast(event);
    }
}