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::marker::PhantomData;
31use std::rc::Rc;
32
33use zrx_scheduler::{Id, Value};
34
35use crate::stream::operator::Operator;
36use crate::stream::Stream;
37
38use super::traits::With;
39use super::WorkspaceRef;
40
41mod schedulable;
42
43pub use schedulable::Schedulable;
44
45// ----------------------------------------------------------------------------
46// Structs
47// ----------------------------------------------------------------------------
48
49/// Workflow.
50#[derive(Debug)]
51pub struct Workflow<I> {
52    /// Shared inner state.
53    inner: Rc<RefCell<WorkflowInner<I>>>,
54}
55
56/// Workflow inner state.
57#[derive(Debug)]
58pub struct WorkflowInner<I> {
59    /// Identifier.
60    id: usize,
61    /// Associated workspace.
62    workspace: WorkspaceRef<I>,
63}
64
65// ----------------------------------------------------------------------------
66// Implementations
67// ----------------------------------------------------------------------------
68
69impl<I> Workflow<I>
70where
71    I: std::fmt::Debug,
72{
73    /// Creates a new workflow.
74    #[must_use]
75    pub fn new(id: usize, workspace: WorkspaceRef<I>) -> Self {
76        Self {
77            inner: Rc::new(RefCell::new(WorkflowInner { id, workspace })),
78        }
79    }
80
81    /// Adds a source stream.
82    #[allow(clippy::missing_panics_doc)]
83    #[must_use]
84    pub fn add_source<T>(&self) -> Stream<I, T>
85    where
86        T: Any,
87    {
88        let id = self.with(|workflow| {
89            let workspace = workflow.workspace.upgrade().expect("invariant");
90            workspace.add_source::<T>()
91        });
92        Stream {
93            id,
94            workflow: self.clone(),
95            marker: PhantomData,
96        }
97    }
98
99    /// Adds an operator to the workflow.
100    #[allow(clippy::missing_panics_doc)]
101    pub fn add_operator<S, O, T, U>(&self, from: S, operator: O) -> Stream<I, U>
102    where
103        I: Id,
104        S: IntoIterator<Item = usize>,
105        O: Operator<I, T> + 'static,
106        T: Value,
107        U: Value,
108    {
109        let id = self.with(|workflow| {
110            let workspace = workflow.workspace.upgrade().expect("invariant");
111            workspace.add_action::<U, _, _>(from, Schedulable::new(operator))
112        });
113        Stream {
114            id,
115            workflow: self.clone(),
116            marker: PhantomData,
117        }
118    }
119}
120
121#[allow(clippy::must_use_candidate)]
122impl<I> Workflow<I> {
123    /// Returns the identifier of the workflow.
124    pub fn id(&self) -> usize {
125        self.with(|workflow| workflow.id)
126    }
127}
128
129// ----------------------------------------------------------------------------
130// Trait implementations
131// ----------------------------------------------------------------------------
132
133impl<I> With for Workflow<I> {
134    type Item = WorkflowInner<I>;
135
136    /// Returns a reference to the inner state.
137    #[inline]
138    fn inner(&self) -> &RefCell<Self::Item> {
139        &self.inner
140    }
141}
142
143// ----------------------------------------------------------------------------
144
145impl<I> Clone for Workflow<I> {
146    /// Returns a copy of the workflow.
147    #[inline]
148    fn clone(&self) -> Self {
149        Self { inner: self.inner.clone() }
150    }
151}
zrx_stream/stream/workspace/workflow.rs

zrx_stream/stream/workspace/
workflow.rs
