1 /*-*- Mode: C; c-basic-offset: 8; indent-tabs-mode: nil -*-*/
4 This file is part of systemd.
6 Copyright 2013 Lennart Poettering
8 systemd is free software; you can redistribute it and/or modify it
9 under the terms of the GNU Lesser General Public License as published by
10 the Free Software Foundation; either version 2.1 of the License, or
11 (at your option) any later version.
13 systemd is distributed in the hope that it will be useful, but
14 WITHOUT ANY WARRANTY; without even the implied warranty of
15 MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
16 Lesser General Public License for more details.
18 You should have received a copy of the GNU Lesser General Public License
19 along with systemd; If not, see <http://www.gnu.org/licenses/>.
39 #include "bus-internal.h"
40 #include "bus-message.h"
42 #include "bus-socket.h"
43 #include "bus-kernel.h"
44 #include "bus-control.h"
45 #include "bus-introspect.h"
46 #include "bus-signature.h"
47 #include "bus-objects.h"
49 #include "bus-container.h"
51 static int bus_poll(sd_bus *bus, bool need_more, uint64_t timeout_usec);
53 static void bus_close_fds(sd_bus *b) {
57 close_nointr_nofail(b->input_fd);
59 if (b->output_fd >= 0 && b->output_fd != b->input_fd)
60 close_nointr_nofail(b->output_fd);
62 b->input_fd = b->output_fd = -1;
65 static void bus_node_destroy(sd_bus *b, struct node *n) {
66 struct node_callback *c;
67 struct node_vtable *v;
68 struct node_enumerator *e;
76 bus_node_destroy(b, n->child);
78 while ((c = n->callbacks)) {
79 LIST_REMOVE(callbacks, n->callbacks, c);
83 while ((v = n->vtables)) {
84 LIST_REMOVE(vtables, n->vtables, v);
89 while ((e = n->enumerators)) {
90 LIST_REMOVE(enumerators, n->enumerators, e);
95 LIST_REMOVE(siblings, n->parent->child, n);
97 assert_se(hashmap_remove(b->nodes, n->path) == n);
102 static void bus_reset_queues(sd_bus *b) {
107 for (i = 0; i < b->rqueue_size; i++)
108 sd_bus_message_unref(b->rqueue[i]);
111 for (i = 0; i < b->wqueue_size; i++)
112 sd_bus_message_unref(b->wqueue[i]);
115 b->rqueue = b->wqueue = NULL;
116 b->rqueue_size = b->wqueue_size = 0;
119 static void bus_free(sd_bus *b) {
120 struct filter_callback *f;
125 sd_bus_detach_event(b);
130 munmap(b->kdbus_buffer, KDBUS_POOL_SIZE);
133 free(b->unique_name);
134 free(b->auth_buffer);
140 strv_free(b->exec_argv);
142 close_many(b->fds, b->n_fds);
147 hashmap_free_free(b->reply_callbacks);
148 prioq_free(b->reply_callbacks_prioq);
150 while ((f = b->filter_callbacks)) {
151 LIST_REMOVE(callbacks, b->filter_callbacks, f);
155 bus_match_free(&b->match_callbacks);
157 hashmap_free_free(b->vtable_methods);
158 hashmap_free_free(b->vtable_properties);
160 while ((n = hashmap_first(b->nodes)))
161 bus_node_destroy(b, n);
163 hashmap_free(b->nodes);
165 bus_kernel_flush_memfd(b);
167 assert_se(pthread_mutex_destroy(&b->memfd_cache_mutex) == 0);
172 _public_ int sd_bus_new(sd_bus **ret) {
175 assert_return(ret, -EINVAL);
181 r->n_ref = REFCNT_INIT;
182 r->input_fd = r->output_fd = -1;
183 r->message_version = 1;
184 r->hello_flags |= KDBUS_HELLO_ACCEPT_FD;
185 r->attach_flags |= KDBUS_ATTACH_NAMES;
186 r->original_pid = getpid();
188 assert_se(pthread_mutex_init(&r->memfd_cache_mutex, NULL) == 0);
190 /* We guarantee that wqueue always has space for at least one
192 r->wqueue = new(sd_bus_message*, 1);
202 _public_ int sd_bus_set_address(sd_bus *bus, const char *address) {
205 assert_return(bus, -EINVAL);
206 assert_return(bus->state == BUS_UNSET, -EPERM);
207 assert_return(address, -EINVAL);
208 assert_return(!bus_pid_changed(bus), -ECHILD);
220 _public_ int sd_bus_set_fd(sd_bus *bus, int input_fd, int output_fd) {
221 assert_return(bus, -EINVAL);
222 assert_return(bus->state == BUS_UNSET, -EPERM);
223 assert_return(input_fd >= 0, -EINVAL);
224 assert_return(output_fd >= 0, -EINVAL);
225 assert_return(!bus_pid_changed(bus), -ECHILD);
227 bus->input_fd = input_fd;
228 bus->output_fd = output_fd;
232 _public_ int sd_bus_set_exec(sd_bus *bus, const char *path, char *const argv[]) {
235 assert_return(bus, -EINVAL);
236 assert_return(bus->state == BUS_UNSET, -EPERM);
237 assert_return(path, -EINVAL);
238 assert_return(!strv_isempty(argv), -EINVAL);
239 assert_return(!bus_pid_changed(bus), -ECHILD);
251 free(bus->exec_path);
252 strv_free(bus->exec_argv);
260 _public_ int sd_bus_set_bus_client(sd_bus *bus, int b) {
261 assert_return(bus, -EINVAL);
262 assert_return(bus->state == BUS_UNSET, -EPERM);
263 assert_return(!bus_pid_changed(bus), -ECHILD);
265 bus->bus_client = !!b;
269 _public_ int sd_bus_negotiate_fds(sd_bus *bus, int b) {
270 assert_return(bus, -EINVAL);
271 assert_return(bus->state == BUS_UNSET, -EPERM);
272 assert_return(!bus_pid_changed(bus), -ECHILD);
274 SET_FLAG(bus->hello_flags, KDBUS_HELLO_ACCEPT_FD, b);
278 _public_ int sd_bus_negotiate_attach_timestamp(sd_bus *bus, int b) {
279 assert_return(bus, -EINVAL);
280 assert_return(bus->state == BUS_UNSET, -EPERM);
281 assert_return(!bus_pid_changed(bus), -ECHILD);
283 SET_FLAG(bus->attach_flags, KDBUS_ATTACH_TIMESTAMP, b);
287 _public_ int sd_bus_negotiate_attach_creds(sd_bus *bus, uint64_t mask) {
288 assert_return(bus, -EINVAL);
289 assert_return(mask <= _SD_BUS_CREDS_MAX, -EINVAL);
290 assert_return(bus->state == BUS_UNSET, -EPERM);
291 assert_return(!bus_pid_changed(bus), -ECHILD);
293 SET_FLAG(bus->attach_flags, KDBUS_ATTACH_CREDS,
294 !!(mask & (SD_BUS_CREDS_UID|SD_BUS_CREDS_GID|SD_BUS_CREDS_PID|SD_BUS_CREDS_PID_STARTTIME|SD_BUS_CREDS_TID)));
296 SET_FLAG(bus->attach_flags, KDBUS_ATTACH_COMM,
297 !!(mask & (SD_BUS_CREDS_COMM|SD_BUS_CREDS_TID_COMM)));
299 SET_FLAG(bus->attach_flags, KDBUS_ATTACH_EXE,
300 !!(mask & SD_BUS_CREDS_EXE));
302 SET_FLAG(bus->attach_flags, KDBUS_ATTACH_CMDLINE,
303 !!(mask & SD_BUS_CREDS_CMDLINE));
305 SET_FLAG(bus->attach_flags, KDBUS_ATTACH_CGROUP,
306 !!(mask & (SD_BUS_CREDS_CGROUP|SD_BUS_CREDS_UNIT|SD_BUS_CREDS_USER_UNIT|SD_BUS_CREDS_SLICE|SD_BUS_CREDS_SESSION|SD_BUS_CREDS_OWNER_UID)));
308 SET_FLAG(bus->attach_flags, KDBUS_ATTACH_CAPS,
309 !!(mask & (SD_BUS_CREDS_EFFECTIVE_CAPS|SD_BUS_CREDS_PERMITTED_CAPS|SD_BUS_CREDS_INHERITABLE_CAPS|SD_BUS_CREDS_BOUNDING_CAPS)));
311 SET_FLAG(bus->attach_flags, KDBUS_ATTACH_SECLABEL,
312 !!(mask & SD_BUS_CREDS_SELINUX_CONTEXT));
314 SET_FLAG(bus->attach_flags, KDBUS_ATTACH_AUDIT,
315 !!(mask & (SD_BUS_CREDS_AUDIT_SESSION_ID|SD_BUS_CREDS_AUDIT_LOGIN_UID)));
317 bus->creds_mask = mask;
322 _public_ int sd_bus_set_server(sd_bus *bus, int b, sd_id128_t server_id) {
323 assert_return(bus, -EINVAL);
324 assert_return(b || sd_id128_equal(server_id, SD_ID128_NULL), -EINVAL);
325 assert_return(bus->state == BUS_UNSET, -EPERM);
326 assert_return(!bus_pid_changed(bus), -ECHILD);
328 bus->is_server = !!b;
329 bus->server_id = server_id;
333 _public_ int sd_bus_set_anonymous(sd_bus *bus, int b) {
334 assert_return(bus, -EINVAL);
335 assert_return(bus->state == BUS_UNSET, -EPERM);
336 assert_return(!bus_pid_changed(bus), -ECHILD);
338 bus->anonymous_auth = !!b;
342 static int hello_callback(sd_bus *bus, sd_bus_message *reply, void *userdata, sd_bus_error *error) {
347 assert(bus->state == BUS_HELLO || bus->state == BUS_CLOSING);
350 r = sd_bus_message_get_errno(reply);
356 r = sd_bus_message_read(reply, "s", &s);
360 if (!service_name_is_valid(s) || s[0] != ':')
363 bus->unique_name = strdup(s);
364 if (!bus->unique_name)
367 if (bus->state == BUS_HELLO)
368 bus->state = BUS_RUNNING;
373 static int bus_send_hello(sd_bus *bus) {
374 _cleanup_bus_message_unref_ sd_bus_message *m = NULL;
379 if (!bus->bus_client || bus->is_kernel)
382 r = sd_bus_message_new_method_call(
384 "org.freedesktop.DBus",
386 "org.freedesktop.DBus",
392 return sd_bus_call_async(bus, m, hello_callback, NULL, 0, &bus->hello_serial);
395 int bus_start_running(sd_bus *bus) {
398 if (bus->bus_client && !bus->is_kernel) {
399 bus->state = BUS_HELLO;
403 bus->state = BUS_RUNNING;
407 static int parse_address_key(const char **p, const char *key, char **value) {
418 if (strncmp(*p, key, l) != 0)
431 while (*a != ';' && *a != ',' && *a != 0) {
449 c = (char) ((x << 4) | y);
456 t = realloc(r, n + 2);
484 static void skip_address_key(const char **p) {
488 *p += strcspn(*p, ",");
494 static int parse_unix_address(sd_bus *b, const char **p, char **guid) {
495 _cleanup_free_ char *path = NULL, *abstract = NULL;
504 while (**p != 0 && **p != ';') {
505 r = parse_address_key(p, "guid", guid);
511 r = parse_address_key(p, "path", &path);
517 r = parse_address_key(p, "abstract", &abstract);
526 if (!path && !abstract)
529 if (path && abstract)
534 if (l > sizeof(b->sockaddr.un.sun_path))
537 b->sockaddr.un.sun_family = AF_UNIX;
538 strncpy(b->sockaddr.un.sun_path, path, sizeof(b->sockaddr.un.sun_path));
539 b->sockaddr_size = offsetof(struct sockaddr_un, sun_path) + l;
540 } else if (abstract) {
541 l = strlen(abstract);
542 if (l > sizeof(b->sockaddr.un.sun_path) - 1)
545 b->sockaddr.un.sun_family = AF_UNIX;
546 b->sockaddr.un.sun_path[0] = 0;
547 strncpy(b->sockaddr.un.sun_path+1, abstract, sizeof(b->sockaddr.un.sun_path)-1);
548 b->sockaddr_size = offsetof(struct sockaddr_un, sun_path) + 1 + l;
554 static int parse_tcp_address(sd_bus *b, const char **p, char **guid) {
555 _cleanup_free_ char *host = NULL, *port = NULL, *family = NULL;
557 struct addrinfo *result, hints = {
558 .ai_socktype = SOCK_STREAM,
559 .ai_flags = AI_ADDRCONFIG,
567 while (**p != 0 && **p != ';') {
568 r = parse_address_key(p, "guid", guid);
574 r = parse_address_key(p, "host", &host);
580 r = parse_address_key(p, "port", &port);
586 r = parse_address_key(p, "family", &family);
599 if (streq(family, "ipv4"))
600 hints.ai_family = AF_INET;
601 else if (streq(family, "ipv6"))
602 hints.ai_family = AF_INET6;
607 r = getaddrinfo(host, port, &hints, &result);
611 return -EADDRNOTAVAIL;
613 memcpy(&b->sockaddr, result->ai_addr, result->ai_addrlen);
614 b->sockaddr_size = result->ai_addrlen;
616 freeaddrinfo(result);
621 static int parse_exec_address(sd_bus *b, const char **p, char **guid) {
623 unsigned n_argv = 0, j;
632 while (**p != 0 && **p != ';') {
633 r = parse_address_key(p, "guid", guid);
639 r = parse_address_key(p, "path", &path);
645 if (startswith(*p, "argv")) {
649 ul = strtoul(*p + 4, (char**) p, 10);
650 if (errno > 0 || **p != '=' || ul > 256) {
660 x = realloc(argv, sizeof(char*) * (ul + 2));
666 memset(x + n_argv, 0, sizeof(char*) * (ul - n_argv + 2));
672 r = parse_address_key(p, NULL, argv + ul);
687 /* Make sure there are no holes in the array, with the
688 * exception of argv[0] */
689 for (j = 1; j < n_argv; j++)
695 if (argv && argv[0] == NULL) {
696 argv[0] = strdup(path);
708 for (j = 0; j < n_argv; j++)
716 static int parse_kernel_address(sd_bus *b, const char **p, char **guid) {
717 _cleanup_free_ char *path = NULL;
725 while (**p != 0 && **p != ';') {
726 r = parse_address_key(p, "guid", guid);
732 r = parse_address_key(p, "path", &path);
751 static int parse_container_address(sd_bus *b, const char **p, char **guid) {
752 _cleanup_free_ char *machine = NULL;
760 while (**p != 0 && **p != ';') {
761 r = parse_address_key(p, "guid", guid);
767 r = parse_address_key(p, "machine", &machine);
780 b->machine = machine;
783 b->sockaddr.un.sun_family = AF_UNIX;
784 strncpy(b->sockaddr.un.sun_path, "/var/run/dbus/system_bus_socket", sizeof(b->sockaddr.un.sun_path));
785 b->sockaddr_size = offsetof(struct sockaddr_un, sun_path) + sizeof("/var/run/dbus/system_bus_socket") - 1;
790 static void bus_reset_parsed_address(sd_bus *b) {
794 b->sockaddr_size = 0;
795 strv_free(b->exec_argv);
799 b->server_id = SD_ID128_NULL;
806 static int bus_parse_next_address(sd_bus *b) {
807 _cleanup_free_ char *guid = NULL;
815 if (b->address[b->address_index] == 0)
818 bus_reset_parsed_address(b);
820 a = b->address + b->address_index;
829 if (startswith(a, "unix:")) {
832 r = parse_unix_address(b, &a, &guid);
837 } else if (startswith(a, "tcp:")) {
840 r = parse_tcp_address(b, &a, &guid);
846 } else if (startswith(a, "unixexec:")) {
849 r = parse_exec_address(b, &a, &guid);
855 } else if (startswith(a, "kernel:")) {
858 r = parse_kernel_address(b, &a, &guid);
863 } else if (startswith(a, "x-container:")) {
866 r = parse_container_address(b, &a, &guid);
879 r = sd_id128_from_string(guid, &b->server_id);
884 b->address_index = a - b->address;
888 static int bus_start_address(sd_bus *b) {
898 r = bus_socket_exec(b);
902 b->last_connect_error = -r;
903 } else if (b->kernel) {
905 r = bus_kernel_connect(b);
909 b->last_connect_error = -r;
911 } else if (b->machine) {
913 r = bus_container_connect(b);
917 b->last_connect_error = -r;
919 } else if (b->sockaddr.sa.sa_family != AF_UNSPEC) {
921 r = bus_socket_connect(b);
925 b->last_connect_error = -r;
928 r = bus_parse_next_address(b);
932 return b->last_connect_error ? -b->last_connect_error : -ECONNREFUSED;
936 int bus_next_address(sd_bus *b) {
939 bus_reset_parsed_address(b);
940 return bus_start_address(b);
943 static int bus_start_fd(sd_bus *b) {
948 assert(b->input_fd >= 0);
949 assert(b->output_fd >= 0);
951 r = fd_nonblock(b->input_fd, true);
955 r = fd_cloexec(b->input_fd, true);
959 if (b->input_fd != b->output_fd) {
960 r = fd_nonblock(b->output_fd, true);
964 r = fd_cloexec(b->output_fd, true);
969 if (fstat(b->input_fd, &st) < 0)
972 if (S_ISCHR(b->input_fd))
973 return bus_kernel_take_fd(b);
975 return bus_socket_take_fd(b);
978 _public_ int sd_bus_start(sd_bus *bus) {
981 assert_return(bus, -EINVAL);
982 assert_return(bus->state == BUS_UNSET, -EPERM);
983 assert_return(!bus_pid_changed(bus), -ECHILD);
985 bus->state = BUS_OPENING;
987 if (bus->is_server && bus->bus_client)
990 if (bus->input_fd >= 0)
991 r = bus_start_fd(bus);
992 else if (bus->address || bus->sockaddr.sa.sa_family != AF_UNSPEC || bus->exec_path || bus->kernel || bus->machine)
993 r = bus_start_address(bus);
1000 return bus_send_hello(bus);
1003 _public_ int sd_bus_open_system(sd_bus **ret) {
1008 assert_return(ret, -EINVAL);
1014 e = secure_getenv("DBUS_SYSTEM_BUS_ADDRESS");
1016 r = sd_bus_set_address(b, e);
1020 b->sockaddr.un.sun_family = AF_UNIX;
1021 strncpy(b->sockaddr.un.sun_path, "/run/dbus/system_bus_socket", sizeof(b->sockaddr.un.sun_path));
1022 b->sockaddr_size = offsetof(struct sockaddr_un, sun_path) + sizeof("/run/dbus/system_bus_socket") - 1;
1025 b->bus_client = true;
1027 r = sd_bus_start(b);
1039 _public_ int sd_bus_open_user(sd_bus **ret) {
1045 assert_return(ret, -EINVAL);
1051 e = secure_getenv("DBUS_SESSION_BUS_ADDRESS");
1053 r = sd_bus_set_address(b, e);
1057 e = secure_getenv("XDG_RUNTIME_DIR");
1064 if (l + 4 > sizeof(b->sockaddr.un.sun_path)) {
1069 b->sockaddr.un.sun_family = AF_UNIX;
1070 memcpy(mempcpy(b->sockaddr.un.sun_path, e, l), "/bus", 4);
1071 b->sockaddr_size = offsetof(struct sockaddr_un, sun_path) + l + 4;
1074 b->bus_client = true;
1076 r = sd_bus_start(b);
1088 _public_ int sd_bus_open_system_remote(const char *host, sd_bus **ret) {
1089 _cleanup_free_ char *e = NULL;
1094 assert_return(host, -EINVAL);
1095 assert_return(ret, -EINVAL);
1097 e = bus_address_escape(host);
1101 p = strjoin("unixexec:path=ssh,argv1=-xT,argv2=", e, ",argv3=systemd-stdio-bridge", NULL);
1105 r = sd_bus_new(&bus);
1112 bus->bus_client = true;
1114 r = sd_bus_start(bus);
1124 _public_ int sd_bus_open_system_container(const char *machine, sd_bus **ret) {
1125 _cleanup_free_ char *e = NULL;
1130 assert_return(machine, -EINVAL);
1131 assert_return(ret, -EINVAL);
1133 e = bus_address_escape(machine);
1137 p = strjoin("x-container:machine=", e, NULL);
1141 r = sd_bus_new(&bus);
1148 bus->bus_client = true;
1150 r = sd_bus_start(bus);
1160 _public_ void sd_bus_close(sd_bus *bus) {
1164 if (bus->state == BUS_CLOSED)
1166 if (bus_pid_changed(bus))
1169 bus->state = BUS_CLOSED;
1171 sd_bus_detach_event(bus);
1173 /* Drop all queued messages so that they drop references to
1174 * the bus object and the bus may be freed */
1175 bus_reset_queues(bus);
1177 if (!bus->is_kernel)
1180 /* We'll leave the fd open in case this is a kernel bus, since
1181 * there might still be memblocks around that reference this
1182 * bus, and they might need to invoke the
1183 * KDBUS_CMD_MSG_RELEASE ioctl on the fd when they are
1187 static void bus_enter_closing(sd_bus *bus) {
1190 if (bus->state != BUS_OPENING &&
1191 bus->state != BUS_AUTHENTICATING &&
1192 bus->state != BUS_HELLO &&
1193 bus->state != BUS_RUNNING)
1196 bus->state = BUS_CLOSING;
1199 _public_ sd_bus *sd_bus_ref(sd_bus *bus) {
1200 assert_return(bus, NULL);
1202 assert_se(REFCNT_INC(bus->n_ref) >= 2);
1207 _public_ sd_bus *sd_bus_unref(sd_bus *bus) {
1208 assert_return(bus, NULL);
1210 if (REFCNT_DEC(bus->n_ref) <= 0)
1216 _public_ int sd_bus_is_open(sd_bus *bus) {
1218 assert_return(bus, -EINVAL);
1219 assert_return(!bus_pid_changed(bus), -ECHILD);
1221 return BUS_IS_OPEN(bus->state);
1224 _public_ int sd_bus_can_send(sd_bus *bus, char type) {
1227 assert_return(bus, -EINVAL);
1228 assert_return(bus->state != BUS_UNSET, -ENOTCONN);
1229 assert_return(!bus_pid_changed(bus), -ECHILD);
1231 if (type == SD_BUS_TYPE_UNIX_FD) {
1232 if (!(bus->hello_flags & KDBUS_HELLO_ACCEPT_FD))
1235 r = bus_ensure_running(bus);
1239 return bus->can_fds;
1242 return bus_type_is_valid(type);
1245 _public_ int sd_bus_get_server_id(sd_bus *bus, sd_id128_t *server_id) {
1248 assert_return(bus, -EINVAL);
1249 assert_return(server_id, -EINVAL);
1250 assert_return(!bus_pid_changed(bus), -ECHILD);
1252 r = bus_ensure_running(bus);
1256 *server_id = bus->server_id;
1260 static int bus_seal_message(sd_bus *b, sd_bus_message *m) {
1263 if (m->header->version > b->message_version)
1267 /* If we copy the same message to multiple
1268 * destinations, avoid using the same serial
1270 b->serial = MAX(b->serial, BUS_MESSAGE_SERIAL(m));
1274 return bus_message_seal(m, ++b->serial);
1277 static int bus_write_message(sd_bus *bus, sd_bus_message *message, size_t *idx) {
1284 r = bus_kernel_write_message(bus, message);
1286 r = bus_socket_write_message(bus, message, idx);
1291 static int dispatch_wqueue(sd_bus *bus) {
1295 assert(bus->state == BUS_RUNNING || bus->state == BUS_HELLO);
1297 while (bus->wqueue_size > 0) {
1299 r = bus_write_message(bus, bus->wqueue[0], &bus->windex);
1303 /* Didn't do anything this time */
1305 else if (bus->is_kernel || bus->windex >= BUS_MESSAGE_SIZE(bus->wqueue[0])) {
1306 /* Fully written. Let's drop the entry from
1309 * This isn't particularly optimized, but
1310 * well, this is supposed to be our worst-case
1311 * buffer only, and the socket buffer is
1312 * supposed to be our primary buffer, and if
1313 * it got full, then all bets are off
1316 sd_bus_message_unref(bus->wqueue[0]);
1317 bus->wqueue_size --;
1318 memmove(bus->wqueue, bus->wqueue + 1, sizeof(sd_bus_message*) * bus->wqueue_size);
1328 static int bus_read_message(sd_bus *bus, sd_bus_message **m) {
1335 r = bus_kernel_read_message(bus, m);
1337 r = bus_socket_read_message(bus, m);
1342 static int dispatch_rqueue(sd_bus *bus, sd_bus_message **m) {
1343 sd_bus_message *z = NULL;
1348 assert(bus->state == BUS_RUNNING || bus->state == BUS_HELLO);
1350 if (bus->rqueue_size > 0) {
1351 /* Dispatch a queued message */
1353 *m = bus->rqueue[0];
1354 bus->rqueue_size --;
1355 memmove(bus->rqueue, bus->rqueue + 1, sizeof(sd_bus_message*) * bus->rqueue_size);
1359 /* Try to read a new message */
1361 r = bus_read_message(bus, &z);
1374 _public_ int sd_bus_send(sd_bus *bus, sd_bus_message *m, uint64_t *serial) {
1377 assert_return(bus, -EINVAL);
1378 assert_return(BUS_IS_OPEN(bus->state), -ENOTCONN);
1379 assert_return(m, -EINVAL);
1380 assert_return(!bus_pid_changed(bus), -ECHILD);
1383 r = sd_bus_can_send(bus, SD_BUS_TYPE_UNIX_FD);
1390 /* If the serial number isn't kept, then we know that no reply
1392 if (!serial && !m->sealed)
1393 m->header->flags |= SD_BUS_MESSAGE_NO_REPLY_EXPECTED;
1395 r = bus_seal_message(bus, m);
1399 /* If this is a reply and no reply was requested, then let's
1400 * suppress this, if we can */
1401 if (m->dont_send && !serial)
1404 if ((bus->state == BUS_RUNNING || bus->state == BUS_HELLO) && bus->wqueue_size <= 0) {
1407 r = bus_write_message(bus, m, &idx);
1410 else if (!bus->is_kernel && idx < BUS_MESSAGE_SIZE(m)) {
1411 /* Wasn't fully written. So let's remember how
1412 * much was written. Note that the first entry
1413 * of the wqueue array is always allocated so
1414 * that we always can remember how much was
1416 bus->wqueue[0] = sd_bus_message_ref(m);
1417 bus->wqueue_size = 1;
1423 /* Just append it to the queue. */
1425 if (bus->wqueue_size >= BUS_WQUEUE_MAX)
1428 q = realloc(bus->wqueue, sizeof(sd_bus_message*) * (bus->wqueue_size + 1));
1433 q[bus->wqueue_size ++] = sd_bus_message_ref(m);
1437 *serial = BUS_MESSAGE_SERIAL(m);
1442 _public_ int sd_bus_send_to(sd_bus *bus, sd_bus_message *m, const char *destination, uint64_t *serial) {
1445 assert_return(bus, -EINVAL);
1446 assert_return(BUS_IS_OPEN(bus->state), -ENOTCONN);
1447 assert_return(m, -EINVAL);
1448 assert_return(!bus_pid_changed(bus), -ECHILD);
1450 if (!streq_ptr(m->destination, destination)) {
1455 r = sd_bus_message_set_destination(m, destination);
1460 return sd_bus_send(bus, m, serial);
1463 static usec_t calc_elapse(uint64_t usec) {
1464 if (usec == (uint64_t) -1)
1468 usec = BUS_DEFAULT_TIMEOUT;
1470 return now(CLOCK_MONOTONIC) + usec;
1473 static int timeout_compare(const void *a, const void *b) {
1474 const struct reply_callback *x = a, *y = b;
1476 if (x->timeout != 0 && y->timeout == 0)
1479 if (x->timeout == 0 && y->timeout != 0)
1482 if (x->timeout < y->timeout)
1485 if (x->timeout > y->timeout)
1491 _public_ int sd_bus_call_async(
1494 sd_bus_message_handler_t callback,
1499 struct reply_callback *c;
1502 assert_return(bus, -EINVAL);
1503 assert_return(BUS_IS_OPEN(bus->state), -ENOTCONN);
1504 assert_return(m, -EINVAL);
1505 assert_return(m->header->type == SD_BUS_MESSAGE_METHOD_CALL, -EINVAL);
1506 assert_return(!(m->header->flags & SD_BUS_MESSAGE_NO_REPLY_EXPECTED), -EINVAL);
1507 assert_return(callback, -EINVAL);
1508 assert_return(!bus_pid_changed(bus), -ECHILD);
1510 r = hashmap_ensure_allocated(&bus->reply_callbacks, uint64_hash_func, uint64_compare_func);
1514 if (usec != (uint64_t) -1) {
1515 r = prioq_ensure_allocated(&bus->reply_callbacks_prioq, timeout_compare);
1520 r = bus_seal_message(bus, m);
1524 c = new0(struct reply_callback, 1);
1528 c->callback = callback;
1529 c->userdata = userdata;
1530 c->serial = BUS_MESSAGE_SERIAL(m);
1531 c->timeout = calc_elapse(usec);
1533 r = hashmap_put(bus->reply_callbacks, &c->serial, c);
1539 if (c->timeout != 0) {
1540 r = prioq_put(bus->reply_callbacks_prioq, c, &c->prioq_idx);
1543 sd_bus_call_async_cancel(bus, c->serial);
1548 r = sd_bus_send(bus, m, serial);
1550 sd_bus_call_async_cancel(bus, c->serial);
1557 _public_ int sd_bus_call_async_cancel(sd_bus *bus, uint64_t serial) {
1558 struct reply_callback *c;
1560 assert_return(bus, -EINVAL);
1561 assert_return(serial != 0, -EINVAL);
1562 assert_return(!bus_pid_changed(bus), -ECHILD);
1564 c = hashmap_remove(bus->reply_callbacks, &serial);
1568 if (c->timeout != 0)
1569 prioq_remove(bus->reply_callbacks_prioq, c, &c->prioq_idx);
1575 int bus_ensure_running(sd_bus *bus) {
1580 if (bus->state == BUS_UNSET || bus->state == BUS_CLOSED || bus->state == BUS_CLOSING)
1582 if (bus->state == BUS_RUNNING)
1586 r = sd_bus_process(bus, NULL);
1589 if (bus->state == BUS_RUNNING)
1594 r = sd_bus_wait(bus, (uint64_t) -1);
1600 _public_ int sd_bus_call(
1604 sd_bus_error *error,
1605 sd_bus_message **reply) {
1612 assert_return(bus, -EINVAL);
1613 assert_return(BUS_IS_OPEN(bus->state), -ENOTCONN);
1614 assert_return(m, -EINVAL);
1615 assert_return(m->header->type == SD_BUS_MESSAGE_METHOD_CALL, -EINVAL);
1616 assert_return(!(m->header->flags & SD_BUS_MESSAGE_NO_REPLY_EXPECTED), -EINVAL);
1617 assert_return(!bus_error_is_dirty(error), -EINVAL);
1618 assert_return(!bus_pid_changed(bus), -ECHILD);
1620 r = bus_ensure_running(bus);
1624 r = sd_bus_send(bus, m, &serial);
1628 timeout = calc_elapse(usec);
1632 sd_bus_message *incoming = NULL;
1637 if (bus->rqueue_size >= BUS_RQUEUE_MAX)
1640 /* Make sure there's room for queuing this
1641 * locally, before we read the message */
1643 q = realloc(bus->rqueue, (bus->rqueue_size + 1) * sizeof(sd_bus_message*));
1651 r = bus_read_message(bus, &incoming);
1657 if (incoming->reply_serial == serial) {
1658 /* Found a match! */
1660 if (incoming->header->type == SD_BUS_MESSAGE_METHOD_RETURN) {
1665 sd_bus_message_unref(incoming);
1670 if (incoming->header->type == SD_BUS_MESSAGE_METHOD_ERROR) {
1673 r = sd_bus_error_copy(error, &incoming->error);
1675 sd_bus_message_unref(incoming);
1679 k = sd_bus_error_get_errno(&incoming->error);
1680 sd_bus_message_unref(incoming);
1684 sd_bus_message_unref(incoming);
1687 } else if (incoming->header->serial == serial &&
1690 streq(bus->unique_name, incoming->sender)) {
1692 /* Our own message? Somebody is trying
1693 * to send its own client a message,
1694 * let's not dead-lock, let's fail
1697 sd_bus_message_unref(incoming);
1701 /* There's already guaranteed to be room for
1702 * this, so need to resize things here */
1703 bus->rqueue[bus->rqueue_size ++] = incoming;
1706 /* Try to read more, right-away */
1715 n = now(CLOCK_MONOTONIC);
1721 left = (uint64_t) -1;
1723 r = bus_poll(bus, true, left);
1727 r = dispatch_wqueue(bus);
1733 _public_ int sd_bus_get_fd(sd_bus *bus) {
1735 assert_return(bus, -EINVAL);
1736 assert_return(bus->input_fd == bus->output_fd, -EPERM);
1737 assert_return(!bus_pid_changed(bus), -ECHILD);
1739 return bus->input_fd;
1742 _public_ int sd_bus_get_events(sd_bus *bus) {
1745 assert_return(bus, -EINVAL);
1746 assert_return(BUS_IS_OPEN(bus->state) || bus->state == BUS_CLOSING, -ENOTCONN);
1747 assert_return(!bus_pid_changed(bus), -ECHILD);
1749 if (bus->state == BUS_OPENING)
1751 else if (bus->state == BUS_AUTHENTICATING) {
1753 if (bus_socket_auth_needs_write(bus))
1758 } else if (bus->state == BUS_RUNNING || bus->state == BUS_HELLO) {
1759 if (bus->rqueue_size <= 0)
1761 if (bus->wqueue_size > 0)
1768 _public_ int sd_bus_get_timeout(sd_bus *bus, uint64_t *timeout_usec) {
1769 struct reply_callback *c;
1771 assert_return(bus, -EINVAL);
1772 assert_return(timeout_usec, -EINVAL);
1773 assert_return(BUS_IS_OPEN(bus->state) || bus->state == BUS_CLOSING, -ENOTCONN);
1774 assert_return(!bus_pid_changed(bus), -ECHILD);
1776 if (bus->state == BUS_CLOSING) {
1781 if (bus->state == BUS_AUTHENTICATING) {
1782 *timeout_usec = bus->auth_timeout;
1786 if (bus->state != BUS_RUNNING && bus->state != BUS_HELLO) {
1787 *timeout_usec = (uint64_t) -1;
1791 if (bus->rqueue_size > 0) {
1796 c = prioq_peek(bus->reply_callbacks_prioq);
1798 *timeout_usec = (uint64_t) -1;
1802 *timeout_usec = c->timeout;
1806 static int process_timeout(sd_bus *bus) {
1807 _cleanup_bus_error_free_ sd_bus_error error_buffer = SD_BUS_ERROR_NULL;
1808 _cleanup_bus_message_unref_ sd_bus_message* m = NULL;
1809 struct reply_callback *c;
1815 c = prioq_peek(bus->reply_callbacks_prioq);
1819 n = now(CLOCK_MONOTONIC);
1823 r = bus_message_new_synthetic_error(
1826 &SD_BUS_ERROR_MAKE_CONST(SD_BUS_ERROR_NO_REPLY, "Method call timed out"),
1831 r = bus_seal_message(bus, m);
1835 assert_se(prioq_pop(bus->reply_callbacks_prioq) == c);
1836 hashmap_remove(bus->reply_callbacks, &c->serial);
1839 bus->iteration_counter ++;
1841 r = c->callback(bus, m, c->userdata, &error_buffer);
1842 r = bus_maybe_reply_error(m, r, &error_buffer);
1845 bus->current = NULL;
1850 static int process_hello(sd_bus *bus, sd_bus_message *m) {
1854 if (bus->state != BUS_HELLO)
1857 /* Let's make sure the first message on the bus is the HELLO
1858 * reply. But note that we don't actually parse the message
1859 * here (we leave that to the usual handling), we just verify
1860 * we don't let any earlier msg through. */
1862 if (m->header->type != SD_BUS_MESSAGE_METHOD_RETURN &&
1863 m->header->type != SD_BUS_MESSAGE_METHOD_ERROR)
1866 if (m->reply_serial != bus->hello_serial)
1872 static int process_reply(sd_bus *bus, sd_bus_message *m) {
1873 _cleanup_bus_error_free_ sd_bus_error error_buffer = SD_BUS_ERROR_NULL;
1874 struct reply_callback *c;
1880 if (m->header->type != SD_BUS_MESSAGE_METHOD_RETURN &&
1881 m->header->type != SD_BUS_MESSAGE_METHOD_ERROR)
1884 c = hashmap_remove(bus->reply_callbacks, &m->reply_serial);
1888 if (c->timeout != 0)
1889 prioq_remove(bus->reply_callbacks_prioq, c, &c->prioq_idx);
1891 r = sd_bus_message_rewind(m, true);
1895 r = c->callback(bus, m, c->userdata, &error_buffer);
1896 r = bus_maybe_reply_error(m, r, &error_buffer);
1902 static int process_filter(sd_bus *bus, sd_bus_message *m) {
1903 _cleanup_bus_error_free_ sd_bus_error error_buffer = SD_BUS_ERROR_NULL;
1904 struct filter_callback *l;
1911 bus->filter_callbacks_modified = false;
1913 LIST_FOREACH(callbacks, l, bus->filter_callbacks) {
1915 if (bus->filter_callbacks_modified)
1918 /* Don't run this more than once per iteration */
1919 if (l->last_iteration == bus->iteration_counter)
1922 l->last_iteration = bus->iteration_counter;
1924 r = sd_bus_message_rewind(m, true);
1928 r = l->callback(bus, m, l->userdata, &error_buffer);
1929 r = bus_maybe_reply_error(m, r, &error_buffer);
1935 } while (bus->filter_callbacks_modified);
1940 static int process_match(sd_bus *bus, sd_bus_message *m) {
1947 bus->match_callbacks_modified = false;
1949 r = bus_match_run(bus, &bus->match_callbacks, m);
1953 } while (bus->match_callbacks_modified);
1958 static int process_builtin(sd_bus *bus, sd_bus_message *m) {
1959 _cleanup_bus_message_unref_ sd_bus_message *reply = NULL;
1965 if (m->header->type != SD_BUS_MESSAGE_METHOD_CALL)
1968 if (!streq_ptr(m->interface, "org.freedesktop.DBus.Peer"))
1971 if (m->header->flags & SD_BUS_MESSAGE_NO_REPLY_EXPECTED)
1974 if (streq_ptr(m->member, "Ping"))
1975 r = sd_bus_message_new_method_return(m, &reply);
1976 else if (streq_ptr(m->member, "GetMachineId")) {
1980 r = sd_id128_get_machine(&id);
1984 r = sd_bus_message_new_method_return(m, &reply);
1988 r = sd_bus_message_append(reply, "s", sd_id128_to_string(id, sid));
1990 r = sd_bus_message_new_method_errorf(
1992 SD_BUS_ERROR_UNKNOWN_METHOD,
1993 "Unknown method '%s' on interface '%s'.", m->member, m->interface);
1999 r = sd_bus_send(bus, reply, NULL);
2006 static int process_message(sd_bus *bus, sd_bus_message *m) {
2013 bus->iteration_counter++;
2015 log_debug("Got message sender=%s object=%s interface=%s member=%s",
2016 strna(sd_bus_message_get_sender(m)),
2017 strna(sd_bus_message_get_path(m)),
2018 strna(sd_bus_message_get_interface(m)),
2019 strna(sd_bus_message_get_member(m)));
2021 r = process_hello(bus, m);
2025 r = process_reply(bus, m);
2029 r = process_filter(bus, m);
2033 r = process_match(bus, m);
2037 r = process_builtin(bus, m);
2041 r = bus_process_object(bus, m);
2044 bus->current = NULL;
2048 static int process_running(sd_bus *bus, sd_bus_message **ret) {
2049 _cleanup_bus_message_unref_ sd_bus_message *m = NULL;
2053 assert(bus->state == BUS_RUNNING || bus->state == BUS_HELLO);
2055 r = process_timeout(bus);
2059 r = dispatch_wqueue(bus);
2063 r = dispatch_rqueue(bus, &m);
2069 r = process_message(bus, m);
2074 r = sd_bus_message_rewind(m, true);
2083 if (m->header->type == SD_BUS_MESSAGE_METHOD_CALL) {
2085 r = sd_bus_reply_method_errorf(
2087 SD_BUS_ERROR_UNKNOWN_OBJECT,
2088 "Unknown object '%s'.", m->path);
2102 static int process_closing(sd_bus *bus, sd_bus_message **ret) {
2103 _cleanup_bus_message_unref_ sd_bus_message *m = NULL;
2104 struct reply_callback *c;
2108 assert(bus->state == BUS_CLOSING);
2110 c = hashmap_first(bus->reply_callbacks);
2112 _cleanup_bus_error_free_ sd_bus_error error_buffer = SD_BUS_ERROR_NULL;
2114 /* First, fail all outstanding method calls */
2115 r = bus_message_new_synthetic_error(
2118 &SD_BUS_ERROR_MAKE_CONST(SD_BUS_ERROR_NO_REPLY, "Connection terminated"),
2123 r = bus_seal_message(bus, m);
2127 if (c->timeout != 0)
2128 prioq_remove(bus->reply_callbacks_prioq, c, &c->prioq_idx);
2130 hashmap_remove(bus->reply_callbacks, &c->serial);
2133 bus->iteration_counter++;
2135 r = c->callback(bus, m, c->userdata, &error_buffer);
2136 r = bus_maybe_reply_error(m, r, &error_buffer);
2142 /* Then, synthesize a Disconnected message */
2143 r = sd_bus_message_new_signal(
2145 "/org/freedesktop/DBus/Local",
2146 "org.freedesktop.DBus.Local",
2152 r = bus_seal_message(bus, m);
2159 bus->iteration_counter++;
2161 r = process_filter(bus, m);
2165 r = process_match(bus, m);
2177 bus->current = NULL;
2181 _public_ int sd_bus_process(sd_bus *bus, sd_bus_message **ret) {
2182 BUS_DONT_DESTROY(bus);
2185 /* Returns 0 when we didn't do anything. This should cause the
2186 * caller to invoke sd_bus_wait() before returning the next
2187 * time. Returns > 0 when we did something, which possibly
2188 * means *ret is filled in with an unprocessed message. */
2190 assert_return(bus, -EINVAL);
2191 assert_return(!bus_pid_changed(bus), -ECHILD);
2193 /* We don't allow recursively invoking sd_bus_process(). */
2194 assert_return(!bus->current, -EBUSY);
2196 switch (bus->state) {
2203 r = bus_socket_process_opening(bus);
2204 if (r == -ECONNRESET || r == -EPIPE) {
2205 bus_enter_closing(bus);
2213 case BUS_AUTHENTICATING:
2214 r = bus_socket_process_authenticating(bus);
2215 if (r == -ECONNRESET || r == -EPIPE) {
2216 bus_enter_closing(bus);
2228 r = process_running(bus, ret);
2229 if (r == -ECONNRESET || r == -EPIPE) {
2230 bus_enter_closing(bus);
2240 return process_closing(bus, ret);
2243 assert_not_reached("Unknown state");
2246 static int bus_poll(sd_bus *bus, bool need_more, uint64_t timeout_usec) {
2247 struct pollfd p[2] = {};
2250 usec_t m = (usec_t) -1;
2254 if (bus->state == BUS_CLOSING)
2257 assert_return(BUS_IS_OPEN(bus->state), -ENOTCONN);
2259 e = sd_bus_get_events(bus);
2264 /* The caller really needs some more data, he doesn't
2265 * care about what's already read, or any timeouts
2270 /* The caller wants to process if there's something to
2271 * process, but doesn't care otherwise */
2273 r = sd_bus_get_timeout(bus, &until);
2278 nw = now(CLOCK_MONOTONIC);
2279 m = until > nw ? until - nw : 0;
2283 if (timeout_usec != (uint64_t) -1 && (m == (uint64_t) -1 || timeout_usec < m))
2286 p[0].fd = bus->input_fd;
2287 if (bus->output_fd == bus->input_fd) {
2291 p[0].events = e & POLLIN;
2292 p[1].fd = bus->output_fd;
2293 p[1].events = e & POLLOUT;
2297 r = ppoll(p, n, m == (uint64_t) -1 ? NULL : timespec_store(&ts, m), NULL);
2301 return r > 0 ? 1 : 0;
2304 _public_ int sd_bus_wait(sd_bus *bus, uint64_t timeout_usec) {
2306 assert_return(bus, -EINVAL);
2307 assert_return(!bus_pid_changed(bus), -ECHILD);
2309 if (bus->state == BUS_CLOSING)
2312 assert_return(BUS_IS_OPEN(bus->state) , -ENOTCONN);
2314 if (bus->rqueue_size > 0)
2317 return bus_poll(bus, false, timeout_usec);
2320 _public_ int sd_bus_flush(sd_bus *bus) {
2323 assert_return(bus, -EINVAL);
2324 assert_return(!bus_pid_changed(bus), -ECHILD);
2326 if (bus->state == BUS_CLOSING)
2329 assert_return(BUS_IS_OPEN(bus->state), -ENOTCONN);
2331 r = bus_ensure_running(bus);
2335 if (bus->wqueue_size <= 0)
2339 r = dispatch_wqueue(bus);
2343 if (bus->wqueue_size <= 0)
2346 r = bus_poll(bus, false, (uint64_t) -1);
2352 _public_ int sd_bus_add_filter(sd_bus *bus,
2353 sd_bus_message_handler_t callback,
2356 struct filter_callback *f;
2358 assert_return(bus, -EINVAL);
2359 assert_return(callback, -EINVAL);
2360 assert_return(!bus_pid_changed(bus), -ECHILD);
2362 f = new0(struct filter_callback, 1);
2365 f->callback = callback;
2366 f->userdata = userdata;
2368 bus->filter_callbacks_modified = true;
2369 LIST_PREPEND(callbacks, bus->filter_callbacks, f);
2373 _public_ int sd_bus_remove_filter(sd_bus *bus,
2374 sd_bus_message_handler_t callback,
2377 struct filter_callback *f;
2379 assert_return(bus, -EINVAL);
2380 assert_return(callback, -EINVAL);
2381 assert_return(!bus_pid_changed(bus), -ECHILD);
2383 LIST_FOREACH(callbacks, f, bus->filter_callbacks) {
2384 if (f->callback == callback && f->userdata == userdata) {
2385 bus->filter_callbacks_modified = true;
2386 LIST_REMOVE(callbacks, bus->filter_callbacks, f);
2395 _public_ int sd_bus_add_match(sd_bus *bus,
2397 sd_bus_message_handler_t callback,
2400 struct bus_match_component *components = NULL;
2401 unsigned n_components = 0;
2402 uint64_t cookie = 0;
2405 assert_return(bus, -EINVAL);
2406 assert_return(match, -EINVAL);
2407 assert_return(!bus_pid_changed(bus), -ECHILD);
2409 r = bus_match_parse(match, &components, &n_components);
2413 if (bus->bus_client) {
2414 cookie = ++bus->match_cookie;
2416 r = bus_add_match_internal(bus, match, components, n_components, cookie);
2421 bus->match_callbacks_modified = true;
2422 r = bus_match_add(&bus->match_callbacks, components, n_components, callback, userdata, cookie, NULL);
2424 if (bus->bus_client)
2425 bus_remove_match_internal(bus, match, cookie);
2429 bus_match_parse_free(components, n_components);
2433 _public_ int sd_bus_remove_match(sd_bus *bus,
2435 sd_bus_message_handler_t callback,
2438 struct bus_match_component *components = NULL;
2439 unsigned n_components = 0;
2441 uint64_t cookie = 0;
2443 assert_return(bus, -EINVAL);
2444 assert_return(match, -EINVAL);
2445 assert_return(!bus_pid_changed(bus), -ECHILD);
2447 r = bus_match_parse(match, &components, &n_components);
2451 bus->match_callbacks_modified = true;
2452 r = bus_match_remove(&bus->match_callbacks, components, n_components, callback, userdata, &cookie);
2454 if (bus->bus_client)
2455 q = bus_remove_match_internal(bus, match, cookie);
2457 bus_match_parse_free(components, n_components);
2459 return r < 0 ? r : q;
2462 bool bus_pid_changed(sd_bus *bus) {
2465 /* We don't support people creating a bus connection and
2466 * keeping it around over a fork(). Let's complain. */
2468 return bus->original_pid != getpid();
2471 static int io_callback(sd_event_source *s, int fd, uint32_t revents, void *userdata) {
2472 sd_bus *bus = userdata;
2477 r = sd_bus_process(bus, NULL);
2484 static int time_callback(sd_event_source *s, uint64_t usec, void *userdata) {
2485 sd_bus *bus = userdata;
2490 r = sd_bus_process(bus, NULL);
2497 static int prepare_callback(sd_event_source *s, void *userdata) {
2498 sd_bus *bus = userdata;
2505 e = sd_bus_get_events(bus);
2509 if (bus->output_fd != bus->input_fd) {
2511 r = sd_event_source_set_io_events(bus->input_io_event_source, e & POLLIN);
2515 r = sd_event_source_set_io_events(bus->output_io_event_source, e & POLLOUT);
2519 r = sd_event_source_set_io_events(bus->input_io_event_source, e);
2524 r = sd_bus_get_timeout(bus, &until);
2530 j = sd_event_source_set_time(bus->time_event_source, until);
2535 r = sd_event_source_set_enabled(bus->time_event_source, r > 0);
2542 static int quit_callback(sd_event_source *event, void *userdata) {
2543 sd_bus *bus = userdata;
2552 _public_ int sd_bus_attach_event(sd_bus *bus, sd_event *event, int priority) {
2555 assert_return(bus, -EINVAL);
2556 assert_return(!bus->event, -EBUSY);
2558 assert(!bus->input_io_event_source);
2559 assert(!bus->output_io_event_source);
2560 assert(!bus->time_event_source);
2563 bus->event = sd_event_ref(event);
2565 r = sd_event_default(&bus->event);
2570 r = sd_event_add_io(bus->event, bus->input_fd, 0, io_callback, bus, &bus->input_io_event_source);
2574 r = sd_event_source_set_priority(bus->input_io_event_source, priority);
2578 if (bus->output_fd != bus->input_fd) {
2579 r = sd_event_add_io(bus->event, bus->output_fd, 0, io_callback, bus, &bus->output_io_event_source);
2583 r = sd_event_source_set_priority(bus->output_io_event_source, priority);
2588 r = sd_event_source_set_prepare(bus->input_io_event_source, prepare_callback);
2592 r = sd_event_add_monotonic(bus->event, 0, 0, time_callback, bus, &bus->time_event_source);
2596 r = sd_event_source_set_priority(bus->time_event_source, priority);
2600 r = sd_event_add_quit(bus->event, quit_callback, bus, &bus->quit_event_source);
2607 sd_bus_detach_event(bus);
2611 _public_ int sd_bus_detach_event(sd_bus *bus) {
2612 assert_return(bus, -EINVAL);
2613 assert_return(bus->event, -ENXIO);
2615 if (bus->input_io_event_source) {
2616 sd_event_source_set_enabled(bus->input_io_event_source, SD_EVENT_OFF);
2617 bus->input_io_event_source = sd_event_source_unref(bus->input_io_event_source);
2620 if (bus->output_io_event_source) {
2621 sd_event_source_set_enabled(bus->output_io_event_source, SD_EVENT_OFF);
2622 bus->output_io_event_source = sd_event_source_unref(bus->output_io_event_source);
2625 if (bus->time_event_source) {
2626 sd_event_source_set_enabled(bus->time_event_source, SD_EVENT_OFF);
2627 bus->time_event_source = sd_event_source_unref(bus->time_event_source);
2630 if (bus->quit_event_source) {
2631 sd_event_source_set_enabled(bus->quit_event_source, SD_EVENT_OFF);
2632 bus->quit_event_source = sd_event_source_unref(bus->quit_event_source);
2636 bus->event = sd_event_unref(bus->event);
2641 _public_ sd_event* sd_bus_get_event(sd_bus *bus) {
2642 assert_return(bus, NULL);
2647 _public_ sd_bus_message* sd_bus_get_current(sd_bus *bus) {
2648 assert_return(bus, NULL);
2650 return bus->current;
2653 static int bus_default(int (*bus_open)(sd_bus **), sd_bus **default_bus, sd_bus **ret) {
2658 assert(default_bus);
2661 return !!*default_bus;
2664 *ret = sd_bus_ref(*default_bus);
2672 b->default_bus_ptr = default_bus;
2680 _public_ int sd_bus_default_system(sd_bus **ret) {
2681 static __thread sd_bus *default_system_bus = NULL;
2683 return bus_default(sd_bus_open_system, &default_system_bus, ret);
2686 _public_ int sd_bus_default_user(sd_bus **ret) {
2687 static __thread sd_bus *default_user_bus = NULL;
2689 return bus_default(sd_bus_open_user, &default_user_bus, ret);
2692 _public_ int sd_bus_get_tid(sd_bus *b, pid_t *tid) {
2693 assert_return(b, -EINVAL);
2694 assert_return(tid, -EINVAL);
2695 assert_return(!bus_pid_changed(b), -ECHILD);
2703 return sd_event_get_tid(b->event, tid);
2708 _public_ char *sd_bus_label_escape(const char *s) {
2712 assert_return(s, NULL);
2714 /* Escapes all chars that D-Bus' object path cannot deal
2715 * with. Can be reversed with bus_path_unescape(). We special
2716 * case the empty string. */
2721 r = new(char, strlen(s)*3 + 1);
2725 for (f = s, t = r; *f; f++) {
2727 /* Escape everything that is not a-zA-Z0-9. We also
2728 * escape 0-9 if it's the first character */
2730 if (!(*f >= 'A' && *f <= 'Z') &&
2731 !(*f >= 'a' && *f <= 'z') &&
2732 !(f > s && *f >= '0' && *f <= '9')) {
2734 *(t++) = hexchar(*f >> 4);
2735 *(t++) = hexchar(*f);
2745 _public_ char *sd_bus_label_unescape(const char *f) {
2748 assert_return(f, NULL);
2750 /* Special case for the empty string */
2754 r = new(char, strlen(f) + 1);
2758 for (t = r; *f; f++) {
2763 if ((a = unhexchar(f[1])) < 0 ||
2764 (b = unhexchar(f[2])) < 0) {
2765 /* Invalid escape code, let's take it literal then */
2768 *(t++) = (char) ((a << 4) | b);
2780 _public_ int sd_bus_get_peer_creds(sd_bus *bus, uint64_t mask, sd_bus_creds **ret) {
2785 assert_return(bus, -EINVAL);
2786 assert_return(mask <= _SD_BUS_CREDS_MAX, -ENOTSUP);
2787 assert_return(ret, -EINVAL);
2788 assert_return(BUS_IS_OPEN(bus->state), -ENOTCONN);
2789 assert_return(!bus_pid_changed(bus), -ECHILD);
2790 assert_return(!bus->is_kernel, -ENOTSUP);
2792 if (!bus->ucred_valid && !isempty(bus->label))
2795 c = bus_creds_new();
2799 if (bus->ucred_valid) {
2800 pid = c->pid = bus->ucred.pid;
2801 c->uid = bus->ucred.uid;
2802 c->gid = bus->ucred.gid;
2804 c->mask |= ((SD_BUS_CREDS_UID | SD_BUS_CREDS_PID | SD_BUS_CREDS_GID) & mask) & bus->creds_mask;
2807 if (!isempty(bus->label) && (mask & SD_BUS_CREDS_SELINUX_CONTEXT)) {
2808 c->label = strdup(bus->label);
2810 sd_bus_creds_unref(c);
2814 c->mask |= SD_BUS_CREDS_SELINUX_CONTEXT | bus->creds_mask;
2817 r = bus_creds_add_more(c, mask, pid, 0);