krabka_client_streams/processor/
mod.rs

1//! Typed Processor API and the type-erased execution graph used by the runtime.
2
3pub mod api;
4pub mod erased;
5pub(crate) mod factory;
6pub mod fixed_key;
7pub(crate) mod graph;
8pub(crate) mod node;
9pub mod punctuation;
10pub mod record;
11pub mod serde;
12
13#[cfg(not(target_family = "wasm"))]
14pub mod schema_serde;
15
16pub use api::{Processor, ProcessorContext, ProcessorSupplier};
17pub use erased::ProcessorError;
18pub use fixed_key::{
19    FixedKeyProcessor, FixedKeyProcessorContext, FixedKeyProcessorSupplier, FixedKeyRecord,
20};
21pub use punctuation::{Cancellable, PunctuationType, Punctuator};
22pub use record::{Record, RecordContext};
23pub use serde::{
24    BytesSerde, Changed, Consumed, DefaultSerde, I64Serde, Produced, Serde, SerdeError, SerdeRole,
25    StringSerde,
26};
27
28/// Implement [`Processor`] with a compact `(input_key, input_value) ->
29/// (output_key, output_value)` declaration.
30///
31/// ```
32/// use krabka_client_streams::{Record, impl_processor};
33///
34/// struct Upper;
35/// impl_processor! {
36///     impl Upper: (String, String) -> (String, String) {
37///         async fn process(&mut self, ctx, r) {
38///             ctx.forward(Record::new(r.key, r.value.to_uppercase(), r.timestamp));
39///         }
40///     }
41/// }
42/// ```
43///
44/// [`Processor`]: crate::processor::api::Processor
45#[macro_export]
46macro_rules! impl_processor {
47    (
48        impl $processor:ty: ($kin:ty, $vin:ty) -> ($kout:ty, $vout:ty) {
49            $(
50                async fn init(&mut self, $init_ctx:ident) $init_body:block
51            )?
52            async fn process(&mut self, $ctx:ident, $record:ident) $process_body:block
53            $(
54                async fn close(&mut self) $close_body:block
55            )?
56        }
57    ) => {
58        #[$crate::__async_trait]
59        impl $crate::processor::api::Processor<$kin, $vin, $kout, $vout> for $processor {
60            $(
61                async fn init(
62                    &mut self,
63                    $init_ctx: &mut $crate::processor::api::ProcessorContext<'_, '_, $kout, $vout>,
64                ) $init_body
65            )?
66
67            async fn process(
68                &mut self,
69                $ctx: &mut $crate::processor::api::ProcessorContext<'_, '_, $kout, $vout>,
70                $record: $crate::processor::record::Record<$kin, $vin>,
71            ) $process_body
72
73            $(
74                async fn close(&mut self) $close_body
75            )?
76        }
77    };
78}
79
80/// Implement [`FixedKeyProcessor`] with a compact `(key, input_value) ->
81/// output_value` declaration.
82///
83/// ```
84/// use krabka_client_streams::impl_fixed_key_processor;
85///
86/// struct UpperValue;
87/// impl_fixed_key_processor! {
88///     impl UpperValue: (String, String) -> String {
89///         async fn process(&mut self, ctx, r) {
90///             let v = r.value.clone();
91///             ctx.forward(r.with_value(v.to_uppercase()));
92///         }
93///     }
94/// }
95/// ```
96///
97/// [`FixedKeyProcessor`]: crate::processor::fixed_key::FixedKeyProcessor
98#[macro_export]
99macro_rules! impl_fixed_key_processor {
100    (
101        impl $processor:ty: ($kin:ty, $vin:ty) -> $vout:ty {
102            async fn process(&mut self, $ctx:ident, $record:ident) $process_body:block
103        }
104    ) => {
105        #[$crate::__async_trait]
106        impl $crate::processor::fixed_key::FixedKeyProcessor<$kin, $vin, $vout> for $processor {
107            async fn process(
108                &mut self,
109                $ctx: &mut $crate::processor::fixed_key::FixedKeyProcessorContext<
110                    '_,
111                    '_,
112                    '_,
113                    $kin,
114                    $vout,
115                >,
116                $record: $crate::processor::fixed_key::FixedKeyRecord<$kin, $vin>,
117            ) $process_body
118        }
119    };
120}
krabka_client_streams/processor/mod.rs

krabka_client_streams/processor/
mod.rs
