import autobind from 'autobind-decorator' ; import uniqBy from 'lodash.uniqby' ; import { getLogger } from './logging' ; import FifoDispatcher from './dispatchers/fifo' ; import { DispatchListener , IDCRFOptions , IDispatcher , IMessageEvent , ISendQueue , ISerializer , IStreamingAPI , ITransport , StreamingRequestHandler , SubscribeOptions , SubscriptionHandler , } from './interface' ; import UUID from './lib/UUID' ; import FifoQueue from './send_queues/fifo' ; import JSONSerializer from './serializers/json' ; import { SubscriptionPromise } from './subscriptions' ; import WebsocketTransport from './transports/websocket' ; const log = getLogger ( 'dcrf' ) ; interface ISubscriptionDescriptor < S , P extends S > { selector : S ; handler : DispatchListener < P > ; subscribeMessage : object ; unsubscribeMessage : object ; } /**
 * Promise representing the listening for responses during a streaming request, and offering an
 * cancel() method to stop listening for additional responses.
 *
 * This is returned from DCRFClient.streamingRequest.
 */ export class StreamingRequestPromise < T > extends Promise < T > { protected _dispatcher : IDispatcher ; protected _listenerId : number | null ; constructor ( executor , dispatcher , listenerId ) { super ( executor ) ; this . _dispatcher = dispatcher ; this . _listenerId = listenerId ; } public get listenerId ( ) { return this . _listenerId ; } return public async cancel ( ) { if ( this . _listenerId !== null ) { const returnValue = this . _dispatcher . cancel ( this . _listenerId ) ; this . _listenerId = null ; return returnValue ; } return false ; } } export class DCRFClient implements IStreamingAPI { public readonly dispatcher : IDispatcher ; public readonly transport : ITransport ; public readonly queue : ISendQueue ; public readonly serializer : ISerializer ; public readonly options : IDCRFOptions ; public readonly pkField : string ; public readonly ensurePkFieldInDeleteEvents : boolean ; public readonly subscriptions : { [ listenerId ] : ISubscriptionDescriptor < any , any > ; } ; param dispatcher param transport param queue param serializer param options constructor ( dispatcher , transport , queue , serializer , options = { } ) { this . dispatcher = dispatcher ; this . transport = transport ; this . queue = queue ; this . serializer = serializer ; this . options = options ; this . pkField = options . pkField ?? 'pk' ; this . ensurePkFieldInDeleteEvents = options . ensurePkFieldInDeleteEvents ?? true ; if ( this . options . buildMultiplexedMessage ) this . buildMultiplexedMessage = this . options . buildMultiplexedMessage . bind ( this ) ; if ( this . options . buildRequestResponseSelector ) this . buildRequestResponseSelector = this . options . buildRequestResponseSelector . bind ( this ) ; if ( this . options . buildSubscribeCreateSelector ) this . buildSubscribeCreateSelector = this . options . buildSubscribeCreateSelector . bind ( this ) ; if ( this . options . buildSubscribeUpdateSelector ) this . buildSubscribeUpdateSelector = this . options . buildSubscribeUpdateSelector . bind ( this ) ; if ( this . options . buildSubscribeDeleteSelector ) this . buildSubscribeDeleteSelector = this . options . buildSubscribeDeleteSelector . bind ( this ) ; if ( this . options . buildSubscribePayload ) this . buildSubscribePayload = this . options . buildSubscribePayload . bind ( this ) ; if ( this . options . buildUnsubscribePayload ) this . buildUnsubscribePayload = this . options . buildUnsubscribePayload . bind ( this ) ; this . queue . initialize ( this . transport . send , this . transport . isConnected ) ; this . subscriptions = { } ; } public initialize ( ) { this . transport . connect ( ) ; this . transport . on ( 'message' , this . handleTransportMessage ) ; this . transport . on ( 'connect' , this . handleTransportConnect ) ; this . transport . on ( 'reconnect' , this . handleTransportReconnect ) ; } public close ( unsubscribe = true ) { let promise ; if ( unsubscribe ) { promise = this . unsubscribeAll ( ) ; } else { promise = Promise . resolve ( ) ; } return promise . then ( ( ) => { this . transport . disconnect ( ) ; } ) ; } public list ( stream , data = { } , requestId ? ) { return this . request ( stream , { action : 'list' , data , } , requestId ) ; } public create ( stream , props , requestId ? ) { return this . request ( stream , { action : 'create' , data : props , } , requestId ) ; } public retrieve ( stream , pk , data = { } , requestId ? ) { return this . request ( stream , { action : 'retrieve' , pk , data , } , requestId ) ; } public update ( stream , pk , props , requestId ? ) { return this . request ( stream , { action : 'update' , pk , data : props , } , requestId ) ; } public patch ( stream , pk , props , requestId ? ) { return this . request ( stream , { action : 'patch' , pk , data : props , } , requestId ) ; } public delete ( stream , pk , data = { } , requestId ? ) { return this . request ( stream , { action : 'delete' , pk , data , } , requestId ) ; } public subscribe ( stream , pk , callback , options ? ) ; public subscribe ( stream , pk , callback , requestId ? ) ; public subscribe ( stream , args , callback , options ? ) ; public subscribe ( stream , args , callback , requestId ? ) ; public subscribe ( stream , args , callback , options ? ) { if ( callback == null ) { throw new Error ( 'callback must be provided' ) ; } if ( typeof options === 'string' ) { options = { requestId : options } ; } options = options ?? { } ; options . includeCreateEvents = options . includeCreateEvents ?? false ; options . requestId = options . requestId ?? UUID . generate ( ) ; options . subscribeAction = options . subscribeAction ?? 'subscribe_instance' ; options . unsubscribeAction = options . unsubscribeAction ?? 'unsubscribe_instance' ; if ( args !== null && typeof args !== 'object' ) { const pk = args ; args = { [ options . subscribeAction === 'subscribe_instance' ? 'pk' : this . pkField ] : pk } ; } const requestId = options . requestId ; const createSelector = this . buildSubscribeCreateSelector ( stream , requestId ) ; const updateSelector = this . buildSubscribeUpdateSelector ( stream , requestId ) ; const deleteSelector = this . buildSubscribeDeleteSelector ( stream , requestId ) ; const handler = this . buildSubscribeListener ( callback ) ; const subscribePayload = this . buildSubscribePayload ( options . subscribeAction , args , requestId ) ; const unsubscribePayload = this . buildUnsubscribePayload ( options . unsubscribeAction , args , requestId ) ; const subscribeMessage = this . buildMultiplexedMessage ( stream , subscribePayload ) ; const unsubscribeMessage = this . buildMultiplexedMessage ( stream , unsubscribePayload ) ; const listenerIds = [ ] ; const addListener = ( selector ) => { const listenerId = this . dispatcher . listen ( selector , handler ) ; listenerIds . push ( listenerId ) ; this . subscriptions [ listenerId ] = { selector , handler , subscribeMessage , unsubscribeMessage } ; } ; addListener ( updateSelector ) ; addListener ( deleteSelector ) ; if ( options . includeCreateEvents ) { addListener ( createSelector ) ; } const requestPromise = this . request ( stream , subscribePayload , requestId ) ; const unsubscribe = async ( ) => { const wasSubbed = listenerIds . map ( this . unsubscribe ) . some ( Boolean ) ; await this . request ( stream , unsubscribePayload , requestId ) ; return wasSubbed ; } ; return new SubscriptionPromise ( ( resolve , reject ) => { requestPromise . then ( resolve , reject ) ; } , unsubscribe ) ; } public unsubscribeAll ( ) { const subscriptions = Object . values ( this . subscriptions ) ; const unsubscribeMessages = uniqBy ( subscriptions , s => { return s . unsubscribeMessage ?. payload ?. request_id ; } ) ; const listenerIds = Object . keys ( this . subscriptions ) . map ( parseInt ) ; log . info ( 'Removing %d listeners' , listenerIds . length ) ; listenerIds . forEach ( listenerId => this . _unsubscribeUnsafe ( listenerId ) ) ; log . info ( 'Sending %d unsubscription requests' , unsubscribeMessages . length ) ; const unsubscriptionPromises = [ ] ; for ( const { unsubscribeMessage } of unsubscribeMessages ) { const { stream , payload : { request_id , ... payload } } = unsubscribeMessage ; unsubscriptionPromises . push ( this . request ( stream , payload , request_id ) . catch ( ( ) => { } ) ) ; } return Promise . all ( unsubscriptionPromises ) . then ( ( ) => listenerIds . length ) ; } /**
     * Send subscription requests for all registered subscriptions
     */ public resubscribe ( ) { const subscriptions = Object . values ( this . subscriptions ) ; const resubscribeMessages = uniqBy ( subscriptions , s => { return s . subscribeMessage ?. payload ?. request_id ; } ) ; log . info ( 'Resending %d subscription requests' , subscriptions . length ) ; for ( const { subscribeMessage } of resubscribeMessages ) { this . sendNow ( subscribeMessage ) ; } } private sendRequest ( payload , requestId , stream ) { payload = Object . assign ( { } , payload , { request_id : requestId } ) ; if ( this . options . preprocessPayload != null ) { payload = this . options . preprocessPayload ( stream , payload , requestId ) || payload ; } let message = this . buildMultiplexedMessage ( stream , payload ) ; if ( this . options . preprocessMessage != null ) { message = this . options . preprocessMessage ( message ) || message ; } this . send ( message ) ; } public request ( stream , payload , requestId = UUID . generate ( ) ) { return new Promise ( ( resolve , reject ) => { const selector = this . buildRequestResponseSelector ( stream , requestId ) ; this . dispatcher . once ( selector , ( data ) => { const { payload : response } = data ; const responseStatus = response . response_status ; if ( Math . floor ( responseStatus / 100 ) === 2 ) { resolve ( response . data ) ; } else { reject ( response ) ; } } ) ; this . sendRequest ( payload , requestId , stream ) ; } ) ; } public streamingRequest ( stream , payload , callback , requestId = UUID . generate ( ) ) { const selector = this . buildRequestResponseSelector ( stream , requestId ) ; let cancelable ; let listenerId = this . dispatcher . listen ( selector , ( data ) => { const { payload : response } = data ; const responseStatus = response . response_status ; if ( ! cancelable . listenerId ) { return ; } if ( Math . floor ( responseStatus / 100 ) === 2 ) { callback ( null , response . data ) ; } else { cancelable . cancel ( ) . finally ( ( ) => { callback ( response , null ) ; } ) ; } } ) ; cancelable = new StreamingRequestPromise ( ( resolve , reject ) => { try { this . sendRequest ( payload , requestId , stream ) ; resolve ( ) ; } catch ( e ) { reject ( e ) ; } } , this . dispatcher , listenerId ) ; return cancelable ; } public send ( object ) { return this . doSend ( object , this . queue . send ) ; } public sendNow ( object ) { return this . doSend ( object , this . queue . sendNow ) ; } protected doSend ( object , send ) { const bytes = this . serializer . serialize ( object ) ; return send ( bytes ) ; } @ autobind protected unsubscribe ( listenerId ) { if ( this . subscriptions . hasOwnProperty ( listenerId ) ) { this . _unsubscribeUnsafe ( listenerId ) ; return true ; } else { return false ; } } protected _unsubscribeUnsafe ( listenerId ) { this . dispatcher . cancel ( listenerId ) ; delete this . subscriptions [ listenerId ] ; } @ autobind protected handleTransportMessage ( event ) { log . debug ( 'Received message over transport: %s' , event . data ) ; const data = this . serializer . deserialize ( event . data ) ; return this . handleMessage ( data ) ; } protected handleMessage ( data ) { this . dispatcher . dispatch ( data ) ; } @ autobind protected handleTransportConnect ( ) { log . debug ( 'Initial API connection over transport %s' , this . transport . constructor . name ) ; this . queue . processQueue ( ) ; } @ autobind protected handleTransportReconnect ( ) { log . debug ( 'Reestablished API connection' ) ; this . resubscribe ( ) ; this . queue . processQueue ( ) ; } public buildMultiplexedMessage ( stream , payload ) { return { stream , payload } ; } public buildRequestResponseSelector ( stream , requestId ) { return { stream , payload : { request_id : requestId } , } ; } public buildSubscribeCreateSelector ( stream , requestId ) { return { stream , payload : { action : 'create' , request_id : requestId , } , } ; } public buildSubscribeUpdateSelector ( stream , requestId ) { return { stream , payload : { action : 'update' , request_id : requestId , } , } ; } public buildSubscribeDeleteSelector ( stream , requestId ) { return { stream , payload : { action : 'delete' , request_id : requestId , } , } ; } public buildSubscribePayload ( action , args , requestId ) { return { action , request_id : requestId , ... args , } ; } public buildUnsubscribePayload ( action , args , requestId ) { return { action , request_id : requestId , ... args , } ; } /**
     * Build a function which will take an entire JSON message and return only
     * the relevant payload (usually an object).
     */ protected buildListener ( callback ) { return ( data ) => { return callback ( data . payload . data , data ) ; } ; } /**
     * Build a function which will take an entire JSON message and return only
     * the relevant payload (usually an object).
     */ protected buildSubscribeListener ( callback ) { return this . buildListener ( ( data , response ) => { const action = response . payload . action ; if ( action === 'delete' && this . ensurePkFieldInDeleteEvents && ! data . hasOwnProperty ( this . pkField ) ) { data [ this . pkField ] = data . pk ; delete data . pk ; } return callback ( data , action ) ; } ) ; } } export default { param url param options connect ( url , options = { } ) { const client = this . createClient ( url , options ) ; client . initialize ( ) ; return client ; } , param url param options createClient ( url , options = { } ) { const dispatcher = options . dispatcher || new FifoDispatcher ( ) ; const transport = options . transport || new WebsocketTransport ( url , options . websocket ) ; const queue = options . queue || new FifoQueue ( ) ; const serializer = options . serializer || new JSONSerializer ( ) ; return new DCRFClient ( dispatcher , transport , queue , serializer , options ) ; } , } ;