timely/dataflow/
stream.rs1use crate::progress::{Source, Target};
8
9use crate::communication::Push;
10use crate::dataflow::Scope;
11use crate::dataflow::channels::pushers::tee::TeeHelper;
12use crate::dataflow::channels::Message;
13use std::fmt::{self, Debug};
14
15pub struct StreamCore<S: Scope, C> {
22 name: Source,
24 scope: S,
26 ports: TeeHelper<S::Timestamp, C>,
28}
29
30impl<S: Scope, C> Clone for StreamCore<S, C> {
31 fn clone(&self) -> Self {
32 Self {
33 name: self.name,
34 scope: self.scope.clone(),
35 ports: self.ports.clone(),
36 }
37 }
38
39 fn clone_from(&mut self, source: &Self) {
40 self.name.clone_from(&source.name);
41 self.scope.clone_from(&source.scope);
42 self.ports.clone_from(&source.ports);
43 }
44}
45
46pub type Stream<S, D> = StreamCore<S, Vec<D>>;
48
49impl<S: Scope, C> StreamCore<S, C> {
50 pub fn connect_to<P: Push<Message<S::Timestamp, C>>+'static>(&self, target: Target, pusher: P, identifier: usize) {
55
56 let mut logging = self.scope().logging();
57 logging.as_mut().map(|l| l.log(crate::logging::ChannelsEvent {
58 id: identifier,
59 scope_addr: self.scope.addr().to_vec(),
60 source: (self.name.node, self.name.port),
61 target: (target.node, target.port),
62 typ: std::any::type_name::<C>().to_string(),
63 }));
64
65 self.scope.add_edge(self.name, target);
66 self.ports.add_pusher(pusher);
67 }
68 pub fn new(source: Source, output: TeeHelper<S::Timestamp, C>, scope: S) -> Self {
70 Self { name: source, ports: output, scope }
71 }
72 pub fn name(&self) -> &Source { &self.name }
74 pub fn scope(&self) -> S { self.scope.clone() }
76
77 pub fn container<C2>(self) -> StreamCore<S, C2> where Self: AsStream<S, C2> { self.as_stream() }
79}
80
81pub trait AsStream<S: Scope, C> {
83 fn as_stream(self) -> StreamCore<S, C>;
85}
86
87impl<S: Scope, C> AsStream<S, C> for StreamCore<S, C> {
88 fn as_stream(self) -> Self { self }
89}
90
91impl<S, C> Debug for StreamCore<S, C>
92where
93 S: Scope,
94{
95 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
96 f.debug_struct("Stream")
97 .field("source", &self.name)
98 .finish_non_exhaustive()
99 }
100}