pub fn throttle_while_syncing(
        self,
        logger: &Logger,
        store: Arc<impl Store>,
        deployment: SubgraphDeploymentId,
        interval: Duration,
    ) -> StoreEventStreamBox {
        // We refresh the synced flag every SYNC_REFRESH_FREQ*interval to
        // avoid hitting the database too often to see if the subgraph has
        // been synced in the meantime. The only downside of this approach is
        // that we might continue throttling subscription updates for a little
        // bit longer than we really should
        static SYNC_REFRESH_FREQ: u32 = 4;

        // Check whether a deployment is marked as synced in the store. The
        // special 'subgraphs' subgraph is never considered synced so that
        // we always throttle it
        let check_synced = |store: &Store, deployment: &SubgraphDeploymentId| {
            deployment != &*SUBGRAPHS_ID
                && store
                    .is_deployment_synced(deployment.clone())
                    .unwrap_or(false)
        };
        let mut synced = check_synced(&*store, &deployment);
        let synced_check_interval = interval.checked_mul(SYNC_REFRESH_FREQ).unwrap();
        let mut synced_last_refreshed = Instant::now();

        let mut pending_event: Option<StoreEvent> = None;
        let mut source = self.source.fuse();
        let mut had_err = false;
        let mut delay = tokio_timer::Delay::new(Instant::now() + interval);
        let logger = logger.clone();

        let source = Box::new(poll_fn(move || -> Poll<Option<StoreEvent>, ()> {
            if had_err {
                // We had an error the last time through, but returned the pending
                // event first. Indicate the error now
                had_err = false;
                return Err(());
            }

            if !synced && synced_last_refreshed.elapsed() > synced_check_interval {
                synced = check_synced(&*store, &deployment);
                synced_last_refreshed = Instant::now();
            }

            if synced {
                return source.poll();
            }

            // Check if interval has passed since the last time we sent something.
            // If it has, start a new delay timer
            let should_send = match delay.poll() {
                Ok(Async::NotReady) => false,
                // Timer errors are harmless. Treat them as if the timer had
                // become ready.
                Ok(Async::Ready(())) | Err(_) => {
                    delay = tokio_timer::Delay::new(Instant::now() + interval);
                    true
                }
            };

            // Get as many events as we can off of the source stream
            loop {
                match source.poll() {
                    Ok(Async::NotReady) => {
                        if should_send && pending_event.is_some() {
                            return Ok(Async::Ready(pending_event.take()));
                        } else {
                            return Ok(Async::NotReady);
                        }
                    }
                    Ok(Async::Ready(None)) => {
                        return Ok(Async::Ready(pending_event.take()));
                    }
                    Ok(Async::Ready(Some(event))) => {
                        StoreEvent::accumulate(&logger, &mut pending_event, event);
                    }
                    Err(()) => {
                        // Before we report the error, deliver what we have accumulated so far.
                        // We will report the error the next time poll() is called
                        if pending_event.is_some() {
                            had_err = true;
                            return Ok(Async::Ready(pending_event.take()));
                        } else {
                            return Err(());
                        }
                    }
                };
            }
        }));
        StoreEventStream::new(source)
    }
/// Reduce the frequency with which events are generated while a
/// subgraph deployment is syncing. While the given `deployment` is not
/// synced yet, events from `source` are reported at most every
/// `interval`. At the same time, no event is held for longer than
/// `interval`. The `StoreEvents` that arrive during an interval appear
/// on the returned stream as a single `StoreEvent`; the events are
/// combined by using the maximum of all sources and the concatenation
/// of the changes of the `StoreEvents` received during the interval.
Reduce the frequency with which events are generated while a
subgraph deployment is syncing.
Reduce the frequency with which events are generated while a subgraph deployment is syncing .
pub fn throttle_while_syncing(
        self,
        logger: &Logger,
        store: Arc<impl Store>,
        deployment: SubgraphDeploymentId,
        interval: Duration,
    ) -> StoreEventStreamBox {
        // We refresh the synced flag every SYNC_REFRESH_FREQ*interval to
        // avoid hitting the database too often to see if the subgraph has
        // been synced in the meantime. The only downside of this approach is
        // that we might continue throttling subscription updates for a little
        // bit longer than we really should
        static SYNC_REFRESH_FREQ: u32 = 4;
        // Check whether a deployment is marked as synced in the store. The
        // special 'subgraphs' subgraph is never considered synced so that
        // we always throttle it
        let check_synced = |store: &Store, deployment: &SubgraphDeploymentId| {
            deployment != &*SUBGRAPHS_ID
                && store
                    .is_deployment_synced(deployment.clone())
                    .unwrap_or(false)
        };
        let mut synced = check_synced(&*store, &deployment);
        let synced_check_interval = interval.checked_mul(SYNC_REFRESH_FREQ).unwrap();
        let mut synced_last_refreshed = Instant::now();
        let mut pending_event: Option<StoreEvent> = None;
        let mut source = self.source.fuse();
        let mut had_err = false;
        let mut delay = tokio_timer::Delay::new(Instant::now() + interval);
        let logger = logger.clone();
        let source = Box::new(poll_fn(move || -> Poll<Option<StoreEvent>, ()> {
            if had_err {
                // We had an error the last time through, but returned the pending
                // event first. Indicate the error now
                had_err = false;
                return Err(());
            }
            if !synced && synced_last_refreshed.elapsed() > synced_check_interval {
                synced = check_synced(&*store, &deployment);
                synced_last_refreshed = Instant::now();
            }
            if synced {
                return source.poll();
            }
            // Check if interval has passed since the last time we sent something.
            // If it has, start a new delay timer
            let should_send = match delay.poll() {
                Ok(Async::NotReady) => false,
                // Timer errors are harmless. Treat them as if the timer had
                // become ready.
                Ok(Async::Ready(())) | Err(_) => {
                    delay = tokio_timer::Delay::new(Instant::now() + interval);
                    true
                }
            };
            // Get as many events as we can off of the source stream
            loop {
                match source.poll() {
                    Ok(Async::NotReady) => {
                        if should_send && pending_event.is_some() {
                            return Ok(Async::Ready(pending_event.take()));
                        } else {
                            return Ok(Async::NotReady);
                        }
                    }
                    Ok(Async::Ready(None)) => {
                        return Ok(Async::Ready(pending_event.take()));
                    }
                    Ok(Async::Ready(Some(event))) => {
                        StoreEvent::accumulate(&logger, &mut pending_event, event);
                    }
                    Err(()) => {
                        // Before we report the error, deliver what we have accumulated so far.
                        // We will report the error the next time poll() is called
                        if pending_event.is_some() {
                            had_err = true;
                            return Ok(Async::Ready(pending_event.take()));
                        } else {
                            return Err(());
                        }
                    }
                };
            }
        }));
        StoreEventStream::new(source)
    }
pub fn throttle_while_syncing ( self , logger : & Logger , store : Arc < impl Store > , deployment : SubgraphDeploymentId , interval : Duration , ) -> StoreEventStreamBox { // We refresh the synced flag every SYNC_REFRESH_FREQ*interval to // avoid hitting the database too often to see if the subgraph has // been synced in the meantime. The only downside of this approach is // that we might continue throttling subscription updates for a little // bit longer than we really should static SYNC_REFRESH_FREQ : u32 = 4 ; // Check whether a deployment is marked as synced in the store. The // special 'subgraphs' subgraph is never considered synced so that // we always throttle it let check_synced = | store : & Store , deployment : & SubgraphDeploymentId | { deployment != & * SUBGRAPHS_ID && store . is_deployment_synced ( deployment . clone ( ) ) . unwrap_or ( false ) } ; let mut synced = check_synced ( & * store , & deployment ) ; let synced_check_interval = interval . checked_mul ( SYNC_REFRESH_FREQ ) . unwrap ( ) ; let mut synced_last_refreshed = Instant :: now ( ) ; let mut pending_event : Option < StoreEvent > = None ; let mut source = self . source . fuse ( ) ; let mut had_err = false ; let mut delay = tokio_timer :: Delay :: new ( Instant :: now ( ) + interval ) ; let logger = logger . clone ( ) ; let source = Box :: new ( poll_fn ( move | | -> Poll < Option < StoreEvent > , ( ) > { if had_err { // We had an error the last time through, but returned the pending // event first. Indicate the error now had_err = false ; return Err ( ( ) ) ; } if ! synced && synced_last_refreshed . elapsed ( ) > synced_check_interval { synced = check_synced ( & * store , & deployment ) ; synced_last_refreshed = Instant :: now ( ) ; } if synced { return source . poll ( ) ; } // Check if interval has passed since the last time we sent something. // If it has, start a new delay timer let should_send = match delay . poll ( ) { Ok ( Async :: NotReady ) => false , // Timer errors are harmless. Treat them as if the timer had // become ready. Ok ( Async :: Ready ( ( ) ) ) | Err ( _ ) => { delay = tokio_timer :: Delay :: new ( Instant :: now ( ) + interval ) ; true } } ; // Get as many events as we can off of the source stream loop { match source . poll ( ) { Ok ( Async :: NotReady ) => { if should_send && pending_event . is_some ( ) { return Ok ( Async :: Ready ( pending_event . take ( ) ) ) ; } else { return Ok ( Async :: NotReady ) ; } } Ok ( Async :: Ready ( None ) ) => { return Ok ( Async :: Ready ( pending_event . take ( ) ) ) ; } Ok ( Async :: Ready ( Some ( event ) ) ) => { StoreEvent :: accumulate ( & logger , & mut pending_event , event ) ; } Err ( ( ) ) => { // Before we report the error, deliver what we have accumulated so far. // We will report the error the next time poll() is called if pending_event . is_some ( ) { had_err = true ; return Ok ( Async :: Ready ( pending_event . take ( ) ) ) ; } else { return Err ( ( ) ) ; } } } ; } } ) ) ; StoreEventStream :: new ( source ) }
Reduce the frequency with which events are generated while a
subgraph deployment is syncing.
Reduce the frequency with which events are generated while a subgraph deployment is syncing .
self
logger
&Logger
store
Arc<impl Store>
deployment
SubgraphDeploymentId
interval
Duration
self
logger
&Logger
store
Arc<impl Store>
deployment
SubgraphDeploymentId
interval
Duration