async def _run_loop(self, executor):
        """Main loop."""
        loop = asyncio.get_event_loop()

        # Get first heap in each stream (should be empty).
        self._log.info("Waiting for %d streams to start...", self._num_streams)
        for stream in self._streams:
            await stream.get(loop=loop)

        i_block = 0
        receive_buffer = [None, None]
        while True:
            # Process the previous buffer, if available.
            processing_tasks = None
            if i_block > 0:
                i_buffer_proc = (i_block - 1) % 2
                processing_tasks = loop.run_in_executor(
                    executor, self.process_buffer, i_block - 1,
                    receive_buffer[i_buffer_proc])

            # Set up asynchronous receive on all streams.
            i_buffer_recv = i_block % 2
            receive_tasks = []
            for stream in self._streams:
                for _ in range(self._num_buffer_times):
                    # NOTE: loop.create_task() is needed to schedule these
                    # in the right order!
                    receive_tasks.append(
                        loop.create_task(stream.get(loop=loop)))
            receive_buffer[i_buffer_recv] = asyncio.gather(*receive_tasks)

            # Ensure asynchronous receives and previous processing tasks are
            # done.
            self._log.info("Receiving block %i", i_block)
            try:
                await receive_buffer[i_buffer_recv]
                self._log.info("Received block %i", i_block)
            except spead2.Stopped:
                self._log.info("Stream Stopped")
                break
            if processing_tasks:
                await processing_tasks

            i_block += 1
async def _run_loop ( self , executor ) : loop = asyncio . get_event_loop ( ) self . _log . info ( "Waiting for %d streams to start..." , self . _num_streams ) for stream in self . _streams : await stream . get ( loop = loop ) i_block = 0 receive_buffer = [ None , None ] while True : processing_tasks = None if i_block > 0 : i_buffer_proc = ( i_block - 1 ) % 2 processing_tasks = loop . run_in_executor ( executor , self . process_buffer , i_block - 1 , receive_buffer [ i_buffer_proc ] ) i_buffer_recv = i_block % 2 receive_tasks = [ ] for stream in self . _streams : for _ in range ( self . _num_buffer_times ) : receive_tasks . append ( loop . create_task ( stream . get ( loop = loop ) ) ) receive_buffer [ i_buffer_recv ] = asyncio . gather ( * receive_tasks ) self . _log . info ( "Receiving block %i" , i_block ) try : await receive_buffer [ i_buffer_recv ] self . _log . info ( "Received block %i" , i_block ) except spead2 . Stopped : self . _log . info ( "Stream Stopped" ) break if processing_tasks : await processing_tasks i_block += 1
Main loop.