import { Observable } from 'rxjs' ; import { IpfsApi } from '@ceramicnetwork/common' ; import { TaskQueue } from './task-queue' ; import { DiagnosticsLogger , ServiceLogger } from '@ceramicnetwork/common' ; import { pipe , MonoTypeOperatorFunction } from 'rxjs' ; import { map , filter , concatMap } from 'rxjs/operators' ; export type IPFSPubsubMessage = { from : string ; seqno : Uint8Array ; data : Uint8Array ; topicIDs : string [ ] ; signature : Uint8Array ; key : Uint8Array ; } ; function buildResubscribeQueue ( logger ) { return new TaskQueue ( ( error , retry ) => { if ( error . message . includes ( 'Already subscribed' ) ) { logger . debug ( error . message ) ; } else if ( error . message . includes ( 'The user aborted a request' ) ) { retry ( ) ; } else { logger . err ( error ) ; } } ) ; } /**
 * Incoming IPFS PubSub message stream as Observable.
 * Ensures that ipfs subscription to a topic is live by periodically re-subscribing.
 *
 * One could simplify this by removing periodic re-subscription and serialization via task-queue.
 * Unfortunately, we can end up with concurrency issues: trying to unsubscribe while (or right after) other instance tries to subscribe.
 * So, better keep it all together.
 */ export class IncomingChannel extends Observable < IPFSPubsubMessage > { readonly tasks : TaskQueue ; constructor ( readonly ipfs , readonly topic , readonly resubscribeEvery , readonly pubsubLogger , readonly logger ) { super ( ( subscriber ) => { const handler = ( message ) => subscriber . next ( message ) ; const complete = ( ) => subscriber . complete ( ) ; this . tasks . add ( ( ) => this . subscribeToIpfs ( handler , complete ) ) ; return ( ) => { this . tasks . clear ( ) ; this . tasks . add ( async ( ) => { await this . ipfs . pubsub ?. unsubscribe ( this . topic , handler ) ; } ) ; } ; } ) ; this . tasks = buildResubscribeQueue ( logger ) ; } private async subscribeToIpfs ( handler , complete ) { const isRunning = this . ipfs && this . ipfs . pubsub ; if ( isRunning ) { const onError = ( error ) => { console . warn ( `IPFS pubsub subscription error, resubscribing: ${ error }` ) ; this . tasks . add ( ( ) => this . subscribeToIpfs ( handler , complete ) ) ; } ; try { const ipfsId = await this . ipfs . id ( ) ; const peerId = ipfsId . id ; await this . ipfs . pubsub . subscribe ( this . topic , handler , { onError } ) ; this . pubsubLogger . log ( { peer : peerId , event : 'subscribed' , topic : this . topic } ) ; } catch ( error ) { console . warn ( `Can not reach ipfs node, resubscribing: ${ error }` ) ; await new Promise ( ( resolve ) => setTimeout ( resolve , this . resubscribeEvery ) ) ; this . tasks . add ( ( ) => this . subscribeToIpfs ( handler , complete ) ) ; } } else { complete ( ) ; } } } param ownPeerId$ export function filterExternal ( ownPeerId$ ) { return pipe ( concatMap ( ( data ) => { return ownPeerId$ . pipe ( map ( ( peerId ) => ( { isOuter : data . from !== peerId , entry : data } ) ) ) ; } ) , filter ( ( data ) => data . isOuter ) , map ( ( data ) => data . entry ) ) ; }