Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

feat: ability to directly write the gRPC frame to the ServerWritableStream #2652

Closed
wants to merge 1 commit into from
Closed

feat: ability to directly write the gRPC frame to the ServerWritableStream #2652

Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
51 changes: 43 additions & 8 deletions packages/grpc-js/src/server-call.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -90,21 +90,32 @@ export type ServerSurfaceCall = {
getPath(): string;
} & EventEmitter;

export type ServerWritableBuffer<ResponseType> = {
_write(
chunk: ResponseType | Buffer,
encoding: string,
callback: Function
): void;
write(chunk: ResponseType | Buffer, cb?: Function): boolean;
write(chunk: ResponseType | Buffer, encoding?: any, cb?: Function): boolean;
};

export type ServerUnaryCall<RequestType, ResponseType> = ServerSurfaceCall & {
request: RequestType;
};
export type ServerReadableStream<RequestType, ResponseType> =
ServerSurfaceCall & ObjectReadable<RequestType>;
export type ServerWritableStream<RequestType, ResponseType> =
ServerSurfaceCall &
ObjectWritable<ResponseType> & {
ObjectWritable<ResponseType> &
ServerWritableBuffer<ResponseType> & {
request: RequestType;
end: (metadata?: Metadata) => void;
};
export type ServerDuplexStream<RequestType, ResponseType> = ServerSurfaceCall &
ObjectReadable<RequestType> &
ObjectWritable<ResponseType> & { end: (metadata?: Metadata) => void };

ObjectWritable<ResponseType> &
ServerWritableBuffer<ResponseType> & { end: (metadata?: Metadata) => void };
export class ServerUnaryCallImpl<RequestType, ResponseType>
extends EventEmitter
implements ServerUnaryCall<RequestType, ResponseType>
Expand Down Expand Up @@ -222,7 +233,7 @@ export class ServerWritableStreamImpl<RequestType, ResponseType>
}

_write(
chunk: ResponseType,
chunk: ResponseType | Buffer,
encoding: string,
// eslint-disable-next-line @typescript-eslint/no-explicit-any
callback: (...args: any[]) => void
Expand Down Expand Up @@ -654,15 +665,39 @@ export class Http2ServerCallStream<
}
}

serializeMessage(value: ResponseType) {
serializeMessage(value: ResponseType | Buffer) {
// TODO(cjihrig): Call compression aware serializeMessage().

if (Buffer.isBuffer(value)) {
return this.serializeMessageHandleBuffer(value);
}

const messageBuffer = this.handler.serialize(value);
return this.addCompressionAndLength(messageBuffer);
}

// TODO(cjihrig): Call compression aware serializeMessage().
const byteLength = messageBuffer.byteLength;
private serializeMessageHandleBuffer(value: Buffer): Buffer {
const byteLength = value.byteLength;
// checking if this is a protobuf message or a gRPC frame
if (
byteLength >= 5 &&
(value.readUInt8(0) === 0 || value.readUint8(0) === 1) &&
value.readUInt32BE(1) === byteLength - 5
) {
return value;
}

return this.addCompressionAndLength(value);
}

private addCompressionAndLength(value: Buffer, compressed = false) {
const byteLength = value.byteLength;
const compressionByte = compressed ? 1 : 0;
const output = Buffer.allocUnsafe(byteLength + 5);
output.writeUInt8(0, 0);
output.writeUInt8(compressionByte, 0);
output.writeUInt32BE(byteLength, 1);
messageBuffer.copy(output, 5);
value.copy(output, 5);
return output;
}

Expand Down