crossflow 0.0.6

Reactive programming and workflow engine in bevy
Documentation 
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

/*
 * Copyright (C) 2025 Open Source Robotics Foundation
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 *
*/

use bevy_ecs::prelude::{Entity, World};

use std::sync::Arc;

use crate::{InnerChannel, OperationRoster, ReportUnhandled, StreamEffect, StreamRequest};

/// Use this channel to stream data using the [`StreamChannel::send`] method.
pub struct StreamChannel<S> {
    target: Option<Entity>,
    inner: Arc<InnerChannel>,
    _ignore: std::marker::PhantomData<fn(S)>,
}

impl<S: StreamEffect> StreamChannel<S> {
    /// Send an instance of data out over a stream.
    pub fn send(&self, data: S::Input) {
        let source = self.inner.source;
        let session = self.inner.session;
        let target = self.target;
        self.inner
            .sender
            .send(Box::new(
                move |world: &mut World, roster: &mut OperationRoster| {
                    let mut request = StreamRequest {
                        source,
                        session,
                        target,
                        world,
                        roster,
                    };

                    S::side_effect(data, &mut request)
                        .and_then(|output| request.send_output(output))
                        .report_unhandled(source, world);
                },
            ))
            .ok();
    }

    pub(crate) fn new(target: Option<Entity>, inner: Arc<InnerChannel>) -> Self {
        Self {
            target,
            inner,
            _ignore: Default::default(),
        }
    }
}

impl<S> Clone for StreamChannel<S> {
    fn clone(&self) -> Self {
        Self {
            target: ::std::clone::Clone::clone(&self.target),
            inner: Arc::clone(&self.inner),
            _ignore: Default::default(),
        }
    }
}