module license import { ModuleRpcCommon } from '../common' ; import { ClientRpcError } from './errors' ; import { EventEmitter } from 'events' ; /**
 * A stream producer takes an RPC method and a request and returns a stream.
 *
 * Stream producers are implemented by streaming protocols (such as gRPC-Web).
 */ export interface StreamProducer { < Request , Message > ( method , request ) : Stream < Message > ; } /**
 * Streams are `EventEmitter` objects that are "read-only" (i.e. they can only register
 * listeners, their `emit` function is not exposed), can be started and canceled,
 * and expose a number of standard events relevant to the lifecycle of an RPC.
 *
 * They are produced by a [[StreamProducer]] which takes an RPC method name and
 * an RPC request.
 *
 * Streams are used to implement both server streams and unary calls (a unary call
 * is then simply a stream that is expected to send one and only one message).
 */ export interface Stream < Message > { /**
     * Starts the stream.  No event will be processed and emitted before the stream is started:
     * basically, nothing happens before `start()` is called.
     */ start ( ) : this ; /** Cancels the stream.  The `canceled` event will be emitted. */ cancel ( ) : this ; param event event on ( event , callback ) : this ; param event event on ( event , callback ) : this ; param event event on ( event , callback ) : this ; } example export function streamFromArray < T > ( array , error ? ) { return new ArrayStream ( array , error ) ; } class ArrayStream < T > extends EventEmitter implements Stream < T > { constructor ( private readonly array , private readonly error ? ) { super ( ) ; } start ( ) { for ( const message of this . array ) { this . emit ( 'message' , message ) ; } if ( this . error ) { this . emit ( 'error' , this . error ) ; } else { this . emit ( 'complete' ) ; } return this ; } cancel ( ) { console . warn ( '"cancel" called on a stream created with "streamFromArray" does nothing' ) ; this . emit ( 'canceled' ) ; return this ; } } typeparam typeparam param source param fn return example export function transformStream < T , U > ( source , fn ) { return new TransformingStream < T , U > ( source , fn ) ; } class TransformingStream < T , U > extends EventEmitter implements Stream < U > { constructor ( private readonly source , fn ) { super ( ) ; source . on ( 'message' , message => { try { this . emit ( 'message' , fn ( message ) ) ; } catch ( err ) { this . emit ( 'error' , err ) ; } } ) ; source . on ( 'error' , err => { this . emit ( 'error' , err ) ; } ) ; } start ( ) { this . source . start ( ) ; return this ; } cancel ( ) { this . source . cancel ( ) ; return this ; } on ( event , callback ) { if ( event === 'message' || event === 'error' ) { return super . on ( event , callback ) ; } this . source . on ( event as any , callback ) ; return this ; } } typeparam param stream return example export function streamAsPromise < Message > ( stream ) { return new Promise < Message [ ] > ( ( accept , reject ) => { const messages = [ ] ; stream . on ( 'message' , message => { messages . push ( message ) ; } ) . on ( 'error' , err => { reject ( err ) ; } ) . on ( 'canceled' , ( ) => { reject ( new ClientRpcError ( ModuleRpcCommon . RpcErrorType . canceled ) ) ; } ) . on ( 'complete' , ( ) => { accept ( messages ) ; } ) . start ( ) ; } ) ; }