Lint

john-z-yang · Nov 7, 2024 · 4185104 · 4185104
1 parent f11ce16
commit 4185104
Show file tree

Hide file tree

Showing 2 changed files with 7 additions and 2 deletions.
diff --git a/src/lib.rs b/src/lib.rs
@@ -273,7 +273,7 @@ macro_rules! processing_strategy {
                 ));
             }
 
-            let (commit_sender, commit_receiver) = crate::processing_strategy!(
+            let (commit_sender, commit_receiver) = $crate::processing_strategy!(
                 @reducers,
                 ($reduce_first $(,$reduce_rest)*),
                 reduce_receiver,

diff --git a/src/main.rs b/src/main.rs
@@ -2,6 +2,7 @@ use anyhow::{anyhow, Error};
 use chrono::serde::ts_seconds;
 use chrono::{DateTime, Utc};
 use kafka_map_reduce::reducers::clickhouse::ClickhouseBatchWriter;
+use kafka_map_reduce::reducers::noop::NoopReducer;
 use kafka_map_reduce::reducers::os_stream::{OsStream, OsStreamWriter};
 use kafka_map_reduce::{processing_strategy, start_consumer, ReduceShutdownBehaviour};
 use rdkafka::{config::RDKafkaLogLevel, message::OwnedMessage, ClientConfig, Message};
@@ -102,14 +103,18 @@ async fn main() -> Result<(), Error> {
             .set_log_level(RDKafkaLogLevel::Debug),
         processing_strategy!({
             map: parse,
+
             reduce: ClickhouseBatchWriter::new(
                 host,
                 port,
                 table,
                 128,
                 Duration::from_secs(4),
                 ReduceShutdownBehaviour::Flush,
-            ),
+            )
+            => NoopReducer::new("hello")
+            => NoopReducer::new("world"),
+
             err: OsStreamWriter::new(
                 Duration::from_secs(1),
                 OsStream::StdErr,