import { Observable , Subscription , interval } from 'rxjs' ; import { KeepaliveMessage , MsgType , PubsubMessage } from './pubsub-message' ; import { Pubsub } from './pubsub' ; import { ObservableWithNext } from './observable-with-next' ; /**
 * Wraps an instance of Pubsub and ensures that a pubsub message is generated with some minimum
 * frequency.
 */ export class PubsubKeepalive extends Observable < PubsubMessage > implements ObservableWithNext < PubsubMessage > { private lastPublishedMessageDate : number = Date . now ( ) - this . maxPubsubPublishInterval ; param pubsub param maxPubsubPublishInterval constructor ( private readonly pubsub , private readonly maxPubsubPublishInterval ) { super ( ( subscriber ) => { pubsub . subscribe ( subscriber ) ; const pubsubKeepaliveInterval = interval ( this . maxPubsubPublishInterval / 2 ) . subscribe ( ( ) => { this . publishPubsubKeepaliveIfNeeded ( ) ; } ) ; return ( ) => { pubsubKeepaliveInterval . unsubscribe ( ) ; } ; } ) ; } /**
     * Passes on the message to be published by the underlying Pubsub instance, while keeping
     * track of the fact that we sent a message and thus can reset our idea of when the next keepalive
     * message needs to be sent.
     */ next ( message ) { this . lastPublishedMessageDate = Date . now ( ) ; return this . pubsub . next ( message ) ; } /**
     * Called periodically and ensures that if we haven't published a pubsub message in too long,
     * we'll publish one so that we never go longer than MAX_PUBSUB_PUBLISH_INTERVAL without
     * publishing a pubsub message.  This is to work around a bug in IPFS where peer connections
     * get dropped if they haven't had traffic in too long.
     */ publishPubsubKeepaliveIfNeeded ( ) { const now = Date . now ( ) ; if ( now - this . lastPublishedMessageDate < this . maxPubsubPublishInterval / 2 ) { return ; } const message = { typ : MsgType . KEEPALIVE , ts : now } ; this . next ( message ) ; } }