import * as utils from './utils/utils' ; import * as formatters from './utils/formatters' ; import { RequestManager } from './RequestManager' ; import * as config from './utils/config' ; import { FilterOptions , LogObject , TxHash , SHHFilterOptions , Data , SHHFilterMessage } from './Schema' ; import { future , IFuture } from 'fp-future' ; import { stringToUtf8Bytes } from './utils/utf8' ; function safeAsync ( fn ) { return function ( ) { fn ( ) . catch ( ( $ ) => console . error ( $ ) ) ; } ; } param value function toTopic ( value ) { if ( value === null || typeof value === 'undefined' ) return null ; const strValue = String ( value ) . toString ( ) ; if ( strValue . indexOf ( '0x' ) === 0 ) { return strValue ; } else { return utils . bytesToHex ( stringToUtf8Bytes ( strValue ) ) ; } } export type FilterCallback = ( messages : LogObject [ ] | string [ ] ) => void ; export abstract class AbstractFilter < ReceivedLog , TransformedLog = ReceivedLog > { public isStarted = false ; public isDisposed = false ; public formatter : ( x : ReceivedLog ) => TransformedLog = x => x as any ; protected filterId : IFuture < Data > = future ( ) ; protected callbacks : ( ( message : TransformedLog ) => void ) [ ] = [ ] ; protected stopSemaphore = future ( ) ; constructor ( public requestManager ) { } async watch ( callback ) { if ( this . isDisposed ) throw new Error ( 'The filter was disposed' ) ; if ( callback ) { this . callbacks . push ( callback ) ; if ( ! this . isStarted ) { await this . start ( ) ; } } } async start ( ) { if ( this . isDisposed ) throw new Error ( 'The filter was disposed' ) ; if ( this . isStarted ) return ; this . isStarted = true ; try { const id = await this . getNewFilter ( ) ; if ( ! id ) { throw new Error ( 'Could not create a filter, response: ' + JSON . stringify ( id ) ) ; } this . filterId . resolve ( id ) ; } catch ( e ) { throw e ; } this . stopSemaphore = future ( ) ; await this . poll ( ) ; } async stop ( ) { if ( ! this . isStarted ) return ; if ( this . isDisposed ) return ; this . isDisposed = true ; const filterId = await this . filterId ; this . isStarted = false ; if ( this . stopSemaphore ) await this . stopSemaphore ; const didStop = await this . uninstall ( ) ; if ( didStop !== true ) { throw new Error ( `Couldn't stop the eth filter: ${ filterId }` ) ; } } protected abstract getNewFilter ( ) ; protected abstract getChanges ( ) ; protected abstract uninstall ( ) ; /**
     * Adds the callback and sets up the methods, to iterate over the results.
     */ private async poll ( ) { if ( this . isStarted ) { if ( this . callbacks . length ) { const result = await this . getChanges ( ) ; this . callbacks . forEach ( ( cb ) => { if ( this . formatter ) { result . forEach ( ( $ ) => { cb ( this . formatter ! ( $ ) ) ; } ) ; } else { result . forEach ( ( $ ) => cb ( $ as any ) ) ; } } ) ; } this . stopSemaphore . resolve ( 1 ) ; if ( this . isStarted ) { this . stopSemaphore = future ( ) ; setTimeout ( safeAsync ( ( ) => this . poll ( ) ) , config . ETH_POLLING_TIMEOUT ) ; } } else { this . stopSemaphore . resolve ( 1 ) ; } } } export class SHHFilter extends AbstractFilter < SHHFilterMessage > { constructor ( public requestManager , public options ) { super ( requestManager ) ; this . options = this . options || { topics : [ ] } ; this . options . topics = this . options . topics || [ ] ; this . options . topics = this . options . topics . map ( function ( topic ) { return toTopic ( topic ) ; } ) ; this . options = { topics : this . options . topics , to : this . options . to } ; } async getMessages ( ) { const filterId = await this . filterId ; return this . requestManager . shh_getMessages ( filterId ) ; } protected async getNewFilter ( ) { return this . requestManager . shh_newFilter ( this . options ) ; } protected async getChanges ( ) { const filterId = await this . filterId ; return this . requestManager . shh_getFilterChanges ( filterId ) ; } protected async uninstall ( ) { const filterId = await this . filterId ; return this . requestManager . shh_uninstallFilter ( filterId ) ; } } export class EthFilter < TransformedLog = LogObject , ReceivedLog = LogObject > extends AbstractFilter < ReceivedLog , TransformedLog > { constructor ( public requestManager , public options , public formatter = ( x ) => x as any ) { super ( requestManager ) ; this . options = this . options || { } ; this . options . topics = this . options . topics || [ ] ; this . options . topics = this . options . topics . map ( function ( topic ) { return toTopic ( topic ) ; } ) ; this . options = { topics : this . options . topics , address : this . options . address ? this . options . address : undefined , fromBlock : typeof this . options . fromBlock === 'number' || typeof this . options . fromBlock === 'string' ? formatters . inputBlockNumberFormatter ( this . options . fromBlock ) || undefined : 'latest' , toBlock : typeof this . options . toBlock === 'number' || typeof this . options . toBlock === 'string' ? formatters . inputBlockNumberFormatter ( this . options . toBlock ) || undefined : 'latest' } ; } async getLogs ( ) { if ( ! this . isStarted ) { await this . start ( ) ; } const filterId = await this . filterId ; return this . requestManager . eth_getFilterLogs ( filterId ) as any ; } protected async getNewFilter ( ) { return this . requestManager . eth_newFilter ( this . options ) ; } protected async getChanges ( ) { const filterId = await this . filterId ; return this . requestManager . eth_getFilterChanges ( filterId ) as any ; } protected async uninstall ( ) { const filterId = await this . filterId ; return this . requestManager . eth_uninstallFilter ( filterId ) ; } } export class EthPendingTransactionFilter extends EthFilter < TxHash , TxHash > { constructor ( requestManager ) { super ( requestManager , { } ) ; } async getNewFilter ( ) { return this . requestManager . eth_newPendingTransactionFilter ( ) ; } } export class EthBlockFilter extends EthFilter < TxHash , TxHash > { constructor ( requestManager ) { super ( requestManager , { } ) ; } async getNewFilter ( ) { return this . requestManager . eth_newBlockFilter ( ) ; } }