X-Git-Url: http://www.chiark.greenend.org.uk/ucgi/~ianmdlvl/git?p=elogind.git;a=blobdiff_plain;f=src%2Flibsystemd-bus%2Fsd-bus.c;h=fceab505ac49090f747b13841c8dfce17d3cc53e;hp=ac886e9d57fc5533e08067e0ed7a107bce69a02e;hb=d9f644e2cdc81089f76314ee1a264895ca391371;hpb=8efd6381798d0cfcba6f98c5a81f9b296ad49733

diff --git a/src/libsystemd-bus/sd-bus.c b/src/libsystemd-bus/sd-bus.c
index ac886e9d5..fceab505a 100644
--- a/src/libsystemd-bus/sd-bus.c
+++ b/src/libsystemd-bus/sd-bus.c
@@ -45,6 +45,8 @@
 #include "bus-introspect.h"
 #include "bus-signature.h"
 #include "bus-objects.h"
+#include "bus-util.h"
+#include "bus-container.h"
 
 static int bus_poll(sd_bus *bus, bool need_more, uint64_t timeout_usec);
 
@@ -104,6 +106,8 @@ static void bus_free(sd_bus *b) {
 
         assert(b);
 
+        sd_bus_detach_event(b);
+
         bus_close_fds(b);
 
         if (b->kdbus_buffer)
@@ -114,6 +118,7 @@ static void bus_free(sd_bus *b) {
         free(b->auth_buffer);
         free(b->address);
         free(b->kernel);
+        free(b->machine);
 
         free(b->exec_path);
         strv_free(b->exec_argv);
@@ -154,7 +159,7 @@ static void bus_free(sd_bus *b) {
         free(b);
 }
 
-int sd_bus_new(sd_bus **ret) {
+_public_ int sd_bus_new(sd_bus **ret) {
         sd_bus *r;
 
         assert_return(ret, -EINVAL);
@@ -183,7 +188,7 @@ int sd_bus_new(sd_bus **ret) {
         return 0;
 }
 
-int sd_bus_set_address(sd_bus *bus, const char *address) {
+_public_ int sd_bus_set_address(sd_bus *bus, const char *address) {
         char *a;
 
         assert_return(bus, -EINVAL);
@@ -201,7 +206,7 @@ int sd_bus_set_address(sd_bus *bus, const char *address) {
         return 0;
 }
 
-int sd_bus_set_fd(sd_bus *bus, int input_fd, int output_fd) {
+_public_ int sd_bus_set_fd(sd_bus *bus, int input_fd, int output_fd) {
         assert_return(bus, -EINVAL);
         assert_return(bus->state == BUS_UNSET, -EPERM);
         assert_return(input_fd >= 0, -EINVAL);
@@ -213,7 +218,7 @@ int sd_bus_set_fd(sd_bus *bus, int input_fd, int output_fd) {
         return 0;
 }
 
-int sd_bus_set_exec(sd_bus *bus, const char *path, char *const argv[]) {
+_public_ int sd_bus_set_exec(sd_bus *bus, const char *path, char *const argv[]) {
         char *p, **a;
 
         assert_return(bus, -EINVAL);
@@ -241,7 +246,7 @@ int sd_bus_set_exec(sd_bus *bus, const char *path, char *const argv[]) {
         return 0;
 }
 
-int sd_bus_set_bus_client(sd_bus *bus, int b) {
+_public_ int sd_bus_set_bus_client(sd_bus *bus, int b) {
         assert_return(bus, -EINVAL);
         assert_return(bus->state == BUS_UNSET, -EPERM);
         assert_return(!bus_pid_changed(bus), -ECHILD);
@@ -250,7 +255,7 @@ int sd_bus_set_bus_client(sd_bus *bus, int b) {
         return 0;
 }
 
-int sd_bus_negotiate_fds(sd_bus *bus, int b) {
+_public_ int sd_bus_negotiate_fds(sd_bus *bus, int b) {
         assert_return(bus, -EINVAL);
         assert_return(bus->state == BUS_UNSET, -EPERM);
         assert_return(!bus_pid_changed(bus), -ECHILD);
@@ -259,7 +264,7 @@ int sd_bus_negotiate_fds(sd_bus *bus, int b) {
         return 0;
 }
 
-int sd_bus_negotiate_attach_comm(sd_bus *bus, int b) {
+_public_ int sd_bus_negotiate_attach_comm(sd_bus *bus, int b) {
         assert_return(bus, -EINVAL);
         assert_return(bus->state == BUS_UNSET, -EPERM);
         assert_return(!bus_pid_changed(bus), -ECHILD);
@@ -268,7 +273,7 @@ int sd_bus_negotiate_attach_comm(sd_bus *bus, int b) {
         return 0;
 }
 
-int sd_bus_negotiate_attach_exe(sd_bus *bus, int b) {
+_public_ int sd_bus_negotiate_attach_exe(sd_bus *bus, int b) {
         assert_return(bus, -EINVAL);
         assert_return(bus->state == BUS_UNSET, -EPERM);
         assert_return(!bus_pid_changed(bus), -ECHILD);
@@ -277,7 +282,7 @@ int sd_bus_negotiate_attach_exe(sd_bus *bus, int b) {
         return 0;
 }
 
-int sd_bus_negotiate_attach_cmdline(sd_bus *bus, int b) {
+_public_ int sd_bus_negotiate_attach_cmdline(sd_bus *bus, int b) {
         assert_return(bus, -EINVAL);
         assert_return(bus->state == BUS_UNSET, -EPERM);
         assert_return(!bus_pid_changed(bus), -ECHILD);
@@ -286,7 +291,7 @@ int sd_bus_negotiate_attach_cmdline(sd_bus *bus, int b) {
         return 0;
 }
 
-int sd_bus_negotiate_attach_cgroup(sd_bus *bus, int b) {
+_public_ int sd_bus_negotiate_attach_cgroup(sd_bus *bus, int b) {
         assert_return(bus, -EINVAL);
         assert_return(bus->state == BUS_UNSET, -EPERM);
         assert_return(!bus_pid_changed(bus), -ECHILD);
@@ -295,7 +300,7 @@ int sd_bus_negotiate_attach_cgroup(sd_bus *bus, int b) {
         return 0;
 }
 
-int sd_bus_negotiate_attach_caps(sd_bus *bus, int b) {
+_public_ int sd_bus_negotiate_attach_caps(sd_bus *bus, int b) {
         assert_return(bus, -EINVAL);
         assert_return(bus->state == BUS_UNSET, -EPERM);
         assert_return(!bus_pid_changed(bus), -ECHILD);
@@ -304,7 +309,7 @@ int sd_bus_negotiate_attach_caps(sd_bus *bus, int b) {
         return 0;
 }
 
-int sd_bus_negotiate_attach_selinux_context(sd_bus *bus, int b) {
+_public_ int sd_bus_negotiate_attach_selinux_context(sd_bus *bus, int b) {
         assert_return(bus, -EINVAL);
         assert_return(bus->state == BUS_UNSET, -EPERM);
         assert_return(!bus_pid_changed(bus), -ECHILD);
@@ -313,7 +318,7 @@ int sd_bus_negotiate_attach_selinux_context(sd_bus *bus, int b) {
         return 0;
 }
 
-int sd_bus_negotiate_attach_audit(sd_bus *bus, int b) {
+_public_ int sd_bus_negotiate_attach_audit(sd_bus *bus, int b) {
         assert_return(bus, -EINVAL);
         assert_return(bus->state == BUS_UNSET, -EPERM);
         assert_return(!bus_pid_changed(bus), -ECHILD);
@@ -322,7 +327,7 @@ int sd_bus_negotiate_attach_audit(sd_bus *bus, int b) {
         return 0;
 }
 
-int sd_bus_set_server(sd_bus *bus, int b, sd_id128_t server_id) {
+_public_ int sd_bus_set_server(sd_bus *bus, int b, sd_id128_t server_id) {
         assert_return(bus, -EINVAL);
         assert_return(b || sd_id128_equal(server_id, SD_ID128_NULL), -EINVAL);
         assert_return(bus->state == BUS_UNSET, -EPERM);
@@ -333,7 +338,7 @@ int sd_bus_set_server(sd_bus *bus, int b, sd_id128_t server_id) {
         return 0;
 }
 
-int sd_bus_set_anonymous(sd_bus *bus, int b) {
+_public_ int sd_bus_set_anonymous(sd_bus *bus, int b) {
         assert_return(bus, -EINVAL);
         assert_return(bus->state == BUS_UNSET, -EPERM);
         assert_return(!bus_pid_changed(bus), -ECHILD);
@@ -350,9 +355,11 @@ static int hello_callback(sd_bus *bus, sd_bus_message *reply, void *userdata) {
         assert(bus->state == BUS_HELLO);
         assert(reply);
 
-        r = bus_message_to_errno(reply);
+        r = sd_bus_message_get_errno(reply);
         if (r < 0)
                 return r;
+        if (r > 0)
+                return -r;
 
         r = sd_bus_message_read(reply, "s", &s);
         if (r < 0)
@@ -748,6 +755,45 @@ static int parse_kernel_address(sd_bus *b, const char **p, char **guid) {
         return 0;
 }
 
+static int parse_container_address(sd_bus *b, const char **p, char **guid) {
+        _cleanup_free_ char *machine = NULL;
+        int r;
+
+        assert(b);
+        assert(p);
+        assert(*p);
+        assert(guid);
+
+        while (**p != 0 && **p != ';') {
+                r = parse_address_key(p, "guid", guid);
+                if (r < 0)
+                        return r;
+                else if (r > 0)
+                        continue;
+
+                r = parse_address_key(p, "machine", &machine);
+                if (r < 0)
+                        return r;
+                else if (r > 0)
+                        continue;
+
+                skip_address_key(p);
+        }
+
+        if (!machine)
+                return -EINVAL;
+
+        free(b->machine);
+        b->machine = machine;
+        machine = NULL;
+
+        b->sockaddr.un.sun_family = AF_UNIX;
+        strncpy(b->sockaddr.un.sun_path, "/var/run/dbus/system_bus_socket", sizeof(b->sockaddr.un.sun_path));
+        b->sockaddr_size = offsetof(struct sockaddr_un, sun_path) + sizeof("/var/run/dbus/system_bus_socket") - 1;
+
+        return 0;
+}
+
 static void bus_reset_parsed_address(sd_bus *b) {
         assert(b);
 
@@ -760,6 +806,8 @@ static void bus_reset_parsed_address(sd_bus *b) {
         b->server_id = SD_ID128_NULL;
         free(b->kernel);
         b->kernel = NULL;
+        free(b->machine);
+        b->machine = NULL;
 }
 
 static int bus_parse_next_address(sd_bus *b) {
@@ -818,6 +866,14 @@ static int bus_parse_next_address(sd_bus *b) {
                         if (r < 0)
                                 return r;
 
+                        break;
+                } else if (startswith(a, "x-container:")) {
+
+                        a += 12;
+                        r = parse_container_address(b, &a, &guid);
+                        if (r < 0)
+                                return r;
+
                         break;
                 }
 
@@ -844,24 +900,32 @@ static int bus_start_address(sd_bus *b) {
         for (;;) {
                 sd_bus_close(b);
 
-                if (b->sockaddr.sa.sa_family != AF_UNSPEC) {
+                if (b->exec_path) {
 
-                        r = bus_socket_connect(b);
+                        r = bus_socket_exec(b);
                         if (r >= 0)
                                 return r;
 
                         b->last_connect_error = -r;
+                } else if (b->kernel) {
 
-                } else if (b->exec_path) {
+                        r = bus_kernel_connect(b);
+                        if (r >= 0)
+                                return r;
 
-                        r = bus_socket_exec(b);
+                        b->last_connect_error = -r;
+
+                } else if (b->machine) {
+
+                        r = bus_container_connect(b);
                         if (r >= 0)
                                 return r;
 
                         b->last_connect_error = -r;
-                } else if (b->kernel) {
 
-                        r = bus_kernel_connect(b);
+                } else if (b->sockaddr.sa.sa_family != AF_UNSPEC) {
+
+                        r = bus_socket_connect(b);
                         if (r >= 0)
                                 return r;
 
@@ -918,7 +982,7 @@ static int bus_start_fd(sd_bus *b) {
                 return bus_socket_take_fd(b);
 }
 
-int sd_bus_start(sd_bus *bus) {
+_public_ int sd_bus_start(sd_bus *bus) {
         int r;
 
         assert_return(bus, -EINVAL);
@@ -932,7 +996,7 @@ int sd_bus_start(sd_bus *bus) {
 
         if (bus->input_fd >= 0)
                 r = bus_start_fd(bus);
-        else if (bus->address || bus->sockaddr.sa.sa_family != AF_UNSPEC || bus->exec_path || bus->kernel)
+        else if (bus->address || bus->sockaddr.sa.sa_family != AF_UNSPEC || bus->exec_path || bus->kernel || bus->machine)
                 r = bus_start_address(bus);
         else
                 return -EINVAL;
@@ -943,7 +1007,7 @@ int sd_bus_start(sd_bus *bus) {
         return bus_send_hello(bus);
 }
 
-int sd_bus_open_system(sd_bus **ret) {
+_public_ int sd_bus_open_system(sd_bus **ret) {
         const char *e;
         sd_bus *b;
         int r;
@@ -979,7 +1043,7 @@ fail:
         return r;
 }
 
-int sd_bus_open_user(sd_bus **ret) {
+_public_ int sd_bus_open_user(sd_bus **ret) {
         const char *e;
         sd_bus *b;
         size_t l;
@@ -1028,7 +1092,79 @@ fail:
         return r;
 }
 
-void sd_bus_close(sd_bus *bus) {
+_public_ int sd_bus_open_system_remote(const char *host, sd_bus **ret) {
+        _cleanup_free_ char *e = NULL;
+        char *p = NULL;
+        sd_bus *bus;
+        int r;
+
+        assert_return(host, -EINVAL);
+        assert_return(ret, -EINVAL);
+
+        e = bus_address_escape(host);
+        if (!e)
+                return -ENOMEM;
+
+        p = strjoin("unixexec:path=ssh,argv1=-xT,argv2=", e, ",argv3=systemd-stdio-bridge", NULL);
+        if (!p)
+                return -ENOMEM;
+
+        r = sd_bus_new(&bus);
+        if (r < 0) {
+                free(p);
+                return r;
+        }
+
+        bus->address = p;
+        bus->bus_client = true;
+
+        r = sd_bus_start(bus);
+        if (r < 0) {
+                bus_free(bus);
+                return r;
+        }
+
+        *ret = bus;
+        return 0;
+}
+
+_public_ int sd_bus_open_system_container(const char *machine, sd_bus **ret) {
+        _cleanup_free_ char *e = NULL;
+        sd_bus *bus;
+        char *p;
+        int r;
+
+        assert_return(machine, -EINVAL);
+        assert_return(ret, -EINVAL);
+
+        e = bus_address_escape(machine);
+        if (!e)
+                return -ENOMEM;
+
+        p = strjoin("x-container:machine=", e, NULL);
+        if (!p)
+                return -ENOMEM;
+
+        r = sd_bus_new(&bus);
+        if (r < 0) {
+                free(p);
+                return r;
+        }
+
+        bus->address = p;
+        bus->bus_client = true;
+
+        r = sd_bus_start(bus);
+        if (r < 0) {
+                bus_free(bus);
+                return r;
+        }
+
+        *ret = bus;
+        return 0;
+}
+
+_public_ void sd_bus_close(sd_bus *bus) {
         if (!bus)
                 return;
         if (bus->state == BUS_CLOSED)
@@ -1038,6 +1174,8 @@ void sd_bus_close(sd_bus *bus) {
 
         bus->state = BUS_CLOSED;
 
+        sd_bus_detach_event(bus);
+
         if (!bus->is_kernel)
                 bus_close_fds(bus);
 
@@ -1048,18 +1186,16 @@ void sd_bus_close(sd_bus *bus) {
          * freed. */
 }
 
-sd_bus *sd_bus_ref(sd_bus *bus) {
-        if (!bus)
-                return NULL;
+_public_ sd_bus *sd_bus_ref(sd_bus *bus) {
+        assert_return(bus, NULL);
 
         assert_se(REFCNT_INC(bus->n_ref) >= 2);
 
         return bus;
 }
 
-sd_bus *sd_bus_unref(sd_bus *bus) {
-        if (!bus)
-                return NULL;
+_public_ sd_bus *sd_bus_unref(sd_bus *bus) {
+        assert_return(bus, NULL);
 
         if (REFCNT_DEC(bus->n_ref) <= 0)
                 bus_free(bus);
@@ -1067,7 +1203,7 @@ sd_bus *sd_bus_unref(sd_bus *bus) {
         return NULL;
 }
 
-int sd_bus_is_open(sd_bus *bus) {
+_public_ int sd_bus_is_open(sd_bus *bus) {
 
         assert_return(bus, -EINVAL);
         assert_return(!bus_pid_changed(bus), -ECHILD);
@@ -1075,7 +1211,7 @@ int sd_bus_is_open(sd_bus *bus) {
         return BUS_IS_OPEN(bus->state);
 }
 
-int sd_bus_can_send(sd_bus *bus, char type) {
+_public_ int sd_bus_can_send(sd_bus *bus, char type) {
         int r;
 
         assert_return(bus, -EINVAL);
@@ -1096,7 +1232,7 @@ int sd_bus_can_send(sd_bus *bus, char type) {
         return bus_type_is_valid(type);
 }
 
-int sd_bus_get_server_id(sd_bus *bus, sd_id128_t *server_id) {
+_public_ int sd_bus_get_server_id(sd_bus *bus, sd_id128_t *server_id) {
         int r;
 
         assert_return(bus, -EINVAL);
@@ -1203,7 +1339,7 @@ static int dispatch_rqueue(sd_bus *bus, sd_bus_message **m) {
         return ret;
 }
 
-int sd_bus_send(sd_bus *bus, sd_bus_message *m, uint64_t *serial) {
+_public_ int sd_bus_send(sd_bus *bus, sd_bus_message *m, uint64_t *serial) {
         int r;
 
         assert_return(bus, -EINVAL);
@@ -1231,7 +1367,7 @@ int sd_bus_send(sd_bus *bus, sd_bus_message *m, uint64_t *serial) {
         /* If this is a reply and no reply was requested, then let's
          * suppress this, if we can */
         if (m->dont_send && !serial)
-                return 0;
+                return 1;
 
         if ((bus->state == BUS_RUNNING || bus->state == BUS_HELLO) && bus->wqueue_size <= 0) {
                 size_t idx = 0;
@@ -1273,7 +1409,7 @@ int sd_bus_send(sd_bus *bus, sd_bus_message *m, uint64_t *serial) {
         if (serial)
                 *serial = BUS_MESSAGE_SERIAL(m);
 
-        return 0;
+        return 1;
 }
 
 static usec_t calc_elapse(uint64_t usec) {
@@ -1304,7 +1440,7 @@ static int timeout_compare(const void *a, const void *b) {
         return 0;
 }
 
-int sd_bus_send_with_reply(
+_public_ int sd_bus_send_with_reply(
                 sd_bus *bus,
                 sd_bus_message *m,
                 sd_bus_message_handler_t callback,
@@ -1318,7 +1454,7 @@ int sd_bus_send_with_reply(
         assert_return(bus, -EINVAL);
         assert_return(BUS_IS_OPEN(bus->state), -ENOTCONN);
         assert_return(m, -EINVAL);
-        assert_return(m->header->type == SD_BUS_MESSAGE_TYPE_METHOD_CALL, -EINVAL);
+        assert_return(m->header->type == SD_BUS_MESSAGE_METHOD_CALL, -EINVAL);
         assert_return(!(m->header->flags & SD_BUS_MESSAGE_NO_REPLY_EXPECTED), -EINVAL);
         assert_return(callback, -EINVAL);
         assert_return(!bus_pid_changed(bus), -ECHILD);
@@ -1370,7 +1506,7 @@ int sd_bus_send_with_reply(
         return r;
 }
 
-int sd_bus_send_with_reply_cancel(sd_bus *bus, uint64_t serial) {
+_public_ int sd_bus_send_with_reply_cancel(sd_bus *bus, uint64_t serial) {
         struct reply_callback *c;
 
         assert_return(bus, -EINVAL);
@@ -1413,7 +1549,7 @@ int bus_ensure_running(sd_bus *bus) {
         }
 }
 
-int sd_bus_send_with_reply_and_block(
+_public_ int sd_bus_send_with_reply_and_block(
                 sd_bus *bus,
                 sd_bus_message *m,
                 uint64_t usec,
@@ -1428,7 +1564,7 @@ int sd_bus_send_with_reply_and_block(
         assert_return(bus, -EINVAL);
         assert_return(BUS_IS_OPEN(bus->state), -ENOTCONN);
         assert_return(m, -EINVAL);
-        assert_return(m->header->type == SD_BUS_MESSAGE_TYPE_METHOD_CALL, -EINVAL);
+        assert_return(m->header->type == SD_BUS_MESSAGE_METHOD_CALL, -EINVAL);
         assert_return(!(m->header->flags & SD_BUS_MESSAGE_NO_REPLY_EXPECTED), -EINVAL);
         assert_return(!bus_error_is_dirty(error), -EINVAL);
         assert_return(!bus_pid_changed(bus), -ECHILD);
@@ -1475,17 +1611,17 @@ int sd_bus_send_with_reply_and_block(
                         if (incoming->reply_serial == serial) {
                                 /* Found a match! */
 
-                                if (incoming->header->type == SD_BUS_MESSAGE_TYPE_METHOD_RETURN) {
+                                if (incoming->header->type == SD_BUS_MESSAGE_METHOD_RETURN) {
 
                                         if (reply)
                                                 *reply = incoming;
                                         else
                                                 sd_bus_message_unref(incoming);
 
-                                        return 0;
+                                        return 1;
                                 }
 
-                                if (incoming->header->type == SD_BUS_MESSAGE_TYPE_METHOD_ERROR) {
+                                if (incoming->header->type == SD_BUS_MESSAGE_METHOD_ERROR) {
                                         int k;
 
                                         r = sd_bus_error_copy(error, &incoming->error);
@@ -1494,9 +1630,9 @@ int sd_bus_send_with_reply_and_block(
                                                 return r;
                                         }
 
-                                        k = bus_error_to_errno(&incoming->error);
+                                        k = sd_bus_error_get_errno(&incoming->error);
                                         sd_bus_message_unref(incoming);
-                                        return k;
+                                        return -k;
                                 }
 
                                 sd_bus_message_unref(incoming);
@@ -1535,7 +1671,7 @@ int sd_bus_send_with_reply_and_block(
         }
 }
 
-int sd_bus_get_fd(sd_bus *bus) {
+_public_ int sd_bus_get_fd(sd_bus *bus) {
 
         assert_return(bus, -EINVAL);
         assert_return(BUS_IS_OPEN(bus->state), -ENOTCONN);
@@ -1545,7 +1681,7 @@ int sd_bus_get_fd(sd_bus *bus) {
         return bus->input_fd;
 }
 
-int sd_bus_get_events(sd_bus *bus) {
+_public_ int sd_bus_get_events(sd_bus *bus) {
         int flags = 0;
 
         assert_return(bus, -EINVAL);
@@ -1571,7 +1707,7 @@ int sd_bus_get_events(sd_bus *bus) {
         return flags;
 }
 
-int sd_bus_get_timeout(sd_bus *bus, uint64_t *timeout_usec) {
+_public_ int sd_bus_get_timeout(sd_bus *bus, uint64_t *timeout_usec) {
         struct reply_callback *c;
 
         assert_return(bus, -EINVAL);
@@ -1623,7 +1759,7 @@ static int process_timeout(sd_bus *bus) {
         r = bus_message_new_synthetic_error(
                         bus,
                         c->serial,
-                        &SD_BUS_ERROR_MAKE("org.freedesktop.DBus.Error.Timeout", "Timed out"),
+                        &SD_BUS_ERROR_MAKE(SD_BUS_ERROR_NO_REPLY, "Method call timed out"),
                         &m);
         if (r < 0)
                 return r;
@@ -1649,8 +1785,8 @@ static int process_hello(sd_bus *bus, sd_bus_message *m) {
          * here (we leave that to the usual handling), we just verify
          * we don't let any earlier msg through. */
 
-        if (m->header->type != SD_BUS_MESSAGE_TYPE_METHOD_RETURN &&
-            m->header->type != SD_BUS_MESSAGE_TYPE_METHOD_ERROR)
+        if (m->header->type != SD_BUS_MESSAGE_METHOD_RETURN &&
+            m->header->type != SD_BUS_MESSAGE_METHOD_ERROR)
                 return -EIO;
 
         if (m->reply_serial != bus->hello_serial)
@@ -1666,8 +1802,8 @@ static int process_reply(sd_bus *bus, sd_bus_message *m) {
         assert(bus);
         assert(m);
 
-        if (m->header->type != SD_BUS_MESSAGE_TYPE_METHOD_RETURN &&
-            m->header->type != SD_BUS_MESSAGE_TYPE_METHOD_ERROR)
+        if (m->header->type != SD_BUS_MESSAGE_METHOD_RETURN &&
+            m->header->type != SD_BUS_MESSAGE_METHOD_ERROR)
                 return 0;
 
         c = hashmap_remove(bus->reply_callbacks, &m->reply_serial);
@@ -1748,7 +1884,7 @@ static int process_builtin(sd_bus *bus, sd_bus_message *m) {
         assert(bus);
         assert(m);
 
-        if (m->header->type != SD_BUS_MESSAGE_TYPE_METHOD_CALL)
+        if (m->header->type != SD_BUS_MESSAGE_METHOD_CALL)
                 return 0;
 
         if (!streq_ptr(m->interface, "org.freedesktop.DBus.Peer"))
@@ -1775,7 +1911,7 @@ static int process_builtin(sd_bus *bus, sd_bus_message *m) {
         } else {
                 r = sd_bus_message_new_method_errorf(
                                 bus, m, &reply,
-                                "org.freedesktop.DBus.Error.UnknownMethod",
+                                SD_BUS_ERROR_UNKNOWN_METHOD,
                                  "Unknown method '%s' on interface '%s'.", m->member, m->interface);
         }
 
@@ -1795,29 +1931,40 @@ static int process_message(sd_bus *bus, sd_bus_message *m) {
         assert(bus);
         assert(m);
 
+        bus->current = m;
         bus->iteration_counter++;
 
+        log_debug("Got message sender=%s object=%s interface=%s member=%s",
+                  strna(sd_bus_message_get_sender(m)),
+                  strna(sd_bus_message_get_path(m)),
+                  strna(sd_bus_message_get_interface(m)),
+                  strna(sd_bus_message_get_member(m)));
+
         r = process_hello(bus, m);
         if (r != 0)
-                return r;
+                goto finish;
 
         r = process_reply(bus, m);
         if (r != 0)
-                return r;
+                goto finish;
 
         r = process_filter(bus, m);
         if (r != 0)
-                return r;
+                goto finish;
 
         r = process_match(bus, m);
         if (r != 0)
-                return r;
+                goto finish;
 
         r = process_builtin(bus, m);
         if (r != 0)
-                return r;
+                goto finish;
 
-        return bus_process_object(bus, m);
+        r = bus_process_object(bus, m);
+
+finish:
+        bus->current = NULL;
+        return r;
 }
 
 static int process_running(sd_bus *bus, sd_bus_message **ret) {
@@ -1855,11 +2002,11 @@ static int process_running(sd_bus *bus, sd_bus_message **ret) {
                 return 1;
         }
 
-        if (m->header->type == SD_BUS_MESSAGE_TYPE_METHOD_CALL) {
+        if (m->header->type == SD_BUS_MESSAGE_METHOD_CALL) {
 
                 r = sd_bus_reply_method_errorf(
                                 bus, m,
-                                "org.freedesktop.DBus.Error.UnknownObject",
+                                SD_BUS_ERROR_UNKNOWN_OBJECT,
                                 "Unknown object '%s'.", m->path);
                 if (r < 0)
                         return r;
@@ -1874,7 +2021,7 @@ null_message:
         return r;
 }
 
-int sd_bus_process(sd_bus *bus, sd_bus_message **ret) {
+_public_ int sd_bus_process(sd_bus *bus, sd_bus_message **ret) {
         BUS_DONT_DESTROY(bus);
         int r;
 
@@ -1979,7 +2126,7 @@ static int bus_poll(sd_bus *bus, bool need_more, uint64_t timeout_usec) {
         return r > 0 ? 1 : 0;
 }
 
-int sd_bus_wait(sd_bus *bus, uint64_t timeout_usec) {
+_public_ int sd_bus_wait(sd_bus *bus, uint64_t timeout_usec) {
 
         assert_return(bus, -EINVAL);
         assert_return(BUS_IS_OPEN(bus->state), -ENOTCONN);
@@ -1991,7 +2138,7 @@ int sd_bus_wait(sd_bus *bus, uint64_t timeout_usec) {
         return bus_poll(bus, false, timeout_usec);
 }
 
-int sd_bus_flush(sd_bus *bus) {
+_public_ int sd_bus_flush(sd_bus *bus) {
         int r;
 
         assert_return(bus, -EINVAL);
@@ -2019,7 +2166,10 @@ int sd_bus_flush(sd_bus *bus) {
         }
 }
 
-int sd_bus_add_filter(sd_bus *bus, sd_bus_message_handler_t callback, void *userdata) {
+_public_ int sd_bus_add_filter(sd_bus *bus,
+                               sd_bus_message_handler_t callback,
+                               void *userdata) {
+
         struct filter_callback *f;
 
         assert_return(bus, -EINVAL);
@@ -2037,7 +2187,10 @@ int sd_bus_add_filter(sd_bus *bus, sd_bus_message_handler_t callback, void *user
         return 0;
 }
 
-int sd_bus_remove_filter(sd_bus *bus, sd_bus_message_handler_t callback, void *userdata) {
+_public_ int sd_bus_remove_filter(sd_bus *bus,
+                                  sd_bus_message_handler_t callback,
+                                  void *userdata) {
+
         struct filter_callback *f;
 
         assert_return(bus, -EINVAL);
@@ -2056,7 +2209,11 @@ int sd_bus_remove_filter(sd_bus *bus, sd_bus_message_handler_t callback, void *u
         return 0;
 }
 
-int sd_bus_add_match(sd_bus *bus, const char *match, sd_bus_message_handler_t callback, void *userdata) {
+_public_ int sd_bus_add_match(sd_bus *bus,
+                              const char *match,
+                              sd_bus_message_handler_t callback,
+                              void *userdata) {
+
         struct bus_match_component *components = NULL;
         unsigned n_components = 0;
         uint64_t cookie = 0;
@@ -2090,7 +2247,11 @@ finish:
         return r;
 }
 
-int sd_bus_remove_match(sd_bus *bus, const char *match, sd_bus_message_handler_t callback, void *userdata) {
+_public_ int sd_bus_remove_match(sd_bus *bus,
+                                 const char *match,
+                                 sd_bus_message_handler_t callback,
+                                 void *userdata) {
+
         struct bus_match_component *components = NULL;
         unsigned n_components = 0;
         int r = 0, q = 0;
@@ -2123,3 +2284,166 @@ bool bus_pid_changed(sd_bus *bus) {
 
         return bus->original_pid != getpid();
 }
+
+static int io_callback(sd_event_source *s, int fd, uint32_t revents, void *userdata) {
+        void *bus = userdata;
+        int r;
+
+        assert(bus);
+
+        r = sd_bus_process(bus, NULL);
+        if (r < 0)
+                return r;
+
+        return 1;
+}
+
+static int time_callback(sd_event_source *s, uint64_t usec, void *userdata) {
+        void *bus = userdata;
+        int r;
+
+        assert(bus);
+
+        r = sd_bus_process(bus, NULL);
+        if (r < 0)
+                return r;
+
+        return 1;
+}
+
+static int prepare_callback(sd_event_source *s, void *userdata) {
+        sd_bus *bus = userdata;
+        int r, e;
+        usec_t until;
+
+        assert(s);
+        assert(bus);
+
+        e = sd_bus_get_events(bus);
+        if (e < 0)
+                return e;
+
+        if (bus->output_fd != bus->input_fd) {
+
+                r = sd_event_source_set_io_events(bus->input_io_event_source, e & POLLIN);
+                if (r < 0)
+                        return r;
+
+                r = sd_event_source_set_io_events(bus->output_io_event_source, e & POLLOUT);
+                if (r < 0)
+                        return r;
+        } else {
+                r = sd_event_source_set_io_events(bus->input_io_event_source, e);
+                if (r < 0)
+                        return r;
+        }
+
+        r = sd_bus_get_timeout(bus, &until);
+        if (r < 0)
+                return r;
+        if (r > 0) {
+                int j;
+
+                j = sd_event_source_set_time(bus->time_event_source, until);
+                if (j < 0)
+                        return j;
+        }
+
+        r = sd_event_source_set_enabled(bus->time_event_source, r > 0);
+        if (r < 0)
+                return r;
+
+        return 1;
+}
+
+static int quit_callback(sd_event_source *event, void *userdata) {
+        sd_bus *bus = userdata;
+
+        assert(event);
+
+        sd_bus_flush(bus);
+
+        return 1;
+}
+
+_public_ int sd_bus_attach_event(sd_bus *bus, sd_event *event, int priority) {
+        int r;
+
+        assert_return(bus, -EINVAL);
+        assert_return(event, -EINVAL);
+        assert_return(!bus->event, -EBUSY);
+
+        assert(!bus->input_io_event_source);
+        assert(!bus->output_io_event_source);
+        assert(!bus->time_event_source);
+
+        bus->event = sd_event_ref(event);
+
+        r = sd_event_add_io(event, bus->input_fd, 0, io_callback, bus, &bus->input_io_event_source);
+        if (r < 0)
+                goto fail;
+
+        r = sd_event_source_set_priority(bus->input_io_event_source, priority);
+        if (r < 0)
+                goto fail;
+
+        if (bus->output_fd != bus->input_fd) {
+                r = sd_event_add_io(event, bus->output_fd, 0, io_callback, bus, &bus->output_io_event_source);
+                if (r < 0)
+                        goto fail;
+
+                r = sd_event_source_set_priority(bus->output_io_event_source, priority);
+                if (r < 0)
+                        goto fail;
+        }
+
+        r = sd_event_source_set_prepare(bus->input_io_event_source, prepare_callback);
+        if (r < 0)
+                goto fail;
+
+        r = sd_event_add_monotonic(event, 0, 0, time_callback, bus, &bus->time_event_source);
+        if (r < 0)
+                goto fail;
+
+        r = sd_event_source_set_priority(bus->time_event_source, priority);
+        if (r < 0)
+                goto fail;
+
+        r = sd_event_add_quit(event, quit_callback, bus, &bus->quit_event_source);
+        if (r < 0)
+                goto fail;
+
+        return 0;
+
+fail:
+        sd_bus_detach_event(bus);
+        return r;
+}
+
+_public_ int sd_bus_detach_event(sd_bus *bus) {
+        assert_return(bus, -EINVAL);
+        assert_return(bus->event, -ENXIO);
+
+        if (bus->input_io_event_source)
+                bus->input_io_event_source = sd_event_source_unref(bus->input_io_event_source);
+
+        if (bus->output_io_event_source)
+                bus->output_io_event_source = sd_event_source_unref(bus->output_io_event_source);
+
+        if (bus->time_event_source)
+                bus->time_event_source = sd_event_source_unref(bus->time_event_source);
+
+        if (bus->quit_event_source)
+                bus->quit_event_source = sd_event_source_unref(bus->quit_event_source);
+
+        if (bus->event)
+                bus->event = sd_event_unref(bus->event);
+
+        return 0;
+}
+
+sd_bus_message* sd_bus_get_current(sd_bus *bus) {
+        assert_return(bus, NULL);
+
+        return bus->current;
+}