zrx_stream/stream/workspace/
workflow.rs

1// Copyright (c) 2025-2026 Zensical and contributors
2
3// SPDX-License-Identifier: MIT
4// All contributions are certified under the DCO
5
6// Permission is hereby granted, free of charge, to any person obtaining a copy
7// of this software and associated documentation files (the "Software"), to
8// deal in the Software without restriction, including without limitation the
9// rights to use, copy, modify, merge, publish, distribute, sublicense, and/or
10// sell copies of the Software, and to permit persons to whom the Software is
11// furnished to do so, subject to the following conditions:
12
13// The above copyright notice and this permission notice shall be included in
14// all copies or substantial portions of the Software.
15
16// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
17// IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
18// FITNESS FOR A PARTICULAR PURPOSE AND NON-INFRINGEMENT. IN NO EVENT SHALL THE
19// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
20// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING
21// FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS
22// IN THE SOFTWARE.
23
24// ----------------------------------------------------------------------------
25
26//! Workflow.
27
28use std::any::Any;
29use std::cell::RefCell;
30use std::fmt::Debug;
31use std::marker::PhantomData;
32use std::rc::Rc;
33
34use zrx_scheduler::{Id, Value};
35
36use crate::stream::operator::Operator;
37use crate::stream::Stream;
38
39use super::traits::With;
40use super::WorkspaceRef;
41
42mod schedulable;
43
44pub use schedulable::Schedulable;
45
46// ----------------------------------------------------------------------------
47// Structs
48// ----------------------------------------------------------------------------
49
50/// Workflow.
51#[derive(Debug)]
52pub struct Workflow<I> {
53    /// Shared inner state.
54    inner: Rc<RefCell<WorkflowInner<I>>>,
55}
56
57/// Workflow inner state.
58#[derive(Debug)]
59pub struct WorkflowInner<I> {
60    /// Identifier.
61    id: usize,
62    /// Associated workspace.
63    workspace: WorkspaceRef<I>,
64}
65
66// ----------------------------------------------------------------------------
67// Implementations
68// ----------------------------------------------------------------------------
69
70impl<I> Workflow<I>
71where
72    I: Debug,
73{
74    /// Creates a new workflow.
75    #[must_use]
76    pub fn new(id: usize, workspace: WorkspaceRef<I>) -> Self {
77        Self {
78            inner: Rc::new(RefCell::new(WorkflowInner { id, workspace })),
79        }
80    }
81
82    /// Adds a source stream.
83    #[allow(clippy::missing_panics_doc)]
84    #[must_use]
85    pub fn add_source<T>(&self) -> Stream<I, T>
86    where
87        T: Any,
88    {
89        let id = self.with(|workflow| {
90            let workspace = workflow.workspace.upgrade().expect("invariant");
91            workspace.add_source::<T>()
92        });
93        Stream {
94            id,
95            workflow: self.clone(),
96            marker: PhantomData,
97        }
98    }
99
100    /// Adds an operator to the workflow.
101    #[allow(clippy::missing_panics_doc)]
102    pub fn add_operator<S, O, T, U>(&self, from: S, operator: O) -> Stream<I, U>
103    where
104        I: Id,
105        S: IntoIterator<Item = usize>,
106        O: Operator<I, T> + 'static,
107        T: Value,
108        U: Value,
109    {
110        let id = self.with(|workflow| {
111            let workspace = workflow.workspace.upgrade().expect("invariant");
112            workspace.add_action::<U, _, _>(from, Schedulable::new(operator))
113        });
114        Stream {
115            id,
116            workflow: self.clone(),
117            marker: PhantomData,
118        }
119    }
120}
121
122#[allow(clippy::must_use_candidate)]
123impl<I> Workflow<I> {
124    /// Returns the identifier of the workflow.
125    pub fn id(&self) -> usize {
126        self.with(|workflow| workflow.id)
127    }
128}
129
130// ----------------------------------------------------------------------------
131// Trait implementations
132// ----------------------------------------------------------------------------
133
134impl<I> With for Workflow<I> {
135    type Item = WorkflowInner<I>;
136
137    /// Returns a reference to the inner state.
138    #[inline]
139    fn inner(&self) -> &RefCell<Self::Item> {
140        &self.inner
141    }
142}
143
144// ----------------------------------------------------------------------------
145
146impl<I> Clone for Workflow<I> {
147    /// Returns a copy of the workflow.
148    #[inline]
149    fn clone(&self) -> Self {
150        Self { inner: self.inner.clone() }
151    }
152}
zrx_stream/stream/workspace/workflow.rs

zrx_stream/stream/workspace/
workflow.rs
