feat(): add incoming request & response serializers

kamilmysliwiec · kamilmysliwiec · commit 0a7f37d90bbc · 2019-08-26T21:23:46.000+02:00
diff --git a/packages/microservices/client/client-proxy.ts b/packages/microservices/client/client-proxy.ts
@@ -11,7 +11,7 @@ import {
 } from 'rxjs';
 import { map, mergeMap, publish, take } from 'rxjs/operators';
 import { CONNECT_EVENT, ERROR_EVENT } from '../constants';
-import { IdentityDeserializer } from '../deserializers/identity.deserializer';
+import { IncomingResponseDeserializer } from '../deserializers/incoming-response.deserializer';
 import { InvalidMessageException } from '../errors/invalid-message.exception';
 import {
   ClientOptions,
@@ -145,6 +145,6 @@ export abstract class ClientProxy {
           | MqttOptions['options']
           | TcpClientOptions['options']
           | RmqOptions['options']).deserializer) ||
-      new IdentityDeserializer();
+      new IncomingResponseDeserializer();
   }
 }
diff --git a/packages/microservices/deserializers/incoming-request.deserializer.ts b/packages/microservices/deserializers/incoming-request.deserializer.ts
@@ -0,0 +1,44 @@
+import { isUndefined } from '@nestjs/common/utils/shared.utils';
+import {
+  ConsumerDeserializer,
+  IncomingEvent,
+  IncomingRequest,
+} from '../interfaces';
+
+export class IncomingRequestDeserializer implements ConsumerDeserializer {
+  deserialize(
+    value: any,
+    options?: Record<string, any>,
+  ): IncomingRequest | IncomingEvent {
+    return this.isExternal(value) ? this.mapToSchema(value, options) : value;
+  }
+
+  isExternal(value: any): boolean {
+    if (!value) {
+      return true;
+    }
+    if (
+      !isUndefined((value as IncomingRequest).pattern) ||
+      !isUndefined((value as IncomingRequest).data)
+    ) {
+      return false;
+    }
+    return true;
+  }
+
+  mapToSchema(
+    value: any,
+    options?: Record<string, any>,
+  ): IncomingRequest | IncomingEvent {
+    if (!options) {
+      return {
+        pattern: undefined,
+        data: undefined,
+      };
+    }
+    return {
+      pattern: options.channel,
+      data: value,
+    };
+  }
+}
diff --git a/packages/microservices/deserializers/incoming-response.deserializer.ts b/packages/microservices/deserializers/incoming-response.deserializer.ts
@@ -0,0 +1,30 @@
+import { isUndefined } from '@nestjs/common/utils/shared.utils';
+import { IncomingResponse, ProducerDeserializer } from '../interfaces';
+
+export class IncomingResponseDeserializer implements ProducerDeserializer {
+  deserialize(value: any, options?: Record<string, any>): IncomingResponse {
+    return this.isExternal(value) ? this.mapToSchema(value) : value;
+  }
+
+  isExternal(value: any): boolean {
+    if (!value) {
+      return true;
+    }
+    if (
+      !isUndefined((value as IncomingResponse).err) ||
+      !isUndefined((value as IncomingResponse).response) ||
+      !isUndefined((value as IncomingResponse).isDisposed)
+    ) {
+      return false;
+    }
+    return true;
+  }
+
+  mapToSchema(value: any): IncomingResponse {
+    return {
+      id: value && value.id,
+      response: value,
+      isDisposed: true,
+    };
+  }
+}
diff --git a/packages/microservices/interfaces/deserializer.interface.ts b/packages/microservices/interfaces/deserializer.interface.ts
@@ -1,8 +1,15 @@
-import { IncomingRequest, IncomingResponse } from './packet.interface';
+import {
+  IncomingEvent,
+  IncomingRequest,
+  IncomingResponse,
+} from './packet.interface';
 
 export interface Deserializer<TInput = any, TOutput = any> {
-  deserialize(value: TInput): TOutput;
+  deserialize(value: TInput, options?: Record<string, any>): TOutput;
 }
 
 export type ProducerDeserializer = Deserializer<any, IncomingResponse>;
-export type ConsumerDeserializer = Deserializer<any, IncomingRequest>;
+export type ConsumerDeserializer = Deserializer<
+  any,
+  IncomingRequest | IncomingEvent
+>;
diff --git a/packages/microservices/server/server-mqtt.ts b/packages/microservices/server/server-mqtt.ts
@@ -8,7 +8,12 @@ import {
   NO_MESSAGE_HANDLER,
 } from '../constants';
 import { MqttClient } from '../external/mqtt-client.interface';
-import { CustomTransportStrategy, PacketId, ReadPacket } from '../interfaces';
+import {
+  CustomTransportStrategy,
+  IncomingRequest,
+  PacketId,
+  ReadPacket,
+} from '../interfaces';
 import { MqttOptions } from '../interfaces/microservice-configuration.interface';
 import { Server } from './server';
 
@@ -72,18 +77,22 @@ export class ServerMqtt extends Server implements CustomTransportStrategy {
     pub: MqttClient,
   ): Promise<any> {
     const rawPacket = this.parseMessage(buffer.toString());
-    const packet = this.deserializer.deserialize(rawPacket);
-    if (isUndefined(packet.id)) {
+    const packet = this.deserializer.deserialize(rawPacket, { channel });
+    if (isUndefined((packet as IncomingRequest).id)) {
       return this.handleEvent(channel, packet);
     }
     const pattern = channel.replace(/_ack$/, '');
-    const publish = this.getPublisher(pub, pattern, packet.id);
+    const publish = this.getPublisher(
+      pub,
+      pattern,
+      (packet as IncomingRequest).id,
+    );
     const handler = this.getHandlerByPattern(pattern);
 
     if (!handler) {
       const status = 'error';
       const noHandlerPacket = {
-        id: packet.id,
+        id: (packet as IncomingRequest).id,
         status,
         err: NO_MESSAGE_HANDLER,
       };
diff --git a/packages/microservices/server/server-nats.ts b/packages/microservices/server/server-nats.ts
@@ -9,7 +9,7 @@ import {
 import { Client } from '../external/nats-client.interface';
 import { CustomTransportStrategy, PacketId } from '../interfaces';
 import { NatsOptions } from '../interfaces/microservice-configuration.interface';
-import { ReadPacket } from '../interfaces/packet.interface';
+import { IncomingRequest, ReadPacket } from '../interfaces/packet.interface';
 import { Server } from './server';
 
 let natsPackage: any = {};
@@ -85,16 +85,20 @@ export class ServerNats extends Server implements CustomTransportStrategy {
     client: Client,
     replyTo: string,
   ) {
-    const message = this.deserializer.deserialize(rawMessage);
-    if (isUndefined(message.id)) {
+    const message = this.deserializer.deserialize(rawMessage, { channel });
+    if (isUndefined((message as IncomingRequest).id)) {
       return this.handleEvent(channel, message);
     }
-    const publish = this.getPublisher(client, replyTo, message.id);
+    const publish = this.getPublisher(
+      client,
+      replyTo,
+      (message as IncomingRequest).id,
+    );
     const handler = this.getHandlerByPattern(channel);
     if (!handler) {
       const status = 'error';
       const noHandlerPacket = {
-        id: message.id,
+        id: (message as IncomingRequest).id,
         status,
         err: NO_MESSAGE_HANDLER,
       };
diff --git a/packages/microservices/server/server-redis.ts b/packages/microservices/server/server-redis.ts
@@ -12,7 +12,7 @@ import {
   RedisClient,
   RetryStrategyOptions,
 } from '../external/redis.interface';
-import { CustomTransportStrategy } from '../interfaces';
+import { CustomTransportStrategy, IncomingRequest } from '../interfaces';
 import { RedisOptions } from '../interfaces/microservice-configuration.interface';
 import { Server } from './server';
 
@@ -85,18 +85,22 @@ export class ServerRedis extends Server implements CustomTransportStrategy {
     pub: RedisClient,
   ) {
     const rawMessage = this.parseMessage(buffer);
-    const packet = this.deserializer.deserialize(rawMessage);
-    if (isUndefined(packet.id)) {
+    const packet = this.deserializer.deserialize(rawMessage, { channel });
+    if (isUndefined((packet as IncomingRequest).id)) {
       return this.handleEvent(channel, packet);
     }
     const pattern = channel.replace(/_ack$/, '');
-    const publish = this.getPublisher(pub, pattern, packet.id);
+    const publish = this.getPublisher(
+      pub,
+      pattern,
+      (packet as IncomingRequest).id,
+    );
     const handler = this.getHandlerByPattern(pattern);
 
     if (!handler) {
       const status = 'error';
       const noHandlerPacket = {
-        id: packet.id,
+        id: (packet as IncomingRequest).id,
         status,
         err: NO_MESSAGE_HANDLER,
       };
diff --git a/packages/microservices/server/server-rmq.ts b/packages/microservices/server/server-rmq.ts
@@ -12,7 +12,10 @@ import {
   RQM_DEFAULT_URL,
 } from '../constants';
 import { CustomTransportStrategy, RmqOptions } from '../interfaces';
-import { OutgoingResponse } from '../interfaces/packet.interface';
+import {
+  IncomingRequest,
+  OutgoingResponse,
+} from '../interfaces/packet.interface';
 import { Server } from './server';
 
 let rqmPackage: any = {};
@@ -96,15 +99,15 @@ export class ServerRMQ extends Server implements CustomTransportStrategy {
       ? packet.pattern
       : JSON.stringify(packet.pattern);
 
-    if (isUndefined(packet.id)) {
+    if (isUndefined((packet as IncomingRequest).id)) {
       return this.handleEvent(pattern, packet);
     }
     const handler = this.getHandlerByPattern(pattern);
 
     if (!handler) {
       const status = 'error';
       const noHandlerPacket = {
-        id: packet.id,
+        id: (packet as IncomingRequest).id,
         err: NO_MESSAGE_HANDLER,
         status,
       };
diff --git a/packages/microservices/server/server-tcp.ts b/packages/microservices/server/server-tcp.ts
@@ -13,6 +13,7 @@ import {
 import { JsonSocket } from '../helpers/json-socket';
 import {
   CustomTransportStrategy,
+  IncomingRequest,
   PacketId,
   ReadPacket,
   WritePacket,
@@ -61,14 +62,14 @@ export class ServerTCP extends Server implements CustomTransportStrategy {
       ? JSON.stringify(packet.pattern)
       : packet.pattern;
 
-    if (isUndefined(packet.id)) {
+    if (isUndefined((packet as IncomingRequest).id)) {
       return this.handleEvent(pattern, packet);
     }
     const handler = this.getHandlerByPattern(pattern);
     if (!handler) {
       const status = 'error';
       const noHandlerPacket = this.serializer.serialize({
-        id: packet.id,
+        id: (packet as IncomingRequest).id,
         status,
         err: NO_MESSAGE_HANDLER,
       });
@@ -80,7 +81,7 @@ export class ServerTCP extends Server implements CustomTransportStrategy {
 
     response$ &&
       this.send(response$, data => {
-        Object.assign(data, { id: packet.id });
+        Object.assign(data, { id: (packet as IncomingRequest).id });
         const outgoingResponse = this.serializer.serialize(data as WritePacket &
           PacketId);
         socket.sendMessage(outgoingResponse);
diff --git a/packages/microservices/server/server.ts b/packages/microservices/server/server.ts
@@ -10,7 +10,7 @@ import {
   Subscription,
 } from 'rxjs';
 import { catchError, finalize, publish } from 'rxjs/operators';
-import { IdentityDeserializer } from '../deserializers/identity.deserializer';
+import { IncomingRequestDeserializer } from '../deserializers/incoming-request.deserializer';
 import {
   ClientOptions,
   MessageHandler,
@@ -65,7 +65,10 @@ export abstract class Server {
     const scheduleOnNextTick = (data: WritePacket) => {
       if (!dataBuffer) {
         dataBuffer = [data];
-        process.nextTick(() => dataBuffer.forEach(buffer => respond(buffer)));
+        process.nextTick(() => {
+          dataBuffer.forEach(buffer => respond(buffer));
+          dataBuffer = null;
+        });
       } else if (!data.isDisposed) {
         dataBuffer = dataBuffer.concat(data);
       } else {
@@ -145,7 +148,7 @@ export abstract class Server {
           | MqttOptions['options']
           | TcpOptions['options']
           | RmqOptions['options']).deserializer) ||
-      new IdentityDeserializer();
+      new IncomingRequestDeserializer();
   }
 
   private isObservable(input: unknown): input is Observable<any> {
diff --git a/packages/microservices/test/deserializers/identity.deserializer.spec.ts b/packages/microservices/test/deserializers/identity.deserializer.spec.ts
@@ -0,0 +1,15 @@
+import { expect } from 'chai';
+import { IdentityDeserializer } from '../../deserializers/identity.deserializer';
+
+describe('IdentityDeserializer', () => {
+  let instance: IdentityDeserializer;
+  beforeEach(() => {
+    instance = new IdentityDeserializer();
+  });
+  describe('deserialize', () => {
+    it('should return the value unchanged', () => {
+      const value = {};
+      expect(instance.deserialize(value)).to.be.eql(value);
+    });
+  });
+});
diff --git a/packages/microservices/test/deserializers/incoming-request.deserializer.ts b/packages/microservices/test/deserializers/incoming-request.deserializer.ts
@@ -0,0 +1,50 @@
+import { expect } from 'chai';
+import { IncomingRequestDeserializer } from '../../deserializers/incoming-request.deserializer';
+import { IncomingRequest } from '../../interfaces';
+
+describe('IncomingRequestDeserializer', () => {
+  let instance: IncomingRequestDeserializer;
+  beforeEach(() => {
+    instance = new IncomingRequestDeserializer();
+  });
+  describe('deserialize', () => {
+    describe('when response is not external', () => {
+      it('should return the same value unchanged', () => {
+        const incomingRequest: IncomingRequest = {
+          id: '1',
+          pattern: 'pattern',
+          data: [],
+        };
+        expect(instance.deserialize(incomingRequest)).to.be.equal(
+          incomingRequest,
+        );
+      });
+    });
+    describe('otherwise', () => {
+      describe('when options are passed in', () => {
+        it('should map to the internal schema', () => {
+          const externalRequest = {
+            array: [1, 2, 3],
+          };
+          const options = {
+            channel: 'test',
+          };
+          expect(
+            instance.deserialize(externalRequest, options),
+          ).to.be.deep.equal({
+            pattern: options.channel,
+            data: externalRequest,
+          });
+        });
+      });
+      describe('when options are undefined', () => {
+        it('should map to proper schema with undefined values', () => {
+          expect(instance.deserialize({})).to.be.deep.equal({
+            pattern: undefined,
+            data: undefined,
+          });
+        });
+      });
+    });
+  });
+});
diff --git a/packages/microservices/test/deserializers/incoming-response.deserializer.ts b/packages/microservices/test/deserializers/incoming-response.deserializer.ts
diff --git a/packages/microservices/test/serializers/identity.serializer.spec.ts b/packages/microservices/test/serializers/identity.serializer.spec.ts
diff --git a/packages/microservices/test/utils/transform-pattern.utils.spec.ts b/packages/microservices/test/utils/transform-pattern.utils.spec.ts
diff --git a/packages/microservices/utils/transform-pattern.utils.ts b/packages/microservices/utils/transform-pattern.utils.ts
