import { Observable , Observer , ObserverComplete , ObserverError , ObserverNext , Subscription , symbolObservable , } from '../observable' ; import { CommunicationObserver } from '../results/CommunicationObserver' ; import { Cancellable } from '../results/Cancellable' ; /** APIExecutor executes the API and passes its response to the supplied consumer */ export type APIExecutor = ( consumer : CommunicationObserver < Uint8Array > ) => void ; type Decorator < T > = ( observer : Observer < T > ) => Observer < Uint8Array > ; class QuerySubscription implements Subscription { private cancellable ? : Cancellable ; private isClosed = false ; public constructor ( observer , executor ) { try { executor ( { next : value => { observer . next ( value ) ; } , error : e => { this . isClosed = true ; observer . error ( e ) ; } , complete : ( ) => { this . isClosed = true ; observer . complete ( ) ; } , useCancellable : c => { this . cancellable = c ; } , } ) ; } catch ( e ) { this . isClosed = true ; observer . error ( e ) ; } } public get closed ( ) : boolean { return this . isClosed ; } public unsubscribe ( ) { this . cancellable ?. cancel ( ) ; this . isClosed = true ; } } function noop ( ) { } function completeObserver < T > ( observer ) { const { next , error , complete } = observer ; return { next : next ? next . bind ( observer ) : noop , error : error ? error . bind ( observer ) : noop , complete : complete ? complete . bind ( observer ) : noop , } ; } export default class ObservableQuery < T > implements Observable < T > { public constructor ( private readonly executor , private readonly decorator ) { } public subscribe ( observerOrNext ? , error ? , complete ? ) { const observer = completeObserver ( typeof observerOrNext !== 'object' || observerOrNext === null ? { next : observerOrNext , error , complete } : observerOrNext ) ; return new QuerySubscription ( this . decorator ( observer ) , this . executor ) ; } public [ symbolObservable ] ( ) { return this ; } public declare [ Symbol . observable ] : ( ) => this ; }