Skip to content

Commit 907d07b

Browse files
kamilmysliwieckamilmysliwiec
authored
Merge pull request nestjs#3964 from nestjs/feat/client-side-streams
feat(microservices): support passing upstream subject to grpc client
2 parents db0cefb + ced278c commit 907d07b

8 files changed

Lines changed: 264 additions & 26 deletions

File tree

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

Lines changed: 16 additions & 1 deletion
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 => {
@@ -204,7 +219,7 @@ describe('Advanced GRPC transport', () => {
204219
to: 'test1',
205220
carrier: 'test-carrier',
206221
},
207-
})
222+
});
208223
});
209224

210225
return new Promise((resolve, reject) => {

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

Lines changed: 23 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

‎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/test/client/client-grpc.spec.ts‎

Lines changed: 73 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -1,7 +1,7 @@
11
import { Logger } from '@nestjs/common';
22
import { expect } from 'chai';
33
import { join } from 'path';
4-
import { Observable } from 'rxjs';
4+
import { Observable, Subject } from 'rxjs';
55
import * as sinon from 'sinon';
66
import { ClientGrpcProxy } from '../../client/client-grpc';
77
import { InvalidGrpcPackageException } from '../../errors/invalid-grpc-package.exception';
@@ -110,7 +110,11 @@ describe('ClientGrpcProxy', () => {
110110

111111
describe('createStreamServiceMethod', () => {
112112
it('should return observable', () => {
113-
const fn = client.createStreamServiceMethod({}, 'method');
113+
const methodKey = 'method';
114+
const fn = client.createStreamServiceMethod(
115+
{ [methodKey]: {} },
116+
methodKey,
117+
);
114118
expect(fn()).to.be.instanceof(Observable);
115119
});
116120
describe('on subscribe', () => {
@@ -134,6 +138,35 @@ describe('ClientGrpcProxy', () => {
134138
});
135139
});
136140

141+
describe('when stream request', () => {
142+
const methodName = 'm';
143+
const writeSpy = sinon.spy();
144+
const obj = {
145+
[methodName]: () => ({ on: (type, fn) => fn(), write: writeSpy }),
146+
};
147+
148+
let stream$: Observable<any>;
149+
let upstream: Subject<unknown>;
150+
151+
beforeEach(() => {
152+
upstream = new Subject();
153+
(obj[methodName] as any).requestStream = true;
154+
stream$ = client.createStreamServiceMethod(obj, methodName)(upstream);
155+
});
156+
157+
it('should subscribe to request upstream', () => {
158+
const upstreamSubscribe = sinon.spy(upstream, 'subscribe');
159+
stream$.subscribe(
160+
() => ({}),
161+
() => ({}),
162+
);
163+
upstream.next({ test: true });
164+
165+
expect(writeSpy.called).to.be.true;
166+
expect(upstreamSubscribe.called).to.be.true;
167+
});
168+
});
169+
137170
describe('flow-control', () => {
138171
const methodName = 'm';
139172
type EvtCallback = (...args: any[]) => void;
@@ -206,7 +239,11 @@ describe('ClientGrpcProxy', () => {
206239

207240
describe('createUnaryServiceMethod', () => {
208241
it('should return observable', () => {
209-
const fn = client.createUnaryServiceMethod({}, 'method');
242+
const methodKey = 'method';
243+
const fn = client.createUnaryServiceMethod(
244+
{ [methodKey]: {} },
245+
methodKey,
246+
);
210247
expect(fn()).to.be.instanceof(Observable);
211248
});
212249
describe('on subscribe', () => {
@@ -229,6 +266,39 @@ describe('ClientGrpcProxy', () => {
229266
expect(spy.called).to.be.true;
230267
});
231268
});
269+
describe('when stream request', () => {
270+
const writeSpy = sinon.spy();
271+
const methodName = 'm';
272+
const obj = {
273+
[methodName]: callback => {
274+
callback(null, {});
275+
return {
276+
write: writeSpy,
277+
};
278+
},
279+
};
280+
281+
let stream$: Observable<any>;
282+
let upstream: Subject<unknown>;
283+
284+
beforeEach(() => {
285+
upstream = new Subject();
286+
(obj[methodName] as any).requestStream = true;
287+
stream$ = client.createUnaryServiceMethod(obj, methodName)(upstream);
288+
});
289+
290+
it('should subscribe to request upstream', () => {
291+
const upstreamSubscribe = sinon.spy(upstream, 'subscribe');
292+
stream$.subscribe(
293+
() => ({}),
294+
() => ({}),
295+
);
296+
upstream.next({ test: true });
297+
298+
expect(writeSpy.called).to.be.true;
299+
expect(upstreamSubscribe.called).to.be.true;
300+
});
301+
});
232302
});
233303

234304
describe('createClients', () => {

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

Lines changed: 43 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -6,6 +6,7 @@ import { of } from 'rxjs';
66
import * as sinon from 'sinon';
77
import { InvalidGrpcPackageException } from '../../errors/invalid-grpc-package.exception';
88
import { ServerGrpc } from '../../server/server-grpc';
9+
import { CANCEL_EVENT } from '../../constants';
910

1011
class NoopLogger extends Logger {
1112
log(message: any, context?: string): void {}
@@ -441,6 +442,48 @@ describe('ServerGrpc', () => {
441442

442443
expect(handler.called).to.be.true;
443444
});
445+
describe('when response is not a stream', () => {
446+
it('should call callback', async () => {
447+
const handler = async () => ({ test: true });
448+
const fn = server.createRequestStreamMethod(handler, false);
449+
const call = {
450+
on: (event, callback) => {
451+
if (event !== CANCEL_EVENT) {
452+
callback();
453+
}
454+
},
455+
off: sinon.spy(),
456+
end: sinon.spy(),
457+
write: sinon.spy(),
458+
};
459+
460+
const responseCallback = sinon.spy();
461+
await fn(call as any, responseCallback);
462+
463+
expect(responseCallback.called).to.be.true;
464+
});
465+
describe('when response is a stream', () => {
466+
it('should call write() and end()', async () => {
467+
const handler = async () => ({ test: true });
468+
const fn = server.createRequestStreamMethod(handler, true);
469+
const call = {
470+
on: (event, callback) => {
471+
if (event !== CANCEL_EVENT) {
472+
callback();
473+
}
474+
},
475+
off: sinon.spy(),
476+
end: sinon.spy(),
477+
write: sinon.spy(),
478+
};
479+
480+
await fn(call as any, null);
481+
482+
expect(call.write.called).to.be.true;
483+
expect(call.end.called).to.be.true;
484+
});
485+
});
486+
});
444487
});
445488

446489
describe('createStreamCallMethod', () => {

0 commit comments

Comments
 (0)