malstrom/operators/
sink.rs

1use crate::{
2    stream::StreamBuilder,
3    types::{Data, MaybeKey, Timestamp},
4};
5
6/// Output messages from a Malstrom stream somewhere
7pub trait Sink<K, V, T, S>: super::sealed::Sealed {
8    /// Sink all messages in this stream to the given output.
9    /// This will consume the messages. If you whish to write to multiple outputs,
10    /// consider calling [.cloned()](crate::operators::Cloned::cloned) on the stream.
11    ///
12    /// # Example
13    ///
14    /// ```
15    /// use malstrom::operators::*;
16    /// use malstrom::runtime::SingleThreadRuntime;
17    /// use malstrom::snapshot::NoPersistence;
18    /// use malstrom::sources::{SingleIteratorSource, StatelessSource};
19    /// use malstrom::worker::StreamProvider;
20    /// use malstrom::sinks::{VecSink, StatelessSink};
21    ///
22    /// let sink = VecSink::new();
23    /// let sink_clone = sink.clone();
24    ///
25    /// SingleThreadRuntime::builder()
26    ///     .persistence(NoPersistence)
27    ///     .build(move |provider: &mut dyn StreamProvider| {
28    ///         provider.new_stream()
29    ///         .source("numbers", StatelessSource::new(SingleIteratorSource::new(0..10)))
30    ///         .sink("sink", StatelessSink::new(sink_clone));
31    ///     })
32    ///     .execute()
33    ///     .unwrap();
34    /// let expected: Vec<i32> = (0..10).collect();
35    /// let out: Vec<i32> = sink.into_iter().map(|x| x.value).collect();
36    /// assert_eq!(out, expected);
37    /// ```
38    fn sink(self, name: &str, sink: S);
39}
40
41/// A stream output which takes messages, usually producing them to some external system.
42/// For users it is normally not necessary to implement this trait unless they are writing
43/// custom outputs for sinks which Malstrom does not (yet) support.
44#[diagnostic::on_unimplemented(message = "Not a Sink: 
45    You might need to wrap this in `StatefulSink::new` or `StatelessSink::new`")]
46pub trait StreamSink<K, V, T> {
47    /// Consume a datastream to the end.
48    fn consume_stream(self, name: &str, builder: StreamBuilder<K, V, T>);
49}
50
51impl<K, V, T, S> Sink<K, V, T, S> for StreamBuilder<K, V, T>
52where
53    K: MaybeKey,
54    V: Data,
55    T: Timestamp,
56    S: StreamSink<K, V, T>,
57{
58    fn sink(self, name: &str, sink: S) {
59        sink.consume_stream(name, self)
60    }
61}
malstrom/operators/sink.rs

malstrom/operators/
sink.rs
