Skip to content

Commit fca5411

Browse files
kamilmysliwieckamilmysliwiec
committed
Merge branch 'grpc-proto-tests-upgrade' of https://github.com/anton-alation/nest into anton-alation-grpc-proto-tests-upgrade
2 parents a3ad69d + bf17608 commit fca5411

7 files changed

Lines changed: 300 additions & 0 deletions

File tree

‎integration/microservices/e2e/orders-grpc.spec.ts‎

Lines changed: 166 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,166 @@
1+
import { INestApplication } from '@nestjs/common';
2+
import { Transport } from '@nestjs/microservices';
3+
import { Test } from '@nestjs/testing';
4+
import * as express from 'express';
5+
import { join } from 'path';
6+
import * as request from 'supertest';
7+
import * as ProtoLoader from '@grpc/proto-loader';
8+
import * as GRPC from 'grpc';
9+
import { expect } from 'chai';
10+
import { fail } from 'assert';
11+
import { AdvancedGrpcController } from '../src/grpc-advanced/advanced.grpc.controller';
12+
13+
describe('Advanced GRPC transport', () => {
14+
let server;
15+
let app: INestApplication;
16+
let client: any;
17+
18+
before(async () => {
19+
const module = await Test.createTestingModule({
20+
controllers: [AdvancedGrpcController],
21+
}).compile();
22+
// Create gRPC + HTTP server
23+
server = express();
24+
app = module.createNestApplication(server);
25+
/*
26+
* Create microservice configuration
27+
*/
28+
app.connectMicroservice({
29+
transport: Transport.GRPC,
30+
options: {
31+
package: 'proto_example',
32+
protoPath: 'root.proto',
33+
loader: {
34+
includeDirs: [join(__dirname, '../src/grpc-advanced/proto')],
35+
keepCase: true,
36+
},
37+
},
38+
});
39+
// Start gRPC microservice
40+
await app.startAllMicroservicesAsync();
41+
await app.init();
42+
// Load proto-buffers for test gRPC dispatch
43+
const proto = ProtoLoader.loadSync('root.proto', {
44+
includeDirs: [join(__dirname, '../src/grpc-advanced/proto')],
45+
}) as any;
46+
// Create Raw gRPC client object
47+
const protoGRPC = GRPC.loadPackageDefinition(proto) as any;
48+
// Create client connected to started services at standard 5000 port
49+
client = new protoGRPC.proto_example.orders.OrderService(
50+
'localhost:5000',
51+
GRPC.credentials.createInsecure(),
52+
);
53+
54+
});
55+
56+
it(`GRPC Sending and Receiving HTTP POST`, () => {
57+
return request(server)
58+
.post('/')
59+
.send('1')
60+
.expect(200, {
61+
id: 1,
62+
itemTypes: [1],
63+
shipmentType: {
64+
from: 'test',
65+
to: 'test1',
66+
carrier: 'test-carrier',
67+
},
68+
});
69+
});
70+
71+
it('GRPC Sending and receiving message', async () => {
72+
73+
// Execute find in Promise
74+
return new Promise(resolve => {
75+
client.find({
76+
id: 1,
77+
}, (err, result) => {
78+
// Compare results
79+
expect(err).to.be.null;
80+
expect(result).to.eql({
81+
id: 1,
82+
itemTypes: [1],
83+
shipmentType: {
84+
from: 'test',
85+
to: 'test1',
86+
carrier: 'test-carrier',
87+
},
88+
});
89+
// Resolve after checkups
90+
resolve();
91+
});
92+
93+
});
94+
95+
});
96+
97+
it('GRPC Sending and receiving Stream from RX handler', async () => {
98+
99+
const callHandler = client.sync();
100+
101+
callHandler.on('data', (msg: number) => {
102+
// Do deep comparison (to.eql)
103+
expect(msg).to.eql({
104+
id: 1,
105+
itemTypes: [1],
106+
shipmentType: {
107+
from: 'test',
108+
to: 'test1',
109+
carrier: 'test-carrier',
110+
},
111+
});
112+
});
113+
114+
callHandler.on('error', (err: any) => {
115+
// We want to fail only on real errors while Cancellation error
116+
// is expected
117+
if (String(err).toLowerCase().indexOf('cancelled') === -1) {
118+
fail('gRPC Stream error happened, error: ' + err);
119+
}
120+
});
121+
122+
return new Promise((resolve, reject) => {
123+
callHandler.write({
124+
id: 1,
125+
});
126+
setTimeout(() => resolve(), 1000);
127+
});
128+
129+
});
130+
131+
it('GRPC Sending and receiving Stream from Call handler', async () => {
132+
133+
const callHandler = client.syncCall();
134+
135+
callHandler.on('data', (msg: number) => {
136+
// Do deep comparison (to.eql)
137+
expect(msg).to.eql({
138+
id: 1,
139+
itemTypes: [1],
140+
shipmentType: {
141+
from: 'test',
142+
to: 'test1',
143+
carrier: 'test-carrier',
144+
},
145+
});
146+
});
147+
148+
callHandler.on('error', (err: any) => {
149+
// We want to fail only on real errors while Cancellation error
150+
// is expected
151+
if (String(err).toLowerCase().indexOf('cancelled') === -1) {
152+
fail('gRPC Stream error happened, error: ' + err);
153+
}
154+
});
155+
156+
return new Promise((resolve, reject) => {
157+
callHandler.write({
158+
id: 1,
159+
});
160+
setTimeout(() => resolve(), 1000);
161+
});
162+
163+
});
164+
165+
166+
});

‎integration/microservices/src/grpc-advanced/advanced.grpc.controller.ts‎

Lines changed: 95 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,95 @@
1+
import { Body, Controller, HttpCode, Post } from '@nestjs/common';
2+
import {
3+
Client, ClientGrpc, GrpcMethod,
4+
GrpcStreamMethod, GrpcStreamCall, Transport,
5+
} from '@nestjs/microservices';
6+
import { join } from 'path';
7+
import { Observable, of, Subject } from 'rxjs';
8+
9+
@Controller()
10+
export class AdvancedGrpcController {
11+
/*
12+
* HTTP Proxy Client defines loading pattern
13+
*/
14+
@Client({
15+
transport: Transport.GRPC,
16+
options: {
17+
package: 'proto_example.orders',
18+
protoPath: 'root.proto',
19+
loader: {
20+
includeDirs: [join(__dirname, './proto')],
21+
keepCase: true,
22+
},
23+
},
24+
})
25+
client: ClientGrpc;
26+
27+
/**
28+
* HTTP Proxy entry for support non-stream find method
29+
* @param id
30+
*/
31+
@Post()
32+
@HttpCode(200)
33+
call(@Body() id: number): Observable<number> {
34+
const svc = this.client.getService<any>('OrderService');
35+
return svc.find({ id });
36+
}
37+
38+
/**
39+
* GRPC stub for Find method
40+
* @param id
41+
*/
42+
@GrpcMethod('orders.OrderService')
43+
async find({ id }: { id: number }): Promise<any> {
44+
return of({
45+
id: 1,
46+
itemTypes: [1],
47+
shipmentType: {
48+
from: 'test',
49+
to: 'test1',
50+
carrier: 'test-carrier',
51+
},
52+
});
53+
}
54+
55+
/**
56+
* GRPC stub implementation for sync stream method
57+
* @param messages
58+
*/
59+
@GrpcStreamMethod('orders.OrderService')
60+
async sync(messages: Observable<any>): Promise<any> {
61+
const s = new Subject();
62+
const o = s.asObservable();
63+
messages.subscribe(msg => {
64+
s.next({
65+
id: 1,
66+
itemTypes: [1],
67+
shipmentType: {
68+
from: 'test',
69+
to: 'test1',
70+
carrier: 'test-carrier',
71+
},
72+
});
73+
});
74+
return o;
75+
}
76+
77+
/**
78+
* GRPC stub implementation for syncCall stream method (implemented through call)
79+
* @param stream
80+
*/
81+
@GrpcStreamCall('orders.OrderService')
82+
async syncCall(stream: any) {
83+
stream.on('data', (msg: any) => {
84+
stream.write({
85+
id: 1,
86+
itemTypes: [1],
87+
shipmentType: {
88+
from: 'test',
89+
to: 'test1',
90+
carrier: 'test-carrier',
91+
},
92+
});
93+
});
94+
}
95+
}

‎integration/microservices/src/grpc-advanced/proto/common/item_types.proto‎

Lines changed: 8 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,8 @@
1+
syntax = "proto3";
2+
package proto_example.common.items;
3+
4+
enum ItemType {
5+
DEFAULT = 0;
6+
SUPERIOR = 1;
7+
FLAWLESS = 2;
8+
}

‎integration/microservices/src/grpc-advanced/proto/common/shipment_types.proto‎

Lines changed: 8 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,8 @@
1+
syntax = "proto3";
2+
package proto_example.common.shipments;
3+
4+
message ShipmentType {
5+
string from = 1;
6+
string to = 2;
7+
string carrier = 3;
8+
}

‎integration/microservices/src/grpc-advanced/proto/orders/message.proto‎

Lines changed: 11 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,11 @@
1+
syntax = "proto3";
2+
package proto_example.orders;
3+
4+
import public "common/item_types.proto";
5+
import public "common/shipment_types.proto";
6+
7+
message Order {
8+
int32 id = 1;
9+
repeated common.items.ItemType itemTypes = 2;
10+
common.shipments.ShipmentType shipmentType = 3;
11+
}

‎integration/microservices/src/grpc-advanced/proto/orders/service.proto‎

Lines changed: 9 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,9 @@
1+
syntax = "proto3";
2+
import "orders/message.proto";
3+
package proto_example.orders;
4+
5+
service OrderService {
6+
rpc Find(Order) returns (Order);
7+
rpc Sync(stream Order) returns (stream Order);
8+
rpc SyncCall(stream Order) returns (stream Order);
9+
}

‎integration/microservices/src/grpc-advanced/proto/root.proto‎

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,3 @@
1+
syntax = "proto3";
2+
package proto_example;
3+
import public "orders/service.proto";

0 commit comments

Comments
 (0)