214 question async consumerfunc (#224)

* ConsumerFunc async

* credit policy

* consumer func return type

* package lock

* timeout

* policy

* package lock

* readme

* removed async in 3 tests

* revert some async tests

* removed async

* add async consumer test

Author: dc1992

README.md

@@ -210,6 +210,21 @@ const consumer = await client.declareConsumer(consumerOptions, (message: Message
 // ...
 ```
 
+### Custom Policy
+By default the client uses the `creditsOnChunkCompleted(1, 1)` policy. This policy grants that messages will be processed in order, as a new chunk will only be requested once the current chunk has been processed. It is possible to override this policy by passing `creditPolicy` to the consumer options. Be aware that modifying this policy can lead to out-of-order message processing.
+
+```typescript
+const consumerOptions = {
+  stream: "stream-name",
+  creditPolicy: creditsOnChunkReceived(2, 1)
+}
+
+await client.declareConsumer(consumerOptions, async (message: Message) => {
+    console.log(message.content)
+  }
+)
+```
+
 ### Clustering
 
 Every time we create a new producer or a new consumer, a new connection object is created. The underlying TCP connections can be shared among different producers and different consumers. Note however that:

example/package-lock.json

@@ -564,9 +564,11 @@ export class Client {
         ? consumer.filter?.postFilterFunc
         : (_msg: Message) => true
 
-    messages.map((message) => {
-      if (messageFilter(message)) consumer.handle(message)
-    })
+    for (const message of messages) {
+      if (messageFilter(message)) {
+        await consumer.handle(message)
+      }
+    }
 
     await consumer.creditPolicy.onChunkCompleted(creditRequestWrapper)
   }

performance_test/package-lock.json

@@ -5,7 +5,7 @@ import { ConsumerCreditPolicy, defaultCreditPolicy } from "./consumer_credit_pol
 import { Message } from "./publisher"
 import { Offset } from "./requests/subscribe_request"
 
-export type ConsumerFunc = (message: Message) => void
+export type ConsumerFunc = (message: Message) => Promise<void> | void
 export const computeExtendedConsumerId = (consumerId: number, connectionId: string) => {
   return `${consumerId}@${connectionId}`
 }
@@ -82,9 +82,9 @@ export class StreamConsumer implements Consumer {
     return this.clientLocalOffset.clone()
   }
 
-  public handle(message: Message) {
+  public async handle(message: Message) {
     if (this.closed || this.isMessageOffsetLessThanConsumers(message)) return
-    this.consumerHandle(message)
+    await this.consumerHandle(message)
     this.maybeUpdateLocalOffset(message)
   }
 

src/client.ts

@@ -54,4 +54,4 @@ export const creditsOnChunkReceived = (startFrom: number, step: number) =>
   new NewCreditsOnChunkReceived(startFrom, step)
 export const creditsOnChunkCompleted = (startFrom: number, step: number) =>
   new NewCreditsOnChunkCompleted(startFrom, step)
-export const defaultCreditPolicy = creditsOnChunkReceived(2, 1)
+export const defaultCreditPolicy = creditsOnChunkCompleted(1, 1)

src/consumer.ts

@@ -40,11 +40,15 @@ describe("address resolver", () => {
   })
 
   it("declaring a consumer - should not throw", async () => {
-    await client.declareConsumer({ stream: streamName, offset: Offset.first() }, () => null)
+    await client.declareConsumer({ stream: streamName, offset: Offset.first() }, () => {
+      return
+    })
   })
 
   it("declaring a consumer - if multiple nodes are present the consumer should be connected to a replica", async () => {
-    const consumer = await client.declareConsumer({ stream: streamName, offset: Offset.first() }, () => null)
+    const consumer = await client.declareConsumer({ stream: streamName, offset: Offset.first() }, () => {
+      return
+    })
 
     const connectionInfo = consumer.getConnectionInfo()
     const queueInfo = await rabbit.getQueueInfo(streamName)

src/consumer_credit_policy.ts

@@ -38,7 +38,9 @@ describe("close consumer", () => {
 
   it("closing a consumer in an existing stream", async () => {
     await client.declarePublisher({ stream: testStreamName })
-    const consumer = await client.declareConsumer({ stream: testStreamName, offset: Offset.first() }, () => null)
+    const consumer = await client.declareConsumer({ stream: testStreamName, offset: Offset.first() }, () => {
+      return
+    })
 
     const response = await client.closeConsumer(consumer.extendedId)
 

test/e2e/address_resolver.test.ts

@@ -29,6 +29,7 @@ import {
   getTestNodesFromEnv,
   password,
   username,
+  waitSleeping,
 } from "../support/util"
 
 describe("declare consumer", () => {
@@ -72,9 +73,21 @@ describe("declare consumer", () => {
     const messages: Buffer[] = []
     await publisher.send(Buffer.from("hello"))
 
-    await client.declareConsumer({ stream: streamName, offset: Offset.first() }, (message: Message) =>
+    await client.declareConsumer({ stream: streamName, offset: Offset.first() }, (message: Message) => {
       messages.push(message.content)
-    )
+    })
+
+    await eventually(() => expect(messages).eql([Buffer.from("hello")]))
+  }).timeout(10000)
+
+  it("declaring an async consumer on an existing stream - the consumer should handle the message", async () => {
+    const messages: Buffer[] = []
+    await publisher.send(Buffer.from("hello"))
+
+    await client.declareConsumer({ stream: streamName, offset: Offset.first() }, async (message: Message) => {
+      await waitSleeping(10)
+      messages.push(message.content)
+    })
 
     await eventually(() => expect(messages).eql([Buffer.from("hello")]))
   }).timeout(10000)
@@ -86,15 +99,21 @@ describe("declare consumer", () => {
     await publisher.send(Buffer.from("hello"))
     await client.declareConsumer(
       { stream: streamName, offset: Offset.first(), singleActive: true, consumerRef: consumerRef },
-      (message: Message) => messages.push(message.content)
+      (message: Message) => {
+        messages.push(message.content)
+      }
     )
     await client.declareConsumer(
       { stream: streamName, offset: Offset.first(), singleActive: true, consumerRef: consumerRef },
-      (message: Message) => messages.push(message.content)
+      (message: Message) => {
+        messages.push(message.content)
+      }
     )
     await client.declareConsumer(
       { stream: streamName, offset: Offset.first(), singleActive: true, consumerRef: consumerRef },
-      (message: Message) => messages.push(message.content)
+      (message: Message) => {
+        messages.push(message.content)
+      }
     )
 
     await eventually(() => expect(messages).eql([Buffer.from("hello")]))
@@ -105,16 +124,20 @@ describe("declare consumer", () => {
     const consumerRef = createConsumerRef()
 
     await publisher.send(Buffer.from("hello"))
-    await client.declareConsumer({ stream: streamName, offset: Offset.first() }, (message: Message) =>
+    await client.declareConsumer({ stream: streamName, offset: Offset.first() }, (message: Message) => {
       messages.push(message.content)
-    )
+    })
     await client.declareConsumer(
       { stream: streamName, offset: Offset.first(), singleActive: true, consumerRef: consumerRef },
-      (message: Message) => messages.push(message.content)
+      (message: Message) => {
+        messages.push(message.content)
+      }
     )
     await client.declareConsumer(
       { stream: streamName, offset: Offset.first(), singleActive: true, consumerRef: consumerRef },
-      (message: Message) => messages.push(message.content)
+      (message: Message) => {
+        messages.push(message.content)
+      }
     )
 
     await eventually(() => expect(messages).eql([Buffer.from("hello"), Buffer.from("hello")]))
@@ -128,19 +151,27 @@ describe("declare consumer", () => {
     await publisher.send(Buffer.from("hello"))
     await client.declareConsumer(
       { stream: streamName, offset: Offset.first(), singleActive: true, consumerRef: consumerRef },
-      (message: Message) => messages.push(message.content)
+      (message: Message) => {
+        messages.push(message.content)
+      }
     )
     await client.declareConsumer(
       { stream: streamName, offset: Offset.first(), singleActive: true, consumerRef: consumerRef },
-      (message: Message) => messages.push(message.content)
+      (message: Message) => {
+        messages.push(message.content)
+      }
     )
     await client.declareConsumer(
       { stream: streamName, offset: Offset.first(), singleActive: true, consumerRef: consumerRef1 },
-      (message: Message) => messages.push(message.content)
+      (message: Message) => {
+        messages.push(message.content)
+      }
     )
     await client.declareConsumer(
       { stream: streamName, offset: Offset.first(), singleActive: true, consumerRef: consumerRef1 },
-      (message: Message) => messages.push(message.content)
+      (message: Message) => {
+        messages.push(message.content)
+      }
     )
 
     await eventually(() => expect(messages).eql([Buffer.from("hello"), Buffer.from("hello")]))
@@ -155,7 +186,9 @@ describe("declare consumer", () => {
       async () => {
         await client.declareConsumer(
           { stream: streamName, offset: Offset.first(), singleActive: true },
-          (message: Message) => messages.push(message.content)
+          (message: Message) => {
+            messages.push(message.content)
+          }
         )
       },
       Error,
@@ -169,9 +202,9 @@ describe("declare consumer", () => {
     await publisher.send(Buffer.from("world"))
     await publisher.send(Buffer.from("world"))
 
-    await client.declareConsumer({ stream: streamName, offset: Offset.first() }, (message: Message) =>
+    await client.declareConsumer({ stream: streamName, offset: Offset.first() }, (message: Message) => {
       messages.push(message.content)
-    )
+    })
 
     await eventually(() => expect(messages).eql([Buffer.from("hello"), Buffer.from("world"), Buffer.from("world")]))
   }).timeout(10000)
@@ -192,7 +225,10 @@ describe("declare consumer", () => {
 
   it("declaring a consumer on a non-existing stream should raise an error", async () => {
     await expectToThrowAsync(
-      () => client.declareConsumer({ stream: nonExistingStreamName, offset: Offset.first() }, () => null),
+      () =>
+        client.declareConsumer({ stream: nonExistingStreamName, offset: Offset.first() }, () => {
+          return
+        }),
       Error,
       "Stream was not found on any node"
     )

test/e2e/close_consumer.test.ts

@@ -74,7 +74,9 @@ describe("filtering", () => {
           matchUnfiltered: true,
         },
       },
-      (msg) => filteredMsg.push(msg.content.toString("utf-8"))
+      (msg) => {
+        filteredMsg.push(msg.content.toString("utf-8"))
+      }
     )
 
     await eventually(async () => {
@@ -107,13 +109,15 @@ describe("filtering", () => {
           matchUnfiltered: false,
         },
       },
-      (msg) => filteredMsg.push(msg.content.toString("utf-8"))
+      (msg) => {
+        filteredMsg.push(msg.content.toString("utf-8"))
+      }
     )
 
     await eventually(async () => {
       expect(filteredMsg.length).eql(2000)
     }, 10000)
-  }).timeout(10000)
+  }).timeout(15000)
 
   it("published messages are filtered on the server side keeping even the ones with filter value", async () => {
     const filteredMsg: string[] = []
@@ -139,7 +143,9 @@ describe("filtering", () => {
           matchUnfiltered: true,
         },
       },
-      (msg) => filteredMsg.push(msg.content.toString("utf-8"))
+      (msg) => {
+        filteredMsg.push(msg.content.toString("utf-8"))
+      }
     )
 
     await eventually(async () => {