flow-graph-interpreter 0.22.0

An intepreter for flow-based programs
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

use futures::StreamExt;
use serde_json::Value;
use wasmrs_rx::Observer;
use wick_packet::{Packet, PacketStream, PayloadFlux};

use crate::{BoxError, BoxFuture, Operation};

#[derive(Default, Debug, Clone, Copy)]
pub struct OneShotComponent {}

impl Operation for OneShotComponent {
  fn handle(&self, payload: wick_packet::StreamMap, _data: Option<Value>) -> BoxFuture<Result<PacketStream, BoxError>> {
    let task = async move {
      let flux = PayloadFluxChannel::new();
      let mut futs = Vec::new();
      for (port, mut stream) in payload.into_iter() {
        futs.push(async move { (port, stream.next().await) });
      }
      let fut = futures::future::join_all(futs).await;
      for (port, message) in fut {
        match message {
          Some(Ok(message)) => {
            flux.send(message);
            flux.send(Packet::done(port));
          }
          Some(Err(_e)) => {
            flux.send(Packet::component_error("Error sending oneshot payload"));
          }
          None => todo!(),
        }
      }
      Ok(PacketStream::new(flux.take_rx().unwrap()))
    };

    Box::pin(task)
  }
}