async def add_signal_receiver(
    callback: Callable,
    session_bus: bool = False,
    signal_name: str | None = None,
    dbus_interface: str | None = None,
    bus_name: str | None = None,
    path: str | None = None,
    check_service: bool = False,
    use_bus: MessageBus | None = None,
    preserve: bool = False,
) -> bool:
    """
    Helper function which aims to recreate python-dbus's add_signal_receiver
    method in dbus_fast with asyncio calls.

    If check_service is `True` the method will raise a wanrning and return False
    if the service is not visible on the bus. If the `bus_name` is None, no
    check will be performed.

    Returns True if subscription is successful.
    """
    if not has_dbus:
        logger.warning("dbus-fast is not installed. Unable to subscribe to signals")
        return False

    if bus_name and check_service:
        found = await find_dbus_service(bus_name, session_bus)
        if not found:
            logger.warning(
                "The %s name was not found on the bus. No callback will be attached.", bus_name
            )
            return False

    match_args = {
        "sender": bus_name,
        "member": signal_name,
        "path": path,
        "interface": dbus_interface,
    }

    rule = "type='signal',"
    rule += ",".join(f"{k}='{v}'" for k, v in match_args.items() if v)

    logger.debug("Adding dbus match rule: %s", rule)

    bus, msg = await _send_dbus_message(
        session_bus,
        MessageType.METHOD_CALL,
        "org.freedesktop.DBus",
        "org.freedesktop.DBus",
        "/org/freedesktop/DBus",
        "AddMatch",
        "s",
        [rule],
        bus=use_bus,
        preserve=preserve,
    )

    # Check if message sent successfully
    if bus and msg and msg.message_type == MessageType.METHOD_RETURN:

        def match_message(msg: Message, match_args: dict[str, str | None]) -> bool:
            return all(getattr(msg, k) == v for k, v in match_args.items() if v)

        async def resolve_sender(signal_msg: Message) -> tuple[str, Message]:
            """Looks up a pretty bus name to retrieve the unique name."""
            _, sender_msg = await _send_dbus_message(
                session_bus,
                MessageType.METHOD_CALL,
                "org.freedesktop.DBus",
                "org.freedesktop.DBus",
                "/org/freedesktop/DBus",
                "GetNameOwner",
                "s",
                [match_args["sender"]],
                bus=bus,
            )

            if sender_msg and sender_msg.message_type == MessageType.METHOD_RETURN:
                return sender_msg.body[0], signal_msg

            return "", signal_msg

        def check_message(task: asyncio.Task) -> None:
            new_match_args = match_args.copy()
            new_sender, signal_message = task.result()
            new_match_args["sender"] = new_sender
            if match_message(signal_message, new_match_args):
                callback(signal_message)

        def signal_callback_wrapper(msg: Message) -> None:
            """Custom wrapper to only run callback if message matches our rule."""
            if msg.message_type == MessageType.SIGNAL:
                if match_message(msg, match_args):
                    callback(msg)
                elif "sender" in match_args:
                    # If the message didn't match and we're trying to match the sender
                    # We may need to convert the pretty name to the bus's unique name first
                    task = create_task(resolve_sender(msg))
                    if task:
                        task.add_done_callback(check_message)

        bus.add_message_handler(signal_callback_wrapper)
        return True

    else:
        return False

    Helper function which aims to recreate python-dbus's add_signal_receiver
    method in dbus_fast with asyncio calls.

    If check_service is `True` the method will raise a wanrning and return False
    if the service is not visible on the bus. If the `bus_name` is None, no
    check will be performed.

    Returns True if subscription is successful.
    
libqtile/utils.py