import { Duplex } from "readable-stream" ; import { JRPCEngineEndCallback , JRPCEngineNextCallback , JRPCEngineReturnHandler , JRPCMiddleware , JRPCRequest , JRPCResponse , serializeError , } from "./jrpc" ; import SafeEventEmitter from "./safeEventEmitter" ; import SerializableError from "./serializableError" ; /**
 * A JSON-RPC request and response processor.
 * Give it a stack of middleware, pass it requests, and get back responses.
 */ export class JRPCEngine extends SafeEventEmitter { private _middleware : JRPCMiddleware < unknown , unknown > [ ] ; constructor ( ) { super ( ) ; this . _middleware = [ ] ; } returns private static async _runAllMiddleware ( req , res , middlewareStack ) { const returnHandlers = [ ] ; let error = null ; let isComplete = false ; for ( const middleware of middlewareStack ) { [ error , isComplete ] = await JRPCEngine . _runMiddleware ( req , res , middleware , returnHandlers ) ; if ( isComplete ) { break ; } } return [ error , isComplete , returnHandlers . reverse ( ) ] ; } returns private static _runMiddleware ( req , res , middleware , returnHandlers ) { return new Promise ( ( resolve ) => { const end = ( err ? ) => { const error = err || res . error ; if ( error ) { res . error = serializeError ( error ) ; } resolve ( [ error , true ] ) ; } ; const next = ( returnHandler ? ) => { if ( res . error ) { end ( res . error ) ; } else { if ( returnHandler ) { if ( typeof returnHandler !== "function" ) { end ( new SerializableError ( { code : - 32603 , message : "JRPCEngine: 'next' return handlers must be functions" } ) ) ; } returnHandlers . push ( returnHandler ) ; } resolve ( [ null , false ] ) ; } } ; try { middleware ( req , res , next , end ) ; } catch ( error ) { end ( error ) ; } } ) ; } /**
     * Serially executes array of return handlers. The request and response are
     * assumed to be in their scope.
     */ private static async _runReturnHandlers ( handlers ) { for ( const handler of handlers ) { await new Promise < void > ( ( resolve , reject ) => { handler ( ( err ) => ( err ? reject ( err ) : resolve ( ) ) ) ; } ) ; } } /**
     * Throws an error if the response has neither a result nor an error, or if
     * the "isComplete" flag is falsy.
     */ private static _checkForCompletion ( req , res , isComplete ) { if ( ! ( "result" in res ) && ! ( "error" in res ) ) { throw new SerializableError ( { code : - 32603 , message : "Response has no error or result for request" } ) ; } if ( ! isComplete ) { throw new SerializableError ( { code : - 32603 , message : "Nothing ended request" } ) ; } } param middleware push < T , U > ( middleware ) { this . _middleware . push ( middleware as JRPCMiddleware < unknown , unknown > ) ; } param request param callback handle < T , U > ( request , callback ) ; param request param callback handle < T , U > ( requests , callback ) ; param request returns handle < T , U > ( request ) ; param request returns handle < T , U > ( requests ) ; handle ( req , cb ? ) { if ( cb && typeof cb !== "function" ) { throw new Error ( '"callback" must be a function if provided.' ) ; } if ( Array . isArray ( req ) ) { if ( cb ) { return this . _handleBatch ( req , cb ) ; } return this . _handleBatch ( req ) ; } if ( cb ) { return this . _handle ( req as JRPCRequest < unknown > , cb ) ; } return this . _promiseHandle ( req as JRPCRequest < unknown > ) ; } returns asMiddleware ( ) { return async ( req , res , next , end ) => { try { const [ middlewareError , isComplete , returnHandlers ] = await JRPCEngine . _runAllMiddleware ( req , res , this . _middleware ) ; if ( isComplete ) { await JRPCEngine . _runReturnHandlers ( returnHandlers ) ; return end ( middlewareError as Error ) ; } return next ( async ( handlerCallback ) => { try { await JRPCEngine . _runReturnHandlers ( returnHandlers ) ; } catch ( error ) { return handlerCallback ( error ) ; } return handlerCallback ( ) ; } ) ; } catch ( error ) { return end ( error ) ; } } ; } /**
     * Like _handle, but for batch requests.
     */ private _handleBatch ( reqs ) ; /**
     * Like _handle, but for batch requests.
     */ private _handleBatch ( reqs , cb ) ; private async _handleBatch ( reqs , cb ? ) { try { const responses = await Promise . all ( reqs . map ( this . _promiseHandle . bind ( this ) ) ) ; if ( cb ) { return cb ( null , responses ) ; } return responses ; } catch ( error ) { if ( cb ) { return cb ( error ) ; } throw error ; } } /**
     * A promise-wrapped _handle.
     */ private _promiseHandle ( req ) { return new Promise ( ( resolve ) => { this . _handle ( req , ( _err , res ) => { resolve ( res ) ; } ) ; } ) ; } /**
     * Ensures that the request object is valid, processes it, and passes any
     * error and the response object to the given callback.
     *
     * Does not reject.
     */ private async _handle ( callerReq , cb ) { if ( ! callerReq || Array . isArray ( callerReq ) || typeof callerReq !== "object" ) { const error = new SerializableError ( { code : - 32603 , message : "request must be plain object" } ) ; return cb ( error , { id : undefined , jsonrpc : "2.0" , error } ) ; } if ( typeof callerReq . method !== "string" ) { const error = new SerializableError ( { code : - 32603 , message : "method must be string" } ) ; return cb ( error , { id : callerReq . id , jsonrpc : "2.0" , error } ) ; } const req = { ... callerReq } ; const res = { id : req . id , jsonrpc : req . jsonrpc , } ; let error = null ; try { await this . _processRequest ( req , res ) ; } catch ( _error ) { error = _error ; } if ( error ) { delete res . result ; if ( ! res . error ) { res . error = serializeError ( error ) ; } } return cb ( error , res as JRPCResponse < unknown > ) ; } /**
     * For the given request and response, runs all middleware and their return
     * handlers, if any, and ensures that internal request processing semantics
     * are satisfied.
     */ private async _processRequest ( req , res ) { const [ error , isComplete , returnHandlers ] = await JRPCEngine . _runAllMiddleware ( req , res , this . _middleware ) ; JRPCEngine . _checkForCompletion ( req , res , isComplete ) ; await JRPCEngine . _runReturnHandlers ( returnHandlers ) ; if ( error ) { throw error ; } } } export function mergeMiddleware ( middlewareStack ) { const engine = new JRPCEngine ( ) ; middlewareStack . forEach ( ( middleware ) => engine . push ( middleware ) ) ; return engine . asMiddleware ( ) ; } export interface EngineStreamOptions { engine : JRPCEngine ; } export function createEngineStream ( opts ) { if ( ! opts || ! opts . engine ) { throw new Error ( "Missing engine parameter!" ) ; } const { engine } = opts ; let stream ; function read ( ) { return undefined ; } function write ( req , _encoding , cb ) { engine . handle ( req , ( _err , res ) => { stream . push ( res ) ; } ) ; cb ( ) ; } stream = new Duplex ( { objectMode : true , read , write } ) ; if ( engine . on ) { engine . on ( "notification" , ( message ) => { stream . push ( message ) ; } ) ; } return stream ; }