import EventEmitter from 'events' ; import async from 'async' ; import Debug from 'debug' ; import { AsyncAPIDocument , Server } from '@asyncapi/parser' ; import GleeAdapter from './adapter.js' ; import GleeRouter , { ChannelErrorMiddlewareTuple , ChannelMiddlewareTuple , GenericMiddleware } from './router.js' ; import GleeMessage from './message.js' ; import { matchChannel , duplicateMessage , getParams } from './util.js' ; import { GleeConfig } from './index.js' ; import GleeConnection from './connection.js' ; import { MiddlewareCallback } from '../middlewares/index.js' ; const debug = Debug ( 'glee' ) ; type AdapterRecord = { Adapter : typeof GleeAdapter ; instance ? : GleeAdapter ; serverName : string ; server : Server ; parsedAsyncAPI : AsyncAPIDocument ; } ; export default class Glee extends EventEmitter { private _options : GleeConfig ; private _router : GleeRouter ; private _adapters : AdapterRecord [ ] ; param { Object } options constructor ( options = { } ) { super ( ) ; this . _options = options ; this . _router = new GleeRouter ( ) ; this . _adapters = [ ] ; } get options ( ) : GleeConfig { return this . _options ; } param { GleeAdapter } adapter param { String } serverName param { AsyncAPIServer } server param { AsyncAPIDocument } parsedAsyncAPI addAdapter ( Adapter , { serverName , server , parsedAsyncAPI } ) { this . _adapters . push ( { Adapter , serverName , server , parsedAsyncAPI } ) ; } param { String } channel param { Function | GleeRouter } . . middlewares use ( ... middlewares ) ; use ( channel , ... middlewares ) ; use ( channel , ... middlewares ) { this . _router . use ( ... arguments ) ; } param { String } channel param { Function | GleeRouter } . . middlewares useOutbound ( ... middlewares ) ; useOutbound ( channel , ... middlewares ) ; useOutbound ( channel , ... middlewares ) { this . _router . useOutbound ( ... arguments ) ; } param { Object | GleeMessage } message send ( message ) { message . setOutbound ( ) ; this . _processMessage ( this . _router . getOutboundMiddlewares ( ) , this . _router . getOutboundErrorMiddlewares ( ) , message ) ; } /**
     * Tells the adapters to connect.
     */ async connect ( ) { const promises = [ ] ; this . _adapters . forEach ( a => { a . instance = new a . Adapter ( this , a . serverName , a . server , a . parsedAsyncAPI ) ; promises . push ( a . instance . connect ( ) ) ; } ) ; return Promise . all ( promises ) ; } /**
     * Alias for `connect`.
     */ async listen ( ) { return this . connect ( ) ; } param { GleeMessage } message param { String } serverName param { GleeConnection } connection injectMessage ( message , serverName , connection ) { message . serverName = serverName ; message . connection = connection ; message . setInbound ( ) ; this . _processMessage ( this . _router . getMiddlewares ( ) , this . _router . getErrorMiddlewares ( ) , message ) ; } param { Any } error param { String } channel injectError ( error , channel ? ) { this . _processError ( this . _router . getErrorMiddlewares ( ) , error , new GleeMessage ( { channel } ) ) ; } param { ChannelMiddlewareTuple } middlewares param { ChannelErrorMiddlewareTuple } errorMiddlewares param { GleeMessage } message private _processMessage ( middlewares , errorMiddlewares , message ) { const mws = middlewares . filter ( mw => matchChannel ( mw . channel , message . channel ) ) . map ( mw => ( msg , next ) => { const msgForMiddleware = duplicateMessage ( msg ) ; msgForMiddleware . params = getParams ( mw . channel , msgForMiddleware . channel ) ; msgForMiddleware . on ( 'send' , ( m ) => { m . setOutbound ( ) ; this . _processMessage ( this . _router . getOutboundMiddlewares ( ) , this . _router . getOutboundErrorMiddlewares ( ) , m ) ; } ) ; mw . fn . call ( mw . fn , msgForMiddleware , ( err , newMessage ) => { const nextMessage = newMessage || msgForMiddleware ; nextMessage . channel = message . channel ; next ( err , nextMessage ) ; } ) ; } ) ; async . seq ( ... mws ) ( message , ( err , msg ) => { if ( err ) { this . _processError ( errorMiddlewares , err , msg ) ; return ; } if ( middlewares === this . _router . getOutboundMiddlewares ( ) ) { debug ( 'Outbound pipeline finished. Sending message...' ) ; debug ( msg ) ; this . _adapters . forEach ( ( a ) => { if ( a . instance && ( ! msg . serverName || msg . serverName === a . serverName ) ) { a . instance . send ( msg ) . catch ( ( e ) => { this . _processError ( errorMiddlewares , e , msg ) ; } ) ; } } ) ; } else { debug ( 'Inbound pipeline finished.' ) ; } } ) ; } param { Array } errorMiddlewares param { Any } error param { GleeMessage } message private _processError ( errorMiddlewares , error , message ) { const emws = errorMiddlewares . filter ( emw => matchChannel ( emw . channel , message . channel ) ) ; if ( ! emws . length ) return ; this . _execErrorMiddleware ( emws , 0 , error , message ) ; } _execErrorMiddleware ( emws , index , error , message ) { emws . at ( index ) . fn ( error , message , ( err ) => { if ( ! emws [ index + 1 ] ) return ; this . _execErrorMiddleware . call ( null , emws , index + 1 , err , message ) ; } ) ; } }