test(microservices): add e2e tests for record builders

integration/microservices/e2e/mqtt-record-builder.spec.ts

@@ -0,0 +1,46 @@
+import { INestApplication } from '@nestjs/common';
+import { Transport } from '@nestjs/microservices';
+import { Test } from '@nestjs/testing';
+import * as request from 'supertest';
+import { MqttController } from '../src/mqtt/mqtt.controller';
+
+describe('MQTT transport', () => {
+  let server;
+  let app: INestApplication;
+
+  beforeEach(async () => {
+    const module = await Test.createTestingModule({
+      controllers: [MqttController],
+    }).compile();
+
+    app = module.createNestApplication();
+    server = app.getHttpAdapter().getInstance();
+
+    app.connectMicroservice({
+      transport: Transport.MQTT,
+      options: {
+        url: 'mqtt://0.0.0.0:1883',
+        subscribeOptions: {
+          qos: 1,
+        },
+      },
+    });
+    await app.startAllMicroservices();
+    await app.init();
+  });
+
+  it(`/POST (setting packet options with "RecordBuilder")`, () => {
+    const payload = { items: [1, 2, 3] };
+    return request(server)
+      .post('/record-builder-duplex')
+      .send(payload)
+      .expect(200, {
+        data: payload,
+        qos: 1,
+      });
+  });
+
+  afterEach(async () => {
+    await app.close();
+  });
+});

integration/microservices/e2e/sum-nats.spec.ts

@@ -96,6 +96,19 @@ describe('NATS transport', () => {
       });
   });
 
+  it(`/POST (sending headers with "RecordBuilder")`, () => {
+    const payload = { items: [1, 2, 3] };
+    return request(server)
+      .post('/record-builder-duplex')
+      .send(payload)
+      .expect(200, {
+        data: payload,
+        headers: {
+          ['x-version']: '1.0.0',
+        },
+      });
+  });
+
   afterEach(async () => {
     await app.close();
   });

integration/microservices/e2e/sum-rmq.spec.ts

@@ -89,6 +89,20 @@ describe('RabbitMQ transport', () => {
       });
   });
 
+  it(`/POST (sending options with "RecordBuilder")`, () => {
+    const payload = { items: [1, 2, 3] };
+    return request(server)
+      .post('/record-builder-duplex')
+      .send(payload)
+      .expect(200, {
+        data: payload,
+        headers: {
+          ['x-version']: '1.0.0',
+        },
+        priority: 3,
+      });
+  });
+
   afterEach(async () => {
     await app.close();
   });

integration/microservices/src/mqtt/mqtt.controller.ts

@@ -2,8 +2,12 @@ import { Body, Controller, HttpCode, Post, Query } from '@nestjs/common';
 import {
   Client,
   ClientProxy,
+  Ctx,
   EventPattern,
   MessagePattern,
+  MqttContext,
+  MqttRecordBuilder,
+  Payload,
   Transport,
 } from '@nestjs/microservices';
 import { from, lastValueFrom, Observable, of } from 'rxjs';
@@ -32,7 +36,7 @@ export class MqttController {
   @HttpCode(200)
   async stream(@Body() data: number[]) {
     const result = lastValueFrom(
-      await this.client
+      this.client
         .send<number>({ cmd: 'streaming' }, data)
         .pipe(scan((a, b) => a + b, 0)),
     );
@@ -86,6 +90,25 @@ export class MqttController {
     return this.client.send<number>('wildcard-message2/test/test', data);
   }
 
+  @Post('record-builder-duplex')
+  @HttpCode(200)
+  useRecordBuilderDuplex(@Body() data: Record<string, any>) {
+    const record = new MqttRecordBuilder(data).setQoS(2).build();
+    return this.client.send('record-builder-duplex', record);
+  }
+
+  @MessagePattern('record-builder-duplex')
+  handleRecordBuilderDuplex(
+    @Payload() data: Record<string, any>,
+    @Ctx() context: MqttContext,
+  ) {
+    const { qos } = context.getPacket();
+    return {
+      data,
+      qos,
+    };
+  }
+
   @MessagePattern('wildcard-message/#')
   wildcardMessageHandler(data: number[]): number {
     if ((data as any).response) {

integration/microservices/src/nats/nats.controller.ts

@@ -6,10 +6,12 @@ import {
   EventPattern,
   MessagePattern,
   NatsContext,
+  NatsRecordBuilder,
   Payload,
   RpcException,
   Transport,
 } from '@nestjs/microservices';
+import * as nats from 'nats';
 import { from, lastValueFrom, Observable, of, throwError } from 'rxjs';
 import { catchError, scan } from 'rxjs/operators';
 import { NatsService } from './nats.service';
@@ -62,6 +64,28 @@ export class NatsController {
       .reduce(async (a, b) => (await a) && b);
   }
 
+  @Post('record-builder-duplex')
+  @HttpCode(200)
+  useRecordBuilderDuplex(@Body() data: Record<string, any>) {
+    const headers = nats.headers();
+    headers.set('x-version', '1.0.0');
+    const record = new NatsRecordBuilder(data).setHeaders(headers).build();
+    return this.client.send('record-builder-duplex', record);
+  }
+
+  @MessagePattern('record-builder-duplex')
+  handleRecordBuilderDuplex(
+    @Payload() data: Record<string, any>,
+    @Ctx() context: NatsContext,
+  ) {
+    return {
+      data,
+      headers: {
+        ['x-version']: context.getHeaders().get('x-version'),
+      },
+    };
+  }
+
   @MessagePattern('math.*')
   sum(@Payload() data: number[], @Ctx() context: NatsContext): number {
     return (data || []).reduce((a, b) => a + b);

integration/microservices/src/rmq/rmq.controller.ts

@@ -2,8 +2,12 @@ import { Body, Controller, HttpCode, Post, Query } from '@nestjs/common';
 import {
   ClientProxy,
   ClientProxyFactory,
+  Ctx,
   EventPattern,
   MessagePattern,
+  Payload,
+  RmqContext,
+  RmqRecordBuilder,
   Transport,
 } from '@nestjs/microservices';
 import { from, lastValueFrom, Observable, of } from 'rxjs';
@@ -57,6 +61,34 @@ export class RMQController {
       .reduce(async (a, b) => (await a) && b);
   }
 
+  @Post('record-builder-duplex')
+  @HttpCode(200)
+  useRecordBuilderDuplex(@Body() data: Record<string, any>) {
+    const record = new RmqRecordBuilder(data)
+      .setOptions({
+        headers: {
+          ['x-version']: '1.0.0',
+        },
+        priority: 3,
+      })
+      .build();
+
+    return this.client.send('record-builder-duplex', record);
+  }
+
+  @MessagePattern('record-builder-duplex')
+  handleRecordBuilderDuplex(
+    @Payload() data: Record<string, any>,
+    @Ctx() context: RmqContext,
+  ) {
+    const originalMessage = context.getMessage();
+    return {
+      data,
+      headers: originalMessage.properties.headers,
+      priority: originalMessage.properties.priority,
+    };
+  }
+
   @MessagePattern({ cmd: 'sum' })
   sum(data: number[]): number {
     return (data || []).reduce((a, b) => a + b);

packages/common/index.ts

@@ -1,6 +1,6 @@
 /*
  * Nest @common
- * Copyright(c) 2017 - 2020 Kamil Mysliwiec
+ * Copyright(c) 2017 - 2021 Kamil Mysliwiec
  * https://nestjs.com
  * MIT Licensed
  */

packages/core/index.ts

@@ -1,6 +1,6 @@
 /*
  * Nest @core
- * Copyright(c) 2017 - 2020 Kamil Mysliwiec
+ * Copyright(c) 2017 - 2021 Kamil Mysliwiec
  * https://nestjs.com
  * MIT Licensed
  */

packages/index.ts

@@ -1,6 +1,6 @@
 /*
  * Nest
- * Copyright(c) 2017 - 2020 Kamil Mysliwiec
+ * Copyright(c) 2017 - 2021 Kamil Mysliwiec
  * https://nestjs.com
  * MIT Licensed
  */

packages/microservices/index.ts

@@ -1,6 +1,6 @@
 /*
  * Nest @microservices
- * Copyright(c) 2017 - 2020 Kamil Mysliwiec
+ * Copyright(c) 2017 - 2021 Kamil Mysliwiec
  * https://nestjs.com
  * MIT Licensed
  */
@@ -15,5 +15,6 @@ export * from './helpers';
 export * from './interfaces';
 export * from './module';
 export * from './nest-microservice';
+export * from './record-builders';
 export * from './server';
 export * from './tokens';

packages/microservices/interfaces/microservice-configuration.interface.ts

@@ -8,7 +8,7 @@ import {
   ProducerConfig,
   ProducerRecord,
 } from '../external/kafka.interface';
-import { MqttClientOptions } from '../external/mqtt-options.interface';
+import { MqttClientOptions, QoS } from '../external/mqtt-options.interface';
 import { ClientOpts } from '../external/redis.interface';
 import { RmqUrl } from '../external/rmq-url.interface';
 import { CustomTransportStrategy } from './custom-transport-strategy.interface';
@@ -96,6 +96,24 @@ export interface MqttOptions {
     url?: string;
     serializer?: Serializer;
     deserializer?: Deserializer;
+    subscribeOptions?: {
+      /**
+       * The QoS
+       */
+      qos: QoS;
+      /*
+       * No local flag
+       * */
+      nl?: boolean;
+      /*
+       * Retain as Published flag
+       * */
+      rap?: boolean;
+      /*
+       * Retain Handling option
+       * */
+      rh?: number;
+    };
   };
 }
 

packages/microservices/server/server-mqtt.ts

@@ -70,6 +70,7 @@ export class ServerMqtt extends Server implements CustomTransportStrategy {
       const { isEventHandler } = this.messageHandlers.get(pattern);
       mqttClient.subscribe(
         isEventHandler ? pattern : this.getRequestPattern(pattern),
+        this.getOptionsProp(this.options, 'subscribeOptions'),
       );
     });
   }

packages/platform-express/index.ts

@@ -1,6 +1,6 @@
 /*
  * Nest @platform-express
- * Copyright(c) 2017 - 2020 Kamil Mysliwiec
+ * Copyright(c) 2017 - 2021 Kamil Mysliwiec
  * https://nestjs.com
  * MIT Licensed
  */

packages/platform-fastify/index.ts

@@ -1,6 +1,6 @@
 /*
  * Nest @platform-express
- * Copyright(c) 2017 - 2020 Kamil Mysliwiec
+ * Copyright(c) 2017 - 2021 Kamil Mysliwiec
  * https://nestjs.com
  * MIT Licensed
  */

packages/platform-socket.io/index.ts

@@ -1,6 +1,6 @@
 /*
  * Nest @platform-socket.io
- * Copyright(c) 2017 - 2020 Kamil Mysliwiec
+ * Copyright(c) 2017 - 2021 Kamil Mysliwiec
  * https://nestjs.com
  * MIT Licensed
  */

packages/platform-ws/index.ts

@@ -1,6 +1,6 @@
 /*
  * Nest @platform-ws
- * Copyright(c) 2017 - 2020 Kamil Mysliwiec
+ * Copyright(c) 2017 - 2021 Kamil Mysliwiec
  * https://nestjs.com
  * MIT Licensed
  */

packages/testing/index.ts

@@ -1,6 +1,6 @@
 /*
  * Nest @testing
- * Copyright(c) 2017 - 2020 Kamil Mysliwiec
+ * Copyright(c) 2017 - 2021 Kamil Mysliwiec
  * https://nestjs.com
  * MIT Licensed
  */

packages/websockets/index.ts

@@ -1,6 +1,6 @@
 /*
  * Nest @websockets
- * Copyright(c) 2017 - 2020 Kamil Mysliwiec
+ * Copyright(c) 2017 - 2021 Kamil Mysliwiec
  * https://nestjs.com
  * MIT Licensed
  */