Struct erdos::dataflow::stream::IngestStream

pub struct IngestStream<D> where
    for<'a> D: Data + Deserialize<'a>,  { /* private fields */ }

An IngestStream enables drivers to inject data into a running ERDOS application.

Similar to a WriteStream, an IngestStream exposes a send function to allow drivers to send data to the operators of the constructed graph.

Example

The below example shows how to use an IngestStream to send data to a FlatMapOperator, and retrieve the processed values through an ExtractStream.

let args = erdos::new_app("ERDOS").get_matches();
let mut node = Node::new(Configuration::from_args(&args));

// Create an IngestStream.
let mut ingest_stream = IngestStream::new();

// Create an ExtractStream from the ReadStream of the FlatMapOperator.
let output_stream = erdos::connect_one_in_one_out(
    || FlatMapOperator::new(|x: &usize| { std::iter::once(2 * x) }),
    || {},
    OperatorConfig::new().name("MapOperator"),
    &ingest_stream,
);
let mut extract_stream = ExtractStream::new(&output_stream);

node.run_async();

// Send data on the IngestStream.
for i in 1..10 {
    ingest_stream.send(Message::new_message(Timestamp::Time(vec![i as u64]), i)).unwrap();
}

// Retrieve mapped values using an ExtractStream.
for i in 1..10 {
    let message = extract_stream.read().unwrap();
    assert_eq!(*message.data().unwrap(), 2 * i);
}

Implementations

impl<D> IngestStream<D> where
    for<'a> D: Data + Deserialize<'a>, 

pub fn new() -> Self

Returns a new instance of the IngestStream.

pub fn is_closed(&self) -> bool

Returns true if a top watermark message was received or the IngestStream failed to set up.

pub fn send(&mut self, msg: Message<D>) -> Result<(), SendError>

Sends data on the stream.

Arguments

• msg - The message to be sent on the stream.

Trait Implementations

impl<D> Default for IngestStream<D> where
    for<'a> D: Data + Deserialize<'a>, 

fn default() -> Self

Returns the “default value” for a type.

impl<D> Stream<D> for IngestStream<D> where
    for<'a> D: Data + Deserialize<'a>, 

fn id(&self) -> StreamId

fn name(&self) -> String

fn set_name(&mut self, name: &str)

impl<D> WriteStreamT<D> for IngestStream<D> where
    for<'a> D: Data + Deserialize<'a>, 

fn send(&mut self, msg: Message<D>) -> Result<(), SendError>

Blocks until write stream is available