module preferred license /** */ import { Router , Response as ExpressResponse } from 'express' ; import * as bodyParser from 'body-parser' ; import * as compression from 'compression' ; import { GrpcWebCodec } from '../common/codec' ; import { getMetadataFromGrpcWebError , GrpcWebError , getGrpcWebErrorFromError , sendHttpErrorResponse , } from '../private/grpc' ; import { FullMethodParts , getFullMethodUrl , isLowerCamelCase , } from '../private/method_url' ; import { isServerStreamMethod , isUnaryMethod } from '../../private/helpers' ; import { GrpcWebJsonCodec } from '../common/json_codec' ; import { AssertionError } from 'assert' ; import { grpc } from '@improbable-eng/grpc-web' ; import { decodeHeaderValue , encodeHeaderValue } from '../private/headers' ; import { ModuleRpcUtils } from '../../../utils' ; import { ModuleRpcCommon } from '../../../common' ; import { ModuleRpcServer } from '../../../server' ; import { HttpStatus } from '../../private/http_status' ; /** Options for the RPC server of the gRPC-Web protocol. */ export interface GrpcWebServerOptions < serviceDefinition extends ModuleRpcCommon . ServiceDefinition > { /** The express router to start from.  By default, we start from `Router()`. */ router ? : Router ; /** A function to report errors that occurred during RPCs. */ reportError ? : ReportErrorHandler ; /**
     * Use compression on response bodies.  Compression can be disabled as it is a security issue
     * (see https://www.blackhat.com/docs/asia-16/materials/asia-16-Karakostas-Practical-New-Developments-In-The-BREACH-Attack-wp.pdf
     * for a good overview).  To mitigate this we recommend, among other things, to use metadata
     * to pass "master" secrets (such as session secrets) around.
     */ useCompression ? : boolean ; /**
     * The Codec to use.  By default, we use the [[GrpcWebJsonCodec]] codec.  The Codec must match
     * the Codec used by the client (this is enforced through content-type negotiation).
     */ codec ? : GrpcWebCodec < serviceDefinition > ; /**
     * Maximum request size.   If this is a number, then the value specifies the number of bytes;
     * if it is a string, the value is passed to the bytes library for parsing.
     * Defaults to '100kb'.
     */ requestLimit ? : number | string ; } param err param errorContext export type ReportErrorHandler = ( err : Error , errorContext : ErrorContext ) => void ; /**
 * Context of an error that occurred during the processing of an RPC request by an RPC server.
 */ export interface ErrorContext { /** The full URL of the service method that errored (`<remoteAddress>/<method>`). */ url : string ; } /** The size in bytes of the grpc-web header in front of the chunks (containing messages or trailers) */ const HEADER_SIZE = 5 ; /** Flag in the header to identify trailers. */ const TRAILER_FLAG = 0x80 ; /** Default request limit (see [[GrpcWebServerOptions.requestLimit]]). */ const DEFAULT_REQUEST_LIMIT = '100kb' ; see param serviceDefinition param serviceHandler param serverContextConnector param options export function registerGrpcWebRoutes < serviceDefinition extends ModuleRpcCommon . ServiceDefinition , RequestContext > ( serviceDefinition , serviceHandler , serverContextConnector , options = { } ) { const router = options . router || Router ( ) ; const codec = options . codec || new GrpcWebJsonCodec ( ) ; router . use ( bodyParser . raw ( { type : '*/*' , limit : options . requestLimit || DEFAULT_REQUEST_LIMIT , } ) ) ; if ( options . useCompression ) { router . use ( compression ( ) ) ; } for ( const method in serviceDefinition ) { if ( ! isLowerCamelCase ( method ) ) { throw new ModuleRpcCommon . MisspecifiedServiceError ( `method name ${ method } is not lower camel case (example: fooBar; not FooBar or foo_bar)` ) ; } router . use ( `/${ method }` , async ( req , resp ) => { if ( req . method !== 'POST' ) { resp . sendStatus ( HttpStatus . methodNotAllowed ) ; return ; } if ( req . header ( 'accept' ) !== codec . getContentType ( ) ) { resp . sendStatus ( HttpStatus . notAcceptable ) ; return ; } try { const encodedRequestContext = ModuleRpcUtils . mapValuesWithStringKeys ( ModuleRpcUtils . filterNonNullValues < string , string [ ] | string > ( req . headers ) , value => typeof value === 'string' ? decodeHeaderValue ( value ) : value . map ( decodeHeaderValue ) . join ( ',' ) ) ; const requestContext = await serverContextConnector . decodeRequestContext ( encodedRequestContext ) ; let request ; const requestChunk = new Uint8Array ( req . body ) ; try { request = codec . decodeRequest ( method , requestChunk ) ; } catch ( err ) { if ( err instanceof ModuleRpcServer . ServerError ) { throw err ; } throw new ModuleRpcServer . ServerTransportError ( err ) ; } const methodHandler = serviceHandler [ method ] ; const fullMethodParts = { method } ; if ( isServerStreamMethod ( serviceDefinition [ method ] , methodHandler ) ) { await processServerStream ( fullMethodParts , methodHandler , requestContext , request , resp , serverContextConnector , codec , options . reportError ) ; } else if ( isUnaryMethod ( serviceDefinition [ method ] , methodHandler ) ) { await processUnary ( fullMethodParts , methodHandler , requestContext , request , resp , serverContextConnector , codec , options . reportError ) ; } else { throw new AssertionError ( { message : `unexpected type for service method ${ method }` , } ) ; } } catch ( err ) { safeCaptureError ( options . reportError , err , { url : method } ) ; try { await addEncodedResponseContext ( serverContextConnector , resp , err ) ; } catch ( encodingErr ) { safeCaptureError ( options . reportError , encodingErr , { url : method } ) ; } const { status , grpcWebError } = getGrpcWebErrorFromError ( err ) ; const errorMetadata = getMetadataFromGrpcWebError ( grpcWebError ) ; try { if ( ! resp . headersSent ) { sendHttpErrorResponse ( status , errorMetadata , resp ) ; } else if ( ! resp . finished ) { resp . write ( encodeFrame ( TRAILER_FLAG , codec . encodeTrailer ( errorMetadata ) ) ) ; resp . end ( ) ; } } catch ( err ) { safeCaptureError ( options . reportError , err , { url : method } ) ; if ( err instanceof ModuleRpcServer . ServerError ) { throw err ; } throw new ModuleRpcServer . ServerTransportError ( err ) ; } } } ) ; } return router ; } async function processServerStream < serviceDefinition extends ModuleRpcCommon . ServiceDefinition , method extends ModuleRpcCommon . MethodsFor < serviceDefinition > , RequestContext > ( fullMethodParts , methodHandler , requestContext , request , resp , serverContextConnector , codec , captureError ? ) { enum StreamStatus { notReady = 'notReady' , ready = 'ready' , end = 'end' } await addEncodedResponseContext ( serverContextConnector , resp ) ; let status = StreamStatus . notReady ; let close ; resp . on ( 'close' , ( ) => { if ( close ) { close ( ) ; } } ) ; try { await new Promise < void > ( ( resolve , reject ) => { const callbacks = { onReady : c => { try { if ( status === StreamStatus . ready ) { new ModuleRpcServer . HandlerProtocolError ( getFullMethodUrl ( fullMethodParts ) , 'onReady called more than once' ) ; } else if ( status === StreamStatus . end ) { throw new ModuleRpcServer . HandlerProtocolError ( getFullMethodUrl ( fullMethodParts ) , 'onReady called after end of streaming' ) ; } try { resp . status ( 200 ) ; } catch ( err ) { throw new ModuleRpcServer . ServerTransportError ( err ) ; } close = c ; status = StreamStatus . ready ; } catch ( err ) { reject ( err ) ; } } , onMessage : message => { try { if ( status === StreamStatus . notReady ) { throw new ModuleRpcServer . HandlerProtocolError ( getFullMethodUrl ( fullMethodParts ) , 'onMessage called before onReady' ) ; } else if ( status === StreamStatus . end ) { throw new ModuleRpcServer . HandlerProtocolError ( getFullMethodUrl ( fullMethodParts ) , 'onMessage called after end of streaming' ) ; } let encodedMessage ; try { encodedMessage = codec . encodeMessage ( fullMethodParts . method , message ) ; } catch ( err ) { if ( err instanceof ModuleRpcServer . ServerError ) { throw err ; } throw new ModuleRpcServer . HandlerProtocolError ( getFullMethodUrl ( fullMethodParts ) , `cannot encode message: ${ message }` ) ; } if ( resp . finished ) { throw new ModuleRpcServer . ServerTransportError ( new Error ( 'onMessage after end' ) ) ; } try { resp . write ( encodeFrame ( 0 , encodedMessage ) ) ; } catch ( err ) { throw new ModuleRpcServer . ServerTransportError ( err ) ; } } catch ( err ) { reject ( err ) ; } } , } ; methodHandler ( request , callbacks , requestContext ) . then ( ( ) => { try { resp . write ( encodeFrame ( TRAILER_FLAG , codec . encodeTrailer ( new grpc . Metadata ( ) ) ) ) ; } catch ( err ) { if ( err instanceof ModuleRpcServer . ServerError ) { throw err ; } throw new ModuleRpcServer . ServerTransportError ( err ) ; } resolve ( ) ; } ) . catch ( reject ) ; } ) ; } catch ( err ) { status = StreamStatus . end ; safeCaptureError ( captureError , err , { url : getFullMethodUrl ( fullMethodParts ) , } ) ; let grpcWebError ; if ( err instanceof ModuleRpcServer . ServerRpcError ) { grpcWebError = { errorType : err . errorType } ; if ( err . unsafeTransmittedMessage ) { grpcWebError . message = err . unsafeTransmittedMessage ; } } else { grpcWebError = { errorType : ModuleRpcCommon . RpcErrorType . internal , } ; } try { resp . write ( encodeFrame ( TRAILER_FLAG , codec . encodeTrailer ( getMetadataFromGrpcWebError ( grpcWebError ) ) ) ) ; resp . end ( ) ; } catch ( err ) { safeCaptureError ( captureError , err , { url : getFullMethodUrl ( fullMethodParts ) , } ) ; if ( err instanceof ModuleRpcServer . ServerError ) { throw err ; } throw new ModuleRpcServer . ServerTransportError ( err ) ; } return ; } try { resp . end ( ) ; } catch ( err ) { safeCaptureError ( captureError , err , { url : getFullMethodUrl ( fullMethodParts ) , } ) ; throw new ModuleRpcServer . ServerTransportError ( err ) ; } } async function processUnary < serviceDefinition extends ModuleRpcCommon . ServiceDefinition , method extends ModuleRpcCommon . MethodsFor < serviceDefinition > , RequestContext > ( fullMethodParts , methodHandler , requestContext , request , resp , serverContextConnector , codec , captureError ? ) { const response = await methodHandler ( request , requestContext ) ; await addEncodedResponseContext ( serverContextConnector , resp ) ; try { resp . write ( encodeFrame ( 0 , codec . encodeMessage ( fullMethodParts . method , response ) ) ) ; resp . write ( encodeFrame ( TRAILER_FLAG , codec . encodeTrailer ( new grpc . Metadata ( ) ) ) ) ; resp . end ( ) ; } catch ( err ) { safeCaptureError ( captureError , err , { url : getFullMethodUrl ( fullMethodParts ) , } ) ; if ( err instanceof ModuleRpcServer . ServerError ) { throw err ; } throw new ModuleRpcServer . ServerTransportError ( err ) ; } } /**
 * Add an encoded response context to an HTTP response.
 */ async function addEncodedResponseContext < RequestContext > ( serverContextConnector , resp , err ? ) { const encodedResponseContext = await serverContextConnector . provideResponseContext ( err ) ; ModuleRpcUtils . entries ( encodedResponseContext ) . forEach ( ( [ key , value ] ) => { if ( key ) { resp . setHeader ( key . toString ( ) , encodeHeaderValue ( value ) ) ; } } ) ; } /**
 * Safely call a `captureError` function, i.e. ensure that all errors thrown by
 * `captureError` itself are catched.
 */ function safeCaptureError ( captureError , err , errContext ) { if ( ! captureError ) { return ; } try { captureError ( err , errContext ) ; } catch ( err ) { console . error ( 'Cannot capture error' , err . stack ) ; } } /** Encode a frame to be sent on the wire. */ function encodeFrame ( firstByte , encodedPayload ) { const buf = new Buffer ( HEADER_SIZE + encodedPayload . byteLength ) ; buf . writeUInt8 ( firstByte , 0 ) ; buf . writeUInt32BE ( encodedPayload . byteLength , 1 ) ; for ( let i = 0 ; i < encodedPayload . byteLength ; ++ i ) { buf . writeUInt8 ( encodedPayload [ i ] , HEADER_SIZE + i ) ; } return buf ; }