feat: add cdk hub command (#3612)

This is the second PR of a two-part feature related to #3578. The first part can be found in #3611.

Reuse the code from #3611 at `cdk hub`.



CI is expected to fail, see the comment below.

Closes #3578

crates/cdk/Cargo.toml

@@ -35,5 +35,6 @@ fluvio = { workspace = true }
 fluvio-cli-common = { workspace = true, features = ["version-cmd"] }
 fluvio-connector-deployer = { path = "../fluvio-connector-deployer"}
 fluvio-connector-package = { workspace = true,  features = ["toml"]}
+fluvio-extension-common = { workspace = true }
 fluvio-future = { workspace = true, features = ["subscriber"]}
-fluvio-hub-util = { workspace = true  }
+fluvio-hub-util = { workspace = true, features = ["connector-cmds"] }

crates/cdk/src/cmd.rs

@@ -9,6 +9,7 @@ use crate::deploy::DeployCmd;
 use crate::test::TestCmd;
 use crate::publish::PublishCmd;
 use crate::set_public::SetPublicCmd;
+use crate::hub::HubCmd;
 
 /// Connector Development Kit
 #[derive(Debug, Parser)]
@@ -19,6 +20,8 @@ pub enum CdkCommand {
     Deploy(DeployCmd),
     Publish(PublishCmd),
     Version(BasicVersionCmd),
+    #[command(subcommand)]
+    Hub(HubCmd),
 
     #[command(name = "set-public")]
     SetPublic(SetPublicCmd),
@@ -34,6 +37,7 @@ impl CdkCommand {
             CdkCommand::SetPublic(opt) => opt.process(),
             CdkCommand::Generate(opt) => opt.process(),
             CdkCommand::Version(opt) => opt.process("CDK"),
+            CdkCommand::Hub(opt) => opt.process(),
         }
     }
 }

crates/cdk/src/hub.rs

@@ -0,0 +1,26 @@
+use std::sync::Arc;
+
+use anyhow::Result;
+use clap::Parser;
+use fluvio_future::task;
+use fluvio_hub_util::cmd::{ConnectorHubListOpts, ConnectorHubDownloadOpts};
+use fluvio_extension_common::PrintTerminal;
+
+/// Work with Connectors hub
+#[derive(Debug, Parser)]
+pub enum HubCmd {
+    #[command(name = "list")]
+    List(ConnectorHubListOpts),
+    #[command(name = "download")]
+    Download(ConnectorHubDownloadOpts),
+}
+
+impl HubCmd {
+    pub fn process(self) -> Result<()> {
+        let terminal = Arc::new(PrintTerminal::new());
+        match self {
+            HubCmd::List(opt) => task::run_block_on(opt.process(terminal)),
+            HubCmd::Download(opt) => task::run_block_on(opt.process(terminal)),
+        }
+    }
+}

crates/cdk/src/main.rs

@@ -7,6 +7,7 @@ mod publish;
 mod set_public;
 
 pub(crate) mod utils;
+mod hub;
 
 fn main() -> anyhow::Result<()> {
     use clap::Parser;

crates/fluvio-cli/Cargo.toml

@@ -89,7 +89,7 @@ fluvio-package-index = { workspace = true }
 fluvio-extension-common = { workspace = true,  features = ["target"] }
 fluvio-channel = { workspace = true }
 fluvio-cli-common = { workspace = true  }
-fluvio-hub-util = { workspace = true  }
+fluvio-hub-util = { workspace = true, features = ["connector-cmds"] }
 fluvio-smartengine = { workspace = true,  features = ["transformation"]}
 fluvio-protocol = { workspace = true, features=["record","api"] }
 fluvio-smartmodule = { workspace = true  }

crates/fluvio-cli/src/client/hub/connector/mod.rs

@@ -1,7 +1,4 @@
-mod list;
-use list::ConnectorHubListOpts;
-mod download;
-use download::ConnectorHubDownloadOpts;
+use fluvio_hub_util::cmd::{ConnectorHubDownloadOpts, ConnectorHubListOpts};
 
 use std::sync::Arc;
 use std::fmt::Debug;
@@ -11,16 +8,17 @@ use anyhow::Result;
 
 use fluvio_extension_common::Terminal;
 
-use super::{get_pkg_list, get_hub_access};
+const CONNECTOR_HUB_DOWNLOAD_REDIRECT_MESSAGE: &str = r#"Connectors running on `InfinyOn Cloud` are automatically downloaded during `fluvio cloud connector create ...`.
+Connectors running locally require `cdk` to download and deploy:
+1. Install cdk: `fluvio install cdk`
+2. Download connector: `cdk hub download ...`
+3. Deploy connector: `cdk deploy start ...`"#;
 
 /// List available Connectors in the hub
 #[derive(Debug, Parser)]
 pub enum ConnectorHubSubCmd {
-    /// List all available SmartConnectors
     #[command(name = "list")]
     List(ConnectorHubListOpts),
-
-    /// Download SmartConnector to the local folder
     #[command(name = "download")]
     Download(ConnectorHubDownloadOpts),
 }
@@ -29,7 +27,56 @@ impl ConnectorHubSubCmd {
     pub async fn process<O: Terminal + Debug + Send + Sync>(self, out: Arc<O>) -> Result<()> {
         match self {
             ConnectorHubSubCmd::List(opts) => opts.process(out).await,
-            ConnectorHubSubCmd::Download(opts) => opts.process(out).await,
+            ConnectorHubSubCmd::Download(_) => {
+                out.println(CONNECTOR_HUB_DOWNLOAD_REDIRECT_MESSAGE);
+                Ok(())
+            }
+        }
+    }
+}
+
+#[cfg(test)]
+mod test {
+    use std::sync::{RwLock, Arc};
+
+    use clap::Parser;
+    use fluvio_extension_common::Terminal;
+    use fluvio_future::task::run_block_on;
+
+    use crate::client::hub::connector::CONNECTOR_HUB_DOWNLOAD_REDIRECT_MESSAGE;
+
+    use super::ConnectorHubSubCmd;
+
+    #[derive(Default, Debug)]
+    struct MockTerminal(Arc<RwLock<String>>);
+    impl Terminal for MockTerminal {
+        fn print(&self, msg: &str) {
+            self.0
+                .write()
+                .expect("could not print to MockTerminal")
+                .push_str(msg);
+        }
+
+        fn println(&self, msg: &str) {
+            self.0
+                .write()
+                .expect("could not println to MockTerminal")
+                .push_str(&format!("{msg}\n"));
         }
     }
+
+    #[test]
+    fn test_calling_fluvio_hub_download_displays_a_help_message() {
+        let terminal = Arc::new(MockTerminal::default());
+        let cmd =
+            ConnectorHubSubCmd::parse_from(["conn", "download", "infinyon/[email protected]"]);
+
+        assert!(matches!(cmd, ConnectorHubSubCmd::Download(_)));
+
+        run_block_on(cmd.process(terminal.clone())).expect("command failed to process");
+
+        let x = terminal.0.read().unwrap();
+        let x = x.as_str().trim_end();
+        assert_eq!(x, CONNECTOR_HUB_DOWNLOAD_REDIRECT_MESSAGE);
+    }
 }

crates/fluvio-cli/src/client/hub/mod.rs

@@ -59,41 +59,3 @@ mod cmd {
         }
     }
 }
-
-use anyhow::Result;
-
-use fluvio_hub_util as hubutil;
-use hubutil::HubAccess;
-use hubutil::PackageListMeta;
-
-use crate::{CliError};
-
-pub(crate) fn get_hub_access(remote: &Option<String>) -> Result<HubAccess> {
-    let access = HubAccess::default_load(remote).map_err(|_| {
-        CliError::HubError("missing access credentials, try 'fluvio cloud login'".into())
-    })?;
-    Ok(access)
-}
-
-pub(crate) async fn get_pkg_list(
-    endpoint: &str,
-    remote: &Option<String>,
-) -> Result<PackageListMeta> {
-    use hubutil::http;
-
-    let access = get_hub_access(remote)?;
-
-    let action_token = access.get_list_token().await.map_err(|_| {
-        CliError::HubError("rejected access credentials, try 'fluvio cloud login'".into())
-    })?;
-    let url = format!("{}/{endpoint}", &access.remote);
-    let mut res = http::get(&url)
-        .header("Authorization", &action_token)
-        .await
-        .map_err(|e| CliError::HubError(format!("list api access error {e}")))?;
-    let pl: PackageListMeta = res
-        .body_json()
-        .await
-        .map_err(|e| CliError::HubError(format!("list api data parse error {e}")))?;
-    Ok(pl)
-}

crates/fluvio-cli/src/client/hub/smartmodule/download.rs

@@ -16,10 +16,10 @@ use fluvio_extension_common::Terminal;
 use fluvio_extension_common::target::ClusterTarget;
 use fluvio_hub_util as hubutil;
 use hubutil::HubAccess;
+use hubutil::cmd::get_hub_access;
 
 use crate::CliError;
 use crate::client::cmd::ClientCmd;
-use crate::client::hub::get_hub_access;
 
 /// Download a SmartModule from the hub
 #[derive(Debug, Parser)]

crates/fluvio-cli/src/client/hub/smartmodule/list.rs

@@ -6,11 +6,10 @@ use anyhow::Result;
 
 use fluvio_extension_common::Terminal;
 use fluvio_hub_util::HUB_API_LIST_META;
+use fluvio_hub_util::cmd::get_pkg_list;
 
 use crate::common::OutputFormat;
 
-use super::get_pkg_list;
-
 /// List available SmartModules in the hub
 #[derive(Debug, Parser)]
 pub struct SmartModuleHubListOpts {

crates/fluvio-cli/src/client/hub/smartmodule/mod.rs

@@ -14,8 +14,6 @@ use fluvio::Fluvio;
 
 use crate::client::ClientCmd;
 
-use super::get_pkg_list;
-
 /// List available SmartModules in the hub
 #[derive(Debug, Parser)]
 pub enum SmartModuleHubSubCmd {

crates/fluvio-hub-util/Cargo.toml

@@ -8,6 +8,14 @@ authors = ["fluvio.io"]
 repository = "https://github.com/infinyon/fluvio"
 publish = false
 
+[features]
+connector-cmds = [
+    "dep:clap",
+    "dep:current_platform",
+    "dep:comfy-table",
+    "dep:fluvio-extension-common",
+]
+
 [dependencies]
 anyhow = { workspace = true }
 cargo_toml = { workspace = true }
@@ -35,10 +43,14 @@ tracing = { workspace = true }
 thiserror = { workspace = true }
 url = { workspace = true }
 wasmparser = { workspace = true }
+current_platform = { workspace = true, optional = true }
+clap = { workspace = true, optional = true }
+comfy-table = { workspace = true, optional = true }
 
 fluvio-future = { workspace = true, features = ["fixture", "task"] }
 fluvio-hub-protocol = { path = "../fluvio-hub-protocol" }
 fluvio-types = { workspace = true }
+fluvio-extension-common = { workspace = true,  optional = true }
 
 [dev-dependencies]
 tracing-subscriber = { workspace = true,  features = ["env-filter", "fmt"] }

crates/fluvio-cli/src/client/hub/connector/download.rs → crates/fluvio-hub-util/src/cmd/download.rs

@@ -3,15 +3,17 @@ use std::sync::Arc;
 use std::fmt::Debug;
 
 use clap::Parser;
-use anyhow::Result;
+use anyhow::{Result, anyhow};
 
 use fluvio_extension_common::Terminal;
 
-use crate::error::CliError;
+use crate::{cli_pkgname_to_filename, cli_conn_pkgname_to_url, get_package};
 
 use super::get_hub_access;
 
+/// Download SmartConnector to the local folder
 #[derive(Debug, Parser)]
+#[command(arg_required_else_help = true)]
 pub struct ConnectorHubDownloadOpts {
     /// SmartConnector name: e.g. infinyon/[email protected]
     #[arg(value_name = "name", required = true)]
@@ -37,10 +39,8 @@ impl ConnectorHubDownloadOpts {
         let access = get_hub_access(&self.remote)?;
 
         let package_name = self.package_name;
-        let file_name = fluvio_hub_util::cli_pkgname_to_filename(&package_name).map_err(|_| {
-            CliError::HubError(format!(
-                "invalid package name format {package_name}, is it the form infinyon/[email protected]"
-            ))
+        let file_name = cli_pkgname_to_filename(&package_name).map_err(|_| {
+            anyhow!("invalid package name format {package_name}, is it the form infinyon/[email protected]")
         })?;
 
         let file_path = if let Some(mut output) = self.output {
@@ -51,28 +51,18 @@ impl ConnectorHubDownloadOpts {
         } else {
             PathBuf::from(file_name)
         };
-        println!(
-            "downloading {package_name} to {}",
-            file_path.to_string_lossy()
-        );
+        let path = file_path.to_string_lossy();
+        println!("downloading {package_name} to {path}");
 
-        let url =
-            fluvio_hub_util::cli_conn_pkgname_to_url(&package_name, &access.remote, &self.target)
-                .map_err(|_| CliError::HubError(format!("invalid pkgname {package_name}")))?;
+        let url = cli_conn_pkgname_to_url(&package_name, &access.remote, &self.target)
+            .map_err(|_| anyhow!("invalid pkgname {package_name}"))?;
 
-        let data = fluvio_hub_util::get_package(&url, &access)
+        let data = get_package(&url, &access)
             .await
-            .map_err(|err| {
-                CliError::HubError(format!(
-                    "downloading {package_name} failed\nHub error: {err}"
-                ))
-            })?;
+            .map_err(|err| anyhow!("downloading {package_name} failed\nServer: {err}"))?;
 
-        std::fs::write(file_path, data).map_err(|err| {
-            CliError::Other(format!(
-                "unable to write downloaded package to the disk: {err}"
-            ))
-        })?;
+        std::fs::write(file_path, data)
+            .map_err(|err| anyhow!("unable to write downloaded package to the disk: {err}"))?;
         println!("... downloading complete");
         Ok(())
     }