Add storage backup support for identity function

Author: HoKim98

dash/pipe/functions/identity/Cargo.toml

@@ -17,4 +17,6 @@ repository = "https://github.com/ulagbulag/OpenARK"
 dash-pipe-provider = { path = "../../provider" }
 
 anyhow = { workspace = true }
+clap = { workspace = true }
+serde = { workspace = true }
 serde_json = { workspace = true }

dash/pipe/functions/identity/src/main.rs

@@ -1,11 +1,41 @@
+use std::sync::Arc;
+
 use anyhow::Result;
-use dash_pipe_provider::{PipeEngine, PipeMessages};
+use clap::{ArgAction, Parser};
+use dash_pipe_provider::{PipeEngine, PipeMessage, PipeMessages, PipePayload};
+use serde::{Deserialize, Serialize};
 use serde_json::Value;
 
 fn main() {
     PipeEngine::from_env().loop_forever(tick)
 }
 
-async fn tick(input: PipeMessages<Value>) -> Result<PipeMessages<Value>> {
-    Ok(input)
+#[derive(Clone, Debug, Parser, Serialize, Deserialize)]
+pub struct FunctionArgs {
+    #[arg(long, env = "PIPE_IDENTITY_WRITE_TO_PERSISTENT_STORAGE", action = ArgAction::SetTrue)]
+    #[serde(default)]
+    write_to_persistent_storage: Option<bool>,
+}
+
+async fn tick(
+    args: Arc<FunctionArgs>,
+    input: PipeMessages<Value>,
+) -> Result<Option<PipeMessages<Value>>> {
+    match args.write_to_persistent_storage {
+        Some(true) => Ok(Some(match input {
+            PipeMessages::Single(message) => PipeMessages::Single(pack_payload(message)?),
+            PipeMessages::Batch(messages) => PipeMessages::Batch(
+                messages
+                    .into_iter()
+                    .map(pack_payload)
+                    .collect::<Result<_>>()?,
+            ),
+        })),
+        Some(false) | None => Ok(Some(input)),
+    }
+}
+
+fn pack_payload(mut message: PipeMessage<Value>) -> Result<PipeMessage<Value>> {
+    message.payloads = vec![PipePayload::new("test".into(), message.to_json_bytes()?)];
+    Ok(message)
 }

dash/pipe/functions/loader-webcam/Cargo.toml

@@ -17,3 +17,5 @@ repository = "https://github.com/ulagbulag/OpenARK"
 dash-pipe-provider = { path = "../../provider" }
 
 anyhow = { workspace = true }
+clap = { workspace = true }
+serde = { workspace = true }

dash/pipe/functions/loader-webcam/src/main.rs

@@ -1,12 +1,22 @@
+use std::sync::Arc;
+
 use anyhow::Result;
+use clap::Parser;
 use dash_pipe_provider::{PipeEngine, PipeMessages};
+use serde::{Deserialize, Serialize};
 
 fn main() {
     PipeEngine::from_env().loop_forever(tick)
 }
 
-async fn tick(input: PipeMessages<String>) -> Result<PipeMessages<String>> {
+#[derive(Clone, Debug, Parser, Serialize, Deserialize)]
+pub struct FunctionArgs {}
+
+async fn tick(
+    _args: Arc<FunctionArgs>,
+    input: PipeMessages<String>,
+) -> Result<Option<PipeMessages<String>>> {
     // TODO: to be implemented
     dbg!(&input);
-    Ok(input)
+    Ok(Some(input))
 }