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

use super::{Revision, Runtime};
use futures::{
    stream::{Stream, StreamExt},
    task::LocalSpawn,
};
use std::{
    pin::Pin,
    task::{Context as FutContext, Poll, Waker},
};

/// A [`Runtime`] that is bound with a particular root function.
///
/// If running in a context with an async executor, can be consumed as a
/// [`futures::Stream`] of [`crate::runtime::Revision`]s in order to provide
/// the [`super::Runtime`] with a [`std::task::Waker`].
pub struct RunLoop<Root> {
    inner: Runtime,
    root: Root,
}

impl super::Runtime {
    /// Returns this runtime bound with a specific root function it will run in
    /// a loop.
    pub fn looped<Root, Out>(self, root: Root) -> RunLoop<Root>
    where
        Root: FnMut() -> Out,
    {
        RunLoop { inner: self, root }
    }
}

impl<Root, Out> RunLoop<Root>
where
    Root: FnMut() -> Out + Unpin,
{
    /// Creates a new `Runtime` attached to the provided root function.
    pub fn new(root: Root) -> RunLoop<Root> {
        RunLoop { root, inner: Runtime::new() }
    }

    /// Returns the runtime's current Revision.
    pub fn revision(&self) -> Revision {
        self.inner.revision()
    }

    /// Sets the [`std::task::Waker`] which will be called when state variables
    /// change.
    pub fn set_state_change_waker(&mut self, wk: Waker) {
        self.inner.set_state_change_waker(wk);
    }

    /// Sets the executor that will be used to spawn normal priority tasks.
    pub fn set_task_executor(&mut self, sp: impl LocalSpawn + 'static) {
        self.inner.set_task_executor(sp);
    }

    /// Run the root function once within this runtime's context, returning the
    /// result.
    pub fn run_once(&mut self) -> Out {
        self.inner.run_once(&mut self.root)
    }

    /// Poll this runtime without exiting. Discards any value returned from the
    /// root function. The future yields in between revisions and is woken on
    /// state changes.
    pub async fn run_on_state_changes(mut self) {
        loop {
            self.next().await;
        }
    }

    /// Unbinds the runtime from its current root function, returning both.
    pub fn unloop(self) -> (Runtime, Root) {
        (self.inner, self.root)
    }
}

impl<Root, Out> Stream for RunLoop<Root>
where
    Root: FnMut() -> Out + Unpin,
{
    type Item = (Revision, Out);

    /// This `Stream` implementation runs a single revision for each call to
    /// `poll_next`, always returning `Poll::Ready(Some(...))`.
    fn poll_next(self: Pin<&mut Self>, cx: &mut FutContext<'_>) -> Poll<Option<Self::Item>> {
        let this = self.get_mut();
        this.inner.set_state_change_waker(cx.waker().clone());
        let out = this.run_once();
        Poll::Ready(Some((this.inner.revision, out)))
    }
}