Skip to content

Commit c6123e7

Browse files
mkaufmanermkaufmaner
committed
Cleans up extraneous code and cleans up the tests.
1 parent 151e202 commit c6123e7

7 files changed

Lines changed: 128 additions & 480 deletions

File tree

‎integration/docker-compose.yml‎

Lines changed: 9 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -45,11 +45,18 @@ services:
4545
- "15672:15672"
4646
- "5672:5672"
4747
tty: true
48+
zookeeper:
49+
container_name: test-zookeeper
50+
hostname: zookeeper
51+
image: wurstmeister/zookeeper
52+
ports:
53+
- "2181:2181"
4854
kafka:
4955
container_name: test-kafka
5056
hostname: kafka
5157
image: wurstmeister/kafka
5258
ports:
5359
- "9092:9092"
54-
55-
60+
environment:
61+
KAFKA_ADVERTISED_HOST_NAME: localhost
62+
KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181

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

Lines changed: 19 additions & 70 deletions
Original file line numberDiff line numberDiff line change
@@ -29,80 +29,29 @@ describe('Kafka transport', () => {
2929
await app.init();
3030
});
3131

32-
it(`/POST`, () => {
33-
return request(server)
32+
it(`/POST (async command sum)`, done => {
33+
request(server)
3434
.post('/?command=math.sum')
3535
.send([1, 2, 3, 4, 5])
36-
.expect(200, '15');
36+
.end(() => {
37+
setTimeout(() => {
38+
expect(KafkaController.MATH_SUM).to.eq(15);
39+
done();
40+
}, 1000);
41+
});
3742
});
3843

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-
// });
44+
it(`/POST (event notification)`, done => {
45+
request(server)
46+
.post('/notify')
47+
.send([1, 2, 3, 4, 5])
48+
.end(() => {
49+
setTimeout(() => {
50+
expect(KafkaController.IS_NOTIFIED).to.be.true;
51+
done();
52+
}, 1000);
53+
});
54+
});
10655

10756
afterEach(async () => {
10857
await app.close();

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

Lines changed: 0 additions & 27 deletions
This file was deleted.

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

Lines changed: 35 additions & 118 deletions
Original file line numberDiff line numberDiff line change
@@ -1,22 +1,18 @@
1-
import * as util from 'util';
2-
import { Body, Controller, Get, HttpCode, Post, Query, Logger } from '@nestjs/common';
1+
import { Body, Controller, HttpCode, Post, Query, OnModuleInit } from '@nestjs/common';
32
import {
43
Client,
54
ClientProxy,
65
EventPattern,
7-
MessagePattern,
8-
RpcException,
96
Transport,
10-
KafkaConsumer,
11-
ClientKafka,
127
} from '@nestjs/microservices';
13-
import { from, Observable, of, throwError } from 'rxjs';
14-
import { catchError, scan, delay, map } from 'rxjs/operators';
15-
import { Producer } from '@nestjs/common/interfaces/external/kafka-options.interface';
8+
import { Observable } from 'rxjs';
9+
10+
import * as uuid from 'uuid';
1611

1712
@Controller()
18-
export class KafkaController {
13+
export class KafkaController implements OnModuleInit {
1914
static IS_NOTIFIED = false;
15+
static MATH_SUM = 0;
2016

2117
@Client({
2218
transport: Transport.KAFKA,
@@ -27,9 +23,12 @@ export class KafkaController {
2723
},
2824
})
2925
private readonly client: ClientProxy;
30-
// private readonly client: ClientKafka;
3126

32-
private serialize(data: any) {
27+
public async onModuleInit(){
28+
await this.client.connect();
29+
}
30+
31+
private parse(data: any) {
3332
data.message.key = data.message.key.toString();
3433
data.message.value = JSON.parse(data.message.value.toString());
3534
}
@@ -39,116 +38,34 @@ export class KafkaController {
3938
async call(
4039
@Query('command') cmd,
4140
@Body() data: number[],
42-
): Promise<Observable<number>> {
43-
await this.client.connect();
44-
// return this.client.send<number>(cmd, data);
45-
46-
return this.client.emit('test.one', [
47-
{
48-
key: 'sum',
49-
value: JSON.stringify({
50-
values: data,
51-
}),
52-
headers: {
53-
'correlation-id': '2bfb68bb-893a-423b-a7fa-7b568cad5b67',
54-
},
55-
},
56-
]).pipe(map((result) => {
57-
Logger.error(util.format('post() result %o', result));
58-
59-
return 15;
60-
}), delay(500000));
61-
62-
// Logger.error(util.format('post() result %o', result));
63-
64-
// return from(Promise.resolve(15)).pipe(delay(500000));
41+
): Promise<Observable<any>> {
42+
return this.client.emit('math.sum', data.map((num) => {
43+
return {
44+
key: uuid.v4(), // stick to a single partition
45+
value: num.toString(),
46+
};
47+
}));
6548
}
6649

67-
@EventPattern('test.one')
68-
testOne(data: any){
69-
this.serialize(data);
70-
71-
Logger.error(util.format('testOne() data %o', data));
72-
73-
return from(Promise.resolve(15)).pipe(delay(5000));
74-
// throw new Error('test');
50+
@EventPattern('math.sum')
51+
mathSum(data: any){
52+
KafkaController.MATH_SUM += parseFloat(data.message.value);
7553
}
7654

77-
// @MessagePattern(this.consumer)
78-
// testTo(data: any){
79-
// Logger.error(util.format('testOne() data %o', data));
80-
81-
// return from(Promise.resolve(15)).pipe(delay(5000));
82-
// // throw new Error('test');
83-
// }
84-
// testOne(...data: any){
85-
// Logger.error(util.format('testOne() data %o', data));
86-
// throw new Error('test');
87-
// }
88-
89-
// @Post('stream')
90-
// @HttpCode(200)
91-
// stream(@Body() data: number[]): Observable<number> {
92-
// return this.client
93-
// .send<number>('streaming.sum', data)
94-
// .pipe(scan((a, b) => a + b));
95-
// }
96-
97-
// @Post('concurrent')
98-
// @HttpCode(200)
99-
// concurrent(@Body() data: number[][]): Promise<boolean> {
100-
// const send = async (tab: number[]) => {
101-
// const expected = tab.reduce((a, b) => a + b);
102-
// const result = await this.client
103-
// .send<number>('math.sum', tab)
104-
// .toPromise();
105-
106-
// return result === expected;
107-
// };
108-
// return data
109-
// .map(async tab => await send(tab))
110-
// .reduce(async (a, b) => (await a) && (await b));
111-
// }
112-
113-
// @MessagePattern('math.*')
114-
// sum(data: number[]): number {
115-
// return (data || []).reduce((a, b) => a + b);
116-
// }
117-
118-
// @MessagePattern('async.*')
119-
// async asyncSum(data: number[]): Promise<number> {
120-
// return (data || []).reduce((a, b) => a + b);
121-
// }
122-
123-
// @MessagePattern('stream.*')
124-
// streamSum(data: number[]): Observable<number> {
125-
// return of((data || []).reduce((a, b) => a + b));
126-
// }
127-
128-
// @MessagePattern('streaming.*')
129-
// streaming(data: number[]): Observable<number> {
130-
// return from(data);
131-
// }
132-
133-
// @Get('exception')
134-
// async getError() {
135-
// return await this.client
136-
// .send<number>('exception', {})
137-
// .pipe(catchError(err => of(err)));
138-
// }
139-
140-
// @MessagePattern('exception')
141-
// throwError(): Observable<number> {
142-
// return throwError(new RpcException('test'));
143-
// }
55+
@Post('notify')
56+
async sendNotification(): Promise<any> {
57+
return this.client.emit('notification', [{
58+
key: 'notify',
59+
value: JSON.stringify({
60+
notify: true,
61+
}),
62+
}]);
63+
}
14464

145-
// @Post('notify')
146-
// async sendNotification(): Promise<any> {
147-
// return this.client.emit<number>('notification', true);
148-
// }
65+
@EventPattern('notification')
66+
eventHandler(data: any) {
67+
this.parse(data);
14968

150-
// @EventPattern('notification')
151-
// eventHandler(data: boolean) {
152-
// KafkaController.IS_NOTIFIED = data;
153-
// }
69+
KafkaController.IS_NOTIFIED = data.message.value.notify;
70+
}
15471
}

0 commit comments

Comments
 (0)