e79cdc3c-762e-40cd-acb3-21e17deb1cd9

Committed 18 Aug 2024 03:11PM UTC coverage: 92.124% (-0.09%) from 92.213%

Build # e79cdc3c-762e-40cd-acb3-21e17deb1cd9

Build Type

Pull #13485

circleci

Committed by

DylanVeldra

Commit Message

fix(core): merge req context with tenant payload in the request instance

Pull Request Pull Request #13485: fix(core): merge request context with tenant context payload in the request singleton

Run Details

2559 of 3078 branches covered (83.14%)

1 of 2 new or added lines in 2 files covered. (50.0%)

74 existing lines in 13 files now uncovered.

6737 of 7313 relevant lines covered (92.12%)

17.03 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

89.47

/packages/microservices/server/server-tcp.ts

import { Type } from '@nestjs/common';
import { isString, isUndefined } from '@nestjs/common/utils/shared.utils';
import * as net from 'net';
import { Server as NetSocket, Socket } from 'net';
import {
  CLOSE_EVENT,
  EADDRINUSE,
  ECONNREFUSED,
  ERROR_EVENT,
  MESSAGE_EVENT,
  NO_MESSAGE_HANDLER,
  TCP_DEFAULT_HOST,
  TCP_DEFAULT_PORT,
} from '../constants';
import { TcpContext } from '../ctx-host/tcp.context';
import { Transport } from '../enums';
import { JsonSocket, TcpSocket } from '../helpers';
import { createServer as tlsCreateServer } from 'tls';
import {
  CustomTransportStrategy,
  IncomingRequest,
  PacketId,
  ReadPacket,
  WritePacket,
} from '../interfaces';
import { TcpOptions } from '../interfaces/microservice-configuration.interface';
import { Server } from './server';

export class ServerTCP extends Server implements CustomTransportStrategy {
  public readonly transportId = Transport.TCP;

  protected server: NetSocket;

  private readonly port: number;
  private readonly host: string;
  private readonly socketClass: Type<TcpSocket>;
  private isExplicitlyTerminated = false;
  private retryAttemptsCount = 0;
  private tlsOptions?;

  constructor(private readonly options: TcpOptions['options']) {
    super();
    this.port = this.getOptionsProp(options, 'port') || TCP_DEFAULT_PORT;
    this.host = this.getOptionsProp(options, 'host') || TCP_DEFAULT_HOST;
    this.socketClass =
      this.getOptionsProp(options, 'socketClass') || JsonSocket;
    this.tlsOptions = this.getOptionsProp(options, 'tlsOptions');

    this.init();
    this.initializeSerializer(options);
    this.initializeDeserializer(options);
  }

  public listen(
    callback: (err?: unknown, ...optionalParams: unknown[]) => void,
  ) {
    this.server.once(ERROR_EVENT, (err: Record<string, unknown>) => {
      if (err?.code === EADDRINUSE || err?.code === ECONNREFUSED) {
        return callback(err);
      }
    });
    this.server.listen(this.port, this.host, callback as () => void);
  }

  public close() {
    this.isExplicitlyTerminated = true;

    this.server.close();
  }

  public bindHandler(socket: Socket) {
    const readSocket = this.getSocketInstance(socket);
    readSocket.on(MESSAGE_EVENT, async (msg: ReadPacket & PacketId) =>
      this.handleMessage(readSocket, msg),
    );
    readSocket.on(ERROR_EVENT, this.handleError.bind(this));
  }

  public async handleMessage(socket: TcpSocket, rawMessage: unknown) {
    const packet = await this.deserializer.deserialize(rawMessage);
    const pattern = !isString(packet.pattern)
      ? JSON.stringify(packet.pattern)
      : packet.pattern;

    const tcpContext = new TcpContext([socket, pattern]);
    if (isUndefined((packet as IncomingRequest).id)) {
      return this.handleEvent(pattern, packet, tcpContext);
    }
    const handler = this.getHandlerByPattern(pattern);
    if (!handler) {
      const status = 'error';
      const noHandlerPacket = this.serializer.serialize({
        id: (packet as IncomingRequest).id,
        status,
        err: NO_MESSAGE_HANDLER,
      });
      return socket.sendMessage(noHandlerPacket);
    }
    const response$ = this.transformToObservable(
      await handler(packet.data, tcpContext),
    );

    response$ &&
      this.send(response$, data => {
        Object.assign(data, { id: (packet as IncomingRequest).id });
        const outgoingResponse = this.serializer.serialize(
          data as WritePacket & PacketId,
        );
        socket.sendMessage(outgoingResponse);
      });
  }

  public handleClose(): undefined | number | NodeJS.Timer {
    if (
      this.isExplicitlyTerminated ||
      !this.getOptionsProp(this.options, 'retryAttempts') ||
      this.retryAttemptsCount >=
        this.getOptionsProp(this.options, 'retryAttempts')
    ) {
      return undefined;
    }
    ++this.retryAttemptsCount;
    return setTimeout(
      () => this.server.listen(this.port, this.host),
      this.getOptionsProp(this.options, 'retryDelay') || 0,
    );
  }

  private init() {
    if (this.tlsOptions) {
      // TLS enabled, use tls server
      this.server = tlsCreateServer(
        this.tlsOptions,
        this.bindHandler.bind(this),
      );
    } else {
      // TLS disabled, use net server
      this.server = net.createServer(this.bindHandler.bind(this));
    }
    this.server.on(ERROR_EVENT, this.handleError.bind(this));
    this.server.on(CLOSE_EVENT, this.handleClose.bind(this));
  }

  private getSocketInstance(socket: Socket): TcpSocket {
    return new this.socketClass(socket);
  }
}

1	import { Type } from '@nestjs/common';
2	import { isString, isUndefined } from '@nestjs/common/utils/shared.utils';	1✔
3	import * as net from 'net';	1✔
4	import { Server as NetSocket, Socket } from 'net';
5	import {	1✔
6	CLOSE_EVENT,
7	EADDRINUSE,
8	ECONNREFUSED,
9	ERROR_EVENT,
10	MESSAGE_EVENT,
11	NO_MESSAGE_HANDLER,
12	TCP_DEFAULT_HOST,
13	TCP_DEFAULT_PORT,
14	} from '../constants';
15	import { TcpContext } from '../ctx-host/tcp.context';	1✔
16	import { Transport } from '../enums';	1✔
17	import { JsonSocket, TcpSocket } from '../helpers';	1✔
18	import { createServer as tlsCreateServer } from 'tls';	1✔
19	import {
20	CustomTransportStrategy,
21	IncomingRequest,
22	PacketId,
23	ReadPacket,
24	WritePacket,
25	} from '../interfaces';
26	import { TcpOptions } from '../interfaces/microservice-configuration.interface';
27	import { Server } from './server';	1✔
28
29	export class ServerTCP extends Server implements CustomTransportStrategy {	1✔
30	public readonly transportId = Transport.TCP;	13✔
31
32	protected server: NetSocket;
33
34	private readonly port: number;
35	private readonly host: string;
36	private readonly socketClass: Type<TcpSocket>;
37	private isExplicitlyTerminated = false;	13✔
38	private retryAttemptsCount = 0;	13✔
39	private tlsOptions?;
40
41	constructor(private readonly options: TcpOptions['options']) {	13✔
42	super();	13✔
43	this.port = this.getOptionsProp(options, 'port') \|\| TCP_DEFAULT_PORT;	13✔
44	this.host = this.getOptionsProp(options, 'host') \|\| TCP_DEFAULT_HOST;	13✔
45	this.socketClass =	13✔
46	this.getOptionsProp(options, 'socketClass') \|\| JsonSocket;	26✔
47	this.tlsOptions = this.getOptionsProp(options, 'tlsOptions');	13✔
48
49	this.init();	13✔
50	this.initializeSerializer(options);	13✔
51	this.initializeDeserializer(options);	13✔
52	}
53
54	public listen(
55	callback: (err?: unknown, ...optionalParams: unknown[]) => void,
56	) {
57	this.server.once(ERROR_EVENT, (err: Record<string, unknown>) => {	1✔
UNCOV 58	if (err?.code === EADDRINUSE \|\| err?.code === ECONNREFUSED) {	×
UNCOV 59	return callback(err);	×
60	}
61	});
62	this.server.listen(this.port, this.host, callback as () => void);	1✔
63	}
64
65	public close() {
66	this.isExplicitlyTerminated = true;	1✔
67
68	this.server.close();	1✔
69	}
70
71	public bindHandler(socket: Socket) {
72	const readSocket = this.getSocketInstance(socket);	1✔
73	readSocket.on(MESSAGE_EVENT, async (msg: ReadPacket & PacketId) =>	1✔
UNCOV 74	this.handleMessage(readSocket, msg),	×
75	);
76	readSocket.on(ERROR_EVENT, this.handleError.bind(this));	1✔
77	}
78
79	public async handleMessage(socket: TcpSocket, rawMessage: unknown) {
80	const packet = await this.deserializer.deserialize(rawMessage);	2✔
81	const pattern = !isString(packet.pattern)	2✔
82	? JSON.stringify(packet.pattern)	2!
83	: packet.pattern;
84
85	const tcpContext = new TcpContext([socket, pattern]);	2✔
86	if (isUndefined((packet as IncomingRequest).id)) {	2!
UNCOV 87	return this.handleEvent(pattern, packet, tcpContext);	×
88	}
89	const handler = this.getHandlerByPattern(pattern);	2✔
90	if (!handler) {	2✔
91	const status = 'error';	1✔
92	const noHandlerPacket = this.serializer.serialize({	1✔
93	id: (packet as IncomingRequest).id,
94	status,
95	err: NO_MESSAGE_HANDLER,
96	});
97	return socket.sendMessage(noHandlerPacket);	1✔
98	}
99	const response$ = this.transformToObservable(	1✔
100	await handler(packet.data, tcpContext),
101	);
102
103	response$ &&	1✔
104	this.send(response$, data => {
105	Object.assign(data, { id: (packet as IncomingRequest).id });	1✔
106	const outgoingResponse = this.serializer.serialize(	1✔
107	data as WritePacket & PacketId,
108	);
109	socket.sendMessage(outgoingResponse);	1✔
110	});
111	}
112
113	public handleClose(): undefined \| number \| NodeJS.Timer {
114	if (	4✔
115	this.isExplicitlyTerminated \|\|	9✔
116	!this.getOptionsProp(this.options, 'retryAttempts') \|\|
117	this.retryAttemptsCount >=
118	this.getOptionsProp(this.options, 'retryAttempts')
119	) {
120	return undefined;	3✔
121	}
122	++this.retryAttemptsCount;	1✔
123	return setTimeout(	1✔
124	() => this.server.listen(this.port, this.host),	1✔
125	this.getOptionsProp(this.options, 'retryDelay') \|\| 0,	1!
126	);
127	}
128
129	private init() {
130	if (this.tlsOptions) {	13!
131	// TLS enabled, use tls server
UNCOV 132	this.server = tlsCreateServer(	×
133	this.tlsOptions,
134	this.bindHandler.bind(this),
135	);
136	} else {
137	// TLS disabled, use net server
138	this.server = net.createServer(this.bindHandler.bind(this));	13✔
139	}
140	this.server.on(ERROR_EVENT, this.handleError.bind(this));	13✔
141	this.server.on(CLOSE_EVENT, this.handleClose.bind(this));	13✔
142	}
143
144	private getSocketInstance(socket: Socket): TcpSocket {
UNCOV 145	return new this.socketClass(socket);	×
146	}
147	}

nestjs / nest / e79cdc3c-762e-40cd-acb3-21e17deb1cd9

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous