Власні транспортери NestJS

{"id":"1","type":"request","pattern":"math.add","data":{"a":2,"b":3}}

{"id":"1","ok":true,"data":5}

{"id":"1","ok":false,"error":"Некоректні аргументи"}

{"id":"2","type":"event","pattern":"notifications.created","data":{"id":"n-1"}}

npm install @nestjs/common @nestjs/core @nestjs/microservices rxjs
npm install -D ts-node typescript @types/node

import * as net from 'node:net';
import {
  Controller,
  Injectable,
  Module,
} from '@nestjs/common';
import { NestFactory } from '@nestjs/core';
import {
  ClientProxy,
  CustomTransportStrategy,
  MessagePattern,
  ReadPacket,
  Server,
  Transport,
  WritePacket,
} from '@nestjs/microservices';
import { isObservable, lastValueFrom } from 'rxjs';

type TransportRequest = {
  id: string;
  type: 'request' | 'event';
  pattern: string;
  data: unknown;
};

type TransportResponse = {
  id: string;
  ok: boolean;
  data?: unknown;
  error?: string;
};

type LineTransportOptions = {
  host?: string;
  port: number;
};

function serialize(value: unknown): string {
  return JSON.stringify(value === undefined ? null : value);
}

async function resolveHandlerResult(value: unknown): Promise<unknown> {
  if (isObservable(value)) {
    return lastValueFrom(value);
  }

  return await value;
}

class LineServer
  extends Server
  implements CustomTransportStrategy
{
  private readonly host: string;
  private readonly port: number;
  private tcpServer?: net.Server;

  constructor(options: LineTransportOptions) {
    super();

    this.host = options.host ?? '127.0.0.1';
    this.port = options.port;
  }

  listen(callback: () => void): void {
    this.tcpServer = net.createServer((socket) => {
      let buffer = '';

      socket.setEncoding('utf8');

      socket.on('data', (chunk: string) => {
        buffer += chunk;

        let newlineIndex = buffer.indexOf('\n');

        while (newlineIndex !== -1) {
          const line = buffer.slice(0, newlineIndex);
          buffer = buffer.slice(newlineIndex + 1);

          if (line.trim().length > 0) {
            void this.handleLine(socket, line);
          }

          newlineIndex = buffer.indexOf('\n');
        }
      });
    });

    this.tcpServer.listen(this.port, this.host, callback);
  }

  close(): void {
    this.tcpServer?.close();
  }

  private async handleLine(
    socket: net.Socket,
    line: string,
  ): Promise<void> {
    let packet: TransportRequest;

    try {
      packet = JSON.parse(line) as TransportRequest;
    } catch {
      socket.write(
        `${JSON.stringify({
          ok: false,
          error: 'Некоректний JSON-пакет',
        })}\n`,
      );
      return;
    }

    const handler = this.getHandlerByPattern(packet.pattern);

    if (!handler) {
      if (packet.type === 'request') {
        const response: TransportResponse = {
          id: packet.id,
          ok: false,
          error: `Обробник для "${packet.pattern}" не знайдено`,
        };

        socket.write(`${JSON.stringify(response)}\n`);
      }

      return;
    }

    try {
      const result = await resolveHandlerResult(
        handler(packet.data, socket),
      );

      if (packet.type === 'request') {
        const response: TransportResponse = {
          id: packet.id,
          ok: true,
          data: result,
        };

        socket.write(`${JSON.stringify(response)}\n`);
      }
    } catch (error) {
      if (packet.type !== 'request') {
        return;
      }

      const message =
        error instanceof Error
          ? error.message
          : 'Невідома помилка обробника';

      const response: TransportResponse = {
        id: packet.id,
        ok: false,
        error: message,
      };

      socket.write(`${JSON.stringify(response)}\n`);
    }
  }
}

class LineClient extends ClientProxy {
  private readonly host: string;
  private readonly port: number;

  private socket?: net.Socket;
  private buffer = '';
  private sequence = 0;

  private readonly pending = new Map<
    string,
    (packet: WritePacket) => void
  >();

  constructor(options: LineTransportOptions) {
    super();

    this.host = options.host ?? '127.0.0.1';
    this.port = options.port;
  }

  connect(): Promise<net.Socket> {
    if (this.socket && !this.socket.destroyed) {
      return Promise.resolve(this.socket);
    }

    return new Promise((resolve, reject) => {
      const socket = net.createConnection({
        host: this.host,
        port: this.port,
      });

      socket.setEncoding('utf8');

      const onConnect = () => {
        socket.off('error', onInitialError);
        resolve(socket);
      };

      const onInitialError = (error: Error) => {
        socket.off('connect', onConnect);
        reject(error);
      };

      socket.once('connect', onConnect);
      socket.once('error', onInitialError);

      socket.on('data', (chunk: string) => {
        this.handleData(chunk);
      });

      socket.on('error', (error: Error) => {
        for (const callback of this.pending.values()) {
          callback({ err: error.message });
        }

        this.pending.clear();
      });

      socket.on('close', () => {
        this.socket = undefined;
      });

      this.socket = socket;
    });
  }

  close(): void {
    this.socket?.end();
    this.socket = undefined;
  }

  protected publish(
    packet: ReadPacket,
    callback: (packet: WritePacket) => void,
  ): () => void {
    const id = String(++this.sequence);

    void this.connect()
      .then((socket) => {
        this.pending.set(id, callback);

        const request: TransportRequest = {
          id,
          type: 'request',
          pattern: String(packet.pattern),
          data: packet.data,
        };

        socket.write(`${JSON.stringify(request)}\n`);
      })
      .catch((error: Error) => {
        callback({ err: error.message });
      });

    return () => {
      this.pending.delete(id);
    };
  }

  protected async dispatchEvent(
    packet: ReadPacket,
  ): Promise<void> {
    const socket = await this.connect();

    const event: TransportRequest = {
      id: String(++this.sequence),
      type: 'event',
      pattern: String(packet.pattern),
      data: packet.data,
    };

    socket.write(`${JSON.stringify(event)}\n`);
  }

  private handleData(chunk: string): void {
    this.buffer += chunk;

    let newlineIndex = this.buffer.indexOf('\n');

    while (newlineIndex !== -1) {
      const line = this.buffer.slice(0, newlineIndex);
      this.buffer = this.buffer.slice(newlineIndex + 1);

      if (line.trim().length > 0) {
        this.handleResponse(line);
      }

      newlineIndex = this.buffer.indexOf('\n');
    }
  }

  private handleResponse(line: string): void {
    let response: TransportResponse;

    try {
      response = JSON.parse(line) as TransportResponse;
    } catch {
      return;
    }

    const callback = this.pending.get(response.id);

    if (!callback) {
      return;
    }

    this.pending.delete(response.id);

    if (response.ok) {
      callback({
        response: response.data,
        isDisposed: true,
      });
    } else {
      callback({
        err: response.error ?? 'Помилка віддаленого обробника',
      });
    }
  }
}

@Controller()
class MathController {
  @MessagePattern('math.add')
  add(data: { a: number; b: number }): number {
    if (
      typeof data?.a !== 'number' ||
      typeof data?.b !== 'number'
    ) {
      throw new Error('Поля "a" та "b" мають бути числами');
    }

    return data.a + data.b;
  }
}

@Module({
  controllers: [MathController],
})
class AppModule {}

async function startServer(): Promise<void> {
  const app = await NestFactory.createMicroservice(AppModule, {
    strategy: new LineServer({
      host: '127.0.0.1',
      port: 4000,
    }),
  });

  await app.listen();
  console.log('Власний транспортер слухає 127.0.0.1:4000');
}

async function startClient(): Promise<void> {
  const client = new LineClient({
    host: '127.0.0.1',
    port: 4000,
  });

  const result = await client
    .send<number>('math.add', { a: 2, b: 3 })
    .toPromise();

  console.log('Результат:', result);

  client.close();
}

const mode = process.argv[2];

if (mode === 'server') {
  void startServer();
} else if (mode === 'client') {
  void startClient();
} else {
  console.log('Використання:');
  console.log('  npx ts-node main.ts server');
  console.log('  npx ts-node main.ts client');
}

npx ts-node main.ts server

npx ts-node main.ts client

Результат: 5

const handler = this.getHandlerByPattern(packet.pattern);

handler(packet.data, socket)

async function resolveHandlerResult(value: unknown): Promise<unknown> {
  if (isObservable(value)) {
    return lastValueFrom(value);
  }

  return await value;
}

import { Module } from '@nestjs/common';

@Module({
  providers: [
    {
      provide: 'LINE_TRANSPORT',
      useFactory: () =>
        new LineClient({
          host: '127.0.0.1',
          port: 4000,
        }),
    },
  ],
  exports: ['LINE_TRANSPORT'],
})
export class TransportModule {}

import { Inject, Injectable } from '@nestjs/common';
import { ClientProxy } from '@nestjs/microservices';
import { firstValueFrom } from 'rxjs';

@Injectable()
export class CalculatorService {
  constructor(
    @Inject('LINE_TRANSPORT')
    private readonly client: ClientProxy,
  ) {}

  async add(a: number, b: number): Promise<number> {
    return firstValueFrom(
      this.client.send<number>('math.add', { a, b }),
    );
  }
}

const result$ = client.send('math.add', { a: 2, b: 3 });

client.emit('notifications.created', {
  id: 'notification-1',
});

let buffer = '';

socket.on('data', (chunk: string) => {
  buffer += chunk;

  let newlineIndex = buffer.indexOf('\n');

  while (newlineIndex !== -1) {
    const line = buffer.slice(0, newlineIndex);
    buffer = buffer.slice(newlineIndex + 1);

    // Обробка одного повного повідомлення

    newlineIndex = buffer.indexOf('\n');
  }
});

{
  id: '1',
  ok: false,
  error: 'Текст помилки'
}

callback({
  err: response.error ?? 'Помилка віддаленого обробника',
});

type TransportError = {
  code: string;
  message: string;
  details?: unknown;
};

if (this.socket && !this.socket.destroyed) {
  return Promise.resolve(this.socket);
}