Switch to awilix 13 and amqp 1.x

kibertoad · kibertoad · commit da3395c2b364 · 2026-04-02T14:11:04.000+03:00
diff --git a/packages/amqp/lib/AbstractAmqpConsumer.ts b/packages/amqp/lib/AbstractAmqpConsumer.ts
@@ -306,7 +306,7 @@ export abstract class AbstractAmqpConsumer<
     }
 
     // Empty content for whatever reason
-    if (!resolveMessageResult.result || !resolveMessageResult.result.body) {
+    if (!resolveMessageResult.result?.body) {
       return ABORT_EARLY_EITHER
     }
 
@@ -347,7 +347,7 @@ export abstract class AbstractAmqpConsumer<
     const resolvedMessage = resolveMessageResult.result
 
     // Empty content for whatever reason
-    if (!resolvedMessage || !resolvedMessage.body) return ABORT_EARLY_EITHER
+    if (!resolvedMessage?.body) return ABORT_EARLY_EITHER
 
     // @ts-expect-error
     if (this.messageIdField in resolvedMessage.body) {
diff --git a/packages/amqp/package.json b/packages/amqp/package.json
@@ -33,24 +33,24 @@
     "peerDependencies": {
         "@message-queue-toolkit/core": ">=25.0.0",
         "@message-queue-toolkit/schemas": ">=7.0.0",
-        "amqplib": "^0.10.8",
+        "amqplib": "^1.0.3",
         "zod": ">=3.25.76 <5.0.0"
     },
     "devDependencies": {
-        "@biomejs/biome": "^2.3.8",
+        "@biomejs/biome": "^2.3.15",
         "@lokalise/biome-config": "^3.1.0",
         "@lokalise/tsconfig": "^3.0.0",
         "@message-queue-toolkit/core": "*",
-        "@types/amqplib": "0.10.8",
-        "@types/node": "^25.0.2",
-        "@vitest/coverage-v8": "^4.0.15",
-        "amqplib": "^0.10.8",
-        "awilix": "^12.0.5",
-        "awilix-manager": "^6.1.0",
-        "rimraf": "^6.0.1",
-        "typescript": "^5.9.3",
-        "vitest": "^4.0.15",
-        "zod": "^4.1.13"
+        "@types/node": "^25.5.0",
+        "@types/amqplib": "^0.10.8",
+        "@vitest/coverage-v8": "^4.0.18",
+        "amqplib": "^1.0.3",
+        "awilix": "^13.0.3",
+        "awilix-manager": "^6.4.0",
+        "rimraf": "^6.1.3",
+        "typescript": "^6.0.2",
+        "vitest": "^4.0.18",
+        "zod": "^4.3.6"
     },
     "homepage": "https://github.com/kibertoad/message-queue-toolkit",
     "repository": {
diff --git a/packages/amqp/test/utils/testContext.ts b/packages/amqp/test/utils/testContext.ts
@@ -94,7 +94,7 @@ export async function registerDependencies(
   dependencyOverrides: DependencyOverrides = {},
   queuesEnabled = true,
 ) {
-  const diContainer = createContainer({
+  const diContainer = createContainer<Dependencies>({
     injectionMode: 'PROXY',
   })
   const awilixManager = new AwilixManager({
@@ -104,7 +104,7 @@ export async function registerDependencies(
     eagerInject: true,
   })
 
-  const diConfig: DiConfig = {
+  const diConfig = {
     logger: asFunction(() => {
       return TestLogger
     }, SINGLETON_CONFIG),
@@ -116,7 +116,6 @@ export async function registerDependencies(
         return new AmqpConnectionManager(config, logger)
       },
       {
-        lifetime: Lifetime.SINGLETON,
         asyncInit: 'init',
         asyncDispose: 'close',
         asyncDisposePriority: 1,
@@ -217,7 +216,7 @@ export async function registerDependencies(
         report: () => {},
       } satisfies ErrorReporter
     }, SINGLETON_CONFIG),
-  }
+  } satisfies DiConfig
   diContainer.register(diConfig)
 
   for (const [dependencyKey, dependencyValue] of Object.entries(dependencyOverrides)) {
diff --git a/packages/core/package.json b/packages/core/package.json
@@ -44,7 +44,7 @@
         "@types/node": "^25.0.2",
         "@types/tmp": "^0.2.6",
         "@vitest/coverage-v8": "^4.0.15",
-        "awilix": "^12.0.5",
+        "awilix": "^13.0.3",
         "awilix-manager": "^6.1.0",
         "rimraf": "^6.0.1",
         "typescript": "^5.9.2",
diff --git a/packages/gcp-pubsub/package.json b/packages/gcp-pubsub/package.json
@@ -46,7 +46,7 @@
         "@message-queue-toolkit/schemas": "*",
         "@types/node": "^25.0.2",
         "@vitest/coverage-v8": "^4.0.15",
-        "awilix": "^12.0.5",
+        "awilix": "^13.0.3",
         "awilix-manager": "^6.1.0",
         "ioredis": "^5.7.0",
         "rimraf": "^6.0.1",
diff --git a/packages/kafka/lib/utils/KafkaMessageBatchStream.ts b/packages/kafka/lib/utils/KafkaMessageBatchStream.ts
@@ -61,6 +61,19 @@ export class KafkaMessageBatchStream<TMessage extends MessageWithTopicAndPartiti
   private pendingCallback: CallbackFunction | undefined
   private isBackPressured: boolean
 
+  // Debug counters for diagnosing flow issues
+  public readonly debug = {
+    writeCalls: 0,
+    readCalls: 0,
+    flushCalls: 0,
+    pushCalls: 0,
+    pushBackpressured: 0,
+    callbacksHeld: 0,
+    callbacksReleased: 0,
+    flushDeferredByBackpressure: 0,
+    finalCalled: false,
+  }
+
   constructor(options: KafkaMessageBatchOptions) {
     super({ objectMode: true, readableHighWaterMark: options.readableHighWaterMark })
     this.batchSize = options.batchSize
@@ -75,9 +88,11 @@ export class KafkaMessageBatchStream<TMessage extends MessageWithTopicAndPartiti
    * by calling the pending callback that was held during backpressure.
    */
   override _read() {
+    this.debug.readCalls++
     this.isBackPressured = false
     if (!this.pendingCallback) return
 
+    this.debug.callbacksReleased++
     const cb = this.pendingCallback
     this.pendingCallback = undefined
     cb() // Resume the writable side
@@ -89,6 +104,7 @@ export class KafkaMessageBatchStream<TMessage extends MessageWithTopicAndPartiti
    * Implements backpressure by holding the callback when downstream cannot consume.
    */
   override _write(message: TMessage, _encoding: BufferEncoding, callback: CallbackFunction) {
+    this.debug.writeCalls++
     let canContinue = true
 
     try {
@@ -104,12 +120,17 @@ export class KafkaMessageBatchStream<TMessage extends MessageWithTopicAndPartiti
       }
     } finally {
       // Backpressure handling: hold the callback if push() returned false
-      if (!canContinue) this.pendingCallback = callback
-      else callback()
+      if (!canContinue) {
+        this.debug.callbacksHeld++
+        this.pendingCallback = callback
+      } else {
+        callback()
+      }
     }
   }
 
   override _final(callback: CallbackFunction) {
+    this.debug.finalCalled = true
     // Clean timeout
     clearTimeout(this.existingTimeout)
     this.existingTimeout = undefined
@@ -120,11 +141,21 @@ export class KafkaMessageBatchStream<TMessage extends MessageWithTopicAndPartiti
     callback()
   }
 
+  public getPendingCount(): number {
+    return this.messages.length
+  }
+
+  public hasPendingCallback(): boolean {
+    return this.pendingCallback !== undefined
+  }
+
   private flushMessages(): boolean {
     clearTimeout(this.existingTimeout)
     this.existingTimeout = undefined
+    this.debug.flushCalls++
 
     if (this.isBackPressured) {
+      this.debug.flushDeferredByBackpressure++
       this.existingTimeout = setTimeout(() => this.flushMessages(), this.timeout)
       return false
     }
@@ -146,7 +177,9 @@ export class KafkaMessageBatchStream<TMessage extends MessageWithTopicAndPartiti
     // same tick also return false, so the last value correctly reflects backpressure.
     let canContinue = true
     for (const messagesForKey of Object.values(messagesByTopicPartition)) {
+      this.debug.pushCalls++
       canContinue = this.push(messagesForKey)
+      if (!canContinue) this.debug.pushBackpressured++
     }
 
     if (!canContinue) this.isBackPressured = true
diff --git a/packages/kafka/package.json b/packages/kafka/package.json
@@ -68,7 +68,7 @@
         "@message-queue-toolkit/schemas": ">=7.0.0",
         "@types/node": "^25.0.2",
         "@vitest/coverage-v8": "^4.0.15",
-        "awilix": "^12.0.1",
+        "awilix": "^13.0.3",
         "awilix-manager": "^6.0.0",
         "rimraf": "^6.0.1",
         "typescript": "^5.9.2",
diff --git a/packages/sns/package.json b/packages/sns/package.json
@@ -53,7 +53,7 @@
         "@message-queue-toolkit/sqs": "*",
         "@types/node": "^25.0.2",
         "@vitest/coverage-v8": "^4.0.15",
-        "awilix": "^12.1.1",
+        "awilix": "^13.0.3",
         "awilix-manager": "^6.1.0",
         "fauxqs": "^2.0.0",
         "ioredis": "^5.7.0",
diff --git a/packages/sqs/package.json b/packages/sqs/package.json
@@ -48,7 +48,7 @@
         "@message-queue-toolkit/schemas": "*",
         "@types/node": "^25.0.2",
         "@vitest/coverage-v8": "^4.0.15",
-        "awilix": "^12.0.5",
+        "awilix": "^13.0.3",
         "awilix-manager": "^6.1.0",
         "fauxqs": "^2.0.0",
         "ioredis": "^5.6.1",

Original file line number	Diff line number	Diff line change
`@@ -306,7 +306,7 @@ export abstract class AbstractAmqpConsumer<`
`306`	`306`	`}`
`307`	`307`
`308`	`308`	`// Empty content for whatever reason`
`309`		`- if (!resolveMessageResult.result \|\| !resolveMessageResult.result.body) {`
	`309`	`+ if (!resolveMessageResult.result?.body) {`
`310`	`310`	`return ABORT_EARLY_EITHER`
`311`	`311`	`}`
`312`	`312`
`@@ -347,7 +347,7 @@ export abstract class AbstractAmqpConsumer<`
`347`	`347`	`const resolvedMessage = resolveMessageResult.result`
`348`	`348`
`349`	`349`	`// Empty content for whatever reason`
`350`		`- if (!resolvedMessage \|\| !resolvedMessage.body) return ABORT_EARLY_EITHER`
	`350`	`+ if (!resolvedMessage?.body) return ABORT_EARLY_EITHER`
`351`	`351`
`352`	`352`	`// @ts-expect-error`
`353`	`353`	`if (this.messageIdField in resolvedMessage.body) {`