feat: Map Streamer

.rustfmt.toml

@@ -1 +1 @@
-edition = "2021"
+edition = "2021"

examples/flatmap-stream/Cargo.toml

@@ -0,0 +1,9 @@
+[package]
+name = "flatmap-stream"
+version = "0.1.0"
+edition = "2021"
+
+[dependencies]
+tonic = "0.12.0"
+tokio = { version = "1.0", features = ["macros", "rt-multi-thread"] }
+numaflow = { path = "../../numaflow" }

examples/flatmap-stream/Dockerfile

@@ -0,0 +1,20 @@
+FROM rust:1.82-bullseye AS build
+
+RUN apt-get update
+RUN apt-get install protobuf-compiler -y
+
+WORKDIR /numaflow-rs
+COPY ./ ./
+WORKDIR /numaflow-rs/examples/flatmap-stream
+
+# build for release
+RUN cargo build --release
+
+# our final base
+FROM debian:bullseye AS flatmap-stream
+
+# copy the build artifact from the build stage
+COPY --from=build /numaflow-rs/target/release/flatmap-stream .
+
+# set the startup command to run your binary
+CMD ["./flatmap-stream"]

examples/flatmap-stream/Makefile

@@ -0,0 +1,20 @@
+TAG ?= stable
+PUSH ?= false
+IMAGE_REGISTRY = quay.io/numaio/numaflow-rs/flatmap-stream:${TAG}
+DOCKER_FILE_PATH = examples/flatmap-stream/Dockerfile
+
+.PHONY: update
+update:
+	cargo check
+	cargo update
+
+.PHONY: image
+image: update
+	cd ../../ && docker build \
+	-f ${DOCKER_FILE_PATH}  \
+	-t ${IMAGE_REGISTRY} .
+	@if [ "$(PUSH)" = "true" ]; then docker push ${IMAGE_REGISTRY}; fi
+
+.PHONY: clean
+clean:
+	-rm -rf target

examples/flatmap-stream/manifests/simple-flatmap-stream.yaml

@@ -0,0 +1,29 @@
+apiVersion: numaflow.numaproj.io/v1alpha1
+kind: Pipeline
+metadata:
+  name: simple-flatmap-stream
+spec:
+  vertices:
+    - name: in
+      source:
+        # A self data generating source
+        generator:
+          rpu: 300
+          duration: 1s
+          keyCount: 5
+          value: 5
+    - name: cat
+      scale:
+        min: 1
+      udf:
+        container:
+          image: quay.io/numaio/numaflow-rs/flatmap-stream:stable
+    - name: out
+      sink:
+        # A simple log printing sink
+        log: { }
+  edges:
+    - from: in
+      to: cat
+    - from: cat
+      to: out

examples/flatmap-stream/src/main.rs

@@ -0,0 +1,27 @@
+use numaflow::mapstream;
+use numaflow::mapstream::Message;
+use tokio::sync::mpsc::Sender;
+
+#[tokio::main]
+async fn main() -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
+    mapstream::Server::new(Cat).start().await
+}
+
+struct Cat;
+
+#[tonic::async_trait]
+impl mapstream::MapStreamer for Cat {
+    async fn map_stream(&self, input: mapstream::MapStreamRequest, tx: Sender<Message>) {
+        let payload_str = String::from_utf8(input.value).unwrap_or_default();
+        let splits: Vec<&str> = payload_str.split(',').collect();
+
+        for split in splits {
+            let message = Message::new(split.as_bytes().to_vec())
+                .keys(input.keys.clone())
+                .tags(vec![]);
+            if tx.send(message).await.is_err() {
+                break;
+            }
+        }
+    }
+}

examples/mapt-event-time-filter/src/main.rs

@@ -1,7 +1,8 @@
+use std::error::Error;
+
 use filter_impl::filter_event_time;
 use numaflow::sourcetransform;
 use numaflow::sourcetransform::{Message, SourceTransformRequest};
-use std::error::Error;
 
 #[tokio::main]
 async fn main() -> Result<(), Box<dyn Error + Send + Sync>> {
@@ -41,9 +42,10 @@ mod filter_impl {
 
 #[cfg(test)]
 mod tests {
-    use crate::filter_impl::filter_event_time;
     use chrono::{TimeZone, Utc};
     use numaflow::sourcetransform::SourceTransformRequest;
+
+    use crate::filter_impl::filter_event_time;
     /// Tests that events from 2022 are tagged as within the year 2022.
     #[test]
     fn test_filter_event_time_should_return_within_year_2022() {

examples/sideinput/src/main.rs

@@ -1,7 +1,7 @@
-use numaflow::sideinput::{self, SideInputer};
 use std::sync::Mutex;
 use std::time::{SystemTime, UNIX_EPOCH};
 
+use numaflow::sideinput::{self, SideInputer};
 use tonic::async_trait;
 
 struct SideInputHandler {

examples/simple-source/src/main.rs

@@ -7,12 +7,12 @@ async fn main() -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
 }
 
 pub(crate) mod simple_source {
-    use chrono::Utc;
     use std::sync::atomic::{AtomicUsize, Ordering};
     use std::{collections::HashSet, sync::RwLock};
-    use tokio::sync::mpsc::Sender;
 
+    use chrono::Utc;
     use numaflow::source::{Message, Offset, SourceReadRequest, Sourcer};
+    use tokio::sync::mpsc::Sender;
 
     /// SimpleSource is a data generator which generates monotonically increasing offsets and data. It is a shared state which is protected using Locks
     /// or Atomics to provide concurrent access. Numaflow actually does not require concurrent access but we are forced to do this because the SDK

numaflow/src/batchmap.rs

@@ -3,12 +3,6 @@ use std::fs;
 use std::path::PathBuf;
 use std::sync::Arc;
 
-use crate::error::Error;
-use crate::error::ErrorKind::{InternalError, UserDefinedError};
-use crate::servers::map as proto;
-use crate::servers::map::map_server::Map;
-use crate::servers::map::{MapRequest, MapResponse, ReadyResponse};
-use crate::shared::{self, shutdown_signal, ContainerType};
 use chrono::{DateTime, Utc};
 use tokio::sync::mpsc::channel;
 use tokio::sync::{mpsc, oneshot};
@@ -18,6 +12,13 @@ use tokio_util::sync::CancellationToken;
 use tonic::{Request, Response, Status, Streaming};
 use tracing::{debug, info};
 
+use crate::error::Error;
+use crate::error::ErrorKind::{InternalError, UserDefinedError};
+use crate::servers::map as proto;
+use crate::servers::map::map_server::Map;
+use crate::servers::map::{MapRequest, MapResponse, ReadyResponse};
+use crate::shared::{self, shutdown_signal, ContainerType};
+
 const DEFAULT_MAX_MESSAGE_SIZE: usize = 64 * 1024 * 1024;
 const DEFAULT_SOCK_ADDR: &str = "/var/run/numaflow/batchmap.sock";
 const DEFAULT_SERVER_INFO_FILE: &str = "/var/run/numaflow/mapper-server-info";

numaflow/src/lib.rs

@@ -34,9 +34,12 @@ pub mod sink;
 /// building [side input](https://numaflow.numaproj.io/user-guide/reference/side-inputs/)
 pub mod sideinput;
 
-/// batchmap is for writing the [batch map mode](https://numaflow.numaproj.io/user-guide/user-defined-functions/map/batchmap/) handlers.
+/// batchmap is for writing the map in [batch mode](https://numaflow.numaproj.io/user-guide/user-defined-functions/map/map/#batch-map-mode) handlers.
 pub mod batchmap;
 
+/// mapstream is for writing the map in [stream mode](https://numaflow.numaproj.io/user-guide/user-defined-functions/map/map/#streaming-mode) handlers.
+pub mod mapstream;
+
 mod servers;
 
 // Error handling on Numaflow SDKs!

numaflow/src/map.rs

@@ -538,6 +538,7 @@ impl<C> Drop for Server<C> {
     // same address. UnixListener doesn't implement Drop trait, so we have to manually remove the socket file.
     fn drop(&mut self) {
         let _ = fs::remove_file(&self.sock_addr);
+        let _ = fs::remove_file(&self.server_info_file);
     }
 }
 
@@ -830,6 +831,9 @@ mod tests {
         };
         tx.send(request).await.unwrap();
 
+        let resp = stream.message().await;
+        assert!(resp.is_err());
+
         // server should shut down gracefully because there was a panic in the handler.
         for _ in 0..10 {
             tokio::time::sleep(Duration::from_millis(10)).await;