feat: string interpolation

connector/sink-test-connector/Connector.toml

@@ -13,6 +13,3 @@ dest = true
 
 [deployment]
 binary = "sink-test-connector"
-
-[secret.TEST_API_KEY]
-type = "env"

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

@@ -5,10 +5,12 @@ meta:
   topic: test-topic
   secrets:
     - name: TEST_API_KEY
+    - name: TEST_API_CLIENT_ID
 custom:
   api_key:
     secret:
       name: TEST_API_KEY
+  client_id: ${{ secrets.TEST_API_CLIENT_ID }}
 transforms:
   - uses: infinyon/jolt@0.1.0
     with:

connector/sink-test-connector/secrets.txt

@@ -1 +1,2 @@
 TEST_API_KEY=test value
+TEST_API_CLIENT_ID=client1

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

@@ -18,4 +18,5 @@ async fn start(config: CustomConfig, mut stream: impl ConsumerStream) -> Result<
 #[connector(config)]
 struct CustomConfig {
     api_key: SecretString,
+    client_id: String,
 }

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

@@ -14,7 +14,9 @@ impl TestSink {
     pub(crate) fn new(config: &CustomConfig) -> Result<Self> {
         debug!(?config.api_key);
         let resolved = config.api_key.resolve()?;
+        let resolved_2 = config.client_id.clone();
         debug!(resolved);
+        debug!(resolved_2);
         Ok(Self {})
     }
 }

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

@@ -1,15 +1,14 @@
-pub use fluvio_connector_package::config::ConnectorConfig;
-use tracing::trace;
+use std::io::Read;
 
-use std::{path::PathBuf, fs::File};
-
-use serde::de::DeserializeOwned;
 use anyhow::{Result, Context};
+use serde::de::DeserializeOwned;
 use serde_yaml::Value;
+use tracing::trace;
+
+pub use fluvio_connector_package::config::ConnectorConfig;
 
-pub fn value_from_file<P: Into<PathBuf>>(path: P) -> Result<Value> {
-    let file = File::open(path.into())?;
-    serde_yaml::from_reader(file).context("unable to parse config file into YAML")
+pub fn value_from_reader<R: Read>(reader: R) -> Result<Value> {
+    serde_yaml::from_reader(reader).context("unable to parse config file into YAML")
 }
 
 pub fn from_value<T: DeserializeOwned>(value: Value, root: Option<&str>) -> Result<T> {

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

@@ -4,6 +4,7 @@ pub mod monitoring;
 pub mod consumer;
 pub mod config;
 
+pub use fluvio_connector_package::render_config_str;
 pub use fluvio_connector_package::secret;
 
 #[cfg(feature = "derive")]

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

@@ -113,11 +113,15 @@ fn init_and_parse_config(config_type_path: &Path) -> TokenStream {
 
         ::fluvio_connector_common::tracing::info!("Reading config file from: {}", opts.config.to_string_lossy());
 
-        let config_value = ::fluvio_connector_common::config::value_from_file(opts.config.as_path())?;
-        ::fluvio_connector_common::tracing::trace!("{:#?}", config_value);
+        let config_str = ::std::fs::read_to_string(opts.config.as_path())?;
+        ::fluvio_connector_common::tracing::debug!(%config_str);
+
+        /// Resolve any secrets/env in the config
+        let config_str_resolved =::fluvio_connector_common::render_config_str(&config_str)?;
+
+        let config_value = ::fluvio_connector_common::config::value_from_reader(config_str_resolved.as_bytes())?;
 
         let common_config = ::fluvio_connector_common::config::ConnectorConfig::from_value(config_value.clone())?;
-        ::fluvio_connector_common::tracing::debug!("{:#?}", common_config);
 
         let user_config: #config_type_path = ::fluvio_connector_common::config::from_value(config_value, Some(#config_type_path::__config_name()))?;
 
@@ -129,6 +133,7 @@ pub(crate) fn generate_connector_config(item: &ConnectorConfigStruct) -> TokenSt
     let config_struct = item.item_struct;
     let ident = &item.item_struct.ident;
     let config_name = &item.config_name;
+
     quote! {
         #[derive(serde::Deserialize)]
         #config_struct
@@ -137,6 +142,7 @@ pub(crate) fn generate_connector_config(item: &ConnectorConfigStruct) -> TokenSt
             pub fn __config_name() -> &'static str {
                 #config_name
             }
+
         }
     }
 }

crates/fluvio-connector-package/Cargo.toml

@@ -14,10 +14,12 @@ default = ["toml"]
 anyhow = { workspace = true }
 bytesize = { workspace = true}
 humantime-serde = "1.1.1"
+minijinja = { version = "0.32", default-features = false, features = ["custom_syntax"] }
 openapiv3 = { git = "https://github.com/galibey/openapiv3", rev = "bdd22f046d2bc19ede257504645d31f835545222", default-features = false }
 once_cell = { workspace = true }
 serde = { workspace = true, default-features = false, features = ["derive"] }
 serde_yaml = { workspace = true }
+serde_json = { workspace = true }
 toml = { workspace = true , optional = true, features = ["display", "parse", "preserve_order"] }
 tracing = { workspace = true }
 

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

@@ -1,3 +1,6 @@
 pub mod metadata;
 pub mod config;
 pub mod secret;
+mod render;
+
+pub use render::render_config_str;

crates/fluvio-connector-package/src/render/context.rs

@@ -0,0 +1,69 @@
+use std::collections::HashMap;
+
+use serde::Serialize;
+
+use crate::{secret::SecretStore};
+
+/// Context for the template engine
+/// This is the data that is available to the template engine
+/// when it is rendering the template.
+#[derive(Serialize, Default)]
+pub(crate) struct Context(pub(crate) HashMap<&'static str, serde_json::Value>);
+
+/// ContextStore is a trait that allows for adding additional
+/// context to the template engine.
+pub(crate) trait ContextStore {
+    fn context_values(&self) -> anyhow::Result<serde_json::Value>;
+
+    fn context_name(&self) -> &'static str;
+}
+impl dyn ContextStore {
+    pub(crate) fn add_to_context(&self, context: &mut Context) -> anyhow::Result<()> {
+        let value = self.context_values()?;
+        context.0.insert(self.context_name(), value);
+        Ok(())
+    }
+}
+impl ContextStore for &dyn SecretStore {
+    fn context_values(&self) -> anyhow::Result<serde_json::Value> {
+        let mut values = serde_json::Map::new();
+        for secret_name in self.keys()? {
+            let secret = self.read(&secret_name)?;
+            values.insert(secret_name, serde_json::to_value(secret)?);
+        }
+        Ok(values.into())
+    }
+
+    fn context_name(&self) -> &'static str {
+        "secrets"
+    }
+}
+
+#[cfg(test)]
+mod test {
+
+    use super::ContextStore;
+
+    pub struct TestStore;
+
+    impl ContextStore for TestStore {
+        fn context_values(&self) -> anyhow::Result<serde_json::Value> {
+            Ok(serde_json::json!({"test": "value"}))
+        }
+
+        fn context_name(&self) -> &'static str {
+            "test"
+        }
+    }
+
+    #[test]
+    fn test_context_store() {
+        let store = &TestStore as &dyn ContextStore;
+        let mut context = super::Context::default();
+        store.add_to_context(&mut context).unwrap();
+        assert_eq!(
+            context.0.get("test").unwrap(),
+            &serde_json::json!({"test": "value"})
+        );
+    }
+}