Skip to content

Commit 16dbe21

Browse files
mkaufmanermkaufmaner
committed
Initial work on the design and implementation.
1 parent 6aa4bfd commit 16dbe21

12 files changed

Lines changed: 664 additions & 132 deletions

File tree

‎integration/microservices/e2e/sum-kafka.spec.ts‎

Lines changed: 110 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,110 @@
1+
import { INestApplication } from '@nestjs/common';
2+
import { Transport } from '@nestjs/microservices';
3+
import { Test } from '@nestjs/testing';
4+
import { expect } from 'chai';
5+
import * as request from 'supertest';
6+
import { KafkaController } from '../src/kafka/kafka.controller';
7+
8+
describe('Kafka transport', () => {
9+
let server;
10+
let app: INestApplication;
11+
12+
beforeEach(async () => {
13+
const module = await Test.createTestingModule({
14+
controllers: [KafkaController],
15+
}).compile();
16+
17+
app = module.createNestApplication();
18+
server = app.getHttpAdapter().getInstance();
19+
20+
app.connectMicroservice({
21+
transport: Transport.KAFKA,
22+
options: {
23+
client: {
24+
brokers: ['localhost:9092'],
25+
},
26+
},
27+
});
28+
await app.startAllMicroservicesAsync();
29+
await app.init();
30+
});
31+
32+
it(`/POST`, () => {
33+
return request(server)
34+
.post('/?command=math.sum')
35+
.send([1, 2, 3, 4, 5])
36+
.expect(200, '15');
37+
});
38+
39+
// it(`/POST`, () => {
40+
// return request(server)
41+
// .post('/?command=math.sum')
42+
// .send([1, 2, 3, 4, 5])
43+
// .expect(200, '15');
44+
// });
45+
46+
// it(`/POST (Promise/async)`, () => {
47+
// return request(server)
48+
// .post('/?command=async.sum')
49+
// .send([1, 2, 3, 4, 5])
50+
// .expect(200)
51+
// .expect(200, '15');
52+
// });
53+
54+
// it(`/POST (Observable stream)`, () => {
55+
// return request(server)
56+
// .post('/?command=stream.sum')
57+
// .send([1, 2, 3, 4, 5])
58+
// .expect(200, '15');
59+
// });
60+
61+
// it(`/POST (streaming)`, () => {
62+
// return request(server)
63+
// .post('/stream')
64+
// .send([1, 2, 3, 4, 5])
65+
// .expect(200, '15');
66+
// });
67+
68+
// it(`/POST (concurrent)`, () => {
69+
// return request(server)
70+
// .post('/concurrent')
71+
// .send([
72+
// Array.from({ length: 10 }, (v, k) => k + 1),
73+
// Array.from({ length: 10 }, (v, k) => k + 11),
74+
// Array.from({ length: 10 }, (v, k) => k + 21),
75+
// Array.from({ length: 10 }, (v, k) => k + 31),
76+
// Array.from({ length: 10 }, (v, k) => k + 41),
77+
// Array.from({ length: 10 }, (v, k) => k + 51),
78+
// Array.from({ length: 10 }, (v, k) => k + 61),
79+
// Array.from({ length: 10 }, (v, k) => k + 71),
80+
// Array.from({ length: 10 }, (v, k) => k + 81),
81+
// Array.from({ length: 10 }, (v, k) => k + 91),
82+
// ])
83+
// .expect(200, 'true');
84+
// });
85+
86+
// it(`/GET (exception)`, () => {
87+
// return request(server)
88+
// .get('/exception')
89+
// .expect(200, {
90+
// message: 'test',
91+
// status: 'error',
92+
// });
93+
// });
94+
95+
// it(`/POST (event notification)`, done => {
96+
// request(server)
97+
// .post('/notify')
98+
// .send([1, 2, 3, 4, 5])
99+
// .end(() => {
100+
// setTimeout(() => {
101+
// expect(KafkaController.IS_NOTIFIED).to.be.true;
102+
// done();
103+
// }, 1000);
104+
// });
105+
// });
106+
107+
afterEach(async () => {
108+
await app.close();
109+
});
110+
});

‎integration/microservices/src/kafka/kafka-broadcast.controller.ts‎

Lines changed: 27 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,27 @@
1+
import { Controller, Get } from '@nestjs/common';
2+
import {
3+
Client,
4+
ClientProxy,
5+
MessagePattern,
6+
Transport,
7+
} from '@nestjs/microservices';
8+
import { Observable } from 'rxjs';
9+
import { scan, take } from 'rxjs/operators';
10+
11+
@Controller()
12+
export class NatsBroadcastController {
13+
@Client({ transport: Transport.NATS })
14+
client: ClientProxy;
15+
16+
@Get('broadcast')
17+
multicats() {
18+
return this.client
19+
.send<number>('broadcast.test', {})
20+
.pipe(scan((a, b) => a + b), take(2));
21+
}
22+
23+
@MessagePattern('broadcast.*')
24+
replyBroadcast(): Observable<number> {
25+
return new Observable(observer => observer.next(1));
26+
}
27+
}

‎integration/microservices/src/kafka/kafka.controller.ts‎

Lines changed: 133 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,133 @@
1+
import * as util from 'util';
2+
import { Body, Controller, Get, HttpCode, Post, Query, Logger } from '@nestjs/common';
3+
import {
4+
Client,
5+
ClientProxy,
6+
EventPattern,
7+
MessagePattern,
8+
RpcException,
9+
Transport,
10+
KafkaConsumer,
11+
} from '@nestjs/microservices';
12+
import { from, Observable, of, throwError } from 'rxjs';
13+
import { catchError, scan, delay } from 'rxjs/operators';
14+
15+
@Controller()
16+
export class KafkaController {
17+
static IS_NOTIFIED = false;
18+
19+
private consumer: any = 'hel[pp me!';
20+
21+
// @KafkaConsumer({
22+
// topic: 'test.two'
23+
// })
24+
// public whoknows;
25+
26+
// @Client({
27+
// transport: Transport.KAFKA,
28+
// options: {
29+
// client: {
30+
// brokers: ['localhost:9092'],
31+
// },
32+
// },
33+
// })
34+
// client: ClientProxy;
35+
36+
@Post()
37+
@HttpCode(200)
38+
async call(
39+
@Query('command') cmd,
40+
@Body() data: number[],
41+
): Promise<Observable<number>> {
42+
// await this.client.connect();
43+
// return this.client.send<number>(cmd, data);
44+
45+
return from(Promise.resolve(15)).pipe(delay(500000));
46+
}
47+
48+
@MessagePattern('test.one')
49+
testOne(data: any){
50+
Logger.error(util.format('testOne() data %o', data));
51+
52+
return from(Promise.resolve(15)).pipe(delay(5000));
53+
// throw new Error('test');
54+
}
55+
56+
@MessagePattern(this.consumer)
57+
testTo(data: any){
58+
Logger.error(util.format('testOne() data %o', data));
59+
60+
return from(Promise.resolve(15)).pipe(delay(5000));
61+
// throw new Error('test');
62+
}
63+
// testOne(...data: any){
64+
// Logger.error(util.format('testOne() data %o', data));
65+
// throw new Error('test');
66+
// }
67+
68+
// @Post('stream')
69+
// @HttpCode(200)
70+
// stream(@Body() data: number[]): Observable<number> {
71+
// return this.client
72+
// .send<number>('streaming.sum', data)
73+
// .pipe(scan((a, b) => a + b));
74+
// }
75+
76+
// @Post('concurrent')
77+
// @HttpCode(200)
78+
// concurrent(@Body() data: number[][]): Promise<boolean> {
79+
// const send = async (tab: number[]) => {
80+
// const expected = tab.reduce((a, b) => a + b);
81+
// const result = await this.client
82+
// .send<number>('math.sum', tab)
83+
// .toPromise();
84+
85+
// return result === expected;
86+
// };
87+
// return data
88+
// .map(async tab => await send(tab))
89+
// .reduce(async (a, b) => (await a) && (await b));
90+
// }
91+
92+
// @MessagePattern('math.*')
93+
// sum(data: number[]): number {
94+
// return (data || []).reduce((a, b) => a + b);
95+
// }
96+
97+
// @MessagePattern('async.*')
98+
// async asyncSum(data: number[]): Promise<number> {
99+
// return (data || []).reduce((a, b) => a + b);
100+
// }
101+
102+
// @MessagePattern('stream.*')
103+
// streamSum(data: number[]): Observable<number> {
104+
// return of((data || []).reduce((a, b) => a + b));
105+
// }
106+
107+
// @MessagePattern('streaming.*')
108+
// streaming(data: number[]): Observable<number> {
109+
// return from(data);
110+
// }
111+
112+
// @Get('exception')
113+
// async getError() {
114+
// return await this.client
115+
// .send<number>('exception', {})
116+
// .pipe(catchError(err => of(err)));
117+
// }
118+
119+
// @MessagePattern('exception')
120+
// throwError(): Observable<number> {
121+
// return throwError(new RpcException('test'));
122+
// }
123+
124+
// @Post('notify')
125+
// async sendNotification(): Promise<any> {
126+
// return this.client.emit<number>('notification', true);
127+
// }
128+
129+
// @EventPattern('notification')
130+
// eventHandler(data: boolean) {
131+
// KafkaController.IS_NOTIFIED = data;
132+
// }
133+
}

‎package-lock.json‎

Lines changed: 3 additions & 3 deletions
Some generated files are not rendered by default. Learn more about customizing how changed files appear on GitHub.

‎package.json‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -59,7 +59,7 @@
5959
"grpc": "1.21.1",
6060
"http2": "3.3.7",
6161
"iterare": "1.2.0",
62-
"kafkajs": "^1.9.2",
62+
"kafkajs": "^1.9.3",
6363
"merge-graphql-schemas": "1.5.8",
6464
"mqtt": "3.0.0",
6565
"multer": "1.4.1",

0 commit comments

Comments
 (0)