Lazy init for AMQP

README.md

@@ -40,9 +40,6 @@ They implement the following public methods:
 
 > **_NOTE:_**  See [SqsPermissionPublisher.ts](./packages/sqs/test/publishers/SqsPermissionPublisher.ts) for a practical example.
 
-> **_NOTE:_**  Lazy loading is not supported for AMQP publishers.
-
-
 ### Consumers
 
 `message-queue-toolkit` provides base classes for implementing consumers for each of the supported protocol.

docker-compose.yml

@@ -1,6 +1,6 @@
 services:
   rabbitmq:
-    image: rabbitmq:3.11.20
+    image: rabbitmq:3.12.14
     ports:
       - ${DOCKER_RABBITMQ_CLIENT_PORT:-5672}:5672
       - ${DOCKER_RABBITMQ_MANAGEMENT_PORT:-15672}:15672

packages/amqp/.eslintrc.json

@@ -60,7 +60,7 @@
     ],
     "max-lines": ["error", { "max": 600 }],
     "max-params": ["error", { "max": 4 }],
-    "max-statements": ["error", { "max": 20 }],
+    "max-statements": ["error", { "max": 30 }],
     "complexity": ["error", { "max": 20 }]
   },
   "overrides": [

packages/amqp/docker-compose.yml

@@ -1,6 +1,6 @@
 services:
   rabbitmq:
-    image: rabbitmq:3.11.20
+    image: rabbitmq:3.12.14
     ports:
       - ${DOCKER_RABBITMQ_CLIENT_PORT:-5672}:5672
       - ${DOCKER_RABBITMQ_MANAGEMENT_PORT:-15672}:15672

packages/amqp/lib/AbstractAmqpPublisher.ts

@@ -25,6 +25,8 @@ export abstract class AbstractAmqpPublisher<MessagePayloadType extends object>
 {
   private readonly messageSchemaContainer: MessageSchemaContainer<MessagePayloadType>
 
+  private initPromise?: Promise<void>
+
   constructor(dependencies: AMQPDependencies, options: AMQPPublisherOptions<MessagePayloadType>) {
     super(dependencies, options)
 
@@ -42,6 +44,27 @@ export abstract class AbstractAmqpPublisher<MessagePayloadType extends object>
     }
     resolveSchemaResult.result.parse(message)
 
+    // If it's not initted yet, do the lazy init
+    if (!this.isInitted) {
+      // avoid multiple concurrent inits
+      if (!this.initPromise) {
+        this.initPromise = this.init()
+      }
+
+      /**
+       * it is intentional that this promise is not awaited, that's how we keep the method invocation synchronous
+       * RabbitMQ publish by itself doesn't guarantee that your message is delivered successfully, so this kind of fire-and-forget is not strongly different from how amqp-lib behaves in the first place.
+       */
+      this.initPromise
+        .then(() => {
+          this.publish(message)
+        })
+        .catch((err) => {
+          this.handleError(err)
+        })
+      return
+    }
+
     /**
      * If the message doesn't have a timestamp field -> add it
      * will be used on the consumer to prevent infinite retries on the same message
@@ -108,6 +131,11 @@ export abstract class AbstractAmqpPublisher<MessagePayloadType extends object>
     throw new Error('Not implemented for publisher')
   }
 
+  async close(): Promise<void> {
+    this.initPromise = undefined
+    await super.close()
+  }
+
   override processMessage(): Promise<Either<'retryLater', 'success'>> {
     throw new Error('Not implemented for publisher')
   }

packages/amqp/lib/AbstractAmqpService.ts

@@ -158,6 +158,7 @@ export abstract class AbstractAmqpService<
     if (this.connection) {
       await this.receiveNewConnection(this.connection)
     }
+    this.isInitted = true
   }
 
   public async reconnect() {
@@ -167,5 +168,6 @@ export abstract class AbstractAmqpService<
   async close(): Promise<void> {
     this.isShuttingDown = true
     await this.destroyChannel()
+    this.isInitted = false
   }
 }

packages/amqp/test/publishers/AmqpPermissionPublisher.spec.ts

@@ -159,7 +159,8 @@ describe('PermissionPublisher', () => {
     it('return details if publish failed', async () => {
       expect.assertions(3)
       try {
-        await permissionPublisher.close()
+        // @ts-ignore
+        permissionPublisher.channel = undefined
         permissionPublisher.publish({
           id: '11',
           messageType: 'add',
@@ -243,6 +244,53 @@ describe('PermissionPublisher', () => {
       })
     })
 
+    it('publishes a message with lazy init', async () => {
+      await permissionConsumer.close()
+      await permissionPublisher.close()
+
+      const message = {
+        id: '1',
+        messageType: 'add',
+        userIds: [1],
+        permissions: ['100'],
+        timestamp: new Date(),
+      } satisfies PERMISSIONS_MESSAGE_TYPE
+
+      let receivedMessage: unknown
+      await channel.consume(AmqpPermissionPublisher.QUEUE_NAME, (message) => {
+        if (message === null) {
+          return
+        }
+        const decodedMessage = deserializeAmqpMessage(
+          message,
+          PERMISSIONS_MESSAGE_SCHEMA,
+          new FakeConsumerErrorResolver(),
+        )
+        receivedMessage = decodedMessage.result!
+      })
+
+      permissionPublisher.publish(message)
+
+      await waitAndRetry(() => !!receivedMessage)
+
+      expect(receivedMessage).toEqual({
+        parsedMessage: {
+          id: '1',
+          messageType: 'add',
+          userIds: [1],
+          permissions: ['100'],
+          timestamp: message.timestamp.toISOString(),
+        },
+        originalMessage: {
+          id: '1',
+          messageType: 'add',
+          userIds: [1],
+          permissions: ['100'],
+          timestamp: message.timestamp.toISOString(),
+        },
+      })
+    })
+
     it('publishes a message auto-filling timestamp', async () => {
       await permissionConsumer.close()
 

packages/core/lib/queues/AbstractQueueService.ts

@@ -52,6 +52,7 @@ export abstract class AbstractQueueService<
   protected readonly locatorConfig?: QueueLocatorType
   protected readonly deletionConfig?: DeletionConfig
   protected readonly _handlerSpy?: HandlerSpy<MessagePayloadSchemas>
+  protected isInitted: boolean
 
   get handlerSpy(): PublicHandlerSpy<MessagePayloadSchemas> {
     if (!this._handlerSpy) {
@@ -75,6 +76,7 @@ export abstract class AbstractQueueService<
 
     this.logMessages = options.logMessages ?? false
     this._handlerSpy = resolveHandlerSpy<MessagePayloadSchemas>(options)
+    this.isInitted = false
   }
 
   protected abstract resolveSchema(

packages/sns/lib/sns/AbstractSnsPublisher.ts

@@ -31,7 +31,6 @@ export abstract class AbstractSnsPublisher<MessagePayloadType extends object>
 {
   private readonly messageSchemaContainer: MessageSchemaContainer<MessagePayloadType>
 
-  private isInitted: boolean
   private initPromise?: Promise<void>
 
   constructor(dependencies: SNSDependencies, options: SNSPublisherOptions<MessagePayloadType>) {
@@ -42,7 +41,6 @@ export abstract class AbstractSnsPublisher<MessagePayloadType extends object>
       messageSchemas,
       messageTypeField: options.messageTypeField,
     })
-    this.isInitted = false
   }
 
   async publish(message: MessagePayloadType, options: SNSMessageOptions = {}): Promise<void> {
@@ -58,6 +56,7 @@ export abstract class AbstractSnsPublisher<MessagePayloadType extends object>
         this.initPromise = this.init()
       }
       await this.initPromise
+      this.initPromise = undefined
     }
 
     try {

packages/sns/lib/sns/AbstractSnsService.ts

@@ -71,9 +71,11 @@ export abstract class AbstractSnsService<
 
     const initResult = await initSns(this.snsClient, this.locatorConfig, this.creationConfig)
     this.topicArn = initResult.topicArn
+    this.isInitted = true
   }
 
   public override close(): Promise<void> {
+    this.isInitted = false
     return Promise.resolve()
   }
 }

packages/sns/lib/sns/SnsPublisherManager.spec.ts

@@ -15,7 +15,7 @@ import { CommonSnsPublisher } from './CommonSnsPublisherFactory'
 import type { SnsMessagePublishType, SnsPublisherManager } from './SnsPublisherManager'
 import { FakeConsumer } from './fakes/FakeConsumer'
 
-describe('AutopilotPublisherManager', () => {
+describe('SnsPublisherManager', () => {
   let diContainer: AwilixContainer<Dependencies>
   let publisherManager: SnsPublisherManager<
     CommonSnsPublisher<TestEventPayloadsType>,
@@ -80,7 +80,7 @@ describe('AutopilotPublisherManager', () => {
       await fakeConsumer.close()
     })
 
-    it('publish to a non existing topic will throw error', async () => {
+    it('publish to a non-existing topic will throw error', async () => {
       await expect(
         publisherManager.publish('non-existing-topic', {
           type: 'entity.created',

packages/sqs/lib/sqs/AbstractSqsPublisher.ts

@@ -1,6 +1,7 @@
 import type { SendMessageCommandInput } from '@aws-sdk/client-sqs'
 import { SendMessageCommand } from '@aws-sdk/client-sqs'
 import type { Either } from '@lokalise/node-core'
+import { InternalError } from '@lokalise/node-core'
 import type {
   AsyncPublisher,
   MessageInvalidFormatError,
@@ -26,7 +27,6 @@ export abstract class AbstractSqsPublisher<MessagePayloadType extends object>
   implements AsyncPublisher<MessagePayloadType, SQSMessageOptions>
 {
   private readonly messageSchemaContainer: MessageSchemaContainer<MessagePayloadType>
-  private isInitted: boolean
   private initPromise?: Promise<void>
 
   constructor(
@@ -40,7 +40,6 @@ export abstract class AbstractSqsPublisher<MessagePayloadType extends object>
       messageSchemas,
       messageTypeField: options.messageTypeField,
     })
-    this.isInitted = false
   }
 
   async publish(message: MessagePayloadType, options: SQSMessageOptions = {}): Promise<void> {
@@ -56,6 +55,7 @@ export abstract class AbstractSqsPublisher<MessagePayloadType extends object>
         this.initPromise = this.init()
       }
       await this.initPromise
+      this.initPromise = undefined
     }
 
     try {
@@ -87,8 +87,20 @@ export abstract class AbstractSqsPublisher<MessagePayloadType extends object>
       await this.sqsClient.send(command)
       this.handleMessageProcessed(message, 'published')
     } catch (error) {
-      this.handleError(error)
-      throw error
+      const err = error as Error
+      this.handleError(err)
+      throw new InternalError({
+        message: `Error while publishing to SQS: ${err.message}`,
+        errorCode: 'SQS_PUBLISH_ERROR',
+        details: {
+          publisher: this.constructor.name,
+          queueArn: this.queueArn,
+          queueName: this.queueName,
+          // @ts-ignore
+          messageType: message[this.messageTypeField] ?? 'unknown',
+        },
+        cause: err,
+      })
     }
   }
 

packages/sqs/lib/sqs/AbstractSqsService.ts

@@ -70,8 +70,11 @@ export abstract class AbstractSqsService<
     this.queueName = queueName
     this.queueUrl = queueUrl
     this.queueArn = queueArn
+    this.isInitted = true
   }
 
-  // eslint-disable-next-line @typescript-eslint/require-await
-  public override async close(): Promise<void> {}
+  public override close(): Promise<void> {
+    this.isInitted = false
+    return Promise.resolve()
+  }
 }

packages/sqs/test/consumers/SqsPermisssionConsumer.deadLetterQueue.spec.ts

@@ -4,7 +4,6 @@ import { waitAndRetry } from '@lokalise/node-core'
 import type { AwilixContainer } from 'awilix'
 import { Consumer } from 'sqs-consumer'
 import { afterEach, beforeEach, describe, expect, it } from 'vitest'
-import z from 'zod'
 
 import type { SQSMessage } from '../../lib/types/MessageTypes'
 import { assertQueue, deleteQueue, getQueueAttributes } from '../../lib/utils/sqsUtils'

packages/sqs/test/publishers/SqsPermissionPublisher.spec.ts

@@ -145,16 +145,12 @@ describe('SqsPermissionPublisher', () => {
 
   describe('publish', () => {
     let diContainer: AwilixContainer<Dependencies>
-    let sqsClient: SQSClient
     let permissionPublisher: SqsPermissionPublisher
 
     beforeEach(async () => {
       diContainer = await registerDependencies()
-      sqsClient = diContainer.cradle.sqsClient
       await diContainer.cradle.permissionConsumer.close()
       permissionPublisher = diContainer.cradle.permissionPublisher
-
-      await deleteQueue(sqsClient, SqsPermissionPublisher.QUEUE_NAME)
     })
 
     afterEach(async () => {