bugfix(microservices): let use unicode characters in TCP Microservice messages

add own json-socket implementation

https://github.com/nestjs/nest/issues/1734

package.json

@@ -59,7 +59,6 @@
     "grpc": "1.19.0",
     "http2": "3.3.7",
     "iterare": "1.1.2",
-    "json-socket": "0.3.0",
     "merge-graphql-schemas": "1.5.8",
     "mqtt": "2.18.8",
     "multer": "1.4.1",

packages/microservices/client/client-tcp.ts

@@ -1,5 +1,4 @@
 import { Logger } from '@nestjs/common';
-import * as JsonSocket from 'json-socket';
 import * as net from 'net';
 import { share, tap } from 'rxjs/operators';
 import {
@@ -14,6 +13,7 @@ import {
   ClientOptions,
   TcpClientOptions,
 } from '../interfaces/client-metadata.interface';
+import { JsonSocket } from '../json-socket';
 import { ClientProxy } from './client-proxy';
 import { ECONNREFUSED } from './constants';
 
@@ -42,7 +42,7 @@ export class ClientTCP extends ClientProxy {
     this.socket = this.createSocket();
     this.bindEvents(this.socket);
 
-    const source$ = this.connect$(this.socket._socket).pipe(
+    const source$ = this.connect$(this.socket.netSocket).pipe(
       tap(() => {
         this.isConnected = true;
         this.socket.on(MESSAGE_EVENT, (buffer: WritePacket & PacketId) =>

packages/microservices/constants.ts

@@ -9,6 +9,7 @@ export const RQM_DEFAULT_URL = 'amqp://localhost';
 export const CONNECT_EVENT = 'connect';
 export const DISCONNECT_EVENT = 'disconnect';
 export const MESSAGE_EVENT = 'message';
+export const DATA_EVENT = 'data';
 export const ERROR_EVENT = 'error';
 export const CLOSE_EVENT = 'close';
 export const SUBSCRIBE = 'subscribe';

packages/microservices/json-socket.ts

@@ -0,0 +1,138 @@
+import { Socket } from 'net';
+import { StringDecoder } from 'string_decoder';
+import {
+  CLOSE_EVENT,
+  CONNECT_EVENT,
+  ERROR_EVENT,
+  MESSAGE_EVENT,
+  DATA_EVENT,
+} from './constants';
+
+const stringDecoder = new StringDecoder();
+
+export class JsonSocket {
+  private contentLength: number | null = null;
+  private buffer = '';
+  private closed = false;
+
+  private readonly delimeter = '#';
+
+  public get netSocket() {
+    return this.socket;
+  }
+
+  constructor(public readonly socket: Socket) {
+    this.socket.on(DATA_EVENT, this.onData.bind(this));
+    this.socket.on(CONNECT_EVENT, () => (this.closed = false));
+    this.socket.on(CLOSE_EVENT, () => (this.closed = true));
+    this.socket.on(ERROR_EVENT, () => (this.closed = true));
+  }
+
+  public connect(port: number, host: string) {
+    this.socket.connect(port, host);
+    return this;
+  }
+
+  public on(event: string, callback: (err?: any) => void) {
+    this.socket.on(event, callback);
+    return this;
+  }
+
+  public once(event: string, callback: (err?: any) => void) {
+    this.socket.once(event, callback);
+    return this;
+  }
+
+  public end() {
+    this.socket.end();
+    return this;
+  }
+
+  public sendMessage(message: any, callback?: (err?: any) => void) {
+    if (this.closed) {
+      if (callback) {
+        callback(new Error('The net socket is closed.'));
+      }
+      return;
+    }
+    this.socket.write(this.formatMessageData(message), 'utf-8', callback);
+  }
+
+  private onData(dataRaw: Buffer | string) {
+    const data = Buffer.isBuffer(dataRaw)
+      ? stringDecoder.write(dataRaw)
+      : dataRaw;
+
+    try {
+      this.handleData(data);
+    } catch (e) {
+      this.socket.emit(ERROR_EVENT, e.message);
+      this.socket.end();
+    }
+  }
+
+  private handleData(data: string) {
+    this.buffer += data;
+
+    if (this.contentLength == null) {
+      const i = this.buffer.indexOf(this.delimeter);
+
+      /**
+       * Check if the buffer has the delimeter (#),
+       * 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 Error(
+            `Corrupted length value "${rawContentLength}" supplied in a packet`,
+          );
+        }
+
+        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.onData(rest);
+      }
+    }
+  }
+
+  private handleMessage(data: string) {
+    this.contentLength = null;
+    this.buffer = '';
+    let message: object;
+
+    try {
+      message = JSON.parse(data);
+    } catch (e) {
+      throw new Error(
+        `Could not parse JSON: ${e.message}\nRequest data: ${data}`,
+      );
+    }
+    message = message || {};
+
+    this.socket.emit(MESSAGE_EVENT, message);
+  }
+
+  private formatMessageData(message: any) {
+    const messageData = JSON.stringify(message);
+    const length = messageData.length;
+    const data = length + this.delimeter + messageData;
+    return data;
+  }
+}

packages/microservices/server/server-tcp.ts

@@ -1,7 +1,6 @@
 import { isString, isUndefined } from '@nestjs/common/utils/shared.utils';
-import * as JsonSocket from 'json-socket';
 import * as net from 'net';
-import { Server as NetSocket } from 'net';
+import { Server as NetSocket, Socket } from 'net';
 import { Observable } from 'rxjs';
 import {
   CLOSE_EVENT,
@@ -15,6 +14,7 @@ import {
   MicroserviceOptions,
   TcpOptions,
 } from '../interfaces/microservice-configuration.interface';
+import { JsonSocket } from '../json-socket';
 import { Server } from './server';
 
 export class ServerTCP extends Server implements CustomTransportStrategy {
@@ -39,15 +39,16 @@ export class ServerTCP extends Server implements CustomTransportStrategy {
     this.server.close();
   }
 
-  public bindHandler<T extends Record<string, any>>(socket: T) {
+  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<T extends Record<string, any>>(
-    socket: T,
+  public async handleMessage(
+    socket: JsonSocket,
     packet: ReadPacket & PacketId,
   ) {
     const pattern = !isString(packet.pattern)
@@ -98,7 +99,7 @@ export class ServerTCP extends Server implements CustomTransportStrategy {
     this.server.on(CLOSE_EVENT, this.handleClose.bind(this));
   }
 
-  private getSocketInstance<T>(socket: T): JsonSocket {
+  private getSocketInstance(socket: Socket): JsonSocket {
     return new JsonSocket(socket);
   }
 }

packages/microservices/test/client/client-tcp.spec.ts

@@ -6,18 +6,7 @@ import { ERROR_EVENT } from '../../constants';
 
 describe('ClientTCP', () => {
   let client: ClientTCP;
-  let socket: {
-    connect: sinon.SinonStub;
-    publish: sinon.SinonSpy;
-    _socket: {
-      addListener: sinon.SinonStub;
-      removeListener: sinon.SinonSpy;
-      once: sinon.SinonStub;
-    };
-    on: sinon.SinonStub;
-    end: sinon.SinonSpy;
-    sendMessage: sinon.SinonSpy;
-  };
+  let socket;
   let createSocketStub: sinon.SinonStub;
 
   beforeEach(() => {
@@ -27,9 +16,8 @@ describe('ClientTCP', () => {
 
     socket = {
       connect: sinon.stub(),
-      publish: sinon.spy(),
       on: sinon.stub().callsFake(onFakeCallback),
-      _socket: {
+      netSocket: {
         addListener: sinon.stub().callsFake(onFakeCallback),
         removeListener: sinon.spy(),
         once: sinon.stub().callsFake(onFakeCallback),
@@ -134,7 +122,9 @@ describe('ClientTCP', () => {
           toPromise: () => source,
           pipe: () => source,
         };
-        connect$Stub = sinon.stub(client, 'connect$' as any).callsFake(() => source);
+        connect$Stub = sinon
+          .stub(client, 'connect$' as any)
+          .callsFake(() => source);
         await client.connect();
       });
       afterEach(() => {

packages/microservices/test/json-socket/connection.spec.ts

@@ -0,0 +1,215 @@
+import { AddressInfo, createServer, Socket } from 'net';
+import { CONNECT_EVENT, MESSAGE_EVENT } from '../../constants';
+import { JsonSocket } from '../../json-socket';
+import { longPayload } from './data/long-payload-with-special-chars';
+import * as helpers from './helpers';
+import { ip } from './helpers';
+import { expect } from 'chai';
+// tslint:disable:no-string-literal
+
+describe('JsonSocket connection', () => {
+  it('should connect, send and receive message', done => {
+    helpers.createServerAndClient(
+      (error, server, clientSocket, serverSocket) => {
+        if (error) {
+          return done(error);
+        }
+
+        expect(clientSocket['closed']).to.be.false;
+        expect(serverSocket['closed']).to.be.false;
+
+        Promise.all([
+          new Promise(callback => {
+            clientSocket.sendMessage({ type: 'ping' }, callback);
+          }),
+          new Promise(callback => {
+            clientSocket.on(MESSAGE_EVENT, (message: string) => {
+              expect(message).to.deep.equal({ type: 'pong' });
+              callback();
+            });
+          }),
+          new Promise(callback => {
+            serverSocket.on(MESSAGE_EVENT, (message: string) => {
+              expect(message).to.deep.equal({ type: 'ping' });
+              serverSocket.sendMessage({ type: 'pong' }, callback);
+            });
+          }),
+        ])
+          .then(() => {
+            expect(clientSocket['closed']).to.equal(false);
+            expect(serverSocket['closed']).to.equal(false);
+            clientSocket.end();
+            server.close(done);
+          })
+          .catch(e => done(e));
+      },
+    );
+  });
+
+  it('should send long messages with special characters without issues', done => {
+    helpers.createServerAndClient((err, server, clientSocket, serverSocket) => {
+      if (err) {
+        return done(err);
+      }
+      expect(clientSocket['closed']).to.equal(false);
+      expect(serverSocket['closed']).to.equal(false);
+      Promise.all([
+        new Promise(callback => {
+          clientSocket.sendMessage(longPayload, callback);
+        }),
+        new Promise(callback => {
+          clientSocket.on(MESSAGE_EVENT, (message: { type: 'pong' }) => {
+            expect(message).to.deep.equal({ type: 'pong' });
+            callback();
+          });
+        }),
+        new Promise(callback => {
+          serverSocket.on(MESSAGE_EVENT, (message: { type: 'pong' }) => {
+            expect(message).to.deep.equal(longPayload);
+            serverSocket.sendMessage({ type: 'pong' }, callback);
+          });
+        }),
+      ])
+        .then(() => {
+          expect(clientSocket['closed']).to.equal(false);
+          expect(serverSocket['closed']).to.equal(false);
+          clientSocket.end();
+          server.close(done);
+        })
+        .catch(e => done(e));
+    });
+  });
+
+  it('should send multiple messages', done => {
+    helpers.createServerAndClient((err, server, clientSocket, serverSocket) => {
+      if (err) {
+        return done(err);
+      }
+      Promise.all([
+        new Promise(callback =>
+          Promise.all(
+            helpers
+              .range(1, 100)
+              .map(
+                i =>
+                  new Promise(resolve =>
+                    clientSocket.sendMessage({ number: i }, resolve),
+                  ),
+              ),
+          ).then(_ => callback()),
+        ),
+        new Promise(callback => {
+          let lastNumber = 0;
+          serverSocket.on(MESSAGE_EVENT, (message: { number: number }) => {
+            expect(message.number).to.deep.equal(lastNumber + 1);
+            lastNumber = message.number;
+            if (lastNumber === 100) {
+              callback();
+            }
+          });
+        }),
+      ])
+        .then(() => {
+          clientSocket.end();
+          server.close(done);
+        })
+        .catch(e => done(e));
+    });
+  });
+
+  it('should return true for "closed" when server disconnects', done => {
+    helpers.createServerAndClient((err, server, clientSocket, serverSocket) => {
+      if (err) {
+        return done(err);
+      }
+
+      new Promise(callback => {
+        serverSocket.end();
+        setTimeout(callback, 10);
+      })
+        .then(
+          () =>
+            new Promise(callback => {
+              expect(clientSocket['closed']).to.equal(true);
+              expect(serverSocket['closed']).to.equal(true);
+              callback();
+            }),
+        )
+        .then(() => {
+          clientSocket.end();
+          server.close(done);
+        })
+        .catch(e => done(e));
+    });
+  });
+
+  it('should return true for "closed" when client disconnects', done => {
+    helpers.createServerAndClient((err, server, clientSocket, serverSocket) => {
+      if (err) {
+        return done(err);
+      }
+
+      new Promise(callback => {
+        clientSocket.end();
+        setTimeout(callback, 10);
+      })
+        .then(
+          () =>
+            new Promise(callback => {
+              expect(clientSocket['closed']).to.equal(true);
+              expect(serverSocket['closed']).to.equal(true);
+              callback();
+            }),
+        )
+        .then(() => server.close(done))
+        .catch(e => done(e));
+    });
+  });
+
+  it('should return true for "closed" when client (re)connects', done => {
+    const server = createServer();
+
+    server.on('listening', () => {
+      const clientSocket = new JsonSocket(new Socket());
+
+      server.once('connection', socket => {
+        const serverSocket = new JsonSocket(socket);
+
+        serverSocket.once('end', () => {
+          setTimeout(() => {
+            expect(serverSocket['closed']).to.equal(true);
+            expect(clientSocket['closed']).to.equal(true);
+
+            clientSocket.on(CONNECT_EVENT, () => {
+              setTimeout(() => {
+                expect(clientSocket['closed']).to.equal(false);
+
+                clientSocket.end();
+                server.close(done);
+              }, 10);
+            });
+
+            const address2 = server.address();
+            if (!address2) {
+              throw new Error('server.address() returned null');
+            }
+            const port2 = (address2 as AddressInfo).port;
+
+            clientSocket.connect(port2, ip);
+          }, 10);
+        });
+
+        clientSocket.end();
+      });
+
+      const address1 = server.address();
+      if (!address1) {
+        throw new Error('server.address() returned null');
+      }
+      const port1 = (address1 as AddressInfo).port;
+
+      clientSocket.connect(port1, ip);
+    });
+    server.listen();
+  });
+});

packages/microservices/test/json-socket/data/long-payload-with-special-chars.ts

@@ -0,0 +1,263 @@
+export const longPayload = [
+  {
+    _id: '584f17147fce7ca0a8bacfd2',
+    index: 0,
+    guid: '1d127572-0369-45fb-aa2f-e3bb083ac2b2',
+    isActive: true,
+    balance: '$2,926.06',
+    picture: 'http://placehold.it/32x32',
+    age: 26,
+    eyeColor: 'green',
+    name:
+      'Wçêtson Aguilar [special characters in name that used to fail on long payloads]',
+    gender: 'male',
+    company: 'PROWASTE',
+    email: 'watsonaguilar@prowaste.com',
+    phone: '+1 (821) 517-2430',
+    address: '910 Robert Street, Bangor, Delaware, 4159',
+    about:
+      'Aliqua et irure id do id id non dolore ipsum sit in proident ipsum. Id elit incididunt occaecat do laboris sunt officia fugiat aliquip. Incididunt aute ad minim Lorem cupidatat aute labore enim elit nostrud amet. Tempor sint irure incididunt aliquip amet sunt mollit aliqua Lorem officia pariatur.\r\n',
+    registered: '2014-02-11T08:45:28 +05:00',
+    latitude: 73.891198,
+    longitude: 90.23414,
+    tags: ['veniam', 'nulla', 'cillum', 'tempor', 'sint', 'magna', 'nostrud'],
+    friends: [
+      {
+        id: 0,
+        name: 'Cecelia James'
+      },
+      {
+        id: 1,
+        name: 'Hilary Young'
+      },
+      {
+        id: 2,
+        name: 'Sharron Goodwin'
+      }
+    ],
+    greeting: 'Hello, Watson Aguilar! You have 3 unread messages.',
+    favoriteFruit: 'banana'
+  },
+  {
+    _id: '584f1714b2e945fb30f73892',
+    index: 1,
+    guid: '3ffce1ee-a442-4dae-804f-40c59f19e7ee',
+    isActive: false,
+    balance: '$2,507.49',
+    picture: 'http://placehold.it/32x32',
+    age: 34,
+    eyeColor: 'brown',
+    name: 'Aguirre Salazar',
+    gender: 'male',
+    company: 'EZENTIA',
+    email: 'aguirresalazar@ezentia.com',
+    phone: '+1 (910) 443-3647',
+    address: '629 Burnett Street, Tyhee, West Virginia, 2905',
+    about:
+      'Labore laboris et deserunt aliquip. Occaecat esse officia est eiusmod. Officia tempor cupidatat commodo minim deserunt mollit qui ut culpa. Est occaecat laborum occaecat non mollit ad reprehenderit magna ad. Consequat culpa excepteur qui aliqua dolore occaecat aliqua sunt elit ea nisi. Officia consectetur dolor labore voluptate. Esse ad esse qui id incididunt.\r\n',
+    registered: '2015-01-28T06:47:34 +05:00',
+    latitude: -64.632254,
+    longitude: -116.659127,
+    tags: [
+      'sit',
+      'anim',
+      'quis',
+      'officia',
+      'minim',
+      'cupidatat',
+      'adipisicing'
+    ],
+    friends: [
+      {
+        id: 0,
+        name: 'Olson Mccall'
+      },
+      {
+        id: 1,
+        name: 'Carolina Conway'
+      },
+      {
+        id: 2,
+        name: 'Carlson Pacheco'
+      }
+    ],
+    greeting: 'Hello, Aguirre Salazar! You have 9 unread messages.',
+    favoriteFruit: 'apple'
+  },
+  {
+    _id: '584f17148282bb876fc4e9a2',
+    index: 2,
+    guid: '892ba80c-7149-4904-bd36-22f619d4df0a',
+    isActive: true,
+    balance: '$2,132.56',
+    picture: 'http://placehold.it/32x32',
+    age: 26,
+    eyeColor: 'green',
+    name: 'Hardin Grant',
+    gender: 'male',
+    company: 'CINASTER',
+    email: 'hardingrant@cinaster.com',
+    phone: '+1 (900) 437-2390',
+    address: '180 Ide Court, Gibsonia, Washington, 3027',
+    about:
+      'Ut aliquip officia adipisicing voluptate aliquip aute fugiat ad quis ad eu non consectetur. Laboris labore veniam officia qui eiusmod. Duis aliqua est quis do dolor excepteur ea dolore non. Nisi mollit laboris nostrud nostrud pariatur culpa laboris anim est irure id aute.\r\n',
+    registered: '2016-09-13T10:54:27 +04:00',
+    latitude: 8.651031,
+    longitude: -136.777747,
+    tags: ['consequat', 'deserunt', 'magna', 'enim', 'esse', 'minim', 'ipsum'],
+    friends: [
+      {
+        id: 0,
+        name: 'Lesley Velasquez'
+      },
+      {
+        id: 1,
+        name: 'Natasha Simmons'
+      },
+      {
+        id: 2,
+        name: 'Isabel Avery'
+      }
+    ],
+    greeting: 'Hello, Hardin Grant! You have 7 unread messages.',
+    favoriteFruit: 'strawberry'
+  },
+  {
+    _id: '584f1714d90ff4b8914a69e7',
+    index: 3,
+    guid: '76f37726-1f73-4cf7-aabe-8dadf37d3ddd',
+    isActive: true,
+    balance: '$2,493.04',
+    picture: 'http://placehold.it/32x32',
+    age: 32,
+    eyeColor: 'blue',
+    name: 'Randall Roy',
+    gender: 'male',
+    company: 'ZAJ',
+    email: 'randallroy@zaj.com',
+    phone: '+1 (938) 562-2214',
+    address: '872 Rugby Road, Hoehne, Indiana, 9792',
+    about:
+      'Non laboris id et cupidatat velit ea ipsum ea mollit quis qui dolore nisi laboris. Enim sit irure enim dolor velit proident sunt pariatur proident consequat mollit enim minim. Laboris deserunt cupidatat nisi enim adipisicing officia dolore ex cupidatat anim. Cupidatat labore voluptate non magna est dolor. Occaecat occaecat magna anim laborum adipisicing esse excepteur cillum aute qui eu do excepteur eu. Nostrud consectetur consectetur aliquip deserunt velit culpa sint excepteur mollit nostrud sit ex. Est ex ut laboris pariatur.\r\n',
+    registered: '2016-05-05T05:24:56 +04:00',
+    latitude: 18.943281,
+    longitude: -110.942673,
+    tags: [
+      'eu',
+      'aliqua',
+      'reprehenderit',
+      'amet',
+      'nulla',
+      'consequat',
+      'nisi'
+    ],
+    friends: [
+      {
+        id: 0,
+        name: 'Barron Maynard'
+      },
+      {
+        id: 1,
+        name: 'Lynn Shepard'
+      },
+      {
+        id: 2,
+        name: 'Robin Whitehead'
+      }
+    ],
+    greeting: 'Hello, Randall Roy! You have 3 unread messages.',
+    favoriteFruit: 'strawberry'
+  },
+  {
+    _id: '584f17142a8f47cef0f5401a',
+    index: 4,
+    guid: '9b50ec22-3fbe-40ce-a5b8-b956f1340a77',
+    isActive: false,
+    balance: '$3,234.48',
+    picture: 'http://placehold.it/32x32',
+    age: 33,
+    eyeColor: 'green',
+    name: 'Chandler Vasquez',
+    gender: 'male',
+    company: 'ZILLACTIC',
+    email: 'chandlervasquez@zillactic.com',
+    phone: '+1 (830) 550-3428',
+    address: '610 Hunts Lane, Cazadero, Michigan, 3584',
+    about:
+      'Fugiat in anim adipisicing sint aliquip ea velit do proident eu ad amet. Nulla velit duis ullamco labore ea Lorem velit elit Lorem. Id laboris do mollit exercitation veniam do amet culpa est excepteur reprehenderit consectetur laborum.\r\n',
+    registered: '2014-04-20T05:23:32 +04:00',
+    latitude: -88.088841,
+    longitude: -163.602482,
+    tags: [
+      'sunt',
+      'excepteur',
+      'enim',
+      'incididunt',
+      'officia',
+      'amet',
+      'irure'
+    ],
+    friends: [
+      {
+        id: 0,
+        name: 'Mckee Norton'
+      },
+      {
+        id: 1,
+        name: 'Durham Parrish'
+      },
+      {
+        id: 2,
+        name: 'Stewart Kramer'
+      }
+    ],
+    greeting: 'Hello, Chandler Vasquez! You have 3 unread messages.',
+    favoriteFruit: 'strawberry'
+  },
+  {
+    _id: '584f171450a4e9dda687adc5',
+    index: 5,
+    guid: '68eeea45-ba6e-4740-b89b-10d690c37a02',
+    isActive: false,
+    balance: '$3,771.46',
+    picture: 'http://placehold.it/32x32',
+    age: 25,
+    eyeColor: 'blue',
+    name: 'Fernandez Caldwell',
+    gender: 'male',
+    company: 'SNIPS',
+    email: 'fernandezcaldwell@snips.com',
+    phone: '+1 (911) 544-3684',
+    address: '786 Newel Street, Elliston, Massachusetts, 6683',
+    about:
+      'Voluptate commodo labore aliqua excepteur irure aliquip officia. Incididunt excepteur elit quis reprehenderit voluptate aliqua ad voluptate duis nisi dolor dolor id dolor. Irure sit consequat amet ea magna laborum velit eu in. Sunt occaecat quis consectetur laboris. Duis est do eu consectetur dolore id incididunt incididunt ut esse magna est. Nostrud irure magna nulla fugiat deserunt deserunt enim mollit proident qui sint dolore incididunt. Incididunt incididunt do quis culpa sint ut aliqua id.\r\n',
+    registered: '2015-08-09T09:02:36 +04:00',
+    latitude: -46.941347,
+    longitude: -171.796168,
+    tags: [
+      'sit',
+      'irure',
+      'reprehenderit',
+      'ut',
+      'proident',
+      'aliquip',
+      'labore'
+    ],
+    friends: [
+      {
+        id: 0,
+        name: 'Adela Preston'
+      },
+      {
+        id: 1,
+        name: 'Phillips Moses'
+      },
+      {
+        id: 2,
+        name: 'Neva Wise'
+      }
+    ],
+    greeting: 'Hello, Fernandez Caldwell! You have 10 unread messages.',
+    favoriteFruit: 'apple'
+  }
+];

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

@@ -0,0 +1,82 @@
+import {
+  AddressInfo,
+  createServer as netCreateServer,
+  Server,
+  Socket,
+} from 'net';
+import { ERROR_EVENT } from '../../constants';
+import { JsonSocket } from '../../json-socket';
+
+export const ip = '127.0.0.1';
+
+export function createServer(callback: (err?: any, server?: Server) => void) {
+  const server = netCreateServer();
+  server.listen();
+
+  server.on('listening', () => {
+    callback(null, server);
+  });
+
+  server.on(ERROR_EVENT, (err: any) => {
+    callback(err);
+  });
+}
+
+export function createClient(
+  server: Server,
+  callback: (
+    err?: any,
+    clientSocket?: JsonSocket,
+    serverSocket?: JsonSocket,
+  ) => void,
+) {
+  const clientSocket = new JsonSocket(new Socket());
+
+  const address = server.address();
+  if (!address) {
+    throw new Error('server.address() returned null');
+  }
+  const port = (address as AddressInfo).port;
+
+  clientSocket.connect(port, ip);
+
+  clientSocket.on(ERROR_EVENT, (err: any) => {
+    callback(err);
+  });
+
+  server.once('connection', socket => {
+    const serverSocket = new JsonSocket(socket);
+    callback(null, clientSocket, serverSocket);
+  });
+}
+
+export function createServerAndClient(
+  callback: (
+    err?: any,
+    server?: Server,
+    clientSocket?: JsonSocket,
+    serverSocket?: JsonSocket,
+  ) => void,
+) {
+  createServer((serverErr, server) => {
+    if (serverErr) {
+      return callback(serverErr);
+    }
+
+    createClient(server, (clientErr, clientSocket, serverSocket) => {
+      if (clientErr) {
+        return callback(clientErr);
+      }
+
+      callback(null, server, clientSocket, serverSocket);
+    });
+  });
+}
+
+export function range(start: number, end: number) {
+  const r = [];
+  for (let i = start; i <= end; i++) {
+    r.push(i);
+  }
+  return r;
+}

packages/microservices/test/json-socket/listener-chaining.spec.ts

@@ -0,0 +1,27 @@
+import { CONNECT_EVENT, MESSAGE_EVENT } from '../../constants';
+import { JsonSocket } from '../../json-socket';
+import * as helpers from './helpers';
+import { expect } from 'chai';
+
+describe('JsonSocket chaining', () => {
+  it('should return the instance when subscribing to event', done => {
+    helpers.createServerAndClient((err, server, clientSocket, serverSocket) => {
+      if (err) {
+        return done(err);
+      }
+
+      expect(clientSocket.on(MESSAGE_EVENT, () => {})).to.be.instanceof(
+        JsonSocket,
+      );
+      expect(clientSocket.on(CONNECT_EVENT, () => {})).to.deep.equal(
+        clientSocket,
+      );
+      expect(
+        clientSocket.on(MESSAGE_EVENT, () => {}).on('end', () => {}),
+      ).to.deep.equal(clientSocket);
+
+      clientSocket.end();
+      server.close(done);
+    });
+  });
+});

packages/microservices/test/json-socket/message-parsing.spec.ts

@@ -0,0 +1,186 @@
+import { Socket } from 'net';
+import * as sinon from 'sinon';
+import { ERROR_EVENT, MESSAGE_EVENT } from '../../constants';
+import { JsonSocket } from '../../json-socket';
+import { expect } from 'chai';
+// tslint:disable:no-string-literal
+
+describe('JsonSocket message parsing', () => {
+  const socket = new JsonSocket(new Socket());
+  let messages: string[] = [];
+
+  socket.on(MESSAGE_EVENT, message => {
+    messages.push(message);
+  });
+
+  beforeEach(() => {
+    messages = [];
+    socket['contentLength'] = null;
+    socket['buffer'] = '';
+  });
+
+  it('should parse JSON strings', () => {
+    socket['handleData']('13#"Hello there"');
+    expect(messages.length).to.deep.equal(1);
+    expect(messages[0]).to.deep.equal('Hello there');
+    expect(socket['buffer']).to.deep.equal('');
+  });
+
+  it('should parse JSON numbers', () => {
+    socket['handleData']('5#12.34');
+    expect(messages.length).to.deep.equal(1);
+    expect(messages[0]).to.deep.equal(12.34);
+    expect(socket['buffer']).to.deep.equal('');
+  });
+
+  it('should parse JSON bools', () => {
+    socket['handleData']('4#true');
+    expect(messages.length).to.deep.equal(1);
+    expect(messages[0]).to.deep.equal(true);
+    expect(socket['buffer']).to.deep.equal('');
+  });
+
+  it('should parse JSON objects', () => {
+    socket['handleData']('17#{"a":"yes","b":9}');
+    expect(messages.length).to.deep.equal(1);
+    expect(messages[0]).to.deep.equal({ a: 'yes', b: 9 });
+    expect(socket['buffer']).to.deep.equal('');
+  });
+
+  it('should parse JSON arrays', () => {
+    socket['handleData']('9#["yes",9]');
+    expect(messages.length).to.deep.equal(1);
+    expect(messages[0]).to.deep.equal(['yes', 9]);
+    expect(socket['buffer']).to.deep.equal('');
+  });
+
+  it('should parse multiple messages in one packet', () => {
+    socket['handleData']('5#"hey"4#true');
+    expect(messages.length).to.deep.equal(2);
+    expect(messages[0]).to.deep.equal('hey');
+    expect(messages[1]).to.deep.equal(true);
+    expect(socket['buffer']).to.deep.equal('');
+  });
+
+  it('should parse chunked messages', () => {
+    socket['handleData']('13#"Hel');
+    socket['handleData']('lo there"');
+    expect(messages.length).to.deep.equal(1);
+    expect(messages[0]).to.deep.equal('Hello there');
+    expect(socket['buffer']).to.deep.equal('');
+  });
+
+  it('should parse chunked and multiple messages', () => {
+    socket['handleData']('13#"Hel');
+    socket['handleData']('lo there"4#true');
+    expect(messages.length).to.deep.equal(2);
+    expect(messages[0]).to.deep.equal('Hello there');
+    expect(messages[1]).to.deep.equal(true);
+    expect(socket['buffer']).to.deep.equal('');
+  });
+
+  it('should parse chunked messages with multi-byte characters', () => {
+    // 0x33 0x23 0xd8 0x22 0xa9 0x22 = 3#"ة" (U+00629)
+    socket['onData'](Buffer.from([0x33, 0x23, 0x22, 0xd8]));
+    socket['onData'](Buffer.from([0xa9, 0x22]));
+    expect(messages.length).to.deep.equal(1);
+    expect(messages[0]).to.deep.equal('ة');
+  });
+
+  it('should parse multiple messages with unicode correctly', () => {
+    socket['handleData']('41#"Diese Zeile enthält das Unicode-Zeichen"4#true');
+    expect(messages[0]).to.deep.equal(
+      'Diese Zeile enthält das Unicode-Zeichen',
+    );
+    expect(messages[1]).to.deep.equal(true);
+    expect(socket['buffer']).to.deep.equal('');
+  });
+
+  it('should parse multiple and chunked messages with unicode correctly', () => {
+    socket['handleData']('41#"Diese Zeile enthält ');
+    socket['handleData']('das Unicode-Zeichen"4#true');
+    expect(messages[0]).to.deep.equal(
+      'Diese Zeile enthält das Unicode-Zeichen',
+    );
+    expect(messages[1]).to.deep.equal(true);
+    expect(socket['buffer']).to.deep.equal('');
+  });
+
+  describe('Error handling', () => {
+    describe('JSON Error', () => {
+      const errorMsg = `Could not parse JSON: Unexpected end of JSON input\nRequest data: "Hel`;
+      const packetStrin = '4#"Hel';
+      const packet = Buffer.from(packetStrin);
+
+      it('should fail to parse invalid JSON', () => {
+        try {
+          socket['handleData']('4#"Hel');
+        } catch (err) {
+          expect(err.message).to.deep.equal(errorMsg);
+        }
+        expect(messages.length).to.deep.equal(0);
+        expect(socket['buffer']).to.deep.equal('');
+      });
+
+      it(`should emit ${ERROR_EVENT} event on socket`, () => {
+        const socketEmitSpy: sinon.SinonSpy<any, any> = sinon.spy(
+          socket['socket'],
+          'emit',
+        );
+
+        socket['onData'](packet);
+
+        expect(socketEmitSpy.calledOnceWithExactly(ERROR_EVENT, errorMsg)).to.be
+          .true;
+        socketEmitSpy.restore();
+      });
+
+      it(`should send a FIN packet`, () => {
+        const socketEndSpy = sinon.spy(socket['socket'], 'end');
+
+        socket['onData'](packet);
+
+        expect(socketEndSpy.calledOnce).to.be.true;
+        socketEndSpy.restore();
+      });
+    });
+
+    describe('Corrupted length value', () => {
+      const errorMsg = `Corrupted length value "wtf" supplied in a packet`;
+      const packetStrin = 'wtf#"Hello"';
+      const packet = Buffer.from(packetStrin);
+
+      it('should not accept invalid content length', () => {
+        try {
+          socket['handleData'](packetStrin);
+        } catch (err) {
+          expect(err.message).to.deep.equal(errorMsg);
+        }
+        expect(messages.length).to.deep.equal(0);
+        expect(socket['buffer']).to.deep.equal('');
+      });
+
+      it(`should emit ${ERROR_EVENT} event on socket`, () => {
+        const socketEmitSpy: sinon.SinonSpy<any, any> = sinon.spy(
+          socket['socket'],
+          'emit',
+        );
+
+        socket['onData'](packet);
+
+        expect(socketEmitSpy.calledOnceWithExactly(ERROR_EVENT, errorMsg)).to.be
+          .true;
+        socketEmitSpy.restore();
+      });
+
+      it(`should send a FIN packet`, () => {
+        const socketEndSpy = sinon.spy(socket['socket'], 'end');
+
+        socket['onData'](packet);
+
+        expect(socketEndSpy.calledOnce).to.be.true;
+        socketEndSpy.restore();
+      });
+    });
+  });
+});

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

@@ -21,9 +21,9 @@ describe('ServerTCP', () => {
         .stub(server, 'getSocketInstance' as any)
         .callsFake(() => socket);
     });
-    it('should bind message event to handler', () => {
+    it('should bind message and error events to handler', () => {
       server.bindHandler(null);
-      expect(socket.on.called).to.be.true;
+      expect(socket.on.calledTwice).to.be.true;
     });
   });
   describe('close', () => {