[Merged by Bors] - feat: fluvio connector support update

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

@@ -86,7 +86,12 @@ impl ConnectorConfig {
         let mut file = File::open(path.into())?;
         let mut contents = String::new();
         file.read_to_string(&mut contents)?;
-        let mut connector_config: Self = serde_yaml::from_str(&contents)?;
+        ConnectorConfig::config_from_str(&contents)
+    }
+
+    /// Only parses the meta section of the config
+    pub fn config_from_str(config_str: &str) -> Result<Self> {
+        let mut connector_config: Self = serde_yaml::from_str(config_str)?;
         connector_config.normalize_batch_size()?;
 
         debug!("Using connector config {connector_config:#?}");
@@ -632,4 +637,16 @@ mod tests {
             BTreeMap::from([("regex".to_string(), "\\w".into())])
         );
     }
+
+    #[test]
+    fn sample_yaml_test_files() {
+        let testfiles = vec!["tests/sample-http.yaml", "tests/sample-mqtt.yaml"];
+
+        for tfile in testfiles {
+            let res = ConnectorConfig::from_file(tfile);
+            assert!(res.is_ok(), "failed to load {tfile}");
+            let connector_cfg = res.unwrap();
+            println!("{tfile}: {connector_cfg:?}");
+        }
+    }
 }

crates/fluvio-connector-package/tests/sample-http.yaml

@@ -0,0 +1,10 @@
+meta:
+  version: latest
+  name: cat-facts
+  type: http-source
+  topic: cat-facts
+  create-topic: true
+http:
+  endpoint: "https://catfact.ninja/fact"
+  interval: 10s  
+

crates/fluvio-connector-package/tests/sample-mqtt.yaml

@@ -0,0 +1,15 @@
+meta:
+  version: latest
+  name: my-mqtt-connector
+  type: mqtt-source
+  topic: mqtt-topic
+  create-topic: true
+mqtt:
+  url: "mqtt://test.mosquitto.org/"
+  topic: "mqtt-to-fluvio"
+  client_id: "my_mqtt"
+  timeout:
+    secs: 30
+    nanos: 0
+  payload_output_type: json
+

crates/fluvio-hub-util/src/utils.rs

@@ -81,6 +81,10 @@ pub fn cli_pkgname_to_filename(pkgname: &str) -> Result<String> {
 /// returns recommended name and data
 pub async fn get_package(pkgurl: &str, access: &HubAccess) -> Result<Vec<u8>> {
     let actiontoken = access.get_download_token().await?;
+    get_package_with_token(pkgurl, &actiontoken).await
+}
+
+pub async fn get_package_with_token(pkgurl: &str, actiontoken: &str) -> Result<Vec<u8>> {
     let mut resp = surf::get(pkgurl)
         .header("Authorization", actiontoken)
         .await