import { API , DEFAULT_API_OPTIONS } from '../core/api' ; import { ApiOptions , MobileQueryEvent } from '../models' ; import { ReadableStream } from 'web-streams-polyfill/ponyfill' ; class EventSubscription { cancel : ( ) => void ; constructor ( cancel ) { this . cancel = cancel ; } } /**
 * Events is an API to stream updates from an ipfs pubsub
 */ export default class Events extends API { private pubsubQueryResultListeners : Array < { listener : ( queryId : string , message : string , messageId : string ) => void ; queryId : string ; } > ; private queryDoneListeners : Array < { listener : ( queryId : string ) => void ; queryId : string ; } > ; private queryErrorListeners : Array < { listener : ( queryId : string , error : string ) => void ; queryId : string ; } > ; constructor ( opts = DEFAULT_API_OPTIONS ) { super ( opts ) ; this . pubsubQueryResultListeners = [ ] ; this . queryDoneListeners = [ ] ; this . queryErrorListeners = [ ] ; } addPubsubQueryResultListener ( listener , queryId , queryHandle ) { this . pubsubQueryResultListeners . push ( { listener , queryId } ) ; const isEventSubscription = new EventSubscription ( ( ) => { this . pubsubQueryResultListeners = this . pubsubQueryResultListeners . filter ( ( item ) => item . queryId !== queryId ) ; this . queryDoneListeners = this . queryDoneListeners . filter ( ( item ) => item . queryId !== queryId ) ; this . queryErrorListeners = this . queryErrorListeners . filter ( ( item ) => item . queryId !== queryId ) ; if ( queryHandle instanceof EventSource ) { queryHandle . close ( ) ; } } ) ; const onPubsubQueryEvent = ( queryEvent ) => { const message = queryEvent . data . value . values [ 0 ] ; const messageId = queryEvent . data . id ; for ( const item of this . pubsubQueryResultListeners ) { if ( item . queryId === queryEvent . id ) { item . listener ( queryEvent . id , message , messageId ) ; } } } ; const onQueryError = ( e ) => { for ( const item of this . queryErrorListeners ) { if ( item . queryId === queryId ) { item . listener ( queryId , e . toString ( ) ) ; } } } ; const onQueryDone = ( ) => { for ( const item of this . queryDoneListeners ) { if ( item . queryId === queryId ) { item . listener ( queryId ) ; } } this . pubsubQueryResultListeners = this . pubsubQueryResultListeners . filter ( ( item ) => item . queryId !== queryId ) ; this . queryDoneListeners = this . queryDoneListeners . filter ( ( item ) => item . queryId !== queryId ) ; this . queryErrorListeners = this . queryErrorListeners . filter ( ( item ) => item . queryId !== queryId ) ; } ; if ( queryHandle instanceof EventSource ) { queryHandle . addEventListener ( 'update' , ( e ) => { try { const queryEvent = JSON . parse ( e . data ) ; onPubsubQueryEvent ( queryEvent ) ; } catch ( e ) { onQueryError ( e ) ; } } , false ) ; queryHandle . addEventListener ( 'open' , ( e ) => { } , false ) ; queryHandle . addEventListener ( 'error' , ( e ) => { if ( e . readyState === 2 ) { onQueryDone ( ) ; } else { onQueryError ( e ) ; } } , false ) ; return isEventSubscription ; } else { let isReader ; let cancellationRequest = false ; return new ReadableStream < ArrayBuffer > ( { start ( controller ) { const reader = queryHandle . getReader ( ) ; isReader = reader ; const decoder = new TextDecoder ( ) ; let dataBuffer = '' ; const processResult = ( result ) => { if ( result . done ) { if ( cancellationRequest ) { onQueryDone ( ) ; return ; } dataBuffer = dataBuffer . trim ( ) ; if ( dataBuffer . length !== 0 ) { try { const queryEvent = JSON . parse ( dataBuffer ) ; onPubsubQueryEvent ( queryEvent ) ; } catch ( e ) { controller . error ( e ) ; } } onQueryDone ( ) ; controller . close ( ) ; return ; } const data = decoder . decode ( result . value , { stream : true } ) ; dataBuffer += data ; const lines = dataBuffer . split ( '\n' ) ; for ( const line of lines ) { const l = line . trim ( ) ; if ( l . length > 0 ) { try { const queryEvent = JSON . parse ( l ) ; onPubsubQueryEvent ( queryEvent ) ; } catch ( e ) { onQueryError ( e ) ; controller . error ( e ) ; break ; } } } dataBuffer = lines . length > 1 ? lines [ lines . length - 1 ] : '' ; reader . read ( ) . then ( processResult ) ; } ; reader . read ( ) . then ( processResult ) ; } , cancel ( reason ? ) { isEventSubscription . cancel ( ) ; cancellationRequest = true ; isReader . cancel ( reason ) ; } } ) ; } } addQueryDoneListener ( listener , queryId ) { this . queryDoneListeners . push ( { listener , queryId } ) ; const isEventSubscription = new EventSubscription ( ( ) => ( this . queryDoneListeners = this . queryDoneListeners . filter ( ( item ) => item . queryId !== queryId ) ) ) ; return isEventSubscription ; } addQueryErrorListener ( listener , queryId ) { this . queryErrorListeners . push ( { listener , queryId } ) ; const isEventSubscription = new EventSubscription ( ( ) => ( this . queryErrorListeners = this . queryErrorListeners . filter ( ( item ) => item . queryId !== queryId ) ) ) ; return isEventSubscription ; } }