feat: add new super stream consumer handle

Author: magne

src/super_stream_consumer.ts

@@ -1,8 +1,11 @@
 import { Client } from "./client"
-import { Consumer, ConsumerFunc } from "./consumer"
+import { Consumer } from "./consumer"
 import { ConsumerCreditPolicy, defaultCreditPolicy } from "./consumer_credit_policy"
+import { Message } from "./publisher"
 import { Offset } from "./requests/subscribe_request"
 
+type SuperStreamConsumerFunc = (msg: Message, consumer?: Consumer) => Promise<void> | void
+
 export class SuperStreamConsumer {
   private consumers: Map<string, Consumer> = new Map<string, Consumer>()
   public consumerRef: string
@@ -13,7 +16,7 @@ export class SuperStreamConsumer {
   private creditPolicy: ConsumerCreditPolicy
 
   private constructor(
-    readonly handle: ConsumerFunc,
+    readonly handle: SuperStreamConsumerFunc,
     params: {
       superStream: string
       locator: Client
@@ -42,7 +45,7 @@ export class SuperStreamConsumer {
             singleActive: true,
             creditPolicy: this.creditPolicy,
           },
-          this.handle,
+          (msg) => this.handle(msg, this.consumers.get(p)),
           this
         )
         this.consumers.set(p, partitionConsumer)
@@ -52,7 +55,7 @@ export class SuperStreamConsumer {
   }
 
   static async create(
-    handle: ConsumerFunc,
+    handle: SuperStreamConsumerFunc,
     params: {
       superStream: string
       locator: Client