chiark / gitweb /
bus: add the ability for backends to queue to input messages at the same time
[elogind.git] / src / libsystemd-bus / sd-bus.c
index 0be5a29f367fbe4e53bd2ef6a849feb55a8a1169..86265955bbde42508e48e6adf9acdb94846fc893 100644 (file)
@@ -99,10 +99,26 @@ static void bus_node_destroy(sd_bus *b, struct node *n) {
         free(n);
 }
 
+static void bus_reset_queues(sd_bus *b) {
+        unsigned i;
+
+        assert(b);
+
+        for (i = 0; i < b->rqueue_size; i++)
+                sd_bus_message_unref(b->rqueue[i]);
+        free(b->rqueue);
+
+        for (i = 0; i < b->wqueue_size; i++)
+                sd_bus_message_unref(b->wqueue[i]);
+        free(b->wqueue);
+
+        b->rqueue = b->wqueue = NULL;
+        b->rqueue_size = b->wqueue_size = 0;
+}
+
 static void bus_free(sd_bus *b) {
         struct filter_callback *f;
         struct node *n;
-        unsigned i;
 
         assert(b);
 
@@ -126,13 +142,7 @@ static void bus_free(sd_bus *b) {
         close_many(b->fds, b->n_fds);
         free(b->fds);
 
-        for (i = 0; i < b->rqueue_size; i++)
-                sd_bus_message_unref(b->rqueue[i]);
-        free(b->rqueue);
-
-        for (i = 0; i < b->wqueue_size; i++)
-                sd_bus_message_unref(b->wqueue[i]);
-        free(b->wqueue);
+        bus_reset_queues(b);
 
         hashmap_free_free(b->reply_callbacks);
         prioq_free(b->reply_callbacks_prioq);
@@ -172,6 +182,7 @@ _public_ int sd_bus_new(sd_bus **ret) {
         r->input_fd = r->output_fd = -1;
         r->message_version = 1;
         r->hello_flags |= KDBUS_HELLO_ACCEPT_FD;
+        r->attach_flags |= KDBUS_ATTACH_NAMES;
         r->original_pid = getpid();
 
         assert_se(pthread_mutex_init(&r->memfd_cache_mutex, NULL) == 0);
@@ -269,80 +280,17 @@ _public_ int sd_bus_negotiate_attach_timestamp(sd_bus *bus, int b) {
         assert_return(bus->state == BUS_UNSET, -EPERM);
         assert_return(!bus_pid_changed(bus), -ECHILD);
 
-        SET_FLAG(bus->hello_flags, KDBUS_HELLO_ATTACH_TIMESTAMP, b);
-        return 0;
-}
-
-_public_ int sd_bus_negotiate_attach_creds(sd_bus *bus, int b) {
-        assert_return(bus, -EINVAL);
-        assert_return(bus->state == BUS_UNSET, -EPERM);
-        assert_return(!bus_pid_changed(bus), -ECHILD);
-
-        SET_FLAG(bus->hello_flags, KDBUS_HELLO_ATTACH_CREDS, b);
-        return 0;
-}
-
-_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);
-
-        SET_FLAG(bus->hello_flags, KDBUS_HELLO_ATTACH_COMM, b);
-        return 0;
-}
-
-_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);
-
-        SET_FLAG(bus->hello_flags, KDBUS_HELLO_ATTACH_EXE, b);
+        SET_FLAG(bus->attach_flags, KDBUS_ATTACH_TIMESTAMP, b);
         return 0;
 }
 
-_public_ int sd_bus_negotiate_attach_cmdline(sd_bus *bus, int b) {
+_public_ int sd_bus_negotiate_attach_creds(sd_bus *bus, uint64_t mask) {
         assert_return(bus, -EINVAL);
+        assert_return(mask <= _SD_BUS_CREDS_MAX, -EINVAL);
         assert_return(bus->state == BUS_UNSET, -EPERM);
         assert_return(!bus_pid_changed(bus), -ECHILD);
 
-        SET_FLAG(bus->hello_flags, KDBUS_HELLO_ATTACH_CMDLINE, b);
-        return 0;
-}
-
-_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);
-
-        SET_FLAG(bus->hello_flags, KDBUS_HELLO_ATTACH_CGROUP, b);
-        return 0;
-}
-
-_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);
-
-        SET_FLAG(bus->hello_flags, KDBUS_HELLO_ATTACH_CAPS, b);
-        return 0;
-}
-
-_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);
-
-        SET_FLAG(bus->hello_flags, KDBUS_HELLO_ATTACH_SECLABEL, b);
-        return 0;
-}
-
-_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);
-
-        SET_FLAG(bus->hello_flags, KDBUS_HELLO_ATTACH_AUDIT, b);
-        return 0;
+        return kdbus_translate_attach_flags(mask, &bus->creds_mask);
 }
 
 _public_ int sd_bus_set_server(sd_bus *bus, int b, sd_id128_t server_id) {
@@ -365,12 +313,12 @@ _public_ int sd_bus_set_anonymous(sd_bus *bus, int b) {
         return 0;
 }
 
-static int hello_callback(sd_bus *bus, sd_bus_message *reply, void *userdata) {
+static int hello_callback(sd_bus *bus, sd_bus_message *reply, void *userdata, sd_bus_error *error) {
         const char *s;
         int r;
 
         assert(bus);
-        assert(bus->state == BUS_HELLO);
+        assert(bus->state == BUS_HELLO || bus->state == BUS_CLOSING);
         assert(reply);
 
         r = sd_bus_message_get_errno(reply);
@@ -390,7 +338,8 @@ static int hello_callback(sd_bus *bus, sd_bus_message *reply, void *userdata) {
         if (!bus->unique_name)
                 return -ENOMEM;
 
-        bus->state = BUS_RUNNING;
+        if (bus->state == BUS_HELLO)
+                bus->state = BUS_RUNNING;
 
         return 1;
 }
@@ -1183,6 +1132,7 @@ _public_ int sd_bus_open_system_container(const char *machine, sd_bus **ret) {
 }
 
 _public_ void sd_bus_close(sd_bus *bus) {
+
         if (!bus)
                 return;
         if (bus->state == BUS_CLOSED)
@@ -1194,6 +1144,10 @@ _public_ void sd_bus_close(sd_bus *bus) {
 
         sd_bus_detach_event(bus);
 
+        /* Drop all queued messages so that they drop references to
+         * the bus object and the bus may be freed */
+        bus_reset_queues(bus);
+
         if (!bus->is_kernel)
                 bus_close_fds(bus);
 
@@ -1204,6 +1158,18 @@ _public_ void sd_bus_close(sd_bus *bus) {
          * freed. */
 }
 
+static void bus_enter_closing(sd_bus *bus) {
+        assert(bus);
+
+        if (bus->state != BUS_OPENING &&
+            bus->state != BUS_AUTHENTICATING &&
+            bus->state != BUS_HELLO &&
+            bus->state != BUS_RUNNING)
+                return;
+
+        bus->state = BUS_CLOSING;
+}
+
 _public_ sd_bus *sd_bus_ref(sd_bus *bus) {
         assert_return(bus, NULL);
 
@@ -1265,7 +1231,7 @@ _public_ int sd_bus_get_server_id(sd_bus *bus, sd_id128_t *server_id) {
         return 0;
 }
 
-static int bus_seal_message(sd_bus *b, sd_bus_message *m) {
+int bus_seal_message(sd_bus *b, sd_bus_message *m) {
         assert(m);
 
         if (m->header->version > b->message_version)
@@ -1282,6 +1248,20 @@ static int bus_seal_message(sd_bus *b, sd_bus_message *m) {
         return bus_message_seal(m, ++b->serial);
 }
 
+static int bus_write_message(sd_bus *bus, sd_bus_message *message, size_t *idx) {
+        int r;
+
+        assert(bus);
+        assert(message);
+
+        if (bus->is_kernel)
+                r = bus_kernel_write_message(bus, message);
+        else
+                r = bus_socket_write_message(bus, message, idx);
+
+        return r;
+}
+
 static int dispatch_wqueue(sd_bus *bus) {
         int r, ret = 0;
 
@@ -1290,15 +1270,10 @@ static int dispatch_wqueue(sd_bus *bus) {
 
         while (bus->wqueue_size > 0) {
 
-                if (bus->is_kernel)
-                        r = bus_kernel_write_message(bus, bus->wqueue[0]);
-                else
-                        r = bus_socket_write_message(bus, bus->wqueue[0], &bus->windex);
-
-                if (r < 0) {
-                        sd_bus_close(bus);
+                r = bus_write_message(bus, bus->wqueue[0], &bus->windex);
+                if (r < 0)
                         return r;
-                else if (r == 0)
+                else if (r == 0)
                         /* Didn't do anything this time */
                         return ret;
                 else if (bus->is_kernel || bus->windex >= BUS_MESSAGE_SIZE(bus->wqueue[0])) {
@@ -1324,42 +1299,78 @@ static int dispatch_wqueue(sd_bus *bus) {
         return ret;
 }
 
+static int bus_read_message(sd_bus *bus) {
+        assert(bus);
+
+        if (bus->is_kernel)
+                return bus_kernel_read_message(bus);
+        else
+                return bus_socket_read_message(bus);
+}
+
+int bus_rqueue_make_room(sd_bus *bus, unsigned n) {
+        sd_bus_message **q;
+        unsigned x;
+
+        x = bus->rqueue_size + n;
+
+        if (bus->rqueue_allocated >= x)
+                return 0;
+
+        if (x > BUS_RQUEUE_MAX)
+                return -ENOBUFS;
+
+        q = realloc(bus->rqueue, x * sizeof(sd_bus_message*));
+        if (!q)
+                return -ENOMEM;
+
+        bus->rqueue = q;
+        bus->rqueue_allocated = x;
+
+        return 0;
+}
+
+int bus_rqueue_push(sd_bus *bus, sd_bus_message *m) {
+        int r;
+
+        assert(bus);
+        assert(m);
+
+        r = bus_rqueue_make_room(bus, 1);
+        if (r < 0)
+                return r;
+
+        bus->rqueue[bus->rqueue_size++] = m;
+
+        return 0;
+}
+
 static int dispatch_rqueue(sd_bus *bus, sd_bus_message **m) {
-        sd_bus_message *z = NULL;
         int r, ret = 0;
 
         assert(bus);
         assert(m);
         assert(bus->state == BUS_RUNNING || bus->state == BUS_HELLO);
 
-        if (bus->rqueue_size > 0) {
-                /* Dispatch a queued message */
-
-                *m = bus->rqueue[0];
-                bus->rqueue_size --;
-                memmove(bus->rqueue, bus->rqueue + 1, sizeof(sd_bus_message*) * bus->rqueue_size);
-                return 1;
-        }
+        for (;;) {
+                if (bus->rqueue_size > 0) {
+                        /* Dispatch a queued message */
 
-        /* Try to read a new message */
-        do {
-                if (bus->is_kernel)
-                        r = bus_kernel_read_message(bus, &z);
-                else
-                        r = bus_socket_read_message(bus, &z);
+                        *m = bus->rqueue[0];
+                        bus->rqueue_size --;
+                        memmove(bus->rqueue, bus->rqueue + 1, sizeof(sd_bus_message*) * bus->rqueue_size);
+                        return 1;
+                }
 
-                if (r < 0) {
-                        sd_bus_close(bus);
+                /* Try to read a new message */
+                r = bus_read_message(bus);
+                if (r < 0)
                         return r;
-                }
                 if (r == 0)
                         return ret;
 
                 ret = 1;
-        } while (!z);
-
-        *m = z;
-        return ret;
+        }
 }
 
 _public_ int sd_bus_send(sd_bus *bus, sd_bus_message *m, uint64_t *serial) {
@@ -1395,15 +1406,10 @@ _public_ int sd_bus_send(sd_bus *bus, sd_bus_message *m, uint64_t *serial) {
         if ((bus->state == BUS_RUNNING || bus->state == BUS_HELLO) && bus->wqueue_size <= 0) {
                 size_t idx = 0;
 
-                if (bus->is_kernel)
-                        r = bus_kernel_write_message(bus, m);
-                else
-                        r = bus_socket_write_message(bus, m, &idx);
-
-                if (r < 0) {
-                        sd_bus_close(bus);
+                r = bus_write_message(bus, m, &idx);
+                if (r < 0)
                         return r;
-                else if (!bus->is_kernel && idx < BUS_MESSAGE_SIZE(m))  {
+                else if (!bus->is_kernel && idx < BUS_MESSAGE_SIZE(m))  {
                         /* Wasn't fully written. So let's remember how
                          * much was written. Note that the first entry
                          * of the wqueue array is always allocated so
@@ -1435,6 +1441,27 @@ _public_ int sd_bus_send(sd_bus *bus, sd_bus_message *m, uint64_t *serial) {
         return 1;
 }
 
+_public_ int sd_bus_send_to(sd_bus *bus, sd_bus_message *m, const char *destination, uint64_t *serial) {
+        int r;
+
+        assert_return(bus, -EINVAL);
+        assert_return(BUS_IS_OPEN(bus->state), -ENOTCONN);
+        assert_return(m, -EINVAL);
+        assert_return(!bus_pid_changed(bus), -ECHILD);
+
+        if (!streq_ptr(m->destination, destination)) {
+
+                if (!destination)
+                        return -EEXIST;
+
+                r = sd_bus_message_set_destination(m, destination);
+                if (r < 0)
+                        return r;
+        }
+
+        return sd_bus_send(bus, m, serial);
+}
+
 static usec_t calc_elapse(uint64_t usec) {
         if (usec == (uint64_t) -1)
                 return 0;
@@ -1552,7 +1579,7 @@ int bus_ensure_running(sd_bus *bus) {
 
         assert(bus);
 
-        if (bus->state == BUS_UNSET || bus->state == BUS_CLOSED)
+        if (bus->state == BUS_UNSET || bus->state == BUS_CLOSED || bus->state == BUS_CLOSING)
                 return -ENOTCONN;
         if (bus->state == BUS_RUNNING)
                 return 1;
@@ -1579,10 +1606,10 @@ _public_ int sd_bus_call(
                 sd_bus_error *error,
                 sd_bus_message **reply) {
 
-        int r;
         usec_t timeout;
         uint64_t serial;
-        bool room = false;
+        unsigned i;
+        int r;
 
         assert_return(bus, -EINVAL);
         assert_return(BUS_IS_OPEN(bus->state), -ENOTCONN);
@@ -1601,39 +1628,26 @@ _public_ int sd_bus_call(
                 return r;
 
         timeout = calc_elapse(usec);
+        i = bus->rqueue_size;
 
         for (;;) {
                 usec_t left;
-                sd_bus_message *incoming = NULL;
-
-                if (!room) {
-                        sd_bus_message **q;
-
-                        if (bus->rqueue_size >= BUS_RQUEUE_MAX)
-                                return -ENOBUFS;
-
-                        /* Make sure there's room for queuing this
-                         * locally, before we read the message */
-
-                        q = realloc(bus->rqueue, (bus->rqueue_size + 1) * sizeof(sd_bus_message*));
-                        if (!q)
-                                return -ENOMEM;
-
-                        bus->rqueue = q;
-                        room = true;
-                }
 
-                if (bus->is_kernel)
-                        r = bus_kernel_read_message(bus, &incoming);
-                else
-                        r = bus_socket_read_message(bus, &incoming);
+                r = bus_read_message(bus);
                 if (r < 0)
                         return r;
-                if (incoming) {
+
+                while (i < bus->rqueue_size) {
+                        sd_bus_message *incoming = NULL;
+
+                        incoming = bus->rqueue[i];
 
                         if (incoming->reply_serial == serial) {
                                 /* Found a match! */
 
+                                memmove(bus->rqueue + i, bus->rqueue + i + 1, sizeof(sd_bus_message*) * (bus->rqueue_size - i - 1));
+                                bus->rqueue_size--;
+
                                 if (incoming->header->type == SD_BUS_MESSAGE_METHOD_RETURN) {
 
                                         if (reply)
@@ -1666,6 +1680,9 @@ _public_ int sd_bus_call(
                                    incoming->sender &&
                                    streq(bus->unique_name, incoming->sender)) {
 
+                                memmove(bus->rqueue + i, bus->rqueue + i + 1, sizeof(sd_bus_message*) * (bus->rqueue_size - i - 1));
+                                bus->rqueue_size--;
+
                                 /* Our own message? Somebody is trying
                                  * to send its own client a message,
                                  * let's not dead-lock, let's fail
@@ -1675,15 +1692,11 @@ _public_ int sd_bus_call(
                                 return -ELOOP;
                         }
 
-                        /* There's already guaranteed to be room for
-                         * this, so need to resize things here */
-                        bus->rqueue[bus->rqueue_size ++] = incoming;
-                        room = false;
-
                         /* Try to read more, right-away */
-                        continue;
+                        i++;
                 }
-                if (r != 0)
+
+                if (r > 0)
                         continue;
 
                 if (timeout > 0) {
@@ -1710,7 +1723,6 @@ _public_ int sd_bus_call(
 _public_ int sd_bus_get_fd(sd_bus *bus) {
 
         assert_return(bus, -EINVAL);
-        assert_return(BUS_IS_OPEN(bus->state), -ENOTCONN);
         assert_return(bus->input_fd == bus->output_fd, -EPERM);
         assert_return(!bus_pid_changed(bus), -ECHILD);
 
@@ -1721,7 +1733,7 @@ _public_ int sd_bus_get_events(sd_bus *bus) {
         int flags = 0;
 
         assert_return(bus, -EINVAL);
-        assert_return(BUS_IS_OPEN(bus->state), -ENOTCONN);
+        assert_return(BUS_IS_OPEN(bus->state) || bus->state == BUS_CLOSING, -ENOTCONN);
         assert_return(!bus_pid_changed(bus), -ECHILD);
 
         if (bus->state == BUS_OPENING)
@@ -1748,9 +1760,14 @@ _public_ int sd_bus_get_timeout(sd_bus *bus, uint64_t *timeout_usec) {
 
         assert_return(bus, -EINVAL);
         assert_return(timeout_usec, -EINVAL);
-        assert_return(BUS_IS_OPEN(bus->state), -ENOTCONN);
+        assert_return(BUS_IS_OPEN(bus->state) || bus->state == BUS_CLOSING, -ENOTCONN);
         assert_return(!bus_pid_changed(bus), -ECHILD);
 
+        if (bus->state == BUS_CLOSING) {
+                *timeout_usec = 0;
+                return 1;
+        }
+
         if (bus->state == BUS_AUTHENTICATING) {
                 *timeout_usec = bus->auth_timeout;
                 return 1;
@@ -1777,6 +1794,7 @@ _public_ int sd_bus_get_timeout(sd_bus *bus, uint64_t *timeout_usec) {
 }
 
 static int process_timeout(sd_bus *bus) {
+        _cleanup_bus_error_free_ sd_bus_error error_buffer = SD_BUS_ERROR_NULL;
         _cleanup_bus_message_unref_ sd_bus_message* m = NULL;
         struct reply_callback *c;
         usec_t n;
@@ -1795,18 +1813,30 @@ static int process_timeout(sd_bus *bus) {
         r = bus_message_new_synthetic_error(
                         bus,
                         c->serial,
-                        &SD_BUS_ERROR_MAKE(SD_BUS_ERROR_NO_REPLY, "Method call timed out"),
+                        &SD_BUS_ERROR_MAKE_CONST(SD_BUS_ERROR_NO_REPLY, "Method call timed out"),
                         &m);
         if (r < 0)
                 return r;
 
+        m->sender = "org.freedesktop.DBus";
+
+        r = bus_seal_message(bus, m);
+        if (r < 0)
+                return r;
+
         assert_se(prioq_pop(bus->reply_callbacks_prioq) == c);
         hashmap_remove(bus->reply_callbacks, &c->serial);
 
-        r = c->callback(bus, m, c->userdata);
+        bus->current = m;
+        bus->iteration_counter ++;
+
+        r = c->callback(bus, m, c->userdata, &error_buffer);
+        r = bus_maybe_reply_error(m, r, &error_buffer);
         free(c);
 
-        return r < 0 ? r : 1;
+        bus->current = NULL;
+
+        return r;
 }
 
 static int process_hello(sd_bus *bus, sd_bus_message *m) {
@@ -1832,6 +1862,7 @@ static int process_hello(sd_bus *bus, sd_bus_message *m) {
 }
 
 static int process_reply(sd_bus *bus, sd_bus_message *m) {
+        _cleanup_bus_error_free_ sd_bus_error error_buffer = SD_BUS_ERROR_NULL;
         struct reply_callback *c;
         int r;
 
@@ -1853,13 +1884,15 @@ static int process_reply(sd_bus *bus, sd_bus_message *m) {
         if (r < 0)
                 return r;
 
-        r = c->callback(bus, m, c->userdata);
+        r = c->callback(bus, m, c->userdata, &error_buffer);
+        r = bus_maybe_reply_error(m, r, &error_buffer);
         free(c);
 
         return r;
 }
 
 static int process_filter(sd_bus *bus, sd_bus_message *m) {
+        _cleanup_bus_error_free_ sd_bus_error error_buffer = SD_BUS_ERROR_NULL;
         struct filter_callback *l;
         int r;
 
@@ -1884,7 +1917,8 @@ static int process_filter(sd_bus *bus, sd_bus_message *m) {
                         if (r < 0)
                                 return r;
 
-                        r = l->callback(bus, m, l->userdata);
+                        r = l->callback(bus, m, l->userdata, &error_buffer);
+                        r = bus_maybe_reply_error(m, r, &error_buffer);
                         if (r != 0)
                                 return r;
 
@@ -1930,7 +1964,7 @@ static int process_builtin(sd_bus *bus, sd_bus_message *m) {
                 return 1;
 
         if (streq_ptr(m->member, "Ping"))
-                r = sd_bus_message_new_method_return(bus, m, &reply);
+                r = sd_bus_message_new_method_return(m, &reply);
         else if (streq_ptr(m->member, "GetMachineId")) {
                 sd_id128_t id;
                 char sid[33];
@@ -1939,14 +1973,14 @@ static int process_builtin(sd_bus *bus, sd_bus_message *m) {
                 if (r < 0)
                         return r;
 
-                r = sd_bus_message_new_method_return(bus, m, &reply);
+                r = sd_bus_message_new_method_return(m, &reply);
                 if (r < 0)
                         return r;
 
                 r = sd_bus_message_append(reply, "s", sd_id128_to_string(id, sid));
         } else {
                 r = sd_bus_message_new_method_errorf(
-                                bus, m, &reply,
+                                m, &reply,
                                 SD_BUS_ERROR_UNKNOWN_METHOD,
                                  "Unknown method '%s' on interface '%s'.", m->member, m->interface);
         }
@@ -2041,7 +2075,7 @@ static int process_running(sd_bus *bus, sd_bus_message **ret) {
         if (m->header->type == SD_BUS_MESSAGE_METHOD_CALL) {
 
                 r = sd_bus_reply_method_errorf(
-                                bus, m,
+                                m,
                                 SD_BUS_ERROR_UNKNOWN_OBJECT,
                                 "Unknown object '%s'.", m->path);
                 if (r < 0)
@@ -2057,6 +2091,87 @@ null_message:
         return r;
 }
 
+static int process_closing(sd_bus *bus, sd_bus_message **ret) {
+        _cleanup_bus_message_unref_ sd_bus_message *m = NULL;
+        struct reply_callback *c;
+        int r;
+
+        assert(bus);
+        assert(bus->state == BUS_CLOSING);
+
+        c = hashmap_first(bus->reply_callbacks);
+        if (c) {
+                _cleanup_bus_error_free_ sd_bus_error error_buffer = SD_BUS_ERROR_NULL;
+
+                /* First, fail all outstanding method calls */
+                r = bus_message_new_synthetic_error(
+                                bus,
+                                c->serial,
+                                &SD_BUS_ERROR_MAKE_CONST(SD_BUS_ERROR_NO_REPLY, "Connection terminated"),
+                                &m);
+                if (r < 0)
+                        return r;
+
+                r = bus_seal_message(bus, m);
+                if (r < 0)
+                        return r;
+
+                if (c->timeout != 0)
+                        prioq_remove(bus->reply_callbacks_prioq, c, &c->prioq_idx);
+
+                hashmap_remove(bus->reply_callbacks, &c->serial);
+
+                bus->current = m;
+                bus->iteration_counter++;
+
+                r = c->callback(bus, m, c->userdata, &error_buffer);
+                r = bus_maybe_reply_error(m, r, &error_buffer);
+                free(c);
+
+                goto finish;
+        }
+
+        /* Then, synthesize a Disconnected message */
+        r = sd_bus_message_new_signal(
+                        bus,
+                        "/org/freedesktop/DBus/Local",
+                        "org.freedesktop.DBus.Local",
+                        "Disconnected",
+                        &m);
+        if (r < 0)
+                return r;
+
+        m->sender = "org.freedesktop.DBus.Local";
+
+        r = bus_seal_message(bus, m);
+        if (r < 0)
+                return r;
+
+        sd_bus_close(bus);
+
+        bus->current = m;
+        bus->iteration_counter++;
+
+        r = process_filter(bus, m);
+        if (r != 0)
+                goto finish;
+
+        r = process_match(bus, m);
+        if (r != 0)
+                goto finish;
+
+        if (ret) {
+                *ret = m;
+                m = NULL;
+        }
+
+        r = 1;
+
+finish:
+        bus->current = NULL;
+        return r;
+}
+
 _public_ int sd_bus_process(sd_bus *bus, sd_bus_message **ret) {
         BUS_DONT_DESTROY(bus);
         int r;
@@ -2070,7 +2185,7 @@ _public_ int sd_bus_process(sd_bus *bus, sd_bus_message **ret) {
         assert_return(!bus_pid_changed(bus), -ECHILD);
 
         /* We don't allow recursively invoking sd_bus_process(). */
-        assert_return(!bus->processing, -EBUSY);
+        assert_return(!bus->current, -EBUSY);
 
         switch (bus->state) {
 
@@ -2080,29 +2195,43 @@ _public_ int sd_bus_process(sd_bus *bus, sd_bus_message **ret) {
 
         case BUS_OPENING:
                 r = bus_socket_process_opening(bus);
-                if (r < 0)
+                if (r == -ECONNRESET || r == -EPIPE) {
+                        bus_enter_closing(bus);
+                        r = 1;
+                } else if (r < 0)
                         return r;
                 if (ret)
                         *ret = NULL;
                 return r;
 
         case BUS_AUTHENTICATING:
-
                 r = bus_socket_process_authenticating(bus);
-                if (r < 0)
+                if (r == -ECONNRESET || r == -EPIPE) {
+                        bus_enter_closing(bus);
+                        r = 1;
+                } else if (r < 0)
                         return r;
+
                 if (ret)
                         *ret = NULL;
+
                 return r;
 
         case BUS_RUNNING:
         case BUS_HELLO:
-
-                bus->processing = true;
                 r = process_running(bus, ret);
-                bus->processing = false;
+                if (r == -ECONNRESET || r == -EPIPE) {
+                        bus_enter_closing(bus);
+                        r = 1;
+
+                        if (ret)
+                                *ret = NULL;
+                }
 
                 return r;
+
+        case BUS_CLOSING:
+                return process_closing(bus, ret);
         }
 
         assert_not_reached("Unknown state");
@@ -2115,6 +2244,10 @@ static int bus_poll(sd_bus *bus, bool need_more, uint64_t timeout_usec) {
         usec_t m = (usec_t) -1;
 
         assert(bus);
+
+        if (bus->state == BUS_CLOSING)
+                return 1;
+
         assert_return(BUS_IS_OPEN(bus->state), -ENOTCONN);
 
         e = sd_bus_get_events(bus);
@@ -2165,9 +2298,13 @@ static int bus_poll(sd_bus *bus, bool need_more, 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);
         assert_return(!bus_pid_changed(bus), -ECHILD);
 
+        if (bus->state == BUS_CLOSING)
+                return 0;
+
+        assert_return(BUS_IS_OPEN(bus->state) , -ENOTCONN);
+
         if (bus->rqueue_size > 0)
                 return 0;
 
@@ -2178,9 +2315,13 @@ _public_ int sd_bus_flush(sd_bus *bus) {
         int r;
 
         assert_return(bus, -EINVAL);
-        assert_return(BUS_IS_OPEN(bus->state), -ENOTCONN);
         assert_return(!bus_pid_changed(bus), -ECHILD);
 
+        if (bus->state == BUS_CLOSING)
+                return 0;
+
+        assert_return(BUS_IS_OPEN(bus->state), -ENOTCONN);
+
         r = bus_ensure_running(bus);
         if (r < 0)
                 return r;
@@ -2465,17 +2606,25 @@ _public_ int sd_bus_detach_event(sd_bus *bus) {
         assert_return(bus, -EINVAL);
         assert_return(bus->event, -ENXIO);
 
-        if (bus->input_io_event_source)
+        if (bus->input_io_event_source) {
+                sd_event_source_set_enabled(bus->input_io_event_source, SD_EVENT_OFF);
                 bus->input_io_event_source = sd_event_source_unref(bus->input_io_event_source);
+        }
 
-        if (bus->output_io_event_source)
+        if (bus->output_io_event_source) {
+                sd_event_source_set_enabled(bus->output_io_event_source, SD_EVENT_OFF);
                 bus->output_io_event_source = sd_event_source_unref(bus->output_io_event_source);
+        }
 
-        if (bus->time_event_source)
+        if (bus->time_event_source) {
+                sd_event_source_set_enabled(bus->time_event_source, SD_EVENT_OFF);
                 bus->time_event_source = sd_event_source_unref(bus->time_event_source);
+        }
 
-        if (bus->quit_event_source)
+        if (bus->quit_event_source) {
+                sd_event_source_set_enabled(bus->quit_event_source, SD_EVENT_OFF);
                 bus->quit_event_source = sd_event_source_unref(bus->quit_event_source);
+        }
 
         if (bus->event)
                 bus->event = sd_event_unref(bus->event);
@@ -2483,6 +2632,12 @@ _public_ int sd_bus_detach_event(sd_bus *bus) {
         return 0;
 }
 
+_public_ sd_event* sd_bus_get_event(sd_bus *bus) {
+        assert_return(bus, NULL);
+
+        return bus->event;
+}
+
 _public_ sd_bus_message* sd_bus_get_current(sd_bus *bus) {
         assert_return(bus, NULL);
 
@@ -2543,3 +2698,120 @@ _public_ int sd_bus_get_tid(sd_bus *b, pid_t *tid) {
 
         return -ENXIO;
 }
+
+_public_ char *sd_bus_label_escape(const char *s) {
+        char *r, *t;
+        const char *f;
+
+        assert_return(s, NULL);
+
+        /* Escapes all chars that D-Bus' object path cannot deal
+         * with. Can be reversed with bus_path_unescape(). We special
+         * case the empty string. */
+
+        if (*s == 0)
+                return strdup("_");
+
+        r = new(char, strlen(s)*3 + 1);
+        if (!r)
+                return NULL;
+
+        for (f = s, t = r; *f; f++) {
+
+                /* Escape everything that is not a-zA-Z0-9. We also
+                 * escape 0-9 if it's the first character */
+
+                if (!(*f >= 'A' && *f <= 'Z') &&
+                    !(*f >= 'a' && *f <= 'z') &&
+                    !(f > s && *f >= '0' && *f <= '9')) {
+                        *(t++) = '_';
+                        *(t++) = hexchar(*f >> 4);
+                        *(t++) = hexchar(*f);
+                } else
+                        *(t++) = *f;
+        }
+
+        *t = 0;
+
+        return r;
+}
+
+_public_ char *sd_bus_label_unescape(const char *f) {
+        char *r, *t;
+
+        assert_return(f, NULL);
+
+        /* Special case for the empty string */
+        if (streq(f, "_"))
+                return strdup("");
+
+        r = new(char, strlen(f) + 1);
+        if (!r)
+                return NULL;
+
+        for (t = r; *f; f++) {
+
+                if (*f == '_') {
+                        int a, b;
+
+                        if ((a = unhexchar(f[1])) < 0 ||
+                            (b = unhexchar(f[2])) < 0) {
+                                /* Invalid escape code, let's take it literal then */
+                                *(t++) = '_';
+                        } else {
+                                *(t++) = (char) ((a << 4) | b);
+                                f += 2;
+                        }
+                } else
+                        *(t++) = *f;
+        }
+
+        *t = 0;
+
+        return r;
+}
+
+_public_ int sd_bus_get_peer_creds(sd_bus *bus, uint64_t mask, sd_bus_creds **ret) {
+        sd_bus_creds *c;
+        pid_t pid = 0;
+        int r;
+
+        assert_return(bus, -EINVAL);
+        assert_return(mask <= _SD_BUS_CREDS_MAX, -ENOTSUP);
+        assert_return(ret, -EINVAL);
+        assert_return(BUS_IS_OPEN(bus->state), -ENOTCONN);
+        assert_return(!bus_pid_changed(bus), -ECHILD);
+        assert_return(!bus->is_kernel, -ENOTSUP);
+
+        if (!bus->ucred_valid && !isempty(bus->label))
+                return -ENODATA;
+
+        c = bus_creds_new();
+        if (!c)
+                return -ENOMEM;
+
+        if (bus->ucred_valid) {
+                pid = c->pid = bus->ucred.pid;
+                c->uid = bus->ucred.uid;
+                c->gid = bus->ucred.gid;
+
+                c->mask |= ((SD_BUS_CREDS_UID | SD_BUS_CREDS_PID | SD_BUS_CREDS_GID) & mask) & bus->creds_mask;
+        }
+
+        if (!isempty(bus->label) && (mask & SD_BUS_CREDS_SELINUX_CONTEXT)) {
+                c->label = strdup(bus->label);
+                if (!c->label) {
+                        sd_bus_creds_unref(c);
+                        return -ENOMEM;
+                }
+
+                c->mask |= SD_BUS_CREDS_SELINUX_CONTEXT | bus->creds_mask;
+        }
+
+        r = bus_creds_add_more(c, mask, pid, 0);
+        if (r < 0)
+                return r;
+
+        *ret = c;
+        return 0;
+}