From f08c65ee18f078a894a7965b38952cf58ee4bb64 Mon Sep 17 00:00:00 2001
From: Esteban Borai <estebanborai@gmail.com>
Date: Fri, 27 Sep 2024 10:34:47 -0300
Subject: [PATCH] feat(conn-pkg): use types for compression in conn pkg (#4197)

* feat: use types for compression in conn pkg

* fix: use ref for compression enum

* feat: update compression instances

* fix: version

* feat: use serde lowercase

* fix: clippy

* revert: fluvio crate updates
---
 Cargo.lock                                        | 3 +--
 crates/fluvio-connector-common/src/producer.rs    | 4 ++--
 crates/fluvio-connector-package/Cargo.toml        | 1 -
 crates/fluvio-connector-package/src/config/mod.rs | 2 +-
 crates/fluvio-types/Cargo.toml                    | 2 +-
 crates/fluvio-types/src/compression.rs            | 5 ++++-
 6 files changed, 9 insertions(+), 8 deletions(-)

diff --git a/Cargo.lock b/Cargo.lock
index 4c89091fa7..5bdcec4f9f 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -2388,7 +2388,7 @@ dependencies = [
 
 [[package]]
 name = "fluvio"
-version = "0.23.2"
+version = "0.23.3"
 dependencies = [
  "anyhow",
  "async-channel 1.9.0",
@@ -2714,7 +2714,6 @@ dependencies = [
  "anyhow",
  "bytesize",
  "bytesize-serde",
- "fluvio-compression",
  "fluvio-controlplane-metadata",
  "fluvio-smartengine",
  "fluvio-types",
diff --git a/crates/fluvio-connector-common/src/producer.rs b/crates/fluvio-connector-common/src/producer.rs
index 9147e2bd79..9f4d8b52f9 100644
--- a/crates/fluvio-connector-common/src/producer.rs
+++ b/crates/fluvio-connector-common/src/producer.rs
@@ -18,8 +18,8 @@ pub async fn producer_from_config(config: &ConnectorConfig) -> Result<(Fluvio, T
         };
 
         // Compression
-        if let Some(compression) = producer_params.compression {
-            config_builder = config_builder.compression(compression)
+        if let Some(compression) = &producer_params.compression {
+            config_builder = config_builder.compression(compression.clone())
         };
 
         // Batch size
diff --git a/crates/fluvio-connector-package/Cargo.toml b/crates/fluvio-connector-package/Cargo.toml
index dacfef0ff1..58036b60d2 100644
--- a/crates/fluvio-connector-package/Cargo.toml
+++ b/crates/fluvio-connector-package/Cargo.toml
@@ -28,7 +28,6 @@ fluvio-controlplane-metadata = { path = "../fluvio-controlplane-metadata/", defa
 fluvio-smartengine = { path = "../fluvio-smartengine", default-features = false, features = [
     "transformation",
 ] }
-fluvio-compression = { path = "../fluvio-compression" }
 fluvio-types = { path = "../fluvio-types" }
 bytesize-serde = "0.2.1"
 
diff --git a/crates/fluvio-connector-package/src/config/mod.rs b/crates/fluvio-connector-package/src/config/mod.rs
index 6666c2bccd..47eda45bce 100644
--- a/crates/fluvio-connector-package/src/config/mod.rs
+++ b/crates/fluvio-connector-package/src/config/mod.rs
@@ -15,8 +15,8 @@ pub use bytesize::ByteSize;
 
 pub use fluvio_controlplane_metadata::topic::config as topic_config;
 pub use fluvio_smartengine::transformation::TransformationStep;
-pub use fluvio_compression::Compression;
 pub use fluvio_types::PartitionId;
+pub use fluvio_types::compression::Compression;
 
 use crate::metadata::Direction;
 
diff --git a/crates/fluvio-types/Cargo.toml b/crates/fluvio-types/Cargo.toml
index 80c49e8096..c964d73831 100644
--- a/crates/fluvio-types/Cargo.toml
+++ b/crates/fluvio-types/Cargo.toml
@@ -14,7 +14,7 @@ events = ["event-listener"]
 event-listener = { workspace = true,  optional = true }
 thiserror = { workspace = true }
 tracing = { workspace = true }
-serde = { workspace = true }
+serde = { workspace = true, features = ["derive"], default-features = false }
 toml = { workspace = true, features = ["display", "preserve_order", "parse"] }
 
 [dev-dependencies]
diff --git a/crates/fluvio-types/src/compression.rs b/crates/fluvio-types/src/compression.rs
index 47904effe5..816d07dee4 100644
--- a/crates/fluvio-types/src/compression.rs
+++ b/crates/fluvio-types/src/compression.rs
@@ -1,4 +1,7 @@
-#[derive(Clone, Debug, Eq, PartialEq)]
+use serde::{Serialize, Deserialize};
+
+#[derive(Clone, Debug, Deserialize, Eq, Serialize, PartialEq)]
+#[serde(rename_all = "lowercase")]
 pub enum Compression {
     None,
     Gzip,