feat: Add prefix and suffix and reorganize configuration file structure

README.md

@@ -34,11 +34,21 @@ http:
 | method               | POST                       | String          | POST, PUT                                         |
 | endpoint             | -                          | String          | HTTP URL endpoint                                 |
 | headers              | -                          | Array\<String\> | Request header(s) "Key:Value" pairs               |
-| params               | -                          | Array\<String\> | Dynamic URL parameters gathered from a json response. Set any key from a json message that you want to include into the URL as a parameter. You can also replace the parameter name by adding ": <param name> in the end of the string, like "json_key : new_key_name"|
+| params               | -                          | Array\<Mapping\>| Dynamic URL parameters gathered from a json response|
 | user-agent           | `fluvio/http-sink 0.2.2`   | String          | Request user-agent                                |
 | http_request_timeout | 1s                         | String          | HTTP Request Timeout                              |
 | http_connect_timeout | 15s                        | String          | HTTP Connect Timeout                              |
 
+#### Parameter Configuration
+| Option               | default                    | type            | description                                       |
+| :--------------------| :--------------------------| :-------------- | :-------------------------------------------------|
+| key                  | -                          | String          | The JSON key to get the value for the parameter   |
+| replace              | -                          | String          | Parameter's new name                              |
+| prefix               | -                          | String          | String to be added before the value                |
+| suffix               | -                          | String          | String to be added after the value                |
+
+ Set any key from a json message that you want to include into the URL as a parameter. You can also replace the parameter name by adding ": <param name> in the end of the string, like "json_key : new_key_name"
+
 > By default HTTP headers will use `Content-Type: text/html` unless anothed value
 > is provided to the Headers configuration.
 

crates/http-sink/Cargo.toml

@@ -18,3 +18,4 @@ humantime-serde = { version = "1.1", default-features = false  }
 fluvio = { workspace = true }
 fluvio-connector-common = { workspace = true, features = ["derive"]}
 serde_json = {workspace = true}
+urlencoding = "2.1.3"

crates/http-sink/src/config.rs

@@ -1,6 +1,7 @@
 use std::time::Duration;
 
 use fluvio_connector_common::connector;
+use serde::Deserialize;
 use url::Url;
 
 const DEFAULT_USER_AGENT: &str = concat!("fluvio/http-sink ", env!("CARGO_PKG_VERSION"));
@@ -35,7 +36,15 @@ pub(crate) struct HttpConfig {
 
     //HTTP Parameters that can be gattered from a Message if the message is a json file
     #[serde(default = "default_http_params")]
-    pub params: Vec<String>,
+    pub params: Vec<Parameter>,
+}
+
+#[derive(Deserialize, Debug, Clone)]
+pub struct Parameter{
+    pub key: String,
+    pub replace: Option<String>,
+    pub prefix: Option<String>,
+    pub suffix: Option<String>
 }
 
 #[inline]
@@ -63,6 +72,6 @@ fn default_http_headers() -> Vec<String> {
     DEFAULT_HTTP_HEADERS.map(String::from).into_iter().collect()
 }
 
-fn default_http_params () -> Vec<String>{
+fn default_http_params () -> Vec<Parameter>{
     vec![]
 }

crates/http-sink/src/sink.rs

@@ -3,10 +3,10 @@ use std::collections::HashMap;
 use anyhow::{anyhow, Result};
 use async_trait::async_trait;
 use reqwest::{Client, RequestBuilder};
-
+use urlencoding::encode;
 use fluvio::Offset;
 use fluvio_connector_common::{tracing, LocalBoxSink, Sink};
-use crate::HttpConfig;
+use crate::{config::Parameter, HttpConfig};
 
 #[derive(Debug)]
 pub(crate) struct HttpSink {
@@ -15,7 +15,7 @@ pub(crate) struct HttpSink {
 #[derive(Debug)]
 struct Body{
     request: RequestBuilder,
-    params: Vec<String>,
+    params: Vec<Parameter>,
 }
 impl Clone for Body {
     fn clone(&self) -> Self {
@@ -55,19 +55,16 @@ impl Sink<String> for HttpSink {
                 if params.len() > 0 {
                     if let Ok(json_message) = serde_json::from_str::<HashMap<String, serde_json::Value>>(&record){
                         for param in params.into_iter() {
-                            let (key, replace_key) = match param.split_once(':') {
-                                Some(map) =>{
-                                    (
-                                        map.0.trim().to_owned(),
-                                        map.1.trim().to_owned()
-                                    )
+                            let key = param.replace.unwrap_or(param.key.clone());
+                            if json_message.contains_key(&param.key){
+                                let mut value = json_message.get(&param.key).unwrap().to_string();
+                                if let Some(prefix) = param.prefix{
+                                    value = prefix + &value;
+                                }
+                                if let Some(suffix) = param.suffix{
+                                    value = value + &suffix;
                                 }
-                                None => (param.clone(), param.clone())
-                            };
-                            if json_message.contains_key(&key){
-                                tracing::info!("Using URL parameter: {key}");
-                                let value = json_message.get(&key).unwrap().to_string();
-                                body.request = body.request.query(&[(replace_key, value)]);
+                                body.request = body.request.query(&[(encode(&key), encode(&value))]);
 
                             }
                         }