wick-component 0.17.0

Macros and exports used to create Wick components
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
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91

use tokio_stream::StreamExt;
use wasmrs_runtime::BoxFuture;
use wick_packet::{BoxStream, PacketExt, VPacket};

use crate::adapters::encode;
use crate::{propagate_if_error, SingleOutput};

#[macro_export]
/// This macro will generate the implementations for simple binary operations, operations that take two inputs, produce one output, and are largely want to remain ignorant of stream state.
macro_rules! unary_simple {
  ($name:ident => $handler:ident) => {
    #[cfg_attr(not(target_family = "wasm"), async_trait::async_trait)]
    #[cfg_attr(target_family = "wasm", async_trait::async_trait(?Send))]
    impl $name::Operation for Component {
      type Error = wick_component::AnyError;
      type Inputs = $name::Inputs;
      type Outputs = $name::Outputs;
      type Config = $name::Config;

      async fn $name(
        mut inputs: Self::Inputs,
        mut outputs: Self::Outputs,
        ctx: Context<Self::Config>,
      ) -> Result<(), Self::Error> {
        use wick_packet::UnaryInputs;
        let input = inputs.take_input();

        wick_component::unary::simple(input, &mut outputs, &ctx, &$handler).await?;

        Ok(())
      }
    }
  };
}

/// Operation helper for common binary operations that have one output.
pub async fn simple<'c, INPUT, OUTPUT, CONTEXT, OUTPORT, F, E>(
  input: BoxStream<VPacket<INPUT>>,
  outputs: &mut OUTPORT,
  ctx: &'c CONTEXT,
  func: &'static F,
) -> Result<(), E>
where
  CONTEXT: Clone + wasmrs_runtime::ConditionallySendSync,
  F: Fn(INPUT, CONTEXT) -> BoxFuture<Result<OUTPUT, E>> + wasmrs_runtime::ConditionallySendSync,
  OUTPORT: SingleOutput + wasmrs_runtime::ConditionallySendSync,
  INPUT: serde::de::DeserializeOwned + Clone + wasmrs_runtime::ConditionallySendSync,
  OUTPUT: serde::Serialize + wasmrs_runtime::ConditionallySendSync,
  E: std::fmt::Display + wasmrs_runtime::ConditionallySendSync,
{
  let _ = inner::<INPUT, OUTPUT, CONTEXT, OUTPORT, F, E>(input, outputs, ctx, func).await;
  outputs.single_output().done();

  Ok(())
}

#[cfg_attr(not(target_family = "wasm"), async_recursion::async_recursion)]
#[cfg_attr(target_family = "wasm", async_recursion::async_recursion(?Send))]
async fn inner<'out, 'c, INPUT, OUTPUT, CONTEXT, OUTPORT, F, E>(
  mut input_stream: BoxStream<VPacket<INPUT>>,
  outputs: &'out mut OUTPORT,
  ctx: &'c CONTEXT,
  func: &'static F,
) -> BoxStream<VPacket<INPUT>>
where
  CONTEXT: Clone + wasmrs_runtime::ConditionallySendSync,
  F: Fn(INPUT, CONTEXT) -> BoxFuture<Result<OUTPUT, E>> + wasmrs_runtime::ConditionallySendSync,
  OUTPORT: SingleOutput + wasmrs_runtime::ConditionallySendSync,
  INPUT: serde::de::DeserializeOwned + Clone + wasmrs_runtime::ConditionallySendSync,
  OUTPUT: serde::Serialize + wasmrs_runtime::ConditionallySendSync,
  E: std::fmt::Display + wasmrs_runtime::ConditionallySendSync,
{
  loop {
    let Some(input) = input_stream.next().await else { break };

    if input.is_open_bracket() {
      outputs.broadcast_open();
      input_stream = inner(input_stream, outputs, ctx, func).await;
      outputs.broadcast_close();
    } else if input.is_close_bracket() || input.is_done() {
      break;
    } else {
      let input: INPUT = propagate_if_error!(input.decode(), outputs, continue);
      outputs
        .single_output()
        .send_raw_payload(encode(func(input.clone(), ctx.clone()).await));
    }
  }

  input_stream
}