harmos_runtime/streams/
demand.rs

1/* ***
2 *
3 * Harmos
4 *
5 * SPDX-License-Identifier: Apache-2.0
6 *
7 * ***
8 */
9
10/* === Modules === */
11
12/* === Re-Exports === */
13
14/* === Imports === */
15
16// Standard Library Imports
17use std::ops::{Deref, DerefMut};
18use std::sync::Arc;
19// Third Party Imports
20use tokio::sync::watch;
21// Local Imports
22use super::window::Window;
23use crate::Error;
24
25/* === Constants === */
26
27/* === Macros === */
28
29/* === Type Aliases === */
30
31/* === Structs === */
32
33#[derive(Clone, Debug)]
34pub struct Demand {
35    updates: watch::Receiver<usize>,
36}
37
38pub(crate) struct Holding {
39    window: Arc<Window>,
40}
41
42/* === Enums === */
43
44/* === Traits === */
45
46/* === Implementations === */
47
48impl Demand {
49    pub(crate) fn of(window: &Window) -> Self {
50        Self { updates: window.interest() }
51    }
52
53    pub async fn changed(&mut self) -> Result<(), Error> {
54        self.updates.changed().await.map_err(|_| Error::Unavailable)
55    }
56
57    pub fn borrow(&self) -> impl Deref<Target = usize> + '_ {
58        self.updates.borrow()
59    }
60}
61
62impl Deref for Demand {
63    type Target = watch::Receiver<usize>;
64
65    fn deref(&self) -> &Self::Target {
66        &self.updates
67    }
68}
69
70impl DerefMut for Demand {
71    fn deref_mut(&mut self) -> &mut Self::Target {
72        &mut self.updates
73    }
74}
75
76impl Holding {
77    pub(crate) fn engage(window: Arc<Window>) -> Self {
78        window.engaged();
79        Self { window }
80    }
81}
82
83impl Drop for Holding {
84    fn drop(&mut self) {
85        self.window.released();
86    }
87}
88
89/* === Public Functions === */
90
91/* === Private Functions === */
92
93/* === Tests === */
harmos_runtime/streams/demand.rs

harmos_runtime/streams/
demand.rs
