import { empty , from , Observable , Subscription } from 'rxjs' ; import { MsgType , PubsubMessage } from './pubsub-message' ; import { TaskQueue } from './task-queue' ; import { DiagnosticsLogger } from '@ceramicnetwork/common' ; import { ClockSource } from '../clock-source' ; import { ObservableWithNext } from './observable-with-next' ; /**
 * Wraps an instance of Pubsub and rate limits how often QUERY messages can be sent.  There are two
 * main configuration parameters: 'queriesPerSecond' and 'maxQueuedQueries'. 'queriesPerSecond'
 * controls how many QUERY pubsub messages can be published per second.  If more than that number
 * of query messages are attempted to be published, additional messages will queue up and will be
 * published when doing so will no longer put us over 'queriesPerSecond'.  'maxQueuedQueries'
 * controls how many queries are allowed to queue up before further attempts to publish query
 * messages just start failing outright.
 *
 * Note that other types of pubsub messages that are not QUERY messages are allowed to be published
 * without limit.
 */ export class PubsubRateLimit extends Observable < PubsubMessage > implements ObservableWithNext < PubsubMessage > { private private readonly _recentQueries : Date [ ] = [ ] ; private private readonly _queryQueue : TaskQueue ; private private readonly _clock : ClockSource = new ClockSource ( ) ; param pubsub param logger param queriesPerSecond constructor ( private readonly pubsub , private readonly logger , private readonly queriesPerSecond ) { super ( ( subscriber ) => { pubsub . subscribe ( subscriber ) ; } ) ; this . _queryQueue = new TaskQueue ( ( err ) => { this . logger . err ( `Error while publishing pubsub QUERY message: ${ err }` ) ; } ) ; this . logger . debug ( `Configuring pubsub to rate limit query messages to ${ queriesPerSecond } per second` ) ; } param message next ( message ) { const maxQueuedQueries = this . queriesPerSecond * 10 ; if ( message . typ === MsgType . QUERY ) { if ( this . _queryQueue . size >= maxQueuedQueries ) { this . logger . err ( `Cannot publish query message to pubsub because we have exceeded the maximum allowed rate. Cannot have more than ${ maxQueuedQueries } queued queries.` ) ; return empty ( ) . subscribe ( ) ; } return from ( this . _queryQueue . run ( this . _publishQuery . bind ( this , message ) ) ) . subscribe ( ) ; } else { return this . pubsub . next ( message ) ; } } param message private private async _publishQuery ( message ) { if ( this . _recentQueries . length >= this . queriesPerSecond ) { const now = this . _clock . now ( ) ; const oldestQuery = this . _recentQueries . shift ( ) ; const timeSinceOldestQuery = now . getTime ( ) - oldestQuery . getTime ( ) ; if ( timeSinceOldestQuery < 1000 ) { this . logger . warn ( `More than ${ this . queriesPerSecond } query messages published in less than a second. Query messages will be rate limited` ) ; await this . _clock . waitUntil ( new Date ( oldestQuery . getTime ( ) + 1000 ) ) ; } } this . _recentQueries . push ( this . _clock . now ( ) ) ; return this . pubsub . next ( message ) ; } }