feat(@rhtml/amqp): added amqp connection module

package.json

@@ -34,11 +34,15 @@
   "license": "MIT",
   "devDependencies": {
     "@rhtml/di": "^0.0.132",
+    "@rhtml/schematics": "^0.0.128",
     "@rxdi/bolt": "^0.24.3",
+    "@types/amqplib": "^0.10.6",
     "@types/jest": "^24.0.18",
     "@types/node": "^20.11.24",
     "@typescript-eslint/eslint-plugin": "^7.1.1",
     "@typescript-eslint/parser": "^7.1.1",
+    "commander": "8.2.0",
+    "esbuild": "0.20.1",
     "eslint": "^8.47.0",
     "eslint-config-prettier": "^6.7.0",
     "eslint-plugin-prettier": "^3.1.1",
@@ -47,10 +51,7 @@
     "jest": "^24.9.0",
     "prettier": "^2.0.5",
     "ts-jest": "25.2.1",
-    "typescript": "^5.3.3",
-    "@rhtml/schematics": "^0.0.128",
-    "commander": "8.2.0",
-    "esbuild": "0.20.1"
+    "typescript": "^5.3.3"
   },
   "browserslist": [
     "last 1 chrome versions"
@@ -59,6 +60,7 @@
   "module": "./dist/index.js",
   "typings": "./dist/index.d.ts",
   "dependencies": {
+    "amqplib": "^0.10.5",
     "fastify": "^4.26.2",
     "mongoose": "^8.2.1"
   }

packages/amqp/.npmignore

@@ -0,0 +1 @@
+

packages/amqp/LICENSE

@@ -0,0 +1,28 @@
+BSD 3-Clause License
+
+Copyright (c) 2019, Reactive Solutions LTD 205455032. All rights reserved.
+
+Redistribution and use in source and binary forms, with or without
+modification, are permitted provided that the following conditions are met:
+
+* Redistributions of source code must retain the above copyright notice, this
+  list of conditions and the following disclaimer.
+
+* Redistributions in binary form must reproduce the above copyright notice,
+  this list of conditions and the following disclaimer in the documentation
+  and/or other materials provided with the distribution.
+
+* Neither the name of the copyright holder nor the names of its
+  contributors may be used to endorse or promote products derived from
+  this software without specific prior written permission.
+
+THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
+AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
+IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE
+DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE
+FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL
+DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR
+SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER
+CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY,
+OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
+OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.

packages/amqp/README.md

@@ -0,0 +1,177 @@
+# @rhtml/amqp
+
+`@rhtml/amqp` is an AMQP (Advanced Message Queuing Protocol) integration library designed for seamless interaction with message brokers such as RabbitMQ. It allows developers to easily publish and subscribe to messages, integrating AMQP functionalities into their applications with minimal setup.
+
+---
+
+## Features
+
+- **AMQP Protocol Support**: Full support for AMQP protocol with customizable configurations.
+- **Integration with Fastify**: Easily integrate AMQP with Fastify controllers and routes.
+- **Simple Publish and Subscribe Mechanism**: Simplified API for sending and consuming messages.
+- **Customizable Options**: Configure protocol, hostname, port, authentication, and vhost.
+
+---
+
+## Installation
+
+To use `@rhtml/amqp` in your project, install it via npm:
+
+```bash
+npm i @rhtml/amqp
+```
+
+## Setup and Configuration
+
+You can set up the AMQP connection in your application by using the AmqpModule.forRoot method. This allows you to configure the connection settings such as protocol, hostname, port, credentials, and vhost.
+
+### Basic Configuration
+
+```ts
+import { FastifyModule } from '@rhtml/fastify';
+import { AmqpModule } from '@rhtml/amqp';
+
+@Module({
+  imports: [
+    FastifyModule.forRoot({...}),
+    AmqpModule.forRoot({
+      protocol: 'amqp', // Default protocol: 'amqp'
+      hostname: 'localhost', // Hostname of the RabbitMQ server
+      port: 5672, // Default AMQP port: 5672
+      username: 'amqp user', // AMQP username (default: 'guest')
+      password: 'amqp password', // AMQP password (default: 'guest')
+      vhost: '', // Virtual host to use (default: empty string)
+    }),
+  ],
+})
+export class AppModule {}
+```
+
+### Configuration Options
+
+- Protocol: Communication protocol. Default is amqp.
+- Hostname: RabbitMQ server address. Default is localhost.
+- Port: Port for RabbitMQ. Default is 5672.
+- Username & Password: Authentication credentials. Defaults are guest and guest.
+- vhost: RabbitMQ virtual host. Defaults to an empty string.
+
+## Usage Example
+
+### Controller Example
+
+Integrate AMQP with your Fastify controllers for message subscription and publishing:
+
+```ts
+import { Controller, Route } from '@rhtml/fastify';
+import { FastifyRequest } from 'fastify';
+import {
+  AckCallbackFunction,
+  AmqpService,
+  ConsumeMessage,
+  Subscribe,
+} from '@rhtml/amqp';
+
+@Controller({
+  route: '/',
+})
+export class MyController {
+  constructor(private amqpService: AmqpService) {}
+
+  // Subscription handler for consuming messages
+  @Subscribe({
+    name: 'test-queue',
+    consumeOptions: {
+      noAck: true, // Automatically acknowledge messages
+    },
+  })
+  withAutoAcknowledge(data: ConsumeMessage, ack: AckCallbackFunction) {
+    // Parse the incoming message
+    const message = JSON.parse(data?.content.toString());
+    console.log(message);
+    // Output: { test: '123' }
+  }
+
+  @Subscribe({
+    name: 'test-queue',
+    consumeOptions: {
+      noAck: false,
+    },
+  })
+  withCustomAcknowledge(data: ConsumeMessage, done: AckCallbackFunction) {
+    const message = JSON.parse(data?.content.toString());
+
+    setTimeout(() => {
+      // Long Running Job can be parsing some file
+      console.log(message);
+      done();
+    }, 10000);
+  }
+
+  // Route to trigger message publication
+  @Route({
+    url: '',
+    method: 'GET',
+  })
+  async myRouteTrigger(req: FastifyRequest) {
+    // Publish a message to the 'test-queue'
+    this.amqpService.publish('test-queue', { test: '123' });
+  }
+}
+```
+
+## Closing the AMQP connection after server stops
+
+```ts
+import { AmqpConnection } from '@rhtml/amqp'
+import { InjectionToken, Module } from '@rhtml/di'
+import { Fastify } from '@rhtml/fastify'
+
+import { Connection } from 'amqplib'
+import { FastifyInstance } from 'fastify'
+
+@Module({
+  providers: [
+    /* Close the AMQP Connection when server stops */
+    {
+      provide: new InjectionToken(),
+      deps: [Fastify, AmqpConnection],
+      useFactory: (fastify: FastifyInstance, connection: Connection) =>
+        fastify.addHook('onClose', () => connection.close()),
+    },
+  ]
+})
+
+```
+
+## Key Concepts
+
+1. Message Subscription
+   The @Subscribe decorator listens for messages on a specified queue:
+
+- Queue Name: Specifies the queue to consume messages from.
+- Consume Options: Allows you to define acknowledgment behavior (e.g., noAck for auto-acknowledge).
+
+2. Message Publishing
+   The publish method provided by AmqpService sends messages to a specified queue:
+
+- Queue Name: Specifies the destination queue.
+- Message Payload: Contains the data object to be sent.
+
+3. Message Acknowledgment
+   For more control over message processing, use the ack callback to manually acknowledge messages when noAck is set to false.
+
+## Example Workflow
+
+1. Subscription: The controller subscribes to the test-queue using the @Subscribe decorator. Any message sent to this queue is automatically consumed by the mySubscription method.
+
+2. Route Trigger: The myRouteTrigger route is exposed as a GET endpoint. When accessed, it publishes a message ({ test: '123' }) to the test-queue.
+
+3. Message Consumption: The subscription method processes the published message, logs its content, and acknowledges it if required.
+
+## Advanced Configuration
+
+Extend the configuration to include custom retry policies, connection recovery, or specific consume behaviors tailored to your application.
+
+## License
+
+This library is licensed under the MIT License. Feel free to use, modify, and distribute it for your projects.

packages/amqp/jest.config.js

@@ -0,0 +1,16 @@
+module.exports = {
+  testEnvironment: 'node',
+  testPathIgnorePatterns: ['/node_modules/'],
+  coverageReporters: ['lcov', 'html'],
+  rootDir: './',
+  moduleFileExtensions: ['ts', 'tsx', 'js', 'json', 'node'],
+  globals: {
+    __DEV__: true
+  },
+  transform: {
+    '\\.(ts|tsx)$': 'ts-jest'
+  },
+  testRegex: '/src/.*\\.spec.(ts|tsx|js)$',
+  verbose: true,
+  collectCoverage: true
+};

packages/amqp/package.json

@@ -0,0 +1,40 @@
+{
+  "name": "@rhtml/amqp",
+  "version": "0.0.15",
+  "description": "AMQP module with decorators",
+  "scripts": {
+    "start": "echo START",
+    "patch": "npm run build && npm version patch && npm publish --update-readme --access public && npm run delete-dist",
+    "delete-dist": "rm -rf dist",
+    "clean": "git clean -dxf",
+    "test": "echo 'no tests specified'",
+    "lint": "npx eslint . --ext .ts",
+    "lint-fix": "npx eslint . --fix --ext .ts",
+    "build": "rm -rf dist && tsc",
+    "build-prod": "npx gapi build --local --path=./src/index.ts"
+  },
+  "repository": {
+    "type": "git",
+    "url": "[email protected]:r-html/rhtml.git"
+  },
+  "peerDependencies": {
+    "amqplib": "^0.10.5"
+  },
+  "dependencies": {},
+  "devDependencies": {
+    "@types/jest": "^24.0.18",
+    "jest": "^24.9.0",
+    "ts-jest": "25.2.1",
+    "@rhtml/di": "^0.0.132",
+    "typescript": "^5.3.3"
+  },
+  "author": "Kristiyan Tachev",
+  "license": "MIT",
+  "files": [
+    "dist"
+  ],
+  "main": "./dist/index.js",
+  "types": "./dist/index.d.ts",
+  "module": "./dist/index.js",
+  "typings": "./dist/index.d.ts"
+}

packages/amqp/src/amqp.constants.ts

@@ -0,0 +1,16 @@
+import { InjectionToken } from '@rhtml/di';
+import { Channel, Connection } from 'amqplib';
+
+/**
+ * Injection for AmqpConnection
+ */
+export const AmqpConnection = new InjectionToken<Connection>();
+export type AmqpConnection = Connection;
+
+/**
+ * Injection for AmqpChannel
+ */
+export const AmqpChannel = new InjectionToken<Channel>();
+export type AmqpChannel = Channel;
+
+export type AckCallbackFunction = () => void;

packages/amqp/src/amqp.module.ts

@@ -0,0 +1,27 @@
+import { Module, ModuleWithProviders } from '@rhtml/di';
+import amqpClient, { Connection, Options } from 'amqplib';
+
+import { AmqpChannel, AmqpConnection } from './amqp.constants';
+import { AmqpService } from './amqp.service';
+
+@Module({
+  providers: [AmqpService],
+})
+export class AmqpModule {
+  public static forRoot(config: Options.Connect): ModuleWithProviders {
+    return {
+      module: AmqpModule,
+      providers: [
+        {
+          provide: AmqpConnection,
+          useFactory: () => amqpClient.connect(config),
+        },
+        {
+          provide: AmqpChannel,
+          deps: [AmqpConnection],
+          useFactory: (connection: Connection) => connection.createChannel(),
+        },
+      ],
+    };
+  }
+}

packages/amqp/src/amqp.service.ts

@@ -0,0 +1,35 @@
+import { Inject, Injectable } from '@rhtml/di';
+import { ConsumeMessage, Options } from 'amqplib';
+
+import { AmqpChannel } from './amqp.constants';
+
+@Injectable()
+export class AmqpService {
+  constructor(@Inject(AmqpChannel) private channel: AmqpChannel) {}
+
+  async publish<T = Record<string, string>>(
+    name: string,
+    payload: T,
+    options?: Options.AssertQueue
+  ) {
+    await this.channel.assertQueue(name, options);
+    return this.channel.sendToQueue(name, Buffer.from(JSON.stringify(payload)));
+  }
+
+  async subscribe(
+    name: string,
+    callback: (msg: ConsumeMessage, ack: () => void) => void,
+    options?: {
+      assertOptions?: Options.AssertQueue;
+      consumeOptions?: Options.Consume;
+    }
+  ) {
+    await this.channel.assertQueue(name, options?.assertOptions);
+
+    await this.channel.consume(
+      name,
+      (data) => callback(data!, () => this.channel.ack(data!)),
+      options?.consumeOptions
+    );
+  }
+}