Make AMQP publisher sync

kibertoad · May 24, 2024 · 9704f54 · 9704f54
1 parent 4297191
commit 9704f54
Show file tree

Hide file tree

Showing 3 changed files with 27 additions and 11 deletions.
diff --git a/packages/amqp/lib/AmqpQueuePublisherManager.spec.ts b/packages/amqp/lib/AmqpQueuePublisherManager.spec.ts
@@ -18,7 +18,7 @@ describe('AmqpQueuePublisherManager', () => {
       const fakeConsumer = new FakeConsumer(diContainer.cradle, TestEvents.updated)
       await fakeConsumer.start()
 
-      const publishedMessage = await queuePublisherManager.publish(FakeConsumer.QUEUE_NAME, {
+      const publishedMessage = queuePublisherManager.publishSync(FakeConsumer.QUEUE_NAME, {
         type: 'entity.updated',
         payload: {
           updatedData: 'msg',

diff --git a/packages/amqp/lib/AmqpQueuePublisherManager.ts b/packages/amqp/lib/AmqpQueuePublisherManager.ts
@@ -113,15 +113,25 @@ export class AmqpQueuePublisherManager<
     }
   }
 
-  publish(
+  publish(): Promise<MessageSchemaType<SupportedEventDefinitions[number]>> {
+    throw new Error('Please use `publishSync` method for AMQP publisher managers')
+  }
+
+  publishSync(
     queue: NonNullable<SupportedEventDefinitions[number]['queueName']>,
     message: MessagePublishType<SupportedEventDefinitions[number]>,
     precedingEventMetadata?: Partial<MetadataType>,
     messageOptions?: AmqpQueueMessageOptions,
-  ): Promise<MessageSchemaType<SupportedEventDefinitions[number]>> {
-    // Purpose of this override is to provide better name for the first argument
-    // For AMQP Queues it is going to be queue
-    return super.publish(queue, message, precedingEventMetadata, messageOptions)
+  ): MessageSchemaType<SupportedEventDefinitions[number]> {
+    const publisher = this.targetToPublisherMap[queue]
+    if (!publisher) {
+      throw new Error(`No publisher for queue ${queue}`)
+    }
+
+    const messageDefinition = this.resolveMessageDefinition(queue, message)
+    const resolvedMessage = this.resolveMessage(messageDefinition, message, precedingEventMetadata)
+    publisher.publish(resolvedMessage, messageOptions)
+    return resolvedMessage
   }
 
   protected override resolveEventTarget(

diff --git a/packages/core/lib/queues/AbstractPublisherManager.ts b/packages/core/lib/queues/AbstractPublisherManager.ts
@@ -180,11 +180,7 @@ export abstract class AbstractPublisherManager<
     if (!publisher) {
       throw new Error(`No publisher for target ${eventTarget}`)
     }
-    // ToDo optimize the lookup
-    const messageDefinition = this.targetToEventMap[eventTarget].find(
-      (entry) => entry.consumerSchema.shape.type.value === message.type,
-    )
-
+    const messageDefinition = this.resolveMessageDefinition(eventTarget, message)
     const resolvedMessage = this.resolveMessage(messageDefinition, message, precedingEventMetadata)
 
     if (this.isAsync) {
@@ -196,6 +192,16 @@ export abstract class AbstractPublisherManager<
     return resolvedMessage
   }
 
+  protected resolveMessageDefinition(
+    eventTarget: EventTargets,
+    message: MessagePublishType<SupportedEventDefinitions[number]>,
+  ) {
+    // ToDo optimize the lookup
+    return this.targetToEventMap[eventTarget].find(
+      (entry) => entry.consumerSchema.shape.type.value === message.type,
+    )
+  }
+
   protected resolveMessage(
     messageDefinition: EventDefinitionType | undefined,
     message: MessagePublishType<SupportedEventDefinitions[number]>,