feat(@rhtml/amqp): changed a little bit the implementation to extend …

…more the api
r-html · Jan 10, 2025 · af8c0a2 · af8c0a2
1 parent 91c415d
commit af8c0a2
Show file tree

Hide file tree

Showing 4 changed files with 9 additions and 12 deletions.
diff --git a/packages/amqp/README.md b/packages/amqp/README.md
@@ -65,7 +65,7 @@ Integrate AMQP with your Fastify controllers for message subscription and publis
 import { Controller, Route } from '@rhtml/fastify';
 import { FastifyRequest } from 'fastify';
 import {
-  AckCallbackFunction,
+  AmqpChannel,
   AmqpService,
   ConsumeMessage,
   Subscribe,
@@ -84,7 +84,7 @@ export class MyController {
       noAck: true, // Automatically acknowledge messages
     },
   })
-  withAutoAcknowledge(data: ConsumeMessage, ack: AckCallbackFunction) {
+  withAutoAcknowledge(data: ConsumeMessage, channel: AmqpChannel) {
     // Parse the incoming message
     const message = JSON.parse(data?.content.toString());
     console.log(message);
@@ -97,13 +97,13 @@ export class MyController {
       noAck: false,
     },
   })
-  withCustomAcknowledge(data: ConsumeMessage, done: AckCallbackFunction) {
+  withCustomAcknowledge(data: ConsumeMessage, channel: AmqpChannel) {
     const message = JSON.parse(data?.content.toString());
 
     setTimeout(() => {
       // Long Running Job can be parsing some file
       console.log(message);
-      done();
+      channel.ack(data!);
     }, 10000);
   }
 

diff --git a/packages/amqp/src/amqp.constants.ts b/packages/amqp/src/amqp.constants.ts
@@ -12,5 +12,3 @@ export type AmqpConnection = Connection;
  */
 export const AmqpChannel = new InjectionToken<Channel>();
 export type AmqpChannel = Channel;
-
-export type AckCallbackFunction = () => void;
diff --git a/packages/amqp/src/amqp.service.ts b/packages/amqp/src/amqp.service.ts
@@ -1,5 +1,5 @@
 import { Inject, Injectable } from '@rhtml/di';
-import { ConsumeMessage, Options } from 'amqplib';
+import { Channel, ConsumeMessage, Options } from 'amqplib';
 
 import { AmqpChannel } from './amqp.constants';
 
@@ -18,7 +18,7 @@ export class AmqpService {
 
   async subscribe(
     name: string,
-    callback: (msg: ConsumeMessage, ack: () => void) => void,
+    callback: (msg: ConsumeMessage, channel: Channel) => void,
     options?: {
       assertOptions?: Options.AssertQueue;
       consumeOptions?: Options.Consume;
@@ -28,7 +28,7 @@ export class AmqpService {
 
     await this.channel.consume(
       name,
-      (data) => callback(data!, () => this.channel.ack(data!)),
+      (data) => callback(data!, this.channel),
       options?.consumeOptions
     );
   }

diff --git a/packages/amqp/src/decorators/subscribe.decorator.ts b/packages/amqp/src/decorators/subscribe.decorator.ts
@@ -1,5 +1,5 @@
 import { set } from '@rhtml/di';
-import { ConsumeMessage, Options } from 'amqplib';
+import { Options } from 'amqplib';
 
 import { AmqpService } from '../amqp.service';
 
@@ -12,7 +12,6 @@ export const Subscribe =
     name: string;
     consumeOptions?: Options.Consume;
     assertOptions?: Options.AssertQueue;
-    parser?: <T = never>(msg: ConsumeMessage | null) => T;
   }) =>
   (target: T, memberName: string) => {
     const OnInit =
@@ -25,7 +24,7 @@ export const Subscribe =
         const amqpService = set(AmqpService);
         await amqpService.subscribe(
           name,
-          (msg, ack) => target[memberName].call(this, msg, ack),
+          (msg, channel) => target[memberName].call(this, msg, channel),
           { assertOptions, consumeOptions }
         );
         return OnInit.apply(this, args);