fn start_flusher(&mut self, mut tripwire: Tripwire) {
        let (send, recv) = channel();
        self.flush_channel = Some(send);

        let period = self.config.flush_period;
        let mut timer = Delay::new(Instant::now().add(period));

        let mut counters = Vec::new();
        let flusher = future::poll_fn(move || {
            // Check for shutdown
            while tripwire.poll() == Ok(Async::NotReady) {
                // Check messages
                counters.extend(recv.try_iter());

                // Check timer
                try_ready!(timer.poll().map_err(|_| ()));

                // Reset values
                for counter in counters.iter() {
                    counter.set(0);
                }

                // Reset timer
                timer.reset(Instant::now().add(period));
            }
            Ok(Async::Ready(()))
        });

        tokio::spawn(flusher);
    }
/// Flusher will stop when tripwire is done
Flusher will stop when tripwire is done
Flusher will stop when tripwire is done
fn start_flusher(&mut self, mut tripwire: Tripwire) {
        let (send, recv) = channel();
        self.flush_channel = Some(send);
        let period = self.config.flush_period;
        let mut timer = Delay::new(Instant::now().add(period));
        let mut counters = Vec::new();
        let flusher = future::poll_fn(move || {
            // Check for shutdown
            while tripwire.poll() == Ok(Async::NotReady) {
                // Check messages
                counters.extend(recv.try_iter());
                // Check timer
                try_ready!(timer.poll().map_err(|_| ()));
                // Reset values
                for counter in counters.iter() {
                    counter.set(0);
                }
                // Reset timer
                timer.reset(Instant::now().add(period));
            }
            Ok(Async::Ready(()))
        });
        tokio::spawn(flusher);
    }
fn start_flusher ( & mut self , mut tripwire : Tripwire ) { let ( send , recv ) = channel ( ) ; self . flush_channel = Some ( send ) ; let period = self . config . flush_period ; let mut timer = Delay :: new ( Instant :: now ( ) . add ( period ) ) ; let mut counters = Vec :: new ( ) ; let flusher = future :: poll_fn ( move | | { // Check for shutdown while tripwire . poll ( ) == Ok ( Async :: NotReady ) { // Check messages counters . extend ( recv . try_iter ( ) ) ; // Check timer try_ready ! ( timer poll ( ) map_err ( ( ) ) ) ; // Reset values for counter in counters . iter ( ) { counter . set ( 0 ) ; } // Reset timer timer . reset ( Instant :: now ( ) . add ( period ) ) ; } Ok ( Async :: Ready ( ( ) ) ) } ) ; tokio :: spawn ( flusher ) ; }
Flusher will stop when tripwire is done
Flusher will stop when tripwire is done
&mut self
tripwire
Tripwire