import { EventEmitter } from 'events' ; import { getErrorFromUnknown } from './internals/errors' ; interface SubscriptionEvents < TOutput > { data : ( data : TOutput ) => void ; destroy : ( ) => void ; error : ( error : Error ) => void ; } declare interface SubscriptionEventEmitter < TOutput > { on < U extends keyof SubscriptionEvents < TOutput > > ( event , listener ) : this ; once < U extends keyof SubscriptionEvents < TOutput > > ( event , listener ) : this ; emit < U extends keyof SubscriptionEvents < TOutput > > ( event , ... args ) : boolean ; } class SubscriptionEventEmitter < TOutput > extends EventEmitter { } type UnsubscribeFn = ( ) => void ; type EmitFn < TOutput > = ( data : TOutput ) => void ; export type SubscriptionEmit < TOutput > = { data : EmitFn < TOutput > ; error : EmitFn < Error > ; } ; export type SubscriptionCallback < TOutput > = ( emit : SubscriptionEmit < TOutput > ) => UnsubscribeFn | Promise < UnsubscribeFn > ; beta export class Subscription < TOutput = unknown > { private readonly events : SubscriptionEventEmitter < TOutput > ; private callback ; private isDestroyed : boolean ; constructor ( callback ) { this . isDestroyed = false ; this . events = new SubscriptionEventEmitter < TOutput > ( ) ; this . callback = callback ; } public destroy ( ) { if ( this . isDestroyed ) { return ; } this . isDestroyed = true ; this . events . emit ( 'destroy' ) ; this . events . removeAllListeners ( ) ; } public async start ( ) { if ( this . isDestroyed ) { throw new Error ( 'Called start() on a destroyed subscription' ) ; } try { const emit = { error : ( err ) => this . emitError ( err ) , data : ( data ) => this . emitOutput ( data ) } ; const cancel = await this . callback ( emit ) ; if ( this . isDestroyed ) { cancel ( ) ; } else { this . events . on ( 'destroy' , cancel ) ; } } catch ( cause ) { this . emitError ( getErrorFromUnknown ( cause ) ) ; } } deprecated public output ( ) { throw new Error ( 'Not in use' ) ; } /**
     * Emit data
     */ emitOutput ( data ) { this . events . emit ( 'data' , data ) ; } /**
     * Emit error
     */ emitError ( err ) { this . events . emit ( 'error' , err ) ; } on ( ... args ) { return this . events . on ( ... args ) ; } off ( ... args ) { return this . events . off ( ... args ) ; } } alpha export function subscriptionPullFactory < TOutput > ( opts ) { let timer ; let stopped = false ; async function _pull ( emit ) { if ( stopped ) { return ; } try { await opts . pull ( emit ) ; } catch ( err ) { emit . error ( getErrorFromUnknown ( err ) ) ; } if ( ! stopped ) { timer = setTimeout ( ( ) => _pull ( emit ) , opts . intervalMs ) ; } } return new Subscription < TOutput > ( ( emit ) => { _pull ( emit ) ; return ( ) => { clearTimeout ( timer ) ; stopped = true ; } ; } ) ; }