Skip to content

Commit 748d3aa

Browse files
kamilmysliwieckamilmysliwiec
authored
Merge pull request nestjs#3963 from nestjs/fix/grpc-client-streaming
fix(microservices): fix client-side grpc streaming
2 parents 9bd1a43 + 907d07b commit 748d3aa

10 files changed

Lines changed: 425 additions & 58 deletions

File tree

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

Lines changed: 63 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -69,6 +69,21 @@ describe('Advanced GRPC transport', () => {
6969
});
7070
});
7171

72+
it(`GRPC Streaming and Receiving HTTP POST`, () => {
73+
return request(server)
74+
.post('/client-streaming')
75+
.send('1')
76+
.expect(200, {
77+
id: 1,
78+
itemTypes: [1],
79+
shipmentType: {
80+
from: 'test',
81+
to: 'test1',
82+
carrier: 'test-carrier',
83+
},
84+
});
85+
});
86+
7287
it('GRPC Sending and receiving message', async () => {
7388
// Execute find in Promise
7489
return new Promise(resolve => {
@@ -166,4 +181,52 @@ describe('Advanced GRPC transport', () => {
166181
setTimeout(() => resolve(), 1000);
167182
});
168183
});
184+
185+
it('GRPC Sending Stream and receiving a single message from RX handler', async () => {
186+
const callHandler = client.streamReq((err, res) => {
187+
if (err) {
188+
throw err;
189+
}
190+
expect(res).to.eql({
191+
id: 1,
192+
itemTypes: [1],
193+
shipmentType: {
194+
from: 'test',
195+
to: 'test1',
196+
carrier: 'test-carrier',
197+
},
198+
});
199+
});
200+
201+
return new Promise((resolve, reject) => {
202+
callHandler.write({
203+
id: 1,
204+
});
205+
setTimeout(() => resolve(), 1000);
206+
});
207+
});
208+
209+
it('GRPC Sending Stream and receiving a single message from Call handler', async () => {
210+
const callHandler = client.streamReqCall((err, res) => {
211+
if (err) {
212+
throw err;
213+
}
214+
expect(res).to.eql({
215+
id: 1,
216+
itemTypes: [1],
217+
shipmentType: {
218+
from: 'test',
219+
to: 'test1',
220+
carrier: 'test-carrier',
221+
},
222+
});
223+
});
224+
225+
return new Promise((resolve, reject) => {
226+
callHandler.write({
227+
id: 1,
228+
});
229+
setTimeout(() => resolve(), 1000);
230+
});
231+
});
169232
});

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

Lines changed: 63 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -8,7 +8,7 @@ import {
88
Transport,
99
} from '@nestjs/microservices';
1010
import { join } from 'path';
11-
import { Observable, of, Subject } from 'rxjs';
11+
import { Observable, of, ReplaySubject, Subject } from 'rxjs';
1212

1313
@Controller()
1414
export class AdvancedGrpcController {
@@ -40,6 +40,28 @@ export class AdvancedGrpcController {
4040
return svc.find({ id });
4141
}
4242

43+
/**
44+
* HTTP Proxy entry for support client-side stream find method
45+
* @param id
46+
*/
47+
@Post('client-streaming')
48+
@HttpCode(200)
49+
stream(): Observable<number> {
50+
const svc = this.client.getService<any>('OrderService');
51+
const upstream = new ReplaySubject();
52+
upstream.next({
53+
id: 1,
54+
itemTypes: [1],
55+
shipmentType: {
56+
from: 'test',
57+
to: 'test1',
58+
carrier: 'test-carrier',
59+
},
60+
});
61+
upstream.complete();
62+
return svc.streamReq(upstream);
63+
}
64+
4365
/**
4466
* GRPC stub for Find method
4567
* @param id
@@ -97,4 +119,44 @@ export class AdvancedGrpcController {
97119
});
98120
});
99121
}
122+
123+
@GrpcStreamMethod('orders.OrderService')
124+
async streamReq(messages: Observable<any>): Promise<any> {
125+
const s = new Subject();
126+
const o = s.asObservable();
127+
messages.subscribe(
128+
msg => {
129+
s.next({
130+
id: 1,
131+
itemTypes: [1],
132+
shipmentType: {
133+
from: 'test',
134+
to: 'test1',
135+
carrier: 'test-carrier',
136+
},
137+
});
138+
},
139+
null,
140+
() => s.complete(),
141+
);
142+
return o;
143+
}
144+
145+
@GrpcStreamCall('orders.OrderService')
146+
async streamReqCall(stream: any, callback: Function) {
147+
stream.on('data', (msg: any) => {
148+
// process msg
149+
});
150+
stream.on('end', () => {
151+
callback(null, {
152+
id: 1,
153+
itemTypes: [1],
154+
shipmentType: {
155+
from: 'test',
156+
to: 'test1',
157+
carrier: 'test-carrier',
158+
},
159+
});
160+
});
161+
}
100162
}

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

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -6,4 +6,6 @@ service OrderService {
66
rpc Find(Order) returns (Order);
77
rpc Sync(stream Order) returns (stream Order);
88
rpc SyncCall(stream Order) returns (stream Order);
9+
rpc StreamReq(stream Order) returns (Order);
10+
rpc StreamReqCall(stream Order) returns (Order);
911
}

‎packages/microservices/client/client-grpc.ts‎

Lines changed: 53 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -1,7 +1,7 @@
11
import { Logger } from '@nestjs/common/services/logger.service';
22
import { loadPackage } from '@nestjs/common/utils/load-package.util';
3-
import { isObject } from '@nestjs/common/utils/shared.utils';
4-
import { Observable } from 'rxjs';
3+
import { isFunction, isObject } from '@nestjs/common/utils/shared.utils';
4+
import { Observable, Subscription } from 'rxjs';
55
import {
66
GRPC_DEFAULT_MAX_RECEIVE_MESSAGE_LENGTH,
77
GRPC_DEFAULT_MAX_SEND_MESSAGE_LENGTH,
@@ -111,10 +111,27 @@ export class ClientGrpcProxy extends ClientProxy implements ClientGrpc {
111111
methodName: string,
112112
): (...args: any[]) => Observable<any> {
113113
return (...args: any[]) => {
114-
return new Observable(observer => {
114+
const isRequestStream = client[methodName].requestStream;
115+
const stream = new Observable(observer => {
115116
let isClientCanceled = false;
116-
const call = client[methodName](...args);
117+
let upstreamSubscription: Subscription;
118+
119+
const upstreamSubjectOrData = args[0];
120+
const isUpstreamSubject =
121+
upstreamSubjectOrData && isFunction(upstreamSubjectOrData.subscribe);
122+
123+
const call =
124+
isRequestStream && isUpstreamSubject
125+
? client[methodName]()
126+
: client[methodName](...args);
117127

128+
if (isRequestStream && isUpstreamSubject) {
129+
upstreamSubscription = upstreamSubjectOrData.subscribe(
130+
(val: unknown) => call.write(val),
131+
(err: unknown) => call.emit('error', err),
132+
() => call.end(),
133+
);
134+
}
118135
call.on('data', (data: any) => observer.next(data));
119136
call.on('error', (error: any) => {
120137
if (error.details === GRPC_CANCELLED) {
@@ -126,17 +143,27 @@ export class ClientGrpcProxy extends ClientProxy implements ClientGrpc {
126143
observer.error(error);
127144
});
128145
call.on('end', () => {
146+
if (upstreamSubscription) {
147+
upstreamSubscription.unsubscribe();
148+
upstreamSubscription = null;
149+
}
129150
call.removeAllListeners();
130151
observer.complete();
131152
});
132-
return (): any => {
153+
return () => {
154+
if (upstreamSubscription) {
155+
upstreamSubscription.unsubscribe();
156+
upstreamSubscription = null;
157+
}
158+
133159
if (call.finished) {
134160
return undefined;
135161
}
136162
isClientCanceled = true;
137163
call.cancel();
138164
};
139165
});
166+
return stream;
140167
};
141168
}
142169

@@ -145,6 +172,27 @@ export class ClientGrpcProxy extends ClientProxy implements ClientGrpc {
145172
methodName: string,
146173
): (...args: any[]) => Observable<any> {
147174
return (...args: any[]) => {
175+
const isRequestStream = client[methodName].requestStream;
176+
const upstreamSubjectOrData = args[0];
177+
const isUpstreamSubject =
178+
upstreamSubjectOrData && isFunction(upstreamSubjectOrData.subscribe);
179+
180+
if (isRequestStream && isUpstreamSubject) {
181+
return new Observable(observer => {
182+
const call = client[methodName]((error, data) => {
183+
if (error) {
184+
return observer.error(error);
185+
}
186+
observer.next(data);
187+
observer.complete();
188+
});
189+
upstreamSubjectOrData.subscribe(
190+
(val: unknown) => call.write(val),
191+
(err: unknown) => call.emit('error', err),
192+
() => call.end(),
193+
);
194+
});
195+
}
148196
return new Observable(observer => {
149197
client[methodName](...args, (error: any, data: any) => {
150198
if (error) {

‎packages/microservices/server/server-grpc.ts‎

Lines changed: 65 additions & 25 deletions
Original file line numberDiff line numberDiff line change
@@ -182,11 +182,19 @@ export class ServerGrpc extends Server implements CustomTransportStrategy {
182182
// streaming from the side of requester
183183
if (protoNativeHandler.requestStream) {
184184
// If any handlers were defined with GrpcStreamMethod annotation use RX
185-
if (streamType === GrpcMethodStreamingType.RX_STREAMING)
186-
return this.createStreamDuplexMethod(methodHandler);
185+
if (streamType === GrpcMethodStreamingType.RX_STREAMING) {
186+
return this.createRequestStreamMethod(
187+
methodHandler,
188+
protoNativeHandler.responseStream,
189+
);
190+
}
187191
// If any handlers were defined with GrpcStreamCall annotation
188-
else if (streamType === GrpcMethodStreamingType.PT_STREAMING)
189-
return this.createStreamCallMethod(methodHandler);
192+
else if (streamType === GrpcMethodStreamingType.PT_STREAMING) {
193+
return this.createStreamCallMethod(
194+
methodHandler,
195+
protoNativeHandler.responseStream,
196+
);
197+
}
190198
}
191199
return protoNativeHandler.responseStream
192200
? this.createStreamServiceMethod(methodHandler)
@@ -220,17 +228,23 @@ export class ServerGrpc extends Server implements CustomTransportStrategy {
220228
};
221229
}
222230

223-
public createStreamDuplexMethod(methodHandler: Function) {
224-
return async (call: GrpcCall) => {
231+
public createRequestStreamMethod(
232+
methodHandler: Function,
233+
isResponseStream: boolean,
234+
) {
235+
return async (
236+
call: GrpcCall,
237+
callback: (err: unknown, value: unknown) => void,
238+
) => {
225239
const req = new Subject<any>();
226240
call.on('data', (m: any) => req.next(m));
227241
call.on('error', (e: any) => {
228242
// Check if error means that stream ended on other end
229-
if (
230-
String(e)
231-
.toLowerCase()
232-
.indexOf('cancelled') > -1
233-
) {
243+
const isCancelledError = String(e)
244+
.toLowerCase()
245+
.indexOf('cancelled');
246+
247+
if (isCancelledError) {
234248
call.end();
235249
return;
236250
}
@@ -241,23 +255,49 @@ export class ServerGrpc extends Server implements CustomTransportStrategy {
241255

242256
const handler = methodHandler(req.asObservable());
243257
const res = this.transformToObservable(await handler);
244-
await res
245-
.pipe(
246-
takeUntil(fromEvent(call as any, CANCEL_EVENT)),
247-
catchError(err => {
248-
call.emit('error', err);
249-
return EMPTY;
250-
}),
251-
)
252-
.forEach(m => call.write(m));
253-
254-
call.end();
258+
if (isResponseStream) {
259+
await res
260+
.pipe(
261+
takeUntil(fromEvent(call as any, CANCEL_EVENT)),
262+
catchError(err => {
263+
call.emit('error', err);
264+
return EMPTY;
265+
}),
266+
)
267+
.forEach(m => call.write(m));
268+
269+
call.end();
270+
} else {
271+
const response = await res
272+
.pipe(
273+
takeUntil(fromEvent(call as any, CANCEL_EVENT)),
274+
catchError(err => {
275+
callback(err, null);
276+
return EMPTY;
277+
}),
278+
)
279+
.toPromise();
280+
281+
if (typeof response !== 'undefined') {
282+
callback(null, response);
283+
}
284+
}
255285
};
256286
}
257287

258-
public createStreamCallMethod(methodHandler: Function) {
259-
return async (call: GrpcCall) => {
260-
methodHandler(call);
288+
public createStreamCallMethod(
289+
methodHandler: Function,
290+
isResponseStream: boolean,
291+
) {
292+
return async (
293+
call: GrpcCall,
294+
callback: (err: unknown, value: unknown) => void,
295+
) => {
296+
if (isResponseStream) {
297+
methodHandler(call);
298+
} else {
299+
methodHandler(call, callback);
300+
}
261301
};
262302
}
263303

0 commit comments

Comments
 (0)