Add ability to configure topic producer in node client (#567)

* fix(ci): use ubuntu 22 for compatibility

* feat: make topic producer configurable in node client

* feat: expose use_spu_local_address in node client

Author: arpandaze

.github/workflows/publish.yml

@@ -12,15 +12,15 @@ jobs:
     runs-on: ${{ matrix.os }}
     strategy:
       matrix:
-        os: [ubuntu-latest]
-        node: ["18.13"]
+        os: [ubuntu-22.04]
+        node: ["20"]
         target: [x86_64-unknown-linux-gnu]
         include:
           - os: macos-latest
             target: x86_64-apple-darwin
           - os: macos-latest
             target: aarch64-apple-darwin
-          - os: ubuntu-latest
+          - os: ubuntu-22.04
             target: x86_64-unknown-linux-gnu
     steps:
       - uses: actions/cache@v4

src/connect.rs

@@ -5,10 +5,18 @@ use crate::error::FluvioErrorJS;
 use crate::fluvio::FluvioJS;
 
 #[node_bindgen()]
-async fn connect(host_addr: Option<String>) -> Result<FluvioJS, FluvioErrorJS> {
+async fn connect(
+    host_addr: Option<String>,
+    use_spu_local_address: Option<bool>,
+) -> Result<FluvioJS, FluvioErrorJS> {
     match host_addr {
         Some(host) => {
-            let config = FluvioConfig::new(host);
+            let mut config = FluvioConfig::new(host);
+
+            if let Some(use_spu_local_address) = use_spu_local_address {
+                config.use_spu_local_address = use_spu_local_address;
+            }
+
             let socket = Fluvio::connect_with_config(&config).await?;
             Ok(FluvioJS::from(socket))
         }

src/fluvio.rs

@@ -1,9 +1,15 @@
+use std::time::Duration;
+
 use crate::CLIENT_NOT_FOUND_ERROR_MSG;
 use crate::admin::FluvioAdminJS;
 use crate::consumer::PartitionConsumerJS;
 use crate::producer::TopicProducerJS;
 use crate::error::FluvioErrorJS;
 
+use fluvio::TopicProducerConfig;
+use fluvio::Compression;
+use fluvio::TopicProducerConfigBuilder;
+use node_bindgen::core::JSValue;
 use tracing::debug;
 
 use fluvio::Fluvio;
@@ -14,6 +20,9 @@ use node_bindgen::core::NjError;
 use node_bindgen::core::val::JsEnv;
 use node_bindgen::sys::napi_value;
 use node_bindgen::core::JSClass;
+use node_bindgen::core::val::JsObject;
+
+use crate::optional_property;
 
 impl From<Fluvio> for FluvioJS {
     fn from(inner: Fluvio) -> Self {
@@ -82,4 +91,83 @@ impl FluvioJS {
             Err(FluvioErrorJS::new(CLIENT_NOT_FOUND_ERROR_MSG.to_owned()))
         }
     }
+
+    #[node_bindgen]
+    async fn topic_producer_with_config(
+        &mut self,
+        topic: String,
+        config: TopicProducerConfigWrapper,
+    ) -> Result<TopicProducerJS, FluvioErrorJS> {
+        let config = config.inner;
+
+        if let Some(client) = &mut self.inner {
+            Ok(TopicProducerJS::from(
+                client.topic_producer_with_config(topic, config).await?,
+            ))
+        } else {
+            Err(FluvioErrorJS::new(CLIENT_NOT_FOUND_ERROR_MSG.to_owned()))
+        }
+    }
+}
+
+pub struct TopicProducerConfigWrapper {
+    pub inner: TopicProducerConfig,
+}
+
+impl JSValue<'_> for TopicProducerConfigWrapper {
+    fn convert_to_rust(env: &JsEnv, js_value: napi_value) -> Result<Self, NjError> {
+        debug!("converting topic producer config from JS");
+        if let Ok(js_obj) = env.convert_to_rust::<JsObject>(js_value) {
+            let mut builder = TopicProducerConfigBuilder::default();
+
+            // Extract compression if provided
+            if let Some(compression_str) = optional_property!("compression", String, js_obj) {
+                let compression = match compression_str.as_str() {
+                    "none" => Compression::None,
+                    "gzip" => Compression::Gzip,
+                    "snappy" => Compression::Snappy,
+                    "lz4" => Compression::Lz4,
+                    "zstd" => Compression::Zstd,
+                    _ => {
+                        return Err(NjError::Other(format!(
+                            "Invalid compression type: {}",
+                            compression_str
+                        )))
+                    }
+                };
+                builder.compression(compression);
+            }
+
+            // Extract max_request_size if provided
+            if let Some(max_request_size) = optional_property!("maxRequestSize", i64, js_obj) {
+                builder.max_request_size(max_request_size as usize);
+            }
+
+            // Extract batch_size if provided
+            if let Some(batch_size) = optional_property!("batchSize", i64, js_obj) {
+                builder.batch_size(batch_size as usize);
+            }
+
+            // Extract batch_queue_size if provided
+            if let Some(batch_queue_size) = optional_property!("batchQueueSize", i64, js_obj) {
+                builder.batch_queue_size(batch_queue_size as usize);
+            }
+
+            // Extract linger if provided (in milliseconds)
+            if let Some(linger_ms) = optional_property!("lingerMs", i64, js_obj) {
+                builder.linger(Duration::from_millis(linger_ms as u64));
+            }
+
+            // Build the config
+            match builder.build() {
+                Ok(config) => Ok(Self { inner: config }),
+                Err(err) => Err(NjError::Other(format!(
+                    "Failed to build TopicProducerConfig: {}",
+                    err
+                ))),
+            }
+        } else {
+            Err(NjError::Other("parameter must be a JSON object".to_owned()))
+        }
+    }
 }

src/index.ts

@@ -75,6 +75,38 @@ export interface Options {
     id?: number
     offsetIndex?: number
     offsetFrom?: string
+    use_spu_local_address?: boolean
+}
+
+/**
+ * Configuration options for a topic producer
+ */
+export interface TopicProducerConfig {
+    /**
+     * Maximum size of a batch before it's sent automatically (in bytes)
+     */
+    batchSize?: number
+    /**
+     * Maximum time in milliseconds to wait before sending a batch, even if not full
+     */
+    timeoutMs?: number
+    /**
+     * Maximum number of records to include in a batch
+     */
+    batchQueueSize?: number
+    /**
+     * Time in milliseconds to wait for additional records before sending a batch
+     */
+    lingerMs?: number
+    /**
+     * Compression algorithm to use for batches
+     * Supported values: "none", "gzip", "snappy", "lz4"
+     */
+    compression?: 'none' | 'gzip' | 'snappy' | 'lz4'
+    /**
+     * Maximum size of a single request in bytes
+     */
+    maxRequestSize?: number
 }
 
 /**
@@ -525,6 +557,10 @@ export interface FluvioClient {
     connect(options?: Partial<Options>): Promise<FluvioClient>
     // connectWithConfig(config: any): Promise<this>
     topicProducer(topic: string): Promise<TopicProducer>
+    topicProducerWithConfig(
+        topic: string,
+        config: TopicProducerConfig
+    ): Promise<TopicProducer>
     partitionConsumer(
         topic: string,
         partition: number
@@ -680,6 +716,24 @@ export default class Fluvio implements FluvioClient {
         return TopicProducer.create(inner)
     }
 
+    /**
+     * Creates a new TopicProducer for the given topic name with configuration
+     *
+     * @param {string} topic Name of the topic to create a producer for
+     * @param {TopicProducerConfig} config Configuration for the topic producer
+     */
+    async topicProducerWithConfig(
+        topic: string,
+        config: TopicProducerConfig
+    ): Promise<TopicProducer> {
+        this.checkConnection()
+        const inner = await this.client?.topicProducerWithConfig(topic, config)
+        if (!inner) {
+            throw new Error('Failed to create topic producer with config')
+        }
+        return TopicProducer.create(inner)
+    }
+
     /**
      *
      * Creates a new PartitionConsumer for the given topic and partition