1.1.0

.github/workflows/CI.yml

@@ -76,7 +76,7 @@ jobs:
       - uses: goto-bus-stop/setup-zig@v2
         if: ${{ matrix.settings.target == 'armv7-unknown-linux-gnueabihf' }}
         with:
-          version: 0.10.1
+          version: 0.13.0
       - name: Setup toolchain
         run: ${{ matrix.settings.setup }}
         if: ${{ matrix.settings.setup }}
@@ -106,7 +106,7 @@ jobs:
         if: ${{ !matrix.settings.docker }}
         shell: bash
       - name: Upload artifact
-        uses: actions/upload-artifact@v3
+        uses: actions/upload-artifact@v4
         with:
           name: bindings-${{ matrix.settings.target }}
           path: ${{ env.APP_NAME }}.*.node
@@ -137,7 +137,7 @@ jobs:
         with:
           run_install: true
       - name: Download artifacts
-        uses: actions/download-artifact@v3
+        uses: actions/download-artifact@v4
         with:
           name: bindings-${{ matrix.settings.target }}
           path: .
@@ -169,7 +169,7 @@ jobs:
         with:
           run_install: true
       - name: Download artifacts
-        uses: actions/download-artifact@v3
+        uses: actions/download-artifact@v4
         with:
           name: bindings-x86_64-unknown-linux-gnu
           path: .
@@ -196,7 +196,7 @@ jobs:
         with:
           run_install: true
       - name: Download all artifacts
-        uses: actions/download-artifact@v3
+        uses: actions/download-artifact@v4
         with:
           path: artifacts
       - name: Move artifacts

Cargo.toml

@@ -15,7 +15,7 @@ napi = { version = "2.16", default-features = false, features = [
 ] }
 napi-derive = "2.16"
 
-anyhow = { version = "1.0", features = ["backtrace"] }
+anyhow = { version = "1.0.95", features = ["backtrace"] }
 
 rdkafka = { version = "0.37", features = [
     "libz-static",
@@ -30,7 +30,7 @@ tracing-subscriber = { version = "0.3", features = ["json"] }
 nanoid = "0.4.0"
 
 [build-dependencies]
-napi-build = "2.1"
+napi-build = "2.1.4"
 
 [profile.release]
 lto = true

js-binding.d.ts

@@ -92,8 +92,9 @@ export declare class KafkaConsumer {
   pause(): void
   resume(): void
   unsubscribe(): void
+  shutdownConsumer(): Promise<void>
   seek(topic: string, partition: number, offsetModel: OffsetModel, timeout?: number | undefined | null): void
-  recv(): Promise<Message>
+  recv(): Promise<Message | null>
   commit(topic: string, partition: number, offset: number, commit: CommitMode): void
 }
 export declare class KafkaClientConfig {

js-src/js-binding.d.ts

@@ -92,8 +92,9 @@ export declare class KafkaConsumer {
   pause(): void
   resume(): void
   unsubscribe(): void
+  shutdownConsumer(): Promise<void>
   seek(topic: string, partition: number, offsetModel: OffsetModel, timeout?: number | undefined | null): void
-  recv(): Promise<Message>
+  recv(): Promise<Message | null>
   commit(topic: string, partition: number, offset: number, commit: CommitMode): void
 }
 export declare class KafkaClientConfig {

js-src/kafka-client.ts

@@ -12,7 +12,7 @@ export class KafkaClient {
    * @throws {Error} If the configuration is invalid
    */
   constructor(private readonly kafkaConfiguration: KafkaConfiguration) {
-    this.kafkaClientConfig = new KafkaClientConfig(kafkaConfiguration)
+    this.kafkaClientConfig = new KafkaClientConfig(this.kafkaConfiguration)
   }
 
   /**

js-src/kafka-stream-readable.ts

@@ -1,6 +1,6 @@
 import { Readable } from 'stream'
 
-import { CommitMode } from '../js-binding'
+import { CommitMode } from './js-binding'
 import { KafkaConsumer, OffsetModel, TopicPartitionConfig } from './js-binding'
 
 /**

kafka-client.js

@@ -13,7 +13,7 @@ class KafkaClient {
      */
     constructor(kafkaConfiguration) {
         this.kafkaConfiguration = kafkaConfiguration;
-        this.kafkaClientConfig = new js_binding_js_1.KafkaClientConfig(kafkaConfiguration);
+        this.kafkaClientConfig = new js_binding_js_1.KafkaClientConfig(this.kafkaConfiguration);
     }
     /**
      * Creates a KafkaProducer instance

kafka-stream-readable.d.ts

@@ -1,5 +1,5 @@
 import { Readable } from 'stream';
-import { CommitMode } from '../js-binding';
+import { CommitMode } from './js-binding';
 import { KafkaConsumer, OffsetModel, TopicPartitionConfig } from './js-binding';
 /**
  * KafkaStreamReadable class

package.json

@@ -44,7 +44,7 @@
     "@napi-rs/cli": "^2.18.4",
     "@types/node": "22.10.1",
     "ava": "6.2.0",
-    "dprint": "^0.47.6",
+    "dprint": "^0.48.0",
     "nanoid": "5.0.9",
     "typescript": "5.7.2",
     "copyfiles": "^2.4.1",
@@ -65,5 +65,5 @@
     "version": "napi version",
     "fmt": "dprint fmt"
   },
-  "packageManager": "pnpm@9.14.2"
+  "packageManager": "pnpm@9.15.2"
 }

src/kafka/consumer/kafka_consumer.rs

@@ -1,4 +1,5 @@
 use std::time::Duration;
+use tokio::sync::watch::{self};
 
 use napi::{Either, Result};
 
@@ -27,13 +28,18 @@ use super::{
   },
 };
 
+use tokio::select;
+
 pub const DEFAULT_SEEK_TIMEOUT: i64 = 1500;
 
+type ShutdownSignal = (watch::Sender<()>, watch::Receiver<()>);
+
 #[napi]
 pub struct KafkaConsumer {
   client_config: ClientConfig,
   stream_consumer: StreamConsumer<CustomContext>,
   fecth_metadata_timeout: Duration,
+  shutdown_signal: ShutdownSignal,
 }
 
 #[napi]
@@ -57,6 +63,7 @@ impl KafkaConsumer {
           .fecth_metadata_timeout
           .unwrap_or(DEFAULT_FECTH_METADATA_TIMEOUT.as_millis() as i64) as u64,
       ),
+      shutdown_signal: watch::channel(()),
     })
   }
 
@@ -171,7 +178,27 @@ impl KafkaConsumer {
 
   #[napi]
   pub fn unsubscribe(&self) -> Result<()> {
+    info!("Unsubscribing from topics");
+    self.stream_consumer.unsubscribe();
+    Ok(())
+  }
+
+  #[napi]
+  pub async fn shutdown_consumer(&self) -> Result<()> {
+    info!("Shutting down consumer - this will unsubscribe and stop the consumer from receiving messages");
+
+    // First unsubscribe from topics
     self.stream_consumer.unsubscribe();
+
+    // Then send shutdown signal
+    let tx = self.shutdown_signal.0.clone();
+    tx.send(()).map_err(|e| {
+      napi::Error::new(
+        napi::Status::GenericFailure,
+        format!("Error sending shutdown signal: {:?}", e),
+      )
+    })?;
+
     Ok(())
   }
 
@@ -207,18 +234,24 @@ impl KafkaConsumer {
   }
 
   #[napi]
-  pub async fn recv(&self) -> Result<Message> {
-    self
-      .stream_consumer
-      .recv()
-      .await
-      .map_err(|e| {
-        napi::Error::new(
-          napi::Status::GenericFailure,
-          format!("Error while receiving from stream consumer: {:?}", e),
-        )
-      })
-      .map(|message| create_message(&message, message.payload().unwrap_or(&[])))
+  pub async fn recv(&self) -> Result<Option<Message>> {
+    let mut rx = self.shutdown_signal.1.clone();
+    select! {
+        message = self.stream_consumer.recv() => {
+            message
+                .map_err(|e| {
+                    napi::Error::new(
+                        napi::Status::GenericFailure,
+                        format!("Error while receiving from stream consumer: {:?}", e),
+                    )
+                })
+                .map(|message| Some(create_message(&message, message.payload().unwrap_or(&[]))))
+        }
+        _ = rx.changed() => {
+            info!("Shutdown signal received and this will stop the consumer from receiving messages");
+            Ok(None)
+        }
+    }
   }
 
   #[napi]