def wait_for_iterator(self, connection_observer, connection_observer_future):
        """
        Version of wait_for() intended to be used by Python3 to implement awaitable object.

        Note: we don't have timeout parameter here. If you want to await with timeout please do use asyncio machinery.
        For ex.:  await asyncio.wait_for(connection_observer, timeout=10)

        :param connection_observer: The one we are awaiting for.
        :param connection_observer_future: Future of connection-observer returned from submit().
        :return: iterator
        """
        self.logger.debug("go foreground: {!r}".format(connection_observer))

        # assuming that connection_observer.start() / runner.submit(connection_observer)
        # has already scheduled future via asyncio.ensure_future
        assert asyncio.futures.isfuture(connection_observer_future)

        return connection_observer_future.__iter__()
        # Note: even if code is so simple we can't move it inside ConnectionObserver.__await__() since different runners
        # may provide different iterator implementing awaitable
        # Here we know, connection_observer_future is asyncio.Future (precisely asyncio.tasks.Task)
        # and we know it has __await__() method. 

def cli():
    import argparse
    from asyncio.tasks import ensure_future

    loop = asyncio.get_event_loop()

    parser = argparse.ArgumentParser(
        description="Perform Bluetooth Low Energy device scan"
    )
    parser.add_argument("-i", dest="dev", default="hci0", help="HCI device")
    parser.add_argument(
        "-t", dest="timeout", type=int, default=5, help="Duration to scan for"
    )
    args = parser.parse_args()

    out = loop.run_until_complete(
        ensure_future(discover(device=args.dev, timeout=float(args.timeout)))
    )
    for o in out:
        print(str(o)) 

def __init__(
        self,
        tasks: Sequence[AbstractExecutorTask],
        task_fail_policy=ExecutorExceptionPolicies.propagate,
    ) -> None:
        """
        Init executor.

        :param tasks: sequence of AbstractExecutorTask instances to run.
        :param task_fail_policy: the exception policy of all the tasks
        """
        self._task_fail_policy: ExecutorExceptionPolicies = task_fail_policy
        self._tasks: Sequence[AbstractExecutorTask] = tasks
        self._is_running: bool = False
        self._future_task: Dict[Awaitable, AbstractExecutorTask] = {}
        self._loop: AbstractEventLoop = asyncio.new_event_loop()
        self._executor_pool: Optional[Executor] = None
        self._set_executor_pool() 

def _wait_tasks_complete(self, skip_exceptions: bool = False) -> None:
        """
        Wait tasks execution to complete.

        :param skip_exceptions: skip exceptions if raised in tasks
        """
        done, pending = await asyncio.wait(
            self._future_task.keys(), return_when=FIRST_EXCEPTION
        )

        async def wait_future(future):
            try:
                await future
            except Exception as e:
                if not skip_exceptions:
                    await self._handle_exception(self._future_task[future], e)

        for future in done:
            await wait_future(future)

        if pending:
            done, _ = await asyncio.wait(pending)
            for task in done:
                await wait_future(task) 

def query(self, decision_task: PollForDecisionTaskResponse, query: WorkflowQuery) -> bytes:
        query_args = query.query_args
        if query_args is None:
            args = []
        else:
            args = json_to_args(query_args)
        task = QueryMethodTask(task_id=self.execution_id,
                               workflow_instance=self.workflow_task.workflow_instance,
                               query_name=query.query_type,
                               query_input=args,
                               decider=self)
        self.tasks.append(task)
        task.start()
        self.event_loop.run_event_loop_once()
        if task.status == Status.DONE:
            if task.exception_thrown:
                raise task.exception_thrown
            else:  # ret_value might be None, need to put it in else
                return task.ret_value
        else:
            raise QueryDidNotComplete(f"Query method {query.query_type} with args {query.query_args} did not complete")


# noinspection PyUnusedLocal 

def cancel_remaining_feeders(loop, logger_name="moler.runner.asyncio", in_shutdown=False):
    remaining = [task for task in asyncio.Task.all_tasks(loop=loop) if (not task.done()) and (is_feeder(task))]
    if remaining:
        logger = logging.getLogger(logger_name)
        loop_id = instance_id(loop)
        log_level = logging.WARNING if in_shutdown else logging.DEBUG
        logger.log(level=log_level, msg="cancelling all remaining feeders of loop {}:".format(loop_id))
        remaining_tasks = asyncio.gather(*remaining, loop=loop, return_exceptions=True)
        for feeder in remaining:
            logger.log(level=log_level, msg="  remaining {}:{}".format(instance_id(feeder), feeder))
        remaining_tasks.cancel()
        if not loop.is_running():
            # Keep the event loop running until it is either destroyed or all tasks have really terminated
            loop.run_until_complete(remaining_tasks) 

def cleanup_remaining_tasks(loop, logger):
    # https://stackoverflow.com/questions/30765606/whats-the-correct-way-to-clean-up-after-an-interrupted-event-loop
    # https://medium.com/python-pandemonium/asyncio-coroutine-patterns-beyond-await-a6121486656f
    # Handle shutdown gracefully by waiting for all tasks to be cancelled
    all_tasks = [task for task in asyncio.Task.all_tasks(loop=loop)]
    not_done_tasks = [task for task in asyncio.Task.all_tasks(loop=loop) if not task.done()]
    if not_done_tasks:
        logger.info("cancelling all remaining tasks")
        # NOTE: following code cancels all tasks - possibly not ours as well

        cleanup_selected_tasks(tasks2cancel=not_done_tasks, loop=loop, logger=logger) 

def cleanup_selected_tasks(tasks2cancel, loop, logger):
    logger.debug("tasks to cancel: {}".format(tasks2cancel))
    remaining_tasks = asyncio.gather(*tasks2cancel, loop=loop, return_exceptions=True)
    remaining_tasks.add_done_callback(lambda t: loop.stop())
    remaining_tasks.cancel()

    # Keep the event loop running until it is either destroyed or all
    # tasks have really terminated
    loop.run_until_complete(remaining_tasks) 

def start(self) -> None:
        """Start tasks."""
        self._is_running = True
        self._start_tasks()
        self._loop.run_until_complete(self._wait_tasks_complete()) 

def stop(self) -> None:
        """Stop tasks."""
        self._is_running = False
        for task in self._tasks:
            self._stop_task(task)
        if not self._loop.is_running():
            self._loop.run_until_complete(
                self._wait_tasks_complete(skip_exceptions=True)
            ) 

def _start_tasks(self) -> None:
        """Schedule tasks."""
        for task in self._tasks:
            future = self._start_task(task)
            task.future = future
            self._future_task[future] = task 

def _make_executor(
        self, mode: str, fail_policy: ExecutorExceptionPolicies
    ) -> AbstractMultipleExecutor:
        """
        Make an executor instance to run agents with.

        :param mode: executor mode to use.
        :param fail_policy: one of ExecutorExceptionPolicies to be used with Executor

        :return: aea executor instance
        """
        executor_cls = self.SUPPORTED_MODES[mode]
        return executor_cls(tasks=self._make_tasks(), task_fail_policy=fail_policy) 

def _make_tasks(self) -> Sequence[AbstractExecutorTask]:
        """Make tasks to run with executor."""
        raise NotImplementedError 

def _set_tasks(self) -> None:
        """Set run loop tasks."""
        raise NotImplementedError 

def wait_run_loop_stopped(self) -> None:
        """Wait all tasks stopped."""
        return await asyncio.gather(
            *self._tasks, loop=self._loop, return_exceptions=True
        ) 

def _stop_tasks(self) -> None:
        """Cancel all tasks."""
        for task in self._tasks:
            if task.done():
                continue
            task.cancel() 

def _set_tasks(self):
        """Set run loop tasks."""
        self._tasks = self._create_tasks()
        logger.debug("tasks created!") 

def _create_tasks(self) -> List[Task]:
        """
        Create tasks.

        :return: list of asyncio Tasks
        """
        tasks = [
            self._task_process_inbox(),
            self._task_process_internal_messages(),
            self._task_process_new_behaviours(),
            self._task_wait_for_error(),
        ]
        return list(map(self._loop.create_task, tasks))  # type: ignore  # some issue with map and create_task 

def _set_tasks(self) -> None:
        """Set run loop tasks."""
        self._tasks = [self._loop.create_task(self._agent_loop())] 

def create_task(self, coro):
        Task = asyncio.tasks.Task
        task = Task(coro, loop=self)
        return task 

def run_until_complete(self, fut):
        tasks = asyncio.tasks
        # 获取任务
        fut = tasks.ensure_future(
                    fut, loop=self)
        # 增加任务到self._ready
        fut.add_done_callback(done_callback)
        # 跑全部任务
        self.run_forever()
        # 从self._ready中移除
        fut.remove_done_callback(done_callback) 

def unblock_all(self):
        for t in self.tasks:
            t.unblock() 

def handle_workflow_execution_started(self, event: HistoryEvent):
        start_event_attributes = event.workflow_execution_started_event_attributes
        self.decision_context.set_current_run_id(start_event_attributes.original_execution_run_id)
        if start_event_attributes.input is None or start_event_attributes.input == b'':
            workflow_input = []
        else:
            workflow_input = json_to_args(start_event_attributes.input)
        self.workflow_task = WorkflowMethodTask(task_id=self.execution_id, workflow_input=workflow_input,
                                                worker=self.worker, workflow_type=self.workflow_type, decider=self)
        self.event_loop.run_event_loop_once()
        assert self.workflow_task.workflow_instance
        self.tasks.append(self.workflow_task) 

def complete_signal_execution(self, task: SignalMethodTask):
        task.destroy()
        self.tasks.remove(task) 

def handle_workflow_execution_signaled(self, event: HistoryEvent):
        signaled_event_attributes = event.workflow_execution_signaled_event_attributes
        signal_input = signaled_event_attributes.input
        if not signal_input:
            signal_input = []
        else:
            signal_input = json_to_args(signal_input)

        task = SignalMethodTask(task_id=self.execution_id,
                                workflow_instance=self.workflow_task.workflow_instance,
                                signal_name=signaled_event_attributes.signal_name,
                                signal_input=signal_input,
                                decider=self)
        self.tasks.append(task)
        task.start() 

def _run_until_complete(event_loop, future):
        """Run until the Future is done.

        If the argument is a coroutine, it is wrapped in a Task.

        WARNING: It would be disastrous to call run_until_complete()
        with the same coroutine twice -- it would wrap it in two
        different Tasks and that can't be good.

        Return the Future's result, or raise its exception.
        """
        event_loop._check_closed()

        new_task = not asyncio.futures.isfuture(future)
        fut_id = instance_id(future)
        future = asyncio.tasks.ensure_future(future, loop=event_loop)
        task_id = instance_id(future)
        msg = "task for future id ({}) future = asyncio.tasks.ensure_future: (task_id = {}, {})".format(fut_id, task_id,
                                                                                                        future)
        sys.stderr.write(msg + "\n")
        logging.getLogger("moler").debug(msg)

        if new_task:
            # An exception is raised if the future didn't complete, so there
            # is no need to log the "destroy pending task" message
            future._log_destroy_pending = False

        future.add_done_callback(_run_until_complete_cb)
        try:
            event_loop.run_forever()
        except BaseException:
            if new_task and future.done() and not future.cancelled():
                # The coroutine raised a BaseException. Consume the exception
                # to not log a warning, the caller doesn't have access to the
                # local task.
                future.exception()
            raise
        finally:
            future.remove_done_callback(_run_until_complete_cb)
        if not future.done():
            fut_id = instance_id(future)
            msg = "not done future in _run_until_complete(fut_id = {}, {})".format(fut_id, future)
            sys.stderr.write(msg + "\n")
            logging.getLogger("moler").debug(msg)
            raise RuntimeError('Event loop stopped before Future completed. (fut_id = {}, {})'.format(fut_id, future))

        return future.result()