def run(self):
        """Starts the sender."""
        # Create the thread pool.
        executor = concurrent.futures.ThreadPoolExecutor(
            max_workers=self._config['num_workers'])

        # Wait to ensure multiple senders can be synchronised.
        now = int(datetime.datetime.utcnow().timestamp())
        start_time = ((now + 29) // 30) * 30
        self._log.info('Waiting until {}'.format(
                       datetime.datetime.fromtimestamp(start_time)))
        while int(datetime.datetime.utcnow().timestamp()) < start_time:
            time.sleep(0.1)

        # Run the event loop.
        loop = asyncio.get_event_loop()
        try:
            loop.run_until_complete(self._run_loop(executor))
        except KeyboardInterrupt:
            pass
        finally:
            # Send the end of stream message to each stream.
            self._log.info('Shutting down, closing streams...')
            tasks = []
            for stream, item_group in self._streams:
                tasks.append(stream.async_send_heap(item_group.get_end()))
            loop.run_until_complete(asyncio.gather(*tasks))
            self._log.info('... finished.')
            executor.shutdown()
def run ( self ) : executor = concurrent . futures . ThreadPoolExecutor ( max_workers = self . _config [ 'num_workers' ] ) now = int ( datetime . datetime . utcnow ( ) . timestamp ( ) ) start_time = ( ( now + 29 ) // 30 ) * 30 self . _log . info ( 'Waiting until {}' . format ( datetime . datetime . fromtimestamp ( start_time ) ) ) while int ( datetime . datetime . utcnow ( ) . timestamp ( ) ) < start_time : time . sleep ( 0.1 ) loop = asyncio . get_event_loop ( ) try : loop . run_until_complete ( self . _run_loop ( executor ) ) except KeyboardInterrupt : pass finally : self . _log . info ( 'Shutting down, closing streams...' ) tasks = [ ] for stream , item_group in self . _streams : tasks . append ( stream . async_send_heap ( item_group . get_end ( ) ) ) loop . run_until_complete ( asyncio . gather ( * tasks ) ) self . _log . info ( '... finished.' ) executor . shutdown ( )
Starts the sender.