Controller

async def is_connected(self) -> bool:
    """Checks if there is running tasks"""

    async with self._clients_lock:
        tasks = [not task.done() for client in self._clients.values() for task in client.tasks]
        return all(tasks)

async def send(self, message: Message, broadcast: bool = False) -> None:
    """Sends message to researcher.

    Args:
        message: Message to send
        broadcast: Broadcast the message to all available researcher. This option should be used for general
            node state messages (e.g. general Error)
    """
    if broadcast:
        return await self._broadcast(message)

    async with self._clients_lock:
        async with self._ip_id_map_lock:
            researcher = message.researcher_id
            await self._clients[self._ip_id_map[researcher]].send(message)

async def start(self) -> None:
    """"Starts the tasks for each GrpcClient"""

    tasks = []
    for researcher in self._researchers:
        client = GrpcClient(self._node_id, researcher, self._update_id_ip_map)
        tasks.append(client.start(on_task=self._on_message))
        self._clients[f"{researcher.host}:{researcher.port}"] = client

    self._loop = asyncio.get_running_loop()

    # Create asyncio locks
    self._ip_id_map_lock = asyncio.Lock()
    self._clients_lock = asyncio.Lock()

    self._is_started.set()

    logger.info("Starting task listeners")

    # Run GrpcClient asyncio tasks
    await asyncio.gather(*tasks)

def is_connected(self) -> bool:
    """"Checks GrpcController is connected to any RPC client.

    This method should only be called from different thread than the one that asyncio loop running in.

    Returns:
        Connection status

    Raises:
        FedbiomedCommunicationError: node is not started
    """
    if self._thread is None or not self._is_started.is_set():
        raise FedbiomedCommunicationError(f"{ErrorNumbers.FB628}: Communication client is not initialized.")

    if not self._thread.is_alive():
        return False

    future = asyncio.run_coroutine_threadsafe(
        super().is_connected(), self._loop
    )
    return future.result()

def send(self, message: Message, broadcast: bool = False) -> None:
    """Sends given message to researcher

    Researcher id must exist in the message.

    Args:
        message: Message to send to researcher
        broadcast: If True, broadcasts the given message to all available.
            This does not prevent adding `researcher_id` to the message.
            The attribute `researcher_id` in the message should be `<unknown>`

    Raises:
        FedbiomedCommunicationError: bad argument type
        FedbiomedCommunicationError: node is not started
    """
    if not isinstance(message, Message):
        raise FedbiomedCommunicationError(
            f"{ErrorNumbers.FB628}: bad argument type for message, expected `Message`, got `{type(message)}`")

    if not self._is_started.is_set():
        raise FedbiomedCommunicationError(f"{ErrorNumbers.FB628}: Communication client is not initialized.")

    asyncio.run_coroutine_threadsafe(
        super().send(message, broadcast), self._loop
    )

def start(self, on_finish: Optional[Callable] = None) -> None:
    """Start GRPCClients in a thread.

    Args:
        on_finish: Called when the tasks for handling all known researchers have finished. 
            Callable has no argument. If None, then no action is taken.
    """
    # Adds grpc handler to send node logs to researchers
    logger.add_grpc_handler(on_log=self.send, node_id=self._node_id)

    self._thread = threading.Thread(target=self._run, args=(on_finish,), daemon=True)
    self._thread.start()