grpc
/
grpc-node
mirror of https://github.com/grpc/grpc-node.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

grpc-js: make client interceptors tests pass mostly unmodified

This commit is contained in:
murgatroid99 2019-11-14 15:02:24 -08:00
parent 17126e4640
commit 33875dce4a
7 changed files with 1880 additions and 1704 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

103
packages/grpc-js/src/call-stream.ts
View File

@ -152,6 +152,7 @@ export class Http2CallStream implements Call {
filterStack: Filter;
private http2Stream: http2.ClientHttp2Stream | null = null;
private pendingRead = false;
private isWriteFilterPending = false;
private pendingWrite: Buffer | null = null;
private pendingWriteCallback: WriteCallback | null = null;
private writesClosed = false;
@ -160,12 +161,16 @@ export class Http2CallStream implements Call {
private isReadFilterPending = false;
private canPush = false;
/**
* Indicates that an 'end' event has come from the http2 stream, so there
* will be no more data events.
*/
private readsClosed = false;
private statusOutput = false;
private unpushedReadMessages: Array<Buffer | null> = [];
private unfilteredReadMessages: Array<Buffer | null> = [];
private unpushedReadMessages: Buffer[] = [];
private unfilteredReadMessages: Buffer[] = [];
// Status code mapped from :status. To be used if grpc-status is not received
private mappedStatusCode: Status = Status.UNKNOWN;
@ -200,16 +205,7 @@ export class Http2CallStream implements Call {
/* Precondition: this.finalStatus !== null */
if (!this.statusOutput) {
this.statusOutput = true;
/* We do this asynchronously to ensure that no async function is in the
* call stack when we return control to the application. If an async
* function is in the call stack, any exception thrown by the application
* (or our tests) will bubble up and turn into promise rejection, which
* will result in an UnhandledPromiseRejectionWarning. Because that is
* a warning, the error will be effectively swallowed and execution will
* continue */
process.nextTick(() => {
this.listener!.onReceiveStatus(this.finalStatus!);
});
if (this.subchannel) {
this.subchannel.callUnref();
this.subchannel.removeDisconnectListener(this.disconnectListener);
@ -227,30 +223,24 @@ export class Http2CallStream implements Call {
* deserialization failure), that new status takes priority */
if (this.finalStatus === null || this.finalStatus.code === Status.OK) {
this.finalStatus = status;
/* Then, if an incoming message is still being handled or the status code
* is OK, hold off on emitting the status until that is done */
if (this.readsClosed || this.finalStatus.code !== Status.OK) {
this.maybeOutputStatus();
}
}
private maybeOutputStatus() {
if (this.finalStatus !== null) {
/* The combination check of readsClosed and that the two message buffer
* arrays are empty checks that there all incoming data has been fully
* processed */
if (this.finalStatus.code !== Status.OK || (this.readsClosed && this.unpushedReadMessages.length === 0 && this.unfilteredReadMessages.length === 0 && !this.isReadFilterPending)) {
this.outputStatus();
}
}
}
private push(message: Buffer | null): void {
if (message === null) {
this.readsClosed = true;
if (this.finalStatus) {
this.outputStatus();
}
} else {
private push(message: Buffer): void {
this.listener!.onReceiveMessage(message);
/* Don't wait for the upper layer to ask for a read before pushing null
* to close out the call, because pushing null doesn't actually push
* another message up to the upper layer */
if (this.unpushedReadMessages.length > 0 && this.unpushedReadMessages[0] === null) {
this.unpushedReadMessages.shift();
this.push(null);
}
}
this.maybeOutputStatus();
}
private handleFilterError(error: Error) {
@ -261,7 +251,7 @@ export class Http2CallStream implements Call {
/* If we the call has already ended, we don't want to do anything with
* this message. Dropping it on the floor is correct behavior */
if (this.finalStatus !== null) {
this.push(null);
this.maybeOutputStatus();
return;
}
this.isReadFilterPending = false;
@ -275,24 +265,16 @@ export class Http2CallStream implements Call {
if (this.unfilteredReadMessages.length > 0) {
/* nextMessage is guaranteed not to be undefined because
unfilteredReadMessages is non-empty */
const nextMessage = this.unfilteredReadMessages.shift() as Buffer | null;
const nextMessage = this.unfilteredReadMessages.shift()!;
this.filterReceivedMessage(nextMessage);
}
}
private filterReceivedMessage(framedMessage: Buffer | null) {
private filterReceivedMessage(framedMessage: Buffer) {
/* If we the call has already ended, we don't want to do anything with
* this message. Dropping it on the floor is correct behavior */
if (this.finalStatus !== null) {
this.push(null);
return;
}
if (framedMessage === null) {
if (this.canPush) {
this.push(null);
} else {
this.unpushedReadMessages.push(null);
}
this.maybeOutputStatus();
return;
}
this.isReadFilterPending = true;
@ -304,7 +286,7 @@ export class Http2CallStream implements Call {
);
}
private tryPush(messageBytes: Buffer | null): void {
private tryPush(messageBytes: Buffer): void {
if (this.isReadFilterPending) {
this.unfilteredReadMessages.push(messageBytes);
} else {
@ -411,12 +393,23 @@ export class Http2CallStream implements Call {
}
});
stream.on('end', () => {
this.tryPush(null);
this.readsClosed = true;
this.maybeOutputStatus();
});
stream.on('close', async () => {
stream.on('close', () => {
let code: Status;
let details = '';
switch (stream.rstCode) {
case http2.constants.NGHTTP2_NO_ERROR:
/* If we get a NO_ERROR code and we already have a status, the
* stream completed properly and we just haven't fully processed
* it yet */
if (this.finalStatus !== null) {
return;
}
code = Status.INTERNAL;
details = `Received RST_STREAM with code ${stream.rstCode}`;
break;
case http2.constants.NGHTTP2_REFUSED_STREAM:
code = Status.UNAVAILABLE;
details = 'Stream refused by server';
@ -435,6 +428,7 @@ export class Http2CallStream implements Call {
break;
default:
code = Status.INTERNAL;
details = `Received RST_STREAM with code ${stream.rstCode}`;
}
// This is a no-op if trailers were received at all.
// This is OK, because status codes emitted here correspond to more
@ -456,9 +450,7 @@ export class Http2CallStream implements Call {
}
stream.write(this.pendingWrite, this.pendingWriteCallback);
}
if (this.writesClosed) {
stream.end();
}
this.maybeCloseWrites();
}
}
@ -514,7 +506,7 @@ export class Http2CallStream implements Call {
/* If we have already emitted a status, we should not emit any more
* messages and we should communicate that the stream has ended */
if (this.finalStatus !== null) {
this.push(null);
this.maybeOutputStatus();
return;
}
this.canPush = true;
@ -522,7 +514,7 @@ export class Http2CallStream implements Call {
this.pendingRead = true;
} else {
if (this.unpushedReadMessages.length > 0) {
const nextMessage: Buffer | null = this.unpushedReadMessages.shift() as Buffer | null;
const nextMessage: Buffer = this.unpushedReadMessages.shift()!;
this.push(nextMessage);
this.canPush = false;
return;
@ -534,26 +526,33 @@ export class Http2CallStream implements Call {
}
}
private maybeCloseWrites() {
if (this.writesClosed && !this.isWriteFilterPending && this.http2Stream !== null) {
this.http2Stream.end();
}
}
sendMessageWithContext(context: MessageContext, message: Buffer) {
const writeObj: WriteObject = {
message: message,
flags: context.flags
};
const cb: WriteCallback = context.callback || (() => {});
this.isWriteFilterPending = true;
this.filterStack.sendMessage(Promise.resolve(writeObj)).then(message => {
this.isWriteFilterPending = false;
if (this.http2Stream === null) {
this.pendingWrite = message.message;
this.pendingWriteCallback = cb;
} else {
this.http2Stream.write(message.message, cb);
this.maybeCloseWrites();
}
}, this.handleFilterError.bind(this));
}
halfClose() {
this.writesClosed = true;
if (this.http2Stream !== null) {
this.http2Stream.end();
}
this.maybeCloseWrites();
}
}

50
packages/grpc-js/src/call.ts
View File

@ -18,11 +18,12 @@
import { EventEmitter } from 'events';
import { Duplex, Readable, Writable } from 'stream';
import { Call, StatusObject, WriteObject } from './call-stream';
import { StatusObject, MessageContext } from './call-stream';
import { Status } from './constants';
import { EmitterAugmentation1 } from './events';
import { Metadata } from './metadata';
import { ObjectReadable, ObjectWritable, WriteCallback } from './object-stream';
import { InterceptingCallInterface } from './client-interceptors';
/**
* A type extending the built-in Error object with additional fields.
@ -81,7 +82,7 @@ export function callErrorFromStatus(status: StatusObject): ServiceError {
export class ClientUnaryCallImpl extends EventEmitter
implements ClientUnaryCall {
constructor(private readonly call: Call) {
constructor(private readonly call: InterceptingCallInterface) {
super();
}
@ -97,7 +98,7 @@ export class ClientUnaryCallImpl extends EventEmitter
export class ClientReadableStreamImpl<ResponseType> extends Readable
implements ClientReadableStream<ResponseType> {
constructor(
private readonly call: Call,
private readonly call: InterceptingCallInterface,
readonly deserialize: (chunk: Buffer) => ResponseType
) {
super({ objectMode: true });
@ -116,33 +117,10 @@ export class ClientReadableStreamImpl<ResponseType> extends Readable
}
}
function tryWrite<RequestType>(
call: Call,
serialize: (value: RequestType) => Buffer,
chunk: RequestType,
encoding: string,
cb: WriteCallback
) {
let message: Buffer;
const flags: number = Number(encoding);
try {
message = serialize(chunk);
} catch (e) {
call.cancelWithStatus(Status.INTERNAL, 'Serialization failure');
cb(e);
return;
}
const writeObj: WriteObject = { message };
if (!Number.isNaN(flags)) {
writeObj.flags = flags;
}
call.write(writeObj, cb);
}
export class ClientWritableStreamImpl<RequestType> extends Writable
implements ClientWritableStream<RequestType> {
constructor(
private readonly call: Call,
private readonly call: InterceptingCallInterface,
readonly serialize: (value: RequestType) => Buffer
) {
super({ objectMode: true });
@ -157,12 +135,14 @@ export class ClientWritableStreamImpl<RequestType> extends Writable
}
_write(chunk: RequestType, encoding: string, cb: WriteCallback) {
const writeObj: WriteObject = { message: chunk };
const context: MessageContext = {
callback: cb
}
const flags: number = Number(encoding);
if (!Number.isNaN(flags)) {
writeObj.flags = flags;
context.flags = flags;
}
this.call.write(writeObj, cb);
this.call.sendMessageWithContext(context, chunk);
}
_final(cb: Function) {
@ -174,7 +154,7 @@ export class ClientWritableStreamImpl<RequestType> extends Writable
export class ClientDuplexStreamImpl<RequestType, ResponseType> extends Duplex
implements ClientDuplexStream<RequestType, ResponseType> {
constructor(
private readonly call: Call,
private readonly call: InterceptingCallInterface,
readonly serialize: (value: RequestType) => Buffer,
readonly deserialize: (chunk: Buffer) => ResponseType
) {
@ -194,12 +174,14 @@ export class ClientDuplexStreamImpl<RequestType, ResponseType> extends Duplex
}
_write(chunk: RequestType, encoding: string, cb: WriteCallback) {
const writeObj: WriteObject = { message: chunk };
const context: MessageContext = {
callback: cb
}
const flags: number = Number(encoding);
if (!Number.isNaN(flags)) {
writeObj.flags = flags;
context.flags = flags;
}
this.call.write(writeObj, cb);
this.call.sendMessageWithContext(context, chunk);
}
_final(cb: Function) {

57
packages/grpc-js/src/client-interceptors.ts
View File

@ -155,8 +155,9 @@ export interface InterceptorOptions extends CallOptions {
export interface InterceptingCallInterface {
cancelWithStatus(status: Status, details: string): void;
getPeer(): string;
start(metadata: Metadata, listener: InterceptingListener): void;
start(metadata: Metadata, listener?: Partial<InterceptingListener>): void;
sendMessageWithContext(context: MessageContext, message: any): void;
sendMessage(message: any): void;
startRead(): void;
halfClose(): void;
@ -194,18 +195,23 @@ export class InterceptingCall implements InterceptingCallInterface {
getPeer() {
return this.nextCall.getPeer();
}
start(metadata: Metadata, interceptingListener: InterceptingListener): void {
this.requester.start(metadata, interceptingListener, (md, listener) => {
start(metadata: Metadata, interceptingListener?: Partial<InterceptingListener>): void {
const fullInterceptingListener: InterceptingListener = {
onReceiveMetadata: interceptingListener?.onReceiveMetadata?.bind(interceptingListener) ?? (metadata => {}),
onReceiveMessage: interceptingListener?.onReceiveMessage?.bind(interceptingListener) ?? (message => {}),
onReceiveStatus: interceptingListener?.onReceiveStatus?.bind(interceptingListener) ?? (status => {})
}
this.requester.start(metadata, fullInterceptingListener, (md, listener) => {
let finalInterceptingListener: InterceptingListener;
if (isInterceptingListener(listener)) {
finalInterceptingListener = listener;
} else {
const fullListener: FullListener = {
onReceiveMetadata: listener.onReceiveMetadata || defaultListener.onReceiveMetadata,
onReceiveMessage: listener.onReceiveMessage || defaultListener.onReceiveMessage,
onReceiveStatus: listener.onReceiveStatus || defaultListener.onReceiveStatus
onReceiveMetadata: listener.onReceiveMetadata ?? defaultListener.onReceiveMetadata,
onReceiveMessage: listener.onReceiveMessage ?? defaultListener.onReceiveMessage,
onReceiveStatus: listener.onReceiveStatus ?? defaultListener.onReceiveStatus
};
finalInterceptingListener = new InterceptingListenerImpl(fullListener, interceptingListener);
finalInterceptingListener = new InterceptingListenerImpl(fullListener, fullInterceptingListener);
}
this.nextCall.start(md, finalInterceptingListener);
});
@ -218,7 +224,7 @@ export class InterceptingCall implements InterceptingCallInterface {
if (this.pendingHalfClose) {
this.nextCall.halfClose();
}
})
});
}
sendMessage(message: any): void {
this.sendMessageWithContext({}, message);
@ -308,17 +314,20 @@ class BaseInterceptingCall implements InterceptingCallInterface {
this.call.cancelWithStatus(Status.INTERNAL, 'Serialization failure');
}
}
start(metadata: Metadata, listener: InterceptingListener): void {
sendMessage(message: any) {
this.sendMessageWithContext({}, message);
}
start(metadata: Metadata, interceptingListener?: Partial<InterceptingListener>): void {
let readError: StatusObject | null = null;
this.call.start(metadata, {
onReceiveMetadata: (metadata) => {
listener.onReceiveMetadata(metadata);
interceptingListener?.onReceiveMetadata?.(metadata);
},
onReceiveMessage: (message) => {
let deserialized: any;
try {
deserialized = this.methodDefinition.responseDeserialize(message);
listener.onReceiveMessage(deserialized);
interceptingListener?.onReceiveMessage?.(deserialized);
} catch (e) {
readError = {code: Status.INTERNAL, details: 'Failed to parse server response', metadata: new Metadata()};
this.call.cancelWithStatus(readError.code, readError.details);
@ -326,9 +335,9 @@ class BaseInterceptingCall implements InterceptingCallInterface {
},
onReceiveStatus: (status) => {
if (readError) {
listener.onReceiveStatus(readError);
interceptingListener?.onReceiveStatus?.(readError);
} else {
listener.onReceiveStatus(status);
interceptingListener?.onReceiveStatus?.(status);
}
}
});
@ -345,8 +354,22 @@ class BaseUnaryInterceptingCall extends BaseInterceptingCall implements Intercep
constructor(call: Call, methodDefinition: ClientMethodDefinition<any, any>) {
super(call, methodDefinition);
}
start(metadata: Metadata, listener: InterceptingListener): void {
super.start(metadata, listener);
start(metadata: Metadata, listener?: Partial<InterceptingListener>): void {
let receivedMessage = false;
const wrapperListener: InterceptingListener = {
onReceiveMetadata: listener?.onReceiveMetadata?.bind(listener) ?? (metadata => {}),
onReceiveMessage: (message: any) => {
receivedMessage = true;
listener?.onReceiveMessage?.(message);
},
onReceiveStatus: (status: StatusObject) => {
if (!receivedMessage) {
listener?.onReceiveMessage?.(null);
}
listener?.onReceiveStatus?.(status);
}
}
super.start(metadata, wrapperListener);
this.call.startRead();
}
}
@ -416,8 +439,8 @@ export function getInterceptingCall(interceptorArgs: InterceptorArguments, metho
* initialValue, which is effectively at the end of the list, is a nextCall
* function that invokes getBottomInterceptingCall, which handles
* (de)serialization and also gets the underlying call from the channel */
const getCall: NextCall = interceptors.reduceRight<NextCall>((previousValue: NextCall, currentValue: Interceptor) => {
return currentOptions => currentValue(currentOptions, previousValue);
const getCall: NextCall = interceptors.reduceRight<NextCall>((nextCall: NextCall, nextInterceptor: Interceptor) => {
return currentOptions => nextInterceptor(currentOptions, nextCall);
}, (finalOptions: InterceptorOptions) => getBottomInterceptingCall(channel, methodDefinition.path, finalOptions, methodDefinition));
return getCall(interceptorOptions);
}

44
packages/grpc-js/src/client.ts
View File

@ -29,14 +29,14 @@ import {
SurfaceCall,
} from './call';
import { CallCredentials } from './call-credentials';
import { Call, Deadline, StatusObject, WriteObject, InterceptingListener } from './call-stream';
import { Deadline, StatusObject, WriteObject, InterceptingListener } from './call-stream';
import { Channel, ConnectivityState, ChannelImplementation } from './channel';
import { ChannelCredentials } from './channel-credentials';
import { ChannelOptions } from './channel-options';
import { Status } from './constants';
import { Metadata } from './metadata';
import { ClientMethodDefinition } from './make-client';
import { getInterceptingCall, Interceptor, InterceptorProvider, InterceptorArguments } from './client-interceptors';
import { getInterceptingCall, Interceptor, InterceptorProvider, InterceptorArguments, InterceptingCallInterface } from './client-interceptors';
const CHANNEL_SYMBOL = Symbol();
const INTERCEPTOR_SYMBOL = Symbol();
@ -231,13 +231,13 @@ export class Client {
callInterceptors: options.interceptors || [],
callInterceptorProviders: options.interceptor_providers || []
};
const call: Call = getInterceptingCall(interceptorArgs, methodDefinition, options, this[CHANNEL_SYMBOL]);
const call: InterceptingCallInterface = getInterceptingCall(interceptorArgs, methodDefinition, options, this[CHANNEL_SYMBOL]);
if (options.credentials) {
call.setCredentials(options.credentials);
}
const writeObj: WriteObject = { message: argument };
const emitter = new ClientUnaryCallImpl(call);
let responseMessage: ResponseType | null = null;
let receivedStatus = false;
call.start(metadata, {
onReceiveMetadata: (metadata) => {
emitter.emit('metadata', metadata);
@ -247,9 +247,12 @@ export class Client {
call.cancelWithStatus(Status.INTERNAL, 'Too many responses received');
}
responseMessage = message;
call.startRead();
},
onReceiveStatus(status: StatusObject) {
if (receivedStatus) {
return;
}
receivedStatus = true;
if (status.code === Status.OK) {
callback!(null, responseMessage!);
} else {
@ -258,8 +261,8 @@ export class Client {
emitter.emit('status', status);
}
});
call.write(writeObj, () => {call.halfClose();});
call.startRead();
call.sendMessage(argument);
call.halfClose();
return emitter;
}
@ -315,12 +318,13 @@ export class Client {
callInterceptors: options.interceptors || [],
callInterceptorProviders: options.interceptor_providers || []
};
const call: Call = getInterceptingCall(interceptorArgs, methodDefinition, options, this[CHANNEL_SYMBOL]);
const call: InterceptingCallInterface = getInterceptingCall(interceptorArgs, methodDefinition, options, this[CHANNEL_SYMBOL]);
if (options.credentials) {
call.setCredentials(options.credentials);
}
const emitter = new ClientWritableStreamImpl<RequestType>(call, serialize);
let responseMessage: ResponseType | null = null;
let receivedStatus = false;
call.start(metadata, {
onReceiveMetadata: (metadata) => {
emitter.emit('metadata', metadata);
@ -330,9 +334,12 @@ export class Client {
call.cancelWithStatus(Status.INTERNAL, 'Too many responses received');
}
responseMessage = message;
call.startRead();
},
onReceiveStatus(status: StatusObject) {
if (receivedStatus) {
return;
}
receivedStatus = true;
if (status.code === Status.OK) {
callback!(null, responseMessage!);
} else {
@ -341,7 +348,6 @@ export class Client {
emitter.emit('status', status);
}
});
call.startRead();
return emitter;
}
@ -406,12 +412,12 @@ export class Client {
callInterceptors: options.interceptors || [],
callInterceptorProviders: options.interceptor_providers || []
};
const call: Call = getInterceptingCall(interceptorArgs, methodDefinition, options, this[CHANNEL_SYMBOL]);
const call: InterceptingCallInterface = getInterceptingCall(interceptorArgs, methodDefinition, options, this[CHANNEL_SYMBOL]);
if (options.credentials) {
call.setCredentials(options.credentials);
}
const writeObj: WriteObject = { message: argument };
const stream = new ClientReadableStreamImpl<ResponseType>(call, deserialize);
let receivedStatus = false;
call.start(metadata, {
onReceiveMetadata(metadata: Metadata) {
stream.emit('metadata', metadata);
@ -422,6 +428,10 @@ export class Client {
}
},
onReceiveStatus(status: StatusObject) {
if (receivedStatus) {
return;
}
receivedStatus = true;
stream.push(null);
if (status.code !== Status.OK) {
stream.emit('error', callErrorFromStatus(status));
@ -429,7 +439,8 @@ export class Client {
stream.emit('status', status);
}
});
call.write(writeObj, () => {call.halfClose();});
call.sendMessage(argument);
call.halfClose();
return stream;
}
@ -467,7 +478,7 @@ export class Client {
callInterceptors: options.interceptors || [],
callInterceptorProviders: options.interceptor_providers || []
};
const call: Call = getInterceptingCall(interceptorArgs, methodDefinition, options, this[CHANNEL_SYMBOL]);
const call: InterceptingCallInterface = getInterceptingCall(interceptorArgs, methodDefinition, options, this[CHANNEL_SYMBOL]);
if (options.credentials) {
call.setCredentials(options.credentials);
}
@ -476,6 +487,7 @@ export class Client {
serialize,
deserialize
);
let receivedStatus = false;
call.start(metadata, {
onReceiveMetadata(metadata: Metadata) {
stream.emit('metadata', metadata);
@ -486,6 +498,10 @@ export class Client {
}
},
onReceiveStatus(status: StatusObject) {
if (receivedStatus) {
return;
}
receivedStatus = true;
stream.push(null);
if (status.code !== Status.OK) {
stream.emit('error', callErrorFromStatus(status));

32
packages/grpc-js/src/server-call.ts
View File

@ -351,10 +351,8 @@ export class Http2ServerCallStream<
});
this.stream.once('close', () => {
if (this.stream.rstCode === http2.constants.NGHTTP2_CANCEL) {
this.cancelled = true;
this.emit('cancelled', 'cancelled');
}
});
this.stream.on('drain', () => {
@ -362,7 +360,20 @@ export class Http2ServerCallStream<
});
}
private checkCancelled(): boolean {
/* In some cases the stream can become destroyed before the close event
* fires. That creates a race condition that this check works around */
if (this.stream.destroyed) {
this.cancelled = true;
}
return this.cancelled;
}
sendMetadata(customMetadata?: Metadata) {
if (this.checkCancelled()) {
return;
}
if (this.metadataSent) {
return;
}
@ -397,6 +408,13 @@ export class Http2ServerCallStream<
metadata.remove(GRPC_TIMEOUT_HEADER);
}
// Remove several headers that should not be propagated to the application
metadata.remove(http2.constants.HTTP2_HEADER_ACCEPT_ENCODING);
metadata.remove(http2.constants.HTTP2_HEADER_TE);
metadata.remove(http2.constants.HTTP2_HEADER_CONTENT_TYPE);
metadata.remove('grpc-encoding');
metadata.remove('grpc-accept-encoding');
return metadata;
}
@ -450,6 +468,9 @@ export class Http2ServerCallStream<
metadata?: Metadata,
flags?: number
) {
if (this.checkCancelled()) {
return;
}
if (!metadata) {
metadata = new Metadata();
}
@ -472,7 +493,7 @@ export class Http2ServerCallStream<
}
sendStatus(statusObj: StatusObject) {
if (this.cancelled) {
if (this.checkCancelled()) {
return;
}
@ -497,6 +518,9 @@ export class Http2ServerCallStream<
}
sendError(error: ServerErrorResponse | ServerStatusResponse) {
if (this.checkCancelled()) {
return;
}
const status: StatusObject = {
code: Status.UNKNOWN,
details: 'message' in error ? error.message : 'Unknown Error',
@ -522,7 +546,7 @@ export class Http2ServerCallStream<
}
write(chunk: Buffer) {
if (this.cancelled) {
if (this.checkCancelled()) {
return;
}

1
packages/grpc-js/src/server.ts
View File

@ -346,7 +346,6 @@ export class Server {
const call = new Http2ServerCallStream(stream, handler);
const metadata: Metadata = call.receiveMetadata(headers) as Metadata;
switch (handler.type) {
case 'unary':
handleUnary(call, handler as UntypedUnaryHandler, metadata);

229
test/api/client_interceptors_test.js
View File

@ -93,12 +93,13 @@ CallRegistry.prototype.maybeCallDone = function() {
}
};
describe('Client interceptors', function() {
describe.only(`${anyGrpc.clientName} client -> ${anyGrpc.serverName} server`, function() {
describe('Client interceptors', function() {
var echo_server;
var echo_port;
var client;
function startServer() {
function startServer(done) {
echo_server = new serverGrpc.Server();
echo_server.addService(echo_service, {
echo: function(call, callback) {
@ -166,8 +167,12 @@ describe('Client interceptors', function() {
}
});
var server_credentials = serverGrpc.ServerCredentials.createInsecure();
echo_port = echo_server.bind('localhost:0', server_credentials);
echo_server.bindAsync('localhost:0', server_credentials, (error, port) => {
assert.ifError(error);
echo_port = port;
echo_server.start();
done();
});
}
function stopServer() {
@ -178,8 +183,8 @@ describe('Client interceptors', function() {
client = new EchoClient('localhost:' + echo_port, insecureCreds);
}
before(function() {
startServer();
before(function(done) {
startServer(done);
});
beforeEach(function() {
resetClient();
@ -193,9 +198,9 @@ describe('Client interceptors', function() {
var registry;
var options;
before(function() {
var interceptor_a = function (options, nextCall) {
var stored_listener;
var stored_metadata;
var interceptor_a = function (options, nextCall) {
options.call_number = 1;
registry.addCall('construct a ' + options.call_number);
return new InterceptingCall(nextCall(options), {
@ -313,7 +318,7 @@ describe('Client interceptors', function() {
return new InterceptingCall(nextCall(options), {
start: function (metadata, listener, next) {
next(metadata, {
onReceiveMetadata: function () { },
onReceiveMetadata: function (metadata, next) { },
onReceiveMessage: function (message, next) {
registry.addCall('interceptor_a');
var second_call = nextCall(options);
@ -321,7 +326,7 @@ describe('Client interceptors', function() {
second_call.sendMessage(message);
second_call.halfClose();
},
onReceiveStatus: function () { }
onReceiveStatus: function (status, next) { }
});
}
});
@ -352,9 +357,8 @@ describe('Client interceptors', function() {
var message = {};
message.value = 'foo';
client.echo(message, options, function(err) {
if (!err) {
assert.ifError(err);
registry.addCall('response');
}
});
});
it('with client streaming call', function(done) {
@ -362,16 +366,15 @@ describe('Client interceptors', function() {
var message = {};
message.value = 'foo';
var stream = client.echoClientStream(options, function(err, response) {
if (!err) {
assert.ifError(err);
registry.addCall('response');
}
});
stream.write(message);
stream.end();
});
});
it.skip('will delay operations and short circuit unary requests', function(done) {
it('will delay operations and short circuit unary requests', function(done) {
var registry = new CallRegistry(done, ['foo_miss', 'foo_hit', 'bar_miss',
'foo_hit_done', 'foo_miss_done', 'bar_miss_done']);
var cache = {};
@ -651,12 +654,13 @@ describe('Client interceptors', function() {
var message = {value: 'error'};
client.echo(message, options, function(err, response) {
assert.ifError(err);
assert.strictEqual(response.value, 'foo');
registry.addCall('result');
});
});
describe('handle interceptor errors', function (doneOuter) {
describe('handle interceptor errors', function () {
var options;
before(function () {
var foo_interceptor = function (options, nextCall) {
@ -686,7 +690,6 @@ describe('Client interceptors', function() {
assert.strictEqual(err.message,
'16 UNAUTHENTICATED: Error in foo interceptor');
done();
doneOuter();
});
});
});
@ -696,9 +699,9 @@ describe('Client interceptors', function() {
var options;
before(function () {
var fallback_response = { value: 'fallback' };
var interceptor = function (options, nextCall) {
var savedMessage;
var savedMessageNext;
var interceptor = function (options, nextCall) {
var requester = (new RequesterBuilder())
.withStart(function (metadata, listener, next) {
var new_listener = (new ListenerBuilder())
@ -709,7 +712,7 @@ describe('Client interceptors', function() {
.withOnReceiveStatus(function (status, next) {
if (status.code !== clientGrpc.status.OK) {
savedMessageNext(fallback_response);
next((new StatusBuilder()).withCode(clientGrpc.status.OK));
next((new StatusBuilder()).withCode(clientGrpc.status.OK).build());
} else {
savedMessageNext(savedMessage);
next(status);
@ -726,6 +729,7 @@ describe('Client interceptors', function() {
it('with client streaming call', function (done) {
var registry = new CallRegistry(done, ['foo_result', 'fallback_result']);
var stream = client.echoClientStream(options, function (err, response) {
assert.ifError(err);
assert.strictEqual(response.value, 'foo');
registry.addCall('foo_result');
});
@ -793,7 +797,8 @@ describe('Client interceptors', function() {
method_name = 'echoServerStream';
method_path_last = 'EchoServerStream';
client.echoServerStream(message, metadata, options);
const stream = client.echoServerStream(message, metadata, options);
stream.on('error', () => {});
});
it('with bidi streaming call', function(cb) {
@ -802,7 +807,8 @@ describe('Client interceptors', function() {
method_name = 'echoBidiStream';
method_path_last = 'EchoBidiStream';
client.echoBidiStream(metadata, options);
const stream = client.echoBidiStream(metadata, options);
stream.on('error', () => {});
});
});
@ -850,7 +856,6 @@ describe('Client interceptors', function() {
});
it('with client streaming call', function(done) {
var client_stream_definition = {
path: '/EchoService/EchoClientStream',
requestStream: true,
@ -859,7 +864,7 @@ describe('Client interceptors', function() {
registry = new CallRegistry(done, [
client_stream_definition,
'result_client_stream'
], false, true);
]);
var metadata = new Metadata();
var message = {value: initial_value};
var client_stream = client.echoClientStream(metadata, options,
@ -921,7 +926,8 @@ describe('Client interceptors', function() {
'invocation_interceptor': 1,
'result_unary': 1,
'result_stream': 1,
'result_invocation': 1
'result_invocation': 1,
'status_stream': 1
});
var constructor_interceptor_a = function(options, nextCall) {
@ -969,13 +975,20 @@ describe('Client interceptors', function() {
var int_client = new EchoClient('localhost:' + echo_port, insecureCreds,
constructor_options);
var message = {};
int_client.echo(message, function() {
int_client.echo(message, function(error, response) {
assert.ifError(error);
registry.addCall('result_unary');
});
var stream = int_client.echoServerStream(message);
stream.on('data', function() {
registry.addCall('result_stream');
});
stream.on('status', (status) => {
registry.addCall('status_stream');
});
stream.on('error', (error) => {
assert.ifError(error);
});
var options = { interceptors: [invocation_interceptor] };
int_client.echo(message, options, function() {
@ -1034,7 +1047,7 @@ describe('Client interceptors', function() {
describe('trigger a stack of interceptors in nested order', function() {
var registry;
var expected_calls = ['constructA', 'constructB', 'outboundA', 'outboundB',
'inboundB', 'inboundA'];
'inboundB', 'inboundA', 'callDone'];
var options;
before(function() {
var interceptor_a = function (options, nextCall) {
@ -1074,13 +1087,18 @@ describe('Client interceptors', function() {
it('with unary call', function(done) {
registry = new CallRegistry(done, expected_calls, true);
client.echo(message, metadata, options, function(){});
client.echo(message, metadata, options, function(error, response){
assert.ifError(error);
registry.addCall('callDone');
});
});
it('with client streaming call', function(done) {
registry = new CallRegistry(done, expected_calls, true);
var client_stream = client.echoClientStream(metadata, options,
function() {});
var client_stream = client.echoClientStream(metadata, options, function(error, response) {
assert.ifError(error);
registry.addCall('callDone');
});
client_stream.write(message);
client_stream.end();
});
@ -1089,12 +1107,26 @@ describe('Client interceptors', function() {
registry = new CallRegistry(done, expected_calls, true);
var stream = client.echoServerStream(message, metadata, options);
stream.on('data', function() {});
stream.on('status', (status) => {
assert.strictEqual(status.code, clientGrpc.status.OK);
registry.addCall('callDone');
});
stream.on('error', (error) => {
assert.ifError(error);
});
});
it('with bidi streaming call', function(done) {
registry = new CallRegistry(done, expected_calls, true);
var bidi_stream = client.echoBidiStream(metadata, options);
bidi_stream.on('data', function(){});
bidi_stream.on('status', (status) => {
assert.strictEqual(status.code, clientGrpc.status.OK);
registry.addCall('callDone');
});
bidi_stream.on('error', (error) => {
assert.ifError(error);
});
bidi_stream.write(message);
bidi_stream.end();
});
@ -1105,7 +1137,8 @@ describe('Client interceptors', function() {
'interceptor_a_start',
'interceptor_b_start',
'interceptor_a_send',
'interceptor_b_send'
'interceptor_b_send',
'call_end'
];
var registry;
var options;
@ -1144,13 +1177,18 @@ describe('Client interceptors', function() {
it('with unary call', function(done) {
registry = new CallRegistry(done, expected_calls, true);
client.echo(message, metadata, options, function(){});
client.echo(message, metadata, options, function(error, response){
assert.ifError(error);
registry.addCall('call_end');
});
});
it('with client streaming call', function(done) {
registry = new CallRegistry(done, expected_calls, true);
var client_stream = client.echoClientStream(metadata, options,
function() {});
var client_stream = client.echoClientStream(metadata, options, function(error, response) {
assert.ifError(error);
registry.addCall('call_end');
});
client_stream.write(message);
client_stream.end();
});
@ -1159,12 +1197,26 @@ describe('Client interceptors', function() {
registry = new CallRegistry(done, expected_calls, true);
var stream = client.echoServerStream(message, metadata, options);
stream.on('data', function() {});
stream.on('status', (status) => {
assert.strictEqual(status.code, clientGrpc.status.OK);
registry.addCall('call_end');
});
stream.on('error', (error) => {
assert.ifError(error);
});
});
it('with bidi streaming call', function(done) {
registry = new CallRegistry(done, expected_calls, true);
var bidi_stream = client.echoBidiStream(metadata, options);
bidi_stream.on('data', function(){});
bidi_stream.on('status', (status) => {
assert.strictEqual(status.code, clientGrpc.status.OK);
registry.addCall('call_end');
});
bidi_stream.on('error', (error) => {
assert.ifError(error);
});
bidi_stream.write(message);
bidi_stream.end();
});
@ -1185,7 +1237,7 @@ describe('Client interceptors', function() {
foo: 'foovalue',
bar: 'barvalue'
};
var expected_calls = ['foo', 'bar', 'response'];
var expected_calls = ['foo', 'bar', 'response', 'end'];
var options;
before(function () {
var foo_interceptor = function (options, nextCall) {
@ -1214,7 +1266,10 @@ describe('Client interceptors', function() {
var metadata = new Metadata();
metadata.add(keys.original, values.original);
var unary_call = client.echo(message, metadata, options, function () {});
var unary_call = client.echo(message, metadata, options, function (error, response) {
assert.ifError(error);
registry.addCall('end');
});
unary_call.on('metadata', function (metadata) {
var has_expected_values = _.every(key_names, function (key_name) {
return _.isEqual(metadata.get(keys[key_name]), [values[key_name]]);
@ -1229,7 +1284,9 @@ describe('Client interceptors', function() {
metadata.add(keys.original, values.original);
var client_stream = client.echoClientStream(metadata, options,
function () {
function (error, response) {
assert.ifError(error);
registry.addCall('end');
});
client_stream.write(message);
client_stream.on('metadata', function (metadata) {
@ -1254,6 +1311,13 @@ describe('Client interceptors', function() {
registry.addCall('response');
});
server_stream.on('data', function() { });
server_stream.on('status', (status) => {
assert.strictEqual(status.code, clientGrpc.status.OK);
registry.addCall('end');
});
server_stream.on('error', (error) => {
assert.ifError(error);
});
});
it('with bidi streaming call', function(done) {
registry = new CallRegistry(done, expected_calls, true);
@ -1269,6 +1333,13 @@ describe('Client interceptors', function() {
registry.addCall('response');
});
bidi_stream.on('data', function() { });
bidi_stream.on('status', (status) => {
assert.strictEqual(status.code, clientGrpc.status.OK);
registry.addCall('end');
});
bidi_stream.on('error', (error) => {
assert.ifError(error);
});
bidi_stream.write(message);
bidi_stream.end();
});
@ -1280,7 +1351,7 @@ describe('Client interceptors', function() {
var expectedValue = 'bar';
var options;
var metadata = new Metadata();
var expected_calls = ['messageIntercepted', 'response'];
var expected_calls = ['messageIntercepted', 'response', 'end'];
before(function() {
var foo_interceptor = function (options, nextCall) {
@ -1301,17 +1372,21 @@ describe('Client interceptors', function() {
var message = {value: originalValue};
client.echo(message, metadata, options, function (err, response) {
assert.ifError(err);
assert.strictEqual(response.value, expectedValue);
registry.addCall('response');
registry.addCall('end');
});
});
it('with client streaming call', function(done) {
registry = new CallRegistry(done, expected_calls, true);
registry = new CallRegistry(done, expected_calls, true, true);
var message = {value: originalValue};
var client_stream = client.echoClientStream(metadata, options,
function (err, response) {
assert.ifError(err);
assert.strictEqual(response.value, expectedValue);
registry.addCall('response');
registry.addCall('end');
});
client_stream.write(message);
client_stream.end();
@ -1324,6 +1399,13 @@ describe('Client interceptors', function() {
assert.strictEqual(data.value, expectedValue);
registry.addCall('response');
});
server_stream.on('status', (status) => {
assert.strictEqual(status.code, clientGrpc.status.OK);
registry.addCall('end');
});
server_stream.on('error', (error) => {
assert.ifError(error);
});
});
it('with bidi streaming call', function(done) {
registry = new CallRegistry(done, expected_calls, true);
@ -1333,6 +1415,13 @@ describe('Client interceptors', function() {
assert.strictEqual(data.value, expectedValue);
registry.addCall('response');
});
bidi_stream.on('status', (status) => {
assert.strictEqual(status.code, clientGrpc.status.OK);
registry.addCall('end');
});
bidi_stream.on('error', (error) => {
assert.ifError(error);
});
bidi_stream.write(message);
bidi_stream.end();
});
@ -1341,7 +1430,7 @@ describe('Client interceptors', function() {
describe('trigger when client closes the call', function() {
var registry;
var expected_calls = [
'response', 'halfClose'
'response', 'halfClose', 'end'
];
var message = {};
var options;
@ -1360,20 +1449,20 @@ describe('Client interceptors', function() {
it('with unary call', function (done) {
registry = new CallRegistry(done, expected_calls);
client.echo(message, options, function (err, response) {
if (!err) {
assert.ifError(err);
registry.addCall('response');
}
registry.addCall('end');
});
});
it('with client streaming call', function (done) {
registry = new CallRegistry(done, expected_calls);
var client_stream = client.echoClientStream(options,
function (err, response) { });
client_stream.write(message, function (err) {
if (!err) {
function (err, response) {
assert.ifError(err);
registry.addCall('response');
}
registry.addCall('end');
});
client_stream.write(message);
client_stream.end();
});
it('with server streaming call', function (done) {
@ -1382,6 +1471,13 @@ describe('Client interceptors', function() {
server_stream.on('data', function (data) {
registry.addCall('response');
});
server_stream.on('status', (status) => {
assert.strictEqual(status.code, clientGrpc.status.OK);
registry.addCall('end');
});
server_stream.on('error', (error) => {
assert.ifError(error);
});
});
it('with bidi streaming call', function (done) {
registry = new CallRegistry(done, expected_calls);
@ -1389,6 +1485,13 @@ describe('Client interceptors', function() {
bidi_stream.on('data', function (data) {
registry.addCall('response');
});
bidi_stream.on('status', (status) => {
assert.strictEqual(status.code, clientGrpc.status.OK);
registry.addCall('end');
});
bidi_stream.on('error', (error) => {
assert.ifError(error);
});
bidi_stream.write(message);
bidi_stream.end();
});
@ -1425,11 +1528,17 @@ describe('Client interceptors', function() {
it('with server streaming call', function(cb) {
done = cb;
var stream = client.echoServerStream(message, options);
stream.on('error', (error) => {
assert.strictEqual(error.code, clientGrpc.status.CANCELLED);
});
stream.cancel();
});
it('with bidi streaming call', function(cb) {
done = cb;
var stream = client.echoBidiStream(options);
stream.on('error', (error) => {
assert.strictEqual(error.code, clientGrpc.status.CANCELLED);
});
stream.cancel();
});
});
@ -1481,6 +1590,9 @@ describe('Client interceptors', function() {
assert.strictEqual(metadata.get(expectedKey)[0], expectedValue);
done();
});
server_stream.on('error', (error) => {
assert.ifError(error);
});
server_stream.on('data', function() { });
});
it('with bidi streaming call', function(done) {
@ -1492,6 +1604,9 @@ describe('Client interceptors', function() {
done();
});
bidi_stream.on('data', function() { });
bidi_stream.on('error', (error) => {
assert.ifError(error);
})
bidi_stream.write(message);
});
});
@ -1526,6 +1641,7 @@ describe('Client interceptors', function() {
it('with unary call', function (done) {
var message = { value: originalValue };
client.echo(message, metadata, options, function (err, response) {
assert.ifError(err);
assert.strictEqual(response.value, expectedValue);
done();
});
@ -1534,6 +1650,7 @@ describe('Client interceptors', function() {
var message = { value: originalValue };
var client_stream = client.echoClientStream(metadata, options,
function (err, response) {
assert.ifError(err);
assert.strictEqual(response.value, expectedValue);
done();
});
@ -1547,6 +1664,9 @@ describe('Client interceptors', function() {
assert.strictEqual(data.value, expectedValue);
done();
});
server_stream.on('error', (error) => {
assert.ifError(error);
});
});
it('with bidi streaming call', function (done) {
var message = { value: originalValue };
@ -1555,6 +1675,9 @@ describe('Client interceptors', function() {
assert.strictEqual(data.value, expectedValue);
done();
});
bidi_stream.on('error', (error) => {
assert.ifError(error);
});
bidi_stream.write(message);
bidi_stream.end();
});
@ -1666,10 +1789,12 @@ describe('Client interceptors', function() {
it('with client streaming call', function (done) {
var message = { value: 'foo' };
var client_stream = client.echoClientStream(metadata, options,
function () { });
function (error, response) {
assert.ifError(error);
done();
});
client_stream.on('metadata', function (metadata) {
assert.equal(metadata.get('fromMessage'), 'foo');
done();
});
client_stream.write(message);
client_stream.end();
@ -1679,14 +1804,21 @@ describe('Client interceptors', function() {
var bidi_stream = client.echoBidiStream(metadata, options);
bidi_stream.on('metadata', function (metadata) {
assert.equal(metadata.get('fromMessage'), 'foo');
});
bidi_stream.on('data', () => {});
bidi_stream.on('status', (status) => {
assert.strictEqual(status.code, clientGrpc.status.OK);
done();
});
bidi_stream.on('error', (error) => {
assert.ifError(error);
});
bidi_stream.write(message);
bidi_stream.end();
});
});
describe.only('order of operations enforced for async interceptors', function() {
describe.skip('order of operations enforced for async interceptors', function() {
it('with unary call', function(done) {
var expected_calls = [
'close_b',
@ -1694,7 +1826,7 @@ describe('Client interceptors', function() {
'start_b',
'done'
];
var registry = new CallRegistry(done, expected_calls, true, true);
var registry = new CallRegistry(done, expected_calls, true);
var message = {value: 'foo'};
var interceptor_a = function(options, nextCall) {
return new InterceptingCall(nextCall(options), {
@ -1737,7 +1869,7 @@ describe('Client interceptors', function() {
'start_b',
'done'
];
var registry = new CallRegistry(done, expected_calls, true, true);
var registry = new CallRegistry(done, expected_calls, true);
var message = {value: 'foo'};
var interceptor_a = function(options, nextCall) {
return new InterceptingCall(nextCall(options), {
@ -1775,4 +1907,5 @@ describe('Client interceptors', function() {
});
});
});
});
});