chore: resolve conflicts (merge)

Author: kamilmysliwiec

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

@@ -0,0 +1,49 @@
+import * as express from 'express';
+import * as request from 'supertest';
+import { Test } from '@nestjs/testing';
+import { INestApplication } from '@nestjs/common';
+import { Transport } from '@nestjs/microservices';
+import { RMQController } from '../src/rmq/rmq.controller';
+import { RMQBroadcastController } from '../src/rmq/rmq-broadcast.controller';
+
+describe('RabbitMQ transport', () => {
+  let server;
+  let app: INestApplication;
+
+  beforeEach(async () => {
+    const module = await Test.createTestingModule({
+      controllers: [RMQBroadcastController],
+    }).compile();
+
+    server = express();
+    app = module.createNestApplication(server);
+    app.connectMicroservice({
+      transport: Transport.RMQ,
+      options: {
+        urls: [`amqp://admin:admin@localhost`],
+        queue: 'test',
+        queueOptions: { durable: false },
+      },
+    });
+    app.connectMicroservice({
+      transport: Transport.RMQ,
+      options: {
+        urls: [`amqp://admin:admin@localhost`],
+        queue: 'test',
+        queueOptions: { durable: false },
+      },
+    });
+    await app.startAllMicroservicesAsync();
+    await app.init();
+  });
+
+  it(`Broadcast (2 subscribers)`, () => {
+    return request(server)
+      .get('/broadcast')
+      .expect(200, '2');
+  });
+
+  afterEach(async () => {
+    await app.close();
+  });
+});

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

@@ -0,0 +1,76 @@
+import * as express from 'express';
+import * as request from 'supertest';
+import { Test } from '@nestjs/testing';
+import { INestApplication } from '@nestjs/common';
+import { Transport } from '@nestjs/microservices';
+import { RMQController } from '../src/rmq/rmq.controller';
+
+describe('RabbimMQ transport', () => {
+  let server;
+  let app: INestApplication;
+
+  beforeEach(async () => {
+    const module = await Test.createTestingModule({
+      controllers: [RMQController],
+    }).compile();
+
+    server = express();
+    app = module.createNestApplication(server);
+    app.connectMicroservice({
+        transport: Transport.RMQ,
+        options: {
+          urls: [`amqp://admin:admin@localhost`],
+          queue: 'test',
+          queueOptions: { durable: false },
+        },
+    });
+    await app.startAllMicroservicesAsync();
+    await app.init();
+});
+
+  it(`/POST`, () => {
+    return request(server)
+      .post('/?command=sum')
+      .send([1, 2, 3, 4, 5])
+      .expect(200, '15');
+  });
+
+  it(`/POST (Promise/async)`, () => {
+    return request(server)
+      .post('/?command=asyncSum')
+      .send([1, 2, 3, 4, 5])
+      .expect(200)
+      .expect(200, '15');
+  });
+
+  it(`/POST (Observable stream)`, () => {
+    return request(server)
+      .post('/?command=streamSum')
+      .send([1, 2, 3, 4, 5])
+      .expect(200, '15');
+  });
+
+  it(`/POST (concurrent)`, () => {
+    return request(server)
+      .post('/concurrent')
+      .send([
+        [1, 2, 3, 4, 5],
+        [6, 7, 8, 9, 10],
+        [11, 12, 13, 14, 15],
+        [16, 17, 18, 19, 20],
+        [21, 22, 23, 24, 25],
+      ])
+      .expect(200, 'true');
+  });
+
+  it(`/POST (streaming)`, () => {
+    return request(server)
+      .post('/stream')
+      .send([1, 2, 3, 4, 5])
+      .expect(200, '15');
+  });
+
+  afterEach(async () => {
+    await app.close();
+  });
+});

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

@@ -0,0 +1,38 @@
+import { Controller, Get } from '@nestjs/common';
+import {
+  Client,
+  MessagePattern,
+  ClientProxy,
+  Transport,
+  ClientProxyFactory,
+} from '@nestjs/microservices';
+import { Observable } from 'rxjs';
+import { scan, take } from 'rxjs/operators';
+
+@Controller()
+export class RMQBroadcastController {
+  client: ClientProxy;
+
+  constructor() {
+    this.client = ClientProxyFactory.create({
+        transport: Transport.RMQ,
+        options: {
+          urls: [`amqp://admin:admin@localhost`],
+          queue: 'test',
+          queueOptions: { durable: false },
+        },
+      });
+  }
+
+  @Get('broadcast')
+  multicats() {
+    return this.client
+      .send<number>({ cmd: 'broadcast' }, {})
+      .pipe(scan((a, b) => a + b), take(2));
+  }
+
+  @MessagePattern({ cmd: 'broadcast' })
+  replyBroadcast(): Observable<number> {
+    return new Observable(observer => observer.next(1));
+  }
+}

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

@@ -0,0 +1,76 @@
+import { Controller, Get, Post, Body, Query, HttpCode } from '@nestjs/common';
+import {
+  Client,
+  MessagePattern,
+  ClientProxy,
+  Transport,
+  ClientProxyFactory,
+} from '@nestjs/microservices';
+import { Observable, of, from } from 'rxjs';
+import { scan } from 'rxjs/operators';
+
+@Controller()
+export class RMQController {
+  client: ClientProxy;
+
+  constructor() {
+    this.client = ClientProxyFactory.create({
+        transport: Transport.RMQ,
+        options: {
+          urls: [`amqp://admin:admin@localhost`],
+          queue: 'test',
+          queueOptions: { durable: false },
+        },
+      });
+  }
+
+  @Post()
+  @HttpCode(200)
+  call(@Query('command') cmd, @Body() data: number[]) {
+    return this.client.send<number>({ cmd }, data);
+  }
+
+  @Post('stream')
+  @HttpCode(200)
+  stream(@Body() data: number[]): Observable<number> {
+    return this.client
+      .send<number>({ cmd: 'streaming' }, data)
+      .pipe(scan((a, b) => a + b));
+  }
+
+  @Post('concurrent')
+  @HttpCode(200)
+  concurrent(@Body() data: number[][]): Promise<boolean> {
+    const send = async (tab: number[]) => {
+      const expected = tab.reduce((a, b) => a + b);
+      const result = await this.client
+        .send<number>({ cmd: 'sum' }, tab)
+        .toPromise();
+
+      return result === expected;
+    };
+    return data
+      .map(async tab => await send(tab))
+      .reduce(async (a, b) => (await a) && (await b));
+  }
+
+  @MessagePattern({ cmd: 'sum' })
+  sum(data: number[]): number {
+    return (data || []).reduce((a, b) => a + b);
+  }
+
+  @MessagePattern({ cmd: 'asyncSum' })
+  async asyncSum(data: number[]): Promise<number> {
+    return (data || []).reduce((a, b) => a + b);
+  }
+
+  @MessagePattern({ cmd: 'streamSum' })
+  streamSum(data: number[]): Observable<number> {
+    return of((data || []).reduce((a, b) => a + b));
+  }
+
+  @MessagePattern({ cmd: 'streaming' })
+  streaming(data: number[]): Observable<number> {
+    return from(data);
+  }
+}

package-lock.json

@@ -40,6 +40,8 @@
     "@nestjs/testing": "5.1.0",
     "@nestjs/websockets": "5.1.0",
     "@nuxtjs/opencollective": "^0.1.0",
+    "amqp-connection-manager": "^2.1.1",
+    "amqplib": "^0.5.2",
     "axios": "^0.17.1",
     "cache-manager": "^2.9.0",
     "class-transformer": "^0.1.8",
@@ -75,6 +77,7 @@
     "uuid": "^3.3.2"
   },
   "devDependencies": {
+    "@types/amqplib": "^0.5.7",
     "@types/cache-manager": "^1.2.6",
     "@types/chai": "^3.5.2",
     "@types/chai-as-promised": "0.0.31",

package.json

@@ -4,4 +4,5 @@ export enum Transport {
   NATS,
   MQTT,
   GRPC,
+  RMQ
 }