f7142b24-e394-4028-89c6-717e281431d1

Committed 25 Nov 2024 01:12PM UTC coverage: 90.141% (-1.8%) from 91.989%

Build # f7142b24-e394-4028-89c6-717e281431d1

Build Type

Pull #14177

circleci

Committed by

web-flow

Commit Message

Merge pull request #14200 from nestjs/feat/allow-queue-per-handler

feat(microservices): support nats queue per handler

Pull Request Pull Request #14177: release: version 11.0.0

Run Details

2612 of 3236 branches covered (80.72%)

496 of 704 new or added lines in 48 files covered. (70.45%)

79 existing lines in 11 files now uncovered.

6985 of 7749 relevant lines covered (90.14%)

16.31 hits per line

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

95.35

/packages/microservices/helpers/json-socket.ts

import { Buffer } from 'buffer';
import { StringDecoder } from 'string_decoder';
import { CorruptedPacketLengthException } from '../errors/corrupted-packet-length.exception';
import { MaxPacketLengthExceededException } from '../errors/max-packet-length-exceeded.exception';
import { TcpSocket } from './tcp-socket';

const MAX_BUFFER_SIZE = (512 * 1024 * 1024) / 4; // 512 MBs in characters with 4 bytes per character (32-bit)

export class JsonSocket extends TcpSocket {
  private contentLength: number | null = null;
  private buffer = '';

  private readonly stringDecoder = new StringDecoder();
  private readonly delimiter = '#';

  protected handleSend(message: any, callback?: (err?: any) => void) {
    this.socket.write(this.formatMessageData(message), 'utf-8', callback);
  }

  protected handleData(dataRaw: Buffer | string) {
    const data = Buffer.isBuffer(dataRaw)
      ? this.stringDecoder.write(dataRaw)
      : dataRaw;
    this.buffer += data;

    if (this.buffer.length > MAX_BUFFER_SIZE) {
      this.buffer = '';
      throw new MaxPacketLengthExceededException(this.buffer.length);
    }

    if (this.contentLength === null) {
      const i = this.buffer.indexOf(this.delimiter);
      /**
       * Check if the buffer has the delimiter (#),
       * if not, the end of the buffer string might be in the middle of a content length string
       */
      if (i !== -1) {
        const rawContentLength = this.buffer.substring(0, i);
        this.contentLength = parseInt(rawContentLength, 10);

        if (isNaN(this.contentLength)) {
          this.contentLength = null;
          this.buffer = '';
          throw new CorruptedPacketLengthException(rawContentLength);
        }
        this.buffer = this.buffer.substring(i + 1);
      }
    }

    if (this.contentLength !== null) {
      const length = this.buffer.length;
      if (length === this.contentLength) {
        this.handleMessage(this.buffer);
      } else if (length > this.contentLength) {
        const message = this.buffer.substring(0, this.contentLength);
        const rest = this.buffer.substring(this.contentLength);
        this.handleMessage(message);
        this.handleData(rest);
      }
    }
  }

  private handleMessage(message: any) {
    this.contentLength = null;
    this.buffer = '';
    this.emitMessage(message);
  }

  private formatMessageData(message: any) {
    const messageData = JSON.stringify(message);
    const length = messageData.length;
    const data = length + this.delimiter + messageData;
    return data;
  }
}

1	import { Buffer } from 'buffer';	1✔
2	import { StringDecoder } from 'string_decoder';	1✔
3	import { CorruptedPacketLengthException } from '../errors/corrupted-packet-length.exception';	1✔
4	import { MaxPacketLengthExceededException } from '../errors/max-packet-length-exceeded.exception';	1✔
5	import { TcpSocket } from './tcp-socket';	1✔
6
7	const MAX_BUFFER_SIZE = (512 * 1024 * 1024) / 4; // 512 MBs in characters with 4 bytes per character (32-bit)	1✔
8
9	export class JsonSocket extends TcpSocket {	1✔
10	private contentLength: number \| null = null;	17✔
11	private buffer = '';	17✔
12
13	private readonly stringDecoder = new StringDecoder();	17✔
14	private readonly delimiter = '#';	17✔
15
16	protected handleSend(message: any, callback?: (err?: any) => void) {
17	this.socket.write(this.formatMessageData(message), 'utf-8', callback);	104✔
18	}
19
20	protected handleData(dataRaw: Buffer \| string) {
21	const data = Buffer.isBuffer(dataRaw)	129✔
22	? this.stringDecoder.write(dataRaw)	129✔
23	: dataRaw;
24	this.buffer += data;	129✔
25
26	if (this.buffer.length > MAX_BUFFER_SIZE) {	129!
NEW 27	this.buffer = '';	×
NEW 28	throw new MaxPacketLengthExceededException(this.buffer.length);	×
29	}
30
31	if (this.contentLength === null) {	129✔
32	const i = this.buffer.indexOf(this.delimiter);	125✔
33	/**
34	* Check if the buffer has the delimiter (#),
35	* if not, the end of the buffer string might be in the middle of a content length string
36	*/
37	if (i !== -1) {	125!
38	const rawContentLength = this.buffer.substring(0, i);	125✔
39	this.contentLength = parseInt(rawContentLength, 10);	125✔
40
41	if (isNaN(this.contentLength)) {	125✔
42	this.contentLength = null;	3✔
43	this.buffer = '';	3✔
44	throw new CorruptedPacketLengthException(rawContentLength);	3✔
45	}
46	this.buffer = this.buffer.substring(i + 1);	122✔
47	}
48	}
49
50	if (this.contentLength !== null) {	126!
51	const length = this.buffer.length;	126✔
52	if (length === this.contentLength) {	126✔
53	this.handleMessage(this.buffer);	19✔
54	} else if (length > this.contentLength) {	107✔
55	const message = this.buffer.substring(0, this.contentLength);	103✔
56	const rest = this.buffer.substring(this.contentLength);	103✔
57	this.handleMessage(message);	103✔
58	this.handleData(rest);	103✔
59	}
60	}
61	}
62
63	private handleMessage(message: any) {
64	this.contentLength = null;	122✔
65	this.buffer = '';	122✔
66	this.emitMessage(message);	122✔
67	}
68
69	private formatMessageData(message: any) {
70	const messageData = JSON.stringify(message);	104✔
71	const length = messageData.length;	104✔
72	const data = length + this.delimiter + messageData;	104✔
73	return data;	104✔
74	}
75	}

nestjs / nest / f7142b24-e394-4028-89c6-717e281431d1

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