libuv 中 sd-bus 的事件循环处理

Question

我们有一个来自libuv 的事件循环来处理 unixsocket 和 TCP 套接字。该程序现在还必须处理 DBus，我们决定为此使用 sd-bus。

伦纳特wrote on his blog:

Note that our APIs, including sd-bus, integrate nicely into sd-event
event loops, but do not require it, and may be integrated into other
event loops too, as long as they support watching for time and I/O events.

所以我认为，这一定是可能的。

我可以通过sd_bus_get_fd (sd_bus *bus) 获得dbus 套接字fd。 但是我找不到任何明显的方法来阻止 sd-bus 使用其bus_poll 方法在内部等待事件。

例如，当使用sd_bus_call(...) 调用will block 和ppoll 的方法时。

那么：如何处理 libuv 中的 dbus 事件？

Answer 1

我想通了，下面是一个关于如何联合 C++、libuv 和 sd-bus 的示例：

我建议您阅读http://0pointer.de/blog/the-new-sd-bus-api-of-systemd.html 以大致了解 sd-bus。

这些是我在https://github.com/TheJJ/horst实现的代码sn-ps

然后可以使用不阻塞的sd_bus_call_async 完成方法调用（与sd_bus_call 相对）。 不要忘记在sd_bus_call_async 之后调用update_events()，以便通过套接字发出调用！

/**
 * Callback function that is invoked from libuv
 * once dbus events flowed in.
 */
static void on_dbus_ready(uv_poll_t *handle, int /*status*/, int /*events*/) {
    DBusConnection *connection = (DBusConnection *)handle->data;

    sd_bus *bus = connection->get_bus();

    // let dbus handle the available events request
    while (true) {
        // this will trigger the dbus vtable-registered functions
        int r = sd_bus_process(bus, nullptr);

        if (r < 0) {
            printf("[dbus] Failed to process bus: %s", strerror(-r));
            break;
        }
        else if (r > 0) {
            // try to process another request!
            continue;
        }
        else {
            // no more progress, wait for the next callback.
            break;
        }
    }

    // update the events we watch for on the socket.
    connection->update_events();
}

/**
 * Convert the sdbus-returned poll flags to
 * corresponding libuv flags.
 */
int poll_to_libuv_events(int pollflags) {
    int ret = 0;
    if (pollflags & (POLLIN | POLLPRI)) {
        ret |= UV_READABLE;
    }
    if (pollflags & POLLOUT) {
        ret |= UV_WRITABLE;
    }

    // we also have the non-corresponding UV_DISCONNECT

    return ret;
}


class DBusConnection {
public:
    DBusConnection(Satellite *sat);

    virtual ~DBusConnection() = default;

    /** connect to dbus */
    int connect() {
        int r = sd_bus_open_system(&this->bus);

        if (r < 0) {
            printf("[dbus] Failed to connect to bus: %s", strerror(-r));
            goto clean_return;
        }

        r = sd_bus_add_object_vtable(
            this->bus,
            &this->bus_slot,
            "/rofl/lol",      // object path
            "rofl.lol",       // interface name
            your_vtable,
            this              // this is the userdata that'll be passed
                              // to the dbus methods
        );

        if (r < 0) {
            printf("[dbus] Failed to install the horst sdbus object: %s", strerror(-r));
            goto clean_return;
        }

        // register our service name
        r = sd_bus_request_name(this->bus, "moveii.horst", 0);
        if (r < 0) {
            printf("[dbus] Failed to acquire service name: %s", strerror(-r));
            goto clean_return;
        }

        // register the filedescriptor from
        // sd_bus_get_fd(bus) to libuv
        uv_poll_init(this->loop, &this->connection, sd_bus_get_fd(this->bus));


        // make `this` reachable in callbacks.
        this->connection.data = this;

        // init the dbus-event-timer
        uv_timer_init(this->loop, &this->timer);
        this->timer.data = this;

        // process initial events and set up the
        // events and timers for subsequent calls
        on_dbus_ready(&this->connection, 0, 0);

        printf("[dbus] Listener initialized");
        return 0;

    clean_return:
        sd_bus_slot_unref(this->bus_slot);
        sd_bus_unref(this->bus);
        this->bus = nullptr;

        return 1;
    }



    /** update the events watched for on the filedescriptor */
    void update_events() {
        sd_bus *bus = this->get_bus();

        // prepare the callback for calling us the next time.
        int new_events = poll_to_libuv_events(
            sd_bus_get_events(bus)
        );

        uint64_t usec;
        int r = sd_bus_get_timeout(bus, &usec);

        if (not r) {
            // if the timer is running already, it is stopped automatically
            // inside uv_timer_start.
            uv_timer_start(
                &this->timer,
                [] (uv_timer_t *handle) {
                    // yes, handle is not a poll_t, but
                    // we just care for its -> data member anyway.
                    on_dbus_ready((uv_poll_t *)handle, 0, 0);
                },
                usec / 1000, // time in milliseconds, sd_bus provides µseconds
                0            // don't repeat
            );
        }

        // always watch for disconnects:
        new_events |= UV_DISCONNECT;

        // activate the socket watching,
        // and if active, invoke the callback function
        uv_poll_start(&this->connection, new_events, &on_dbus_ready);
    }


    /** close the connections */
    int close() {
        // TODO: maybe this memoryerrors when the loop actually
        //       does the cleanup. we have to wait for the callback.
        uv_close((uv_handle_t *) &this->timer, nullptr);

        uv_poll_stop(&this->connection);

        sd_bus_close(this->bus);
        sd_bus_slot_unref(this->bus_slot);
        sd_bus_unref(this->bus);
        return 0;
    }

    /**
      * Return the bus handle.
      */
    sd_bus *get_bus() const {
        return this->bus;
    }

protected:
    /**
      * loop handle
      */
    uv_loop_t *loop;

    /**
      * polling object for dbus events
      */
    uv_poll_t connection;

    /**
      * dbus also wants to be called periodically
      */
    uv_timer_t timer;

    /**
      * dbus bus handle
      */
    sd_bus *bus;

    /**
      * dbus slot handle
      */
    sd_bus_slot *bus_slot;
};