[Merged by Bors] - feat: added sink connector support to SDK

Cargo.toml

@@ -49,7 +49,8 @@ members = [
     "crates/smartmodule-development-kit",
     "crates/cdk",
     "crates/cargo-builder",
-    "connector/json-test-connector"
+    "connector/json-test-connector",
+    "connector/sink-test-connector"
 ]
 
 exclude = ["smartmodule/regex-filter"]

connector/sink-test-connector/Cargo.toml

@@ -0,0 +1,19 @@
+[package]
+name = "sink-test-connector"
+version = "0.0.0"
+authors = ["Fluvio Contributors <team@fluvio.io>"]
+description = "Connector that reads data from a topic and prints to stdout"
+edition = "2021"
+publish = false
+
+
+
+[dependencies]
+async-trait = { version = "0.1", default-features = false}
+futures = { version = "0.3", default-features = false }
+anyhow = { workspace = true}
+async-std = { version = "1.12",  default-features = false, features = ["attributes"]}
+
+fluvio = { path = "../../crates/fluvio/", features = ["smartengine"]}
+fluvio-connector-common = { path = "../../crates/fluvio-connector-common/", features = ["derive"] }
+

connector/sink-test-connector/Connector.toml

@@ -0,0 +1,16 @@
+[package]
+name = "sink-test-connector"
+group = "fluvio"
+version = "0.1.0"
+apiVersion = "0.1.0"
+fluvio = "0.10.0"
+description = "Test Sink Connector"
+license = "Apache-2.0"
+
+
+[direction]
+dest = true
+
+[deployment]
+binary = "sink-test-connector"
+

connector/sink-test-connector/config-example.yaml

@@ -0,0 +1,14 @@
+version: 0.1.0
+name: my-sink-test-connector
+type: test-sink
+topic: test-topic
+parameters:
+  interval: 10  
+transforms:
+  - uses: infinyon/jolt@0.1.0
+    with:
+      spec:
+        - operation: default
+          spec:
+            data:
+              source: "sink-test-connector"

connector/sink-test-connector/src/main.rs

@@ -0,0 +1,16 @@
+mod sink;
+
+use fluvio_connector_common::{connector, ConnectorConfig, Result, consumer::ConsumerStream, Sink};
+use futures::SinkExt;
+use sink::TestSink;
+
+#[connector(sink)]
+async fn start(config: ConnectorConfig, mut stream: impl ConsumerStream) -> Result<()> {
+    let sink = TestSink::new(&config)?;
+    let mut sink = sink.connect(None).await?;
+    while let Some(item) = stream.next().await {
+        let str = String::from_utf8(item?.as_ref().to_vec())?;
+        sink.send(str).await?;
+    }
+    Ok(())
+}

connector/sink-test-connector/src/sink.rs

@@ -0,0 +1,26 @@
+use anyhow::Result;
+
+use async_trait::async_trait;
+
+use fluvio::Offset;
+use fluvio_connector_common::{ConnectorConfig, Sink, LocalBoxSink};
+
+#[derive(Debug)]
+pub(crate) struct TestSink {}
+
+impl TestSink {
+    pub(crate) fn new(_config: &ConnectorConfig) -> Result<Self> {
+        Ok(Self {})
+    }
+}
+
+#[async_trait]
+impl Sink<String> for TestSink {
+    async fn connect(self, _offset: Option<Offset>) -> Result<LocalBoxSink<String>> {
+        let unfold = futures::sink::unfold((), |_: (), record: String| async move {
+            println!("Received record: {}", record);
+            Ok::<_, anyhow::Error>(())
+        });
+        Ok(Box::pin(unfold))
+    }
+}

crates/fluvio-connector-common/src/consumer.rs

@@ -0,0 +1,46 @@
+use fluvio::{FluvioConfig, Fluvio};
+use fluvio::dataplane::record::ConsumerRecord;
+use fluvio_sc_schema::errors::ErrorCode;
+use futures::StreamExt;
+use crate::{ConnectorConfig, Result};
+use crate::ensure_topic_exists;
+use crate::smartmodule::smartmodule_vec_from_config;
+
+pub trait ConsumerStream:
+    StreamExt<Item = std::result::Result<ConsumerRecord, ErrorCode>> + std::marker::Unpin
+{
+}
+
+impl<T: StreamExt<Item = std::result::Result<ConsumerRecord, ErrorCode>> + std::marker::Unpin>
+    ConsumerStream for T
+{
+}
+
+pub async fn consumer_stream_from_config(
+    config: &ConnectorConfig,
+) -> Result<(Fluvio, impl ConsumerStream)> {
+    let mut cluster_config = FluvioConfig::load()?;
+    cluster_config.client_id = Some(format!("fluvio_connector_{}", &config.name));
+
+    let fluvio = Fluvio::connect_with_config(&cluster_config).await?;
+    ensure_topic_exists(config).await?;
+    let consumer = fluvio
+        .partition_consumer(
+            &config.topic,
+            config
+                .consumer
+                .as_ref()
+                .and_then(|c| c.partition)
+                .unwrap_or_default(),
+        )
+        .await?;
+
+    let mut builder = fluvio::ConsumerConfig::builder();
+    if let Some(smartmodules) = smartmodule_vec_from_config(config) {
+        builder.smartmodule(smartmodules);
+    }
+    let config = builder.build()?;
+    let offset = fluvio::Offset::end();
+    let stream = consumer.stream_with_config(offset, config).await?;
+    Ok((fluvio, stream))
+}

crates/fluvio-connector-common/src/lib.rs

@@ -1,6 +1,7 @@
 pub mod producer;
 pub mod smartmodule;
 pub mod monitoring;
+pub mod consumer;
 
 #[cfg(feature = "derive")]
 pub use fluvio_connector_derive::connector;
@@ -28,6 +29,13 @@ pub trait Source<'a, I> {
     async fn connect(self, offset: Option<Offset>) -> Result<LocalBoxStream<'a, I>>;
 }
 
+pub type LocalBoxSink<I> = std::pin::Pin<Box<dyn futures::Sink<I, Error = anyhow::Error>>>;
+
+#[async_trait]
+pub trait Sink<I> {
+    async fn connect(self, offset: Option<Offset>) -> Result<LocalBoxSink<I>>;
+}
+
 pub async fn ensure_topic_exists(config: &ConnectorConfig) -> Result<()> {
     let admin = fluvio::FluvioAdmin::connect().await?;
     let topics = admin

crates/fluvio-connector-common/src/smartmodule.rs

@@ -1,4 +1,4 @@
-use fluvio::FluvioConfig;
+use fluvio::{FluvioConfig, SmartModuleInvocation, SmartModuleKind};
 use crate::{ConnectorConfig, Result};
 use fluvio_smartengine::transformation::TransformationConfig;
 
@@ -30,3 +30,24 @@ pub async fn smartmodule_chain_from_config(
         Ok(None)
     }
 }
+
+pub fn smartmodule_vec_from_config(config: &ConnectorConfig) -> Option<Vec<SmartModuleInvocation>> {
+    Some(
+        config
+            .transforms
+            .as_ref()?
+            .transforms
+            .iter()
+            .map(|s| SmartModuleInvocation {
+                wasm: fluvio::SmartModuleInvocationWasm::Predefined(s.uses.clone()),
+                kind: SmartModuleKind::Generic(Default::default()),
+                params: s
+                    .with
+                    .iter()
+                    .map(|(k, v)| (k.clone(), v.clone().into()))
+                    .collect::<std::collections::BTreeMap<String, String>>()
+                    .into(),
+            })
+            .collect(),
+    )
+}

crates/fluvio-connector-derive/src/generator.rs

@@ -13,18 +13,13 @@ pub(crate) fn generate_connector(direction: ConnectorDirection, func: &Connector
 fn generate_source(func: &ConnectorFn) -> TokenStream {
     let user_fn = &func.name;
     let user_code = &func.func;
+
+    let init_and_parse_config = init_and_parse_config();
     quote! {
 
         fn main() -> ::fluvio_connector_common::Result<()> {
-            ::fluvio_connector_common::future::init_logger();
-
-            let opts = ConnectorOpt::parse();
-            ::fluvio_connector_common::tracing::info!("Reading config file from: {}", opts.config.to_string_lossy());
-
-            let config = ::fluvio_connector_common::ConnectorConfig::from_file(opts.config.as_path())?;
-            ::fluvio_connector_common::tracing::debug!("{:#?}", config);
+            #init_and_parse_config
 
-            ::fluvio_connector_common::tracing::info!("starting processing");
             ::fluvio_connector_common::future::run_block_on(async {
                 let (fluvio, producer) = ::fluvio_connector_common::producer::producer_from_config(&config).await?;
 
@@ -37,6 +32,38 @@ fn generate_source(func: &ConnectorFn) -> TokenStream {
             Ok(())
         }
 
+        #user_code
+    }
+}
+
+fn generate_sink(func: &ConnectorFn) -> TokenStream {
+    let user_fn = &func.name;
+    let user_code = &func.func;
+
+    let init_and_parse_config = init_and_parse_config();
+    quote! {
+
+        fn main() -> ::fluvio_connector_common::Result<()> {
+            #init_and_parse_config
+
+            ::fluvio_connector_common::future::run_block_on(async {
+                let (fluvio, stream) = ::fluvio_connector_common::consumer::consumer_stream_from_config(&config).await?;
+
+                let metrics = ::std::sync::Arc::new(::fluvio_connector_common::monitoring::ConnectorMetrics::new(fluvio.metrics()));
+                ::fluvio_connector_common::monitoring::init_monitoring(metrics);
+
+                #user_fn(config, stream).await
+            })?;
+
+            Ok(())
+        }
+
+        #user_code
+    }
+}
+
+fn init_and_parse_config() -> TokenStream {
+    quote! {
         #[derive(Debug)]
         pub struct ConnectorOpt {
             config: ::std::path::PathBuf,
@@ -59,10 +86,14 @@ fn generate_source(func: &ConnectorFn) -> TokenStream {
             }
         }
 
-        #user_code
-    }
-}
+        ::fluvio_connector_common::future::init_logger();
+
+        let opts = ConnectorOpt::parse();
+        ::fluvio_connector_common::tracing::info!("Reading config file from: {}", opts.config.to_string_lossy());
 
-fn generate_sink(_func: &ConnectorFn) -> TokenStream {
-    todo!() // https://github.com/infinyon/fluvio/issues/2884
+        let config = ::fluvio_connector_common::ConnectorConfig::from_file(opts.config.as_path())?;
+        ::fluvio_connector_common::tracing::debug!("{:#?}", config);
+
+        ::fluvio_connector_common::tracing::info!("starting processing");
+    }
 }

crates/fluvio-connector-derive/src/lib.rs

@@ -4,19 +4,14 @@ mod generator;
 use ast::{ConnectorDirection, ConnectorFn};
 use generator::generate_connector;
 use proc_macro::TokenStream;
-use syn::{spanned::Spanned, parse_macro_input, AttributeArgs, Error, ItemFn};
+use syn::{parse_macro_input, AttributeArgs, ItemFn};
 
 #[proc_macro_attribute]
 pub fn connector(args: TokenStream, input: TokenStream) -> TokenStream {
     let args = parse_macro_input!(args as AttributeArgs);
 
     let direction = match ConnectorDirection::from_ast(&args) {
-        Ok(ConnectorDirection::Source) => ConnectorDirection::Source,
-        Ok(ConnectorDirection::Sink) => {
-            return Error::new(args[0].span(), "Sink Connectors are not implemented yet")
-                .to_compile_error()
-                .into()
-        }
+        Ok(dir) => dir,
         Err(e) => return e.into_compile_error().into(),
     };
 

crates/fluvio-connector-derive/tests/mod.rs

@@ -1,7 +1,6 @@
 #[test]
 fn ui() {
     let t = trybuild::TestCases::new();
-    t.compile_fail("tests/ui/sink.rs");
     t.compile_fail("tests/ui/wrong_direction.rs");
     t.compile_fail("tests/ui/wrong_input_len.rs");
     t.compile_fail("tests/ui/not_async.rs");

crates/fluvio-connector-package/Cargo.toml

@@ -23,6 +23,7 @@ fluvio-smartengine = { path = "../fluvio-smartengine", default-features = false,
     "transformation",
 ] }
 fluvio-compression = { path = "../fluvio-compression" }
+fluvio-types = { path = "../fluvio-types" }
 
 tracing = { workspace = true }
 anyhow = { workspace = true }

crates/fluvio-connector-package/src/config.rs

@@ -8,6 +8,7 @@ use std::path::{PathBuf, Path};
 use std::str::FromStr;
 use std::time::Duration;
 
+use fluvio_types::PartitionId;
 use tracing::debug;
 use anyhow::Result;
 use serde::{Deserialize, Serialize};
@@ -52,7 +53,7 @@ pub struct ConnectorConfig {
 #[derive(Debug, Clone, PartialEq, Deserialize, Serialize)]
 pub struct ConsumerParameters {
     #[serde(default, skip_serializing_if = "Option::is_none")]
-    partition: Option<i32>,
+    pub partition: Option<PartitionId>,
 }
 
 #[allow(clippy::derive_partial_eq_without_eq)]