Forem: Diyorbek Sobirov

TCP-based P2P data transmission

Diyorbek Sobirov — Thu, 26 Dec 2024 14:18:03 +0000

ClapPeer - is a library for creating a distributed network of nodes that supports message exchange between them. Nodes can exchange both plain and encrypted messages.
https://github.com/DIY0R/clap-peer

Features
Example
Connecting to a Node
Sending Messages
Receiving Messages
Error Handling

Features:

Message transmission using intermediate nodes.
TTL mechanism to prevent message loops.
Encrypted message exchange (RSA + AES).

Example

const {
  ClapPeer,
  DM,
  CRYPTO_DM,
  ERROR,
  INVALID_CRYPTO_DM,
} = require('clap-peer');

//Node - 1
const node_1 = new ClapPeer(1001, 'A');
node_1.on(DM, msg => console.log(msg));
node_1.on(CRYPTO_DM, msg => console.log(msg));

//Node - 2
const node_2 = new ClapPeer(1002, 'B');
node_2.connect({ host: '127.0.0.1', port: 1001 });
node_2
  .send(node_1.nodeId, { hello: 'hello crypto' })
  .catch(error => console.log(error));
node_2.publish(node_1.nodeId, { hello: 'just hello' });

Connecting to a Node

You can use either of two methods to connect to a node: via the .connect method or by passing a configuration object when creating the node. Choose the method based on how you want to structure your code. Here are the two approaches:

1. Connecting using the `.connect()` method:

const node = new ClapPeer(1001, 'A');
node.connect({ host: '127.0.0.1', port: 1002 });

Here, we create a node and then call the .connect() method, passing the host and port parameters. This allows you to separate the node creation and connection logic.

2. Connecting using a configuration object during node creation:

const node = new ClapPeer(1002, 'A', { host: '127.0.0.1', port: 1002 });

In this case, we pass the connection parameters directly when creating the ClapPeer object. This method is convenient if you need to connect to the node immediately upon creation.

Sending Messages

1. `send` — Sending an Encrypted Message

The send method is used to send encrypted messages. Before sending, it checks if the target node's public key is available:

If the key is available, the message is encrypted and sent immediately.
If the key is not available, the node requests the public key from the target node, then encrypts and sends the message.

Example:

node.send(node_2.nodeId, { text: 'Hello, secure world!' }).catch(error => {
  console.log(error);
});

2. `publish` — Sending a Plain Message

The publish method sends messages without encryption. It simply forwards the data to the specified node.

Example:

node.publish(node_2.nodeId, {
  text: 'Hello, open world!',
});

Differences Between `send` and `publish`:

Method	Encryption	Public Key Check	Request Public Key if Needed	Routing Through Intermediate Nodes
`send`	✅	✅	✅	✅
`publish`	❌	❌	❌	✅

Receiving Messages

A node can subscribe to events to handle both plain and encrypted messages.

1. Handling Messages from `publish` (`DM`)

DM type messages are generated when another node calls the publish method. These messages are transmitted unencrypted.

node.on(DM, msg => console.log(msg));

2. Handling Messages from `send` (`CRYPTO_DM`)

Messages of type CRYPTO_DM are generated when the send method is called by another node. These messages are received in encrypted form.

node.on(CRYPTO_DM, msg => console.log(msg));

Differences Between `DM` and `CRYPTO_DM`:

Event	Method That Generates the Message	Message Type	Description
`DM`	`publish`	Plain text message	Handled as a regular, unencrypted message.
`CRYPTO_DM`	`send`	Encrypted message	Sent and received in an encrypted form.

Error Handling

You can subscribe to the ERROR event to handle all errors.

node.on(ERROR, (messageError, originalError) => {
  console.log(messageError);
  console.error(originalError);
});

The messageError.message parameter may contain one of the following messages:

TIMEOUT_ERROR_MESSAGE:

Neighbor check timed out after ${TIMEOUT_DURATION / 1000} seconds
TIMEOUT_ERROR_REQUEST:

RSA key retrieval timed out.
DECRYPT_ERROR:

Unable to decrypt the message.
SEND_ERROR:

Failed to send the data.
PUBLISH_ERROR:

Failed to publish the data.

Library for storing graphs in files and making queries to them.🔗

Diyorbek Sobirov — Thu, 26 Dec 2024 14:15:50 +0000

The file-graph class provides an abstract implementation for graph operations. It supports graph manipulation, including creating, updating, and deleting vertices, creating arcs and edges, and traversing or querying the graph structure.
https://github.com/DIY0R/file-graph

Installation

To install file-graph from npm, run the following command:

npm install file-graph

Alternatively, you can install it from the GitHub registry.

Creating a Graph Instance

const graph = FileGraph('graph.txt');

FileGraph accepts a path where the graph data will be stored.The returned object graph is an instance of the FileGraphAbstract class, which provides methods for working with the graph.

File Structure

After initializing the graph at the specified path, a file will be created where each line represents a vertex. It is recommended not to modify the file contents manually.

Example of a vertex record: graph.txt

{
  "id": "33ef29be-adaa-4509-b306-62a32db7310e", // UUID - unique identifier of the vertex
  "data": { ... }, // The data associated with the vertex
  "links": [] // List of UUIDs of vertices that this vertex is connected to
}

Public API - Asynchronous Methods

Below is an overview of the methods, their parameters, return values, and examples of how to use them.

`createVertex<T extends object>(data: T): Promise<IVertex<T>>`

Creates a vertex (node) in the graph with the provided data.

Parameters:
- data: T - An object representing the vertex data.
Returns:
- A promise that resolves to the created vertex object { id, data, links }.

Example:

const data = { name: 'Alice', age: 30 };
const createdVertex = await graph.createVertex(data);
console.log(createdVertex);
// Output: { id: 'some-unique-id', data: { name: 'Alice', age: 30 }, links: [] }

`createVertices<T extends object>(data: T[]): Promise<IVertex<T>[]>`

Creates multiple vertices at once.

Parameters:
- data: T[] - An array of objects representing the data for each vertex.
Returns:
- A promise that resolves to an array of created vertex objects.

Example:

const data = [{ name: 'Alice' }, { name: 'Bob' }];
const createdVertices = await graph.createVertices(data);
console.log(createdVertices);
// Output: [{ id: 'id1', data: { name: 'Alice' }, links: [] }, { id: 'id2', data: { name: 'Bob' }, links: [] }]

`updateVertex<T extends object>(updater: IUpdater<T>): Promise<boolean>`

Updates a vertex that matches the given condition. The updater function defines the logic for finding and modifying the vertex.

Parameters:
- updater: IUpdater<T> - A function that takes a vertex and returns an updated vertex if it matches the condition.
Returns:
- A promise that resolves to true if the update was successful, otherwise false.

Example:

const isUpdated = await graph.updateVertex(
  vertex =>
    vertex.id === 'some-unique-id' && { data: { name: 'Alice Updated' } },
);
console.log(isUpdated); // true

`deleteVertex<T extends object>(predicate: IPredicate<T>): Promise<boolean>`

Deletes a vertex that matches the given condition.

Parameters:
- predicate: IPredicate<T> - A function that returns true if the vertex should be deleted.
Returns:
- A promise that resolves to true if the deletion was successful, otherwise false.

Example:

const isDeleted = await graph.deleteVertex(
  vertex => vertex.id === 'some-unique-id',
);
console.log(isDeleted); // true

`findOne<T extends object>(predicate: IPredicate<T>): Promise<IVertex<T> | null>`

Finds a single vertex that matches the given condition.

Parameters:
- predicate: IPredicate<T> - A function that returns true if the vertex matches the search condition.
Returns:
- A promise that resolves to the matching vertex object, or null if no match is found.

Example:

const foundVertex = await graph.findOne(vertex => vertex.data.name === 'Alice');
console.log(foundVertex);
// Output: { id: 'some-unique-id', data: { name: 'Alice', age: 30 }, links: [] }

`findAll<T extends object>(predicate: IPredicate<T>): Promise<IVertex<T>[]>`

Finds all vertices that match the given condition.

Parameters:
- predicate: IPredicate<T> - A function that returns true for each vertex that matches the search condition.
Returns:
- A promise that resolves to an array of matching vertex objects.

Example:

const foundVertices = await graph.findAll(
  vertex => vertex.data.name === 'Alice',
);
console.log(foundVertices);
// Output: [{ id: 'id1', data: { name: 'Alice', age: 30 }, links: [] }, { id: 'id2', data: { name: 'Alice', age: 25 }, links: [] }]

`forEachVertex<T extends object>(callbackVertex: ICallbackVertex<T>): Promise<void>`

Iterates over each vertex in the graph and applies the provided callback function.

Parameters:
- callbackVertex: ICallbackVertex<T> - A callback function that is invoked for each vertex. If it returns true, the iteration stops.
Returns:
- A promise that resolves when the iteration is complete.

Example:

await graph.forEachVertex(vertex => {
  console.log(vertex);
  // Stop iteration if the vertex's name is 'Alice'
  return vertex.data.name === 'Alice';
});

`createEdge(ids: IUuidArray): Promise<boolean>`

Creates edges (links) between the specified vertices.

Parameters:
- ids: IUuidArray - An array of vertex IDs between which to create edges.
Returns:
- A promise that resolves to true if the edge creation was successful.

Example:

const isEdgeCreated = await graph.createEdge(['id1', 'id2', 'id3']);
console.log(isEdgeCreated); // true

`removeEdge(ids: IUuidArray): Promise<boolean>`

Removes the edges (links) between the specified vertices in the graph.

Parameters
- ids: IUuidArray - An array of vertex IDs. The edges between the vertices specified by these IDs will be removed.
Returns
- A promise that resolves to true if the edge removal was successful.

Example:

const isEdgeRemoved = await graph.removeEdge(['id1', 'id2', 'id3']);
console.log(isEdgeRemoved); // true

`createArc(sourceVertexId: uuidType, targetVertexId: uuidType): Promise<boolean>`

Creates an arc (directed edge) between two vertices.

Parameters:
- sourceVertexId: uuidType - The ID of the source vertex.
- targetVertexId: uuidType - The ID of the target vertex.
Returns:
- A promise that resolves to true if the arc creation was successful.

Example:

const isArcCreated = await graph.createArc('id1', 'id2');
console.log(isArcCreated); // true

`createArcs(ids: IUuidArray): Promise<boolean>`

Creates arcs (directed edges) between multiple vertices in the specified order.

Parameters:
- ids: IUuidArray - An array of vertex IDs.
Returns:
- A promise that resolves to true if the arcs creation was successful.

Example:

const isArcsCreated = await graph.createArcs(['id1', 'id2', 'id3']);
console.log(isArcsCreated); // true

`removeArc(sourceVertexId: uuidType, targetVertexId: uuidType): Promise<boolean>`

Removes an arc (edge) between two vertices.

Parameters:
- sourceVertexId: uuidType - The ID of the source vertex.
- targetVertexId: uuidType - The ID of the target vertex.
Returns:
- A promise that resolves to true if the arc was successfully removed.

Example:

const isArcRemoved = await graph.removeArc('id1', 'id2');
console.log(isArcRemoved); // true

`hasArc(sourceVertexId: uuidType, targetVertexId: uuidType): Promise<boolean>`

Checks if an arc exists between two vertices.

Parameters:
- sourceVertexId: uuidType - The ID of the source vertex.
- targetVertexId: uuidType - The ID of the target vertex.
Returns:
- A promise that resolves to true if the arc exists, otherwise false.

Example:

const hasArc = await graph.hasArc('id1', 'id2');
console.log(hasArc); // true or false

`findUpToLevel<T extends object>(vertexId: uuidType, maxLevel?: number): Promise<IVertexTree<T>[]>`

Retrieves vertices up to a specified depth level from a starting vertex.

Parameters:
- vertexId: uuidType - The ID of the starting vertex.
- maxLevel?: number - (Optional) The depth level to limit the search.
Returns:
- A promise that resolves to an array of vertices up to the specified level.

Example:

const graphTree = await graph.findUpToLevel('id1', 2);
console.log(graphTree);
/* Output: [
{ id: 'id1',data:{...},links:[...],level: 0 }, 
{ id: 'id2',data:{...},links:[...], level: 1 }, 
{ id: 'id3', data:{...},links:[...],level: 2 }
 ];  */

`searchVerticesFrom<T extends object>(vertexId: uuidType, predicate: IPredicate<T>): Promise<IVertex<T>[]>`

Performs a search starting from the given vertex and returns vertices that match the predicate.

Parameters:
- vertexId: uuidType - The ID of the starting vertex.
- predicate: IPredicate<T> - A function used to evaluate each vertex. Only vertices that satisfy the predicate will be returned.
Returns:
- A promise that resolves to an array of matching vertices.

Example:

const matchingVertices = await graph.searchVerticesFrom(
  'id1',
  vertex => vertex.data.age > 25,
);
console.log(matchingVertices);
// Output: [{ id: 'id2', data: { name: 'Alice', age: 30 }, links: [] }]

`hasPath(sourceVertexId: uuidType, targetVertexId: uuidType): Promise<boolean>`

Checks if a path exists between two vertices using Depth-First Search (DFS).

Parameters:
- sourceVertexId: uuidType - The ID of the source vertex.
- targetVertexId: uuidType - The ID of the target vertex.
Returns:
- A promise that resolves to true if a path exists between the vertices, otherwise false.

Example:

const pathExists = await graph.hasPath('id1', 'id3');
console.log(pathExists); // true or false

Library for building NestJS microservices with RabbitMQ 🐇

Diyorbek Sobirov — Thu, 26 Dec 2024 14:14:17 +0000

This library simplifies interaction with RabbitMQ using a modular approach, where each module is linked to a specific exchange and queue. It is ideal for scalable applications, such as API Gateways. The library offers flexible initialization methods (ForRoot, ForRootAsync), supports RPC (send), notifications (notify), direct queue messaging, and provides middleware, validation, and interceptors for message processing. Additional features include serialization/deserialization, health check tools, and advanced error handling (RMQErrorHandler). It also includes extensive configuration options (ForFeature, ForFeatureAsync) and support for advanced RabbitMQ pattern topics.

https://github.com/DIY0R/nestjs-rabbitmq/

Contents

Installation
Module Initialization
- ForRoot
- ForRootAsync - Async initialization
- Connect via URI
RmqGlobalService
- notify
- send (RPC)
- Send to Queue
- Access to the channel
- Serialization/Deserialization
Module-Specific Configuration
- ForFeature
- ForFeatureAsync - Async initialization
- Receiving messages
- Validation
- Middleware
- Interceptor
- RmqService
- send(RPC) and notify - methods
- healthCheck, ack and nack - methods
- Serialization/Deserialization in specific module
Throw error and handling
- RMQErrorHandler
- Catch Error
- Throw error

Installation

Start by installing the @diy0r/nestjs-rabbitmq package:

npm i @diy0r/nestjs-rabbitmq

Module Initialization

In your root module, import RmqModule:

ForRoot

import { RmqModule, IRMQExtendedOptions } from '@diy0r/nestjs-rabbitmq';

const extendedOptions: IRMQExtendedOptions = {
  typeChannel: TypeChannel.CONFIR_CHANNEL,
  globalBroker: {
    replyTo: {
      queue: '',
      options: { exclusive: true },
      consumeOptions: { noAck: true },
    },
    //..
  },
  //...
};

@Module({
  imports: [
    RmqModule.forRoot({
      connectOptions: {
        username: 'username',
        password: 'password',
        hostname: 'localhost',
        port: 5672,
        vhost: '/',
        protocol: 'amqp',
      },
      extendedOptions, //optional
    }),
  ],
  providers: [SomeService],
})
export class AppModule {}

ForRootAsync (Async initialization)

@Module({
  imports: [
    RmqModule.forRootAsync({
      useFactory: async (...providers: any[]) => ({
        connectOptions: {
          username: 'username',
          password: 'password',
          hostname: 'localhost',
          port: 5672,
          vhost: '/',
          protocol: 'amqp',
        },
        extendedOptions, //optional
      }),
      inject: [],
      imports: [],
    }),
  ],
  providers: [SomeService],
})
export class AppModule {}

Connect via URI

You can also connect using the standard URI. For more information, see the RabbitMQ URI Specification.

RmqModule.forRoot({
  connectOptions: 'amqp://username:password@localhost:5672/',
  extendedOptions, //optional
});

In forRoot({ connectOptions, extendedOptions }), we pass an object consisting of two parameters. The first, connectOptions, is used for connections, and the second, extendedOptions, is for additional settings.

connectOptions: IRMQConnectConfig - standard parameters amqlib.connect for connection. Parameters can be an object or a URI.
extendedOptions: IRMQExtendedOptions - environment settings (optional)
- typeChannel - channel type, default is TypeChannel.CHANNEL
- prefetch - Configuration for message prefetching. See
- count - The maximum number of messages that can be sent over the channel before receiving an acknowledgment. A falsy value indicates no limit.
- isGlobal - If true, applies the prefetch limit to the entire channel (for versions before RabbitMQ 3.3.0). For newer versions, this parameter is ignored as prefetch is applied per consumer.
- socketOptions - here you can configure SSL/TLS. See SSL/TLS and Client properties
- globalBroker - specify if you want to use RPC from RmqGlobalService
- replyTo - needed for setting up the replyTo queue
  - queue - queue name. It's recommended to leave it as an empty string so RabbitMQ will automatically generate a unique name.
  - options - options passed as the second parameter in assertQueue
  - consumeOptions - configure the consumer with a callback that will be invoked with each message. See
  - errorHandler (class) - custom error handler for dealing with errors from replies that extends RMQErrorHandler.
- serDes (object) - The serDes parameter is an object that defines serialization and deserialization functions for handling messages. See
- messageTimeout - Number of milliseconds the send method will wait for the response before a timeout error. Default is 40,000.
- serviceName - an arbitrary identifier for the originating application

We recommend specifying the TypeChannel.CONFIR_CHANNEL channel type to get more accurate statuses for your requests.

RmqGlobalService

After successful connection, RmqGlobalService will be available globally and you will be able to import your module from any part of your application and send messages.

@Injectable()
export class SomeService {
  constructor(private readonly rmqGlobalService: RmqGlobalService) {}

  //...
}

We recommend specifying interfaces for the objects you send and receive.

interface IReply {
  message: object;
}

interface ISend {
  age: number;
  name: string;
}

Method `notify` If you want to just notify services

async sendNotify(obj: ISend) {
    const { status } = await this.rmqGlobalService.notify<ISend>(
      'exchange',
      'user.login',
      obj,
    );
    return status;
  }

'this.rmqGlobalService.notify(exchange, routingKey, options)' - asynchronous request that returns the message status
- exchange - exchange name
- routingKey - routing key
- options - standard publish options (optional)

Method `send`

If you have defined globalBroker in forRoot, you will have access to the RPC method send,otherwise, you will catch an error when calling it.

  async sendMessage(obj: ISend) {
    const message = await this.rmqGlobalService.send<ISend, IReply>(
      'exchange name',
      'user.rpc',
      obj,
    );
    return message;
  }

'this.rmqGlobalService.send(exchange, routingKey, options)' - asynchronous request.
- exchange - exchange name
- routingKey - routing key
- options - standard publish options (optional)
- timeout - if supplied, the message will have its own timeout.(custom parameter)

Method `sendToQueue`

Unlike the standard sendToQueue method from amqlib, this asynchronous method differs in that messages go through Serialization before being sent.

 async sendToQueue(queue: string, obj:ISend) {
    const status = await this.rmqGlobalService.sendToQueue<ISend>(queue, obj);
    return status;
  }

Return either true, meaning “keep sending”, or false, meaning “please wait for a ‘drain’ event”.See Flow control

'this.rmqGlobalService.sendToQueue()' - asynchronous function
- queue - queue name

Channel

If you want access to the standard RabbitMQ channel, you always have access to async getter this.rmqGlobalService.channel and all its methods See.

 async getChannel() {
    const channel: Channel | ConfirmChannel = await this.rmqGlobalService.channel;
    return channel;
  }

Serialization/Deserialization

By default, messages are parsed using the JSON.parse method when they are received and converted to a string using JSON.stringify when they are published. If you want to change this behavior you can use your own parsers.

const extendedOptions: IRMQExtendedOptions = {
  typeChannel: TypeChannel.CONFIR_CHANNEL,
  globalBroker: {
    serDes: {
      deserialize: (message: Buffer): any => OtherDeserializer(message),
      serialize: (message: any): Buffer => OtherSerialize(message),
    },
    //...
  },
  //...
};

Method `deserialize`

The deserialize method is responsible for converting a message from its raw format (typically a Buffer in Node.js) into a JavaScript object or another suitable format for processing.

Method `serialize`

The serialize method is responsible for converting a JavaScript object or any other data format into a raw format (typically a Buffer) suitable for transmission over RabbitMQ.

In this case, it will be applied to all methods of RmqGlobalService.

Module-Specific Configuration

The forFeature and forFeatureAsync methods allow configuring RabbitMQ parameters at the specific module level in your NestJS application. This is useful for setting up different exchanges, queues, and other RabbitMQ parameters specific to that module.

forFeature

To use forFeature in your module, import RmqModule and configure the parameters in the forFeature method.

@Module({
  imports: [
    RmqModule.forFeature({
      exchange: {
        exchange: 'exchange_name',
        type: 'topic',
        options: {
          durable: true,
          autoDelete: true,
        },
      },
      queue: {
        queue: 'queue',
        options: { durable: true },
        consumeOptions: { noAck: false },
      },
      replyTo: {
        queue: '',
        options: { exclusive: true },
        consumeOptions: { noAck: true },
      },
    }),
  ],
  providers: [MyService, MyRmqEvents],
})
export class MyModule {}

forFeatureAsync - Async initialization

RmqModule.forFeatureAsync({
  inject: [],
  imports: [],
  useFactory: async () => ({
    exchange,
    queue,
    replyTo,
    //...
  }),

  interceptors: [],
  //..
});

useFactory should return an object of type IModuleBroker, and serdes, interceptors, and middlewares come after useFactory, not inside it, because these parameters are needed to extend the functionality. That's why they come after it.

In the forFeature(ModuleBroker) method, you need to pass an object of type IModuleBroker:

exchange - exchange parameters
- exchange - The name of the exchange, which should be unique.
- type - exchange type (Currently only the topic is supported)
- options - The default queue parameters from Options.AssertExchange.(optional) See Channel.assertExchange.
queue - An object containing the queue parameters. (optional)
- queue - The name of the queue, which should be unique.
- options - The default queue parameters from Options.AssertQueue. See Channel AssertQueue.
- consumeOptions - configure the consumer with a callback that will be invoked with each message. See Channel Consume
replyTo - needed for setting up the replyTo queue (optional)
- queue - The name of the queue, which should be unique.
- options - The default queue parameters from Options.AssertQueue. See Channel AssertQueue.
- consumeOptions - configure the consumer with a callback that will be invoked with each message. See Channel Consume
- errorHandler (class) - custom error handler for dealing with errors from replies that extends RMQErrorHandler.
serviceName - The name of your service. (optional)
messageTimeout - The timeout for waiting for a response, with a default value of 40000ms. (optional)

Receiving messages

In @diy0r/nestjs-rabbitmq, requests are processed through the @MessageRoute and @MessageNonRoute decorators. You need to declare the queue to use these decorators.

@Injectable()
export class MyRmqEvents {
  constructor(private readonly rmqService: RmqService) {}
  @MessageRoute('text.text')
  received(obj: any, consumeMessage: ConsumeMessage) {
    this.rmqService.ack(consumeMessage); // call if 'consumeOptions: { noAck: false }'
    return { message: obj };
  }
  @MessageRoute('*.*.rpc') // subscribe with a pattern!
  receivedTopic(obj: any, consumeMessage: ConsumeMessage) {
    return { message: obj };
  }
  @MessageRoute('rpc.#') // subscribe with a pattern!
  receivedTopicPattern(obj: any, consumeMessage: ConsumeMessage) {
    //...
    return { message: obj };
  }
  @MessageRoute('*.rpc.mix.#') //subscribe with a mix pattern!
  recivedMixTopic(obj: any, consumeMessage: ConsumeMessage) {
    return { message: obj };
  }
  @MessageRoute('notify.rpc')
  receivedTopicNotify(obj: any, consumeMessage: ConsumeMessage) {
    this.rmqService.ack(consumeMessage);
    //...
  }
  @MessageNonRoute()
  receivedNonRoute(obj: any, consumeMessage: ConsumeMessage) {
    this.rmqService.ack(consumeMessage);
    return { message: obj };
  }
}

@MessageRoute(RoutingKey: string)
method(message: ISend, consumeMessage: ConsumeMessage) {}

message - what the listener receives
consumeMessage - more information from the message (not just content) from amqlib

The @MessageRoute decorator will automatically bind the queue and RoutingKey to the exchange specified in the forFeature method, and this routing key will only be visible inside the specific module (e.g., MyModule). Just like magic!

In some cases, when a non-processable message arrives in the queue, it will stay there forever unless it is processed. In such cases, the @MessageNonRoute decorator can help. The method bound to this decorator will be called only when the request cannot find a bound routing key to the exchange.

If you want to manually acknowledge messages, set consumeOptions: { noAck: false } in the queue. If you want the library to automatically acknowledge messages, set noAck: true, so you don't have to explicitly call ack.

Here’s a more polished version of your text:

Validation

@diy0r/nestjs-rabbitmq integrates seamlessly with class-validator to validate incoming messages. To use it, annotate your route method with @RMQValidate().

@MessageRoute('message.valid')
@RMQValidate()
getValidMessage(obj: MyClass, consumeMessage: ConsumeMessage) {
  this.rmqService.ack(consumeMessage);
  return { message: obj };
}

Here, MyClass represents the validation schema:

import { IsInt, IsString, MaxLength } from 'class-validator';

export class MyClass {
  @IsString()
  @MaxLength(5, { message: 'The name must be less than 5 characters.' })
  name: string;

  @IsInt()
  age: number;
}

If the input message fails validation, the library will immediately send back an error without invoking your method, middlewares and interceptors.

Middleware

Middleware allows you to execute additional logic before message processing.You can declare middleware at various levels, including the module level, provider level, and specific endpoint level.
It's important to note that the middleware runs before any interceptors.

To declare an middleware, implement the IRmqMiddleware abstract class.

export class EventMiddleware implements IRmqMiddleware {
  async use(message: ConsumeMessage, content: any): Promise<void | any> {
    content.args = '1,2,...';
  }
}

or you can directly return:

export class EventMiddleware implements IRmqMiddleware {
  async use(message: ConsumeMessage, content: any): Promise<void | any> {
    if (content.args !== '2,1') return { status: 'error' };
  }
}

If nothing is returned, the execution of the middleware chain will continue.

EventMiddleware implements the use method, which takes two parameters: message, which is the standard ConsumeMessage from amqplib, and content, which is your message that has passed through deserialization.

Module Level

At the module level, you can declare middleware that will apply to all message handlers within that module.

@Module({
  imports: [
    RmqModule.forFeature({
      exchange: {
        /* exchange parameters */
      },
      queue: {
        /* queue parameters */
      },
      middlewares: [EventMiddleware, OtherMiddleware], // Middleware is applied to all handlers in the module
    }),
  ],
  providers: [MyService, MyRmqEvents],
})
export class MyModule {}

Provider Level

At the provider level, you can declare middleware that will apply to all message handlers within that provider.

@Injectable()
@RmqMiddleware(EventMiddleware) // Middleware is applied to all methods in the MyRmqEvents
export class MyRmqEvents {
  @MessageRoute('text.text')
  received(obj: any, consumeMessage: ConsumeMessage) {
    return { message: obj };
  }

  // Class methods ...
}

Specific Endpoint Level

Middleware can be applied at the specific endpoint (method) level.

@Injectable()
export class MyRmqEvents {
  @MessageRoute('text.text')
  @RmqMiddleware(EventMiddleware) // Middleware is applied only to this method
  handleTextText(obj: any, consumeMessage: ConsumeMessage) {
    return { message: obj };
  }
}

Initially, middleware at the module level will be invoked, then at the class level, followed by the endpoint level.

Interceptor

Interceptors allow you to intercept and modify the processing of a message both before and after it is processed by a handler, just like Interceptors from Nest.js. You can declare interceptors at various levels, including the module level, the provider level, and the specific endpoint level.
It's important to note that interceptors are executed after the middleware!

To declare an interceptor, implement the abstract class IRmqInterceptor with a decorator @Injectable.

@Injectable()
export class EventInterceptor implements IRmqInterceptor {
  constructor(private readonly rmqSerivce: RmqService) {}
  async intercept(message: ConsumeMessage, content: any): Promise<ReverseFunction> {
    console.log('Before...');
    return async (content: any, message: ConsumeMessage) => {
      console.log(`After... ${Date.now() - now}ms`);
    };
  }
}

EventInterceptor implements the intercept method, which takes two parameters: message, which is the standard ConsumeMessage from amqplib, and content, which is your message that has passed through deserialization. The intercept method returns an asynchronous function that will be invoked after processing.

Module Level

At the module level, you can declare interceptors that will apply to all message handlers within that module.

@Module({
  imports: [
    RmqModule.forFeature({
      exchange: {
        /* exchange parameters */
      },
      queue: {
        /* queue parameters */
      },
      interceptors: [EventInterceptor, OtherInterceptor], // Interceptors are applied to all handlers in the module!
    }),
  ],
  providers: [MyService, MyRmqEvents],
})
export class MyModule {}

The interceptor will be invoked from left to right.

Provider Level

At the provider level, you can declare interceptors that will apply to all message handlers within that provider.

@Injectable()
@RmqInterceptor(EventInterceptor) // Interceptor is applied to all methods in the MyRmqEvents
export class MyRmqEvents {
  @MessageRoute('text.text')
  received(obj: any, consumeMessage: ConsumeMessage) {
    return { message: obj };
  }

  // Class methods ...
}

Specific Endpoint Level

At the specific endpoint (method) level, allowing you to configure interceptor for each endpoint individually.

@Injectable()
export class MyRmqEvents {
  @MessageRoute('text.text')
  @RmqInterceptor(EventInterceptor) // Interceptor is applied only to this method
  handleTextText(obj: any, consumeMessage: ConsumeMessage) {
    return { message: obj };
  }
}

Initially, interceptors at the module level will be invoked, then at the class level, followed by the endpoint level, and then in reverse order.

RmqService

If you just want to send messages and receive responses, simply specify the exchange and replyTo, and you will have access to all methods from RmqService as shown below.
It is mainly useful when you are writing an API gateway and want to maintain modularity.

send and notify - methods

@Injectable()
export class MyService {
  constructor(private readonly rmqService: RmqService) {}

  async sendMessage(obj: ISend) {
    const message = await this.rmqGlobalService.send<ISend, IReply>(
      'user.rpc',
      obj,
    );
    return message;
  }

  async sendNotifyService(obj: Record<string, any>) {
    const message = await this.rmqService.notify<ISend>('notify.rpc', obj);
    return message;
  }

'this.rmqService.send(routingKey, options)' - asynchronous request
- routingKey - routing key
- options - standard publish options (optional) - timeout - if supplied, the message will have its own timeout.(custom parameter)

To use send, you must have replyTo declared in forFeature.

'this.rmqService.notify(routingKey, options)' - asynchronous request that returns the message status
- routingKey - routing key
- options - standard publish options (optional)

Note that in these examples we do not specify the exchange in the requests because the send and notify methods use the exchange specified on MyModule in method forFeature.

healthCheck,ack and nack - methods

const isConnected = this.rmqService.healthCheck();

'this.rmqService.healthCheck()' - Check if you are still connected to RMQ.

this.rmqService.ack(message);

'this.rmqService.ack(message:ConsumeMessage)' - Acknowledge the given message, or all messages up to and including the given message. See

this.rmqService.nack(message);

'this.rmqService.nack(message:ConsumeMessage)' - Reject a message. This instructs the server to either requeue the message or throw it away (which may result in it being dead-lettered). See

Serialization/Deserialization in specific module

The serDes allows for both the deserialization and serialization of messages.

It's important to note the execution order, as this is performed only once.
When a message reaches an endpoint, the framework checks for the @SerDes decorator on method(endpoint). If found, the methods specified there are used. If not, it searches for the decorator in the parent class(Provider). If the decorator is still not found, it falls back to using the serDes object from your module. If none of these are defined, the default serialization and deserialization are used.

Module Level

At the module level, the SerDes configuration applies to all message handlers within that module. This is useful for ensuring consistent message processing across the entire module.

@Module({
  imports: [
    RmqModule.forFeature({
      exchange: {
        /* exchange parameters */
      },
      queue: {
        /* queue parameters */
      },
      serDes: {
        deserialize: (message: Buffer): any => OtherDeserializer(message),
        serialize: (message: any): Buffer => OtherSerializer(message),
      }, // serDes is applied to all handlers in the module!
    }),
  ],
})
export class MyModule {}

Provider Level

You can also configure SerDes at the provider level, applying it to all message handlers within a specific class(provider).

@Injectable()
@SerDes({
  deserialize: (message: Buffer): any => OtherDeserializer(message),
  serialize: (message: any): Buffer => OtherSerializer(message),
}) // SerDes is applied to all methods in the MyRmqEvents
export class MyRmqEvents {
  @MessageRoute('text.text')
  received(obj: any, consumeMessage: ConsumeMessage) {
    return { message: obj };
  }

  // Class methods ...
}

Specific Endpoint Level

At the specific endpoint (method) level, allowing you to configure SerDes for each endpoint individually.

@Injectable()
export class MyRmqEvents {
  @MessageRoute('text.text')
  @SerDes({
    deserialize: (message: Buffer): any => OtherDeserializer(message),
    serialize: (message: any): Buffer => OtherSerializer(message),
  }) // SerDes is applied only to this method
  handleTextText(obj: any, consumeMessage: ConsumeMessage) {
    return { message: obj };
  }
}

Throw error and handling

If you want to use a custom error handler to handle errors in responses, use the errorHandler property in the replyTo parameters in forRoot and forFeature, and pass a class that extends RMQErrorHandler

RMQErrorHandler

If you call rmqGlobalService.send, the errorHandler from forRoot will be used, and if you call rmqService.send, the errorHandler from forFeature will be used.

export class MyRMQErrorHandler extends RMQErrorHandler {
  public static handle(headers: IRmqErrorHeaders | MessagePropertyHeaders): Error | RMQError {
    // do something ...
    return new RMQError(
      headers['-x-error'],
      headers['-x-service'],
      headers['-x-status-code'],
      headers['-x-host'],
      headers['-x-date'],
    );
  }
}

Catch Error

async sendMessage (obj: ISend){
  try {
    const message = await this.rmqGlobalService.send<ISend, IReply>(
      'user.rpc',
      obj,
    );
  } catch (error: RMQError) {
    //...
  }
};

Throw error

  @MessageRoute()
  handleTextText(obj: ISend, consumeMessage: ConsumeMessage) {
    throw new RMQError('Error message', 2);
    //or
    throw new Error('Error message');
  }

Forem: Diyorbek Sobirov

TCP-based P2P data transmission

Contents

Features:

Example

Connecting to a Node

1. Connecting using the .connect() method:

2. Connecting using a configuration object during node creation:

Sending Messages

1. send — Sending an Encrypted Message

2. publish — Sending a Plain Message

Differences Between send and publish:

Receiving Messages

1. Handling Messages from publish (DM)

2. Handling Messages from send (CRYPTO_DM)

Differences Between DM and CRYPTO_DM:

Error Handling

Library for storing graphs in files and making queries to them.🔗

Installation

Creating a Graph Instance

File Structure

Public API - Asynchronous Methods

createVertex<T extends object>(data: T): Promise<IVertex<T>>

Example:

createVertices<T extends object>(data: T[]): Promise<IVertex<T>[]>

Example:

updateVertex<T extends object>(updater: IUpdater<T>): Promise<boolean>

Example:

deleteVertex<T extends object>(predicate: IPredicate<T>): Promise<boolean>

Example:

findOne<T extends object>(predicate: IPredicate<T>): Promise<IVertex<T> | null>

Example:

findAll<T extends object>(predicate: IPredicate<T>): Promise<IVertex<T>[]>

Example:

forEachVertex<T extends object>(callbackVertex: ICallbackVertex<T>): Promise<void>

Example:

createEdge(ids: IUuidArray): Promise<boolean>

Example:

removeEdge(ids: IUuidArray): Promise<boolean>

Example:

createArc(sourceVertexId: uuidType, targetVertexId: uuidType): Promise<boolean>

Example:

createArcs(ids: IUuidArray): Promise<boolean>

Example:

removeArc(sourceVertexId: uuidType, targetVertexId: uuidType): Promise<boolean>

Example:

hasArc(sourceVertexId: uuidType, targetVertexId: uuidType): Promise<boolean>

Example:

findUpToLevel<T extends object>(vertexId: uuidType, maxLevel?: number): Promise<IVertexTree<T>[]>

Example:

searchVerticesFrom<T extends object>(vertexId: uuidType, predicate: IPredicate<T>): Promise<IVertex<T>[]>

Example:

hasPath(sourceVertexId: uuidType, targetVertexId: uuidType): Promise<boolean>

Example:

Library for building NestJS microservices with RabbitMQ 🐇

Installation

Module Initialization

ForRoot

ForRootAsync (Async initialization)

Connect via URI

RmqGlobalService

Method notify If you want to just notify services

Method send

Method sendToQueue

Channel

Serialization/Deserialization

Method deserialize

Method serialize

Module-Specific Configuration

forFeature

forFeatureAsync - Async initialization

Receiving messages

Validation

Middleware

Module Level

Provider Level

Specific Endpoint Level

Interceptor

Module Level

Provider Level

1. Connecting using the `.connect()` method:

1. `send` — Sending an Encrypted Message

2. `publish` — Sending a Plain Message

Differences Between `send` and `publish`:

1. Handling Messages from `publish` (`DM`)

2. Handling Messages from `send` (`CRYPTO_DM`)

Differences Between `DM` and `CRYPTO_DM`:

`createVertex<T extends object>(data: T): Promise<IVertex<T>>`

`createVertices<T extends object>(data: T[]): Promise<IVertex<T>[]>`

`updateVertex<T extends object>(updater: IUpdater<T>): Promise<boolean>`

`deleteVertex<T extends object>(predicate: IPredicate<T>): Promise<boolean>`

`findOne<T extends object>(predicate: IPredicate<T>): Promise<IVertex<T> | null>`

`findAll<T extends object>(predicate: IPredicate<T>): Promise<IVertex<T>[]>`

`forEachVertex<T extends object>(callbackVertex: ICallbackVertex<T>): Promise<void>`

`createEdge(ids: IUuidArray): Promise<boolean>`

`removeEdge(ids: IUuidArray): Promise<boolean>`

`createArc(sourceVertexId: uuidType, targetVertexId: uuidType): Promise<boolean>`

`createArcs(ids: IUuidArray): Promise<boolean>`

`removeArc(sourceVertexId: uuidType, targetVertexId: uuidType): Promise<boolean>`

`hasArc(sourceVertexId: uuidType, targetVertexId: uuidType): Promise<boolean>`

`findUpToLevel<T extends object>(vertexId: uuidType, maxLevel?: number): Promise<IVertexTree<T>[]>`

`searchVerticesFrom<T extends object>(vertexId: uuidType, predicate: IPredicate<T>): Promise<IVertex<T>[]>`

`hasPath(sourceVertexId: uuidType, targetVertexId: uuidType): Promise<boolean>`

Method `notify` If you want to just notify services

Method `send`

Method `sendToQueue`

Method `deserialize`

Method `serialize`