import { Observable , EMPTY , pipe , of , from , Subscription , UnaryFunction } from 'rxjs' ; import { deserialize , KeepaliveMessage , MsgType , PubsubMessage , serialize } from './pubsub-message' ; import { IpfsApi } from '@ceramicnetwork/common' ; import { map , catchError , mergeMap , withLatestFrom } from 'rxjs/operators' ; import { IncomingChannel , filterExternal , IPFSPubsubMessage } from './incoming-channel' ; import { DiagnosticsLogger , ServiceLogger } from '@ceramicnetwork/common' ; import { TextDecoder } from 'util' ; import uint8ArrayToString from 'uint8arrays/to-string' ; const textDecoder = new TextDecoder ( 'utf-8' ) ; param peerId$ param pubsubLogger param topic function ipfsToPubsub ( peerId$ , pubsubLogger , topic ) { return pipe ( withLatestFrom ( peerId$ ) , mergeMap ( ( [ incoming , peerId ] ) => of ( incoming ) . pipe ( map ( ( incoming ) => { const message = deserialize ( incoming ) ; const serializedMessage = serialize ( message ) ; const logMessage = { ... incoming , ... JSON . parse ( textDecoder . decode ( serializedMessage ) ) } ; logMessage . seqno = uint8ArrayToString ( logMessage . seqno , 'base16' ) ; delete logMessage . data ; delete logMessage . key ; delete logMessage . signature ; pubsubLogger . log ( { peer : peerId , event : 'received' , topic : topic , message : logMessage } ) ; return message ; } ) , catchError ( ( ) => EMPTY ) ) ) ) ; } /**
 * Receive and publish messages to IPFS pubsub.
 */ export class Pubsub extends Observable < PubsubMessage > { private readonly peerId$ : Observable < string > ; constructor ( private readonly ipfs , private readonly topic , private readonly resubscribeEvery , private readonly pubsubLogger , private readonly logger ) { super ( ( subscriber ) => { const incoming$ = new IncomingChannel ( ipfs , topic , resubscribeEvery , pubsubLogger , logger ) ; incoming$ . pipe ( filterExternal ( this . peerId$ ) , ipfsToPubsub ( this . peerId$ , pubsubLogger , topic ) ) . subscribe ( subscriber ) ; } ) ; this . peerId$ = from < Promise < string > > ( this . ipfs . id ( ) . then ( ( _ ) => _ . id ) ) ; } /**
     * Publish message to IPFS Pubsub as a "fire and forget" command.
     *
     * You could use returned Subscription to react when the operation is finished.
     * Feel free to disregard it though.
     */ next ( message ) { return this . peerId$ . pipe ( mergeMap ( async ( peerId ) => { const serializedMessage = serialize ( message ) ; await this . ipfs . pubsub . publish ( this . topic , serializedMessage ) ; return { peerId , serializedMessage } ; } ) ) . subscribe ( { next : ( { peerId , serializedMessage } ) => { const logMessage = { ... message , ... JSON . parse ( textDecoder . decode ( serializedMessage ) ) } ; this . pubsubLogger . log ( { peer : peerId , event : 'published' , topic : this . topic , message : logMessage , } ) ; } , error : ( error ) => { this . logger . err ( error ) ; } , } ) ; } }