--- /dev/null
+/*
+ * DBUS interface
+ * Copyright (C) 2014 Jaroslav Kysela
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU General Public License as published by
+ * the Free Software Foundation, either version 3 of the License, or
+ * (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+ * GNU General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License
+ * along with this program. If not, see <http://www.gnu.org/licenses/>.
+ */
+
+#include <time.h>
+#include <stdio.h>
+#include <stdlib.h>
+#include <assert.h>
+#include <signal.h>
+#include <fcntl.h>
+
+#include <dbus-1.0/dbus/dbus.h>
+
+#include "tvheadend.h"
+#include "tvhpoll.h"
+#include "dbus.h"
+
+
+typedef struct dbus_sig {
+ TAILQ_ENTRY(dbus_sig) link;
+ char *sig_name;
+ htsmsg_t *msg;
+} dbus_sig_t;
+
+
+TAILQ_HEAD(dbus_signal_queue, dbus_sig);
+static struct dbus_signal_queue dbus_signals;
+static th_pipe_t dbus_pipe;
+static pthread_mutex_t dbus_lock;
+static int dbus_running;
+
+/**
+ *
+ */
+void
+dbus_emit_signal(const char *sig_name, htsmsg_t *msg)
+{
+ dbus_sig_t *ds = calloc(1, sizeof(dbus_sig_t));
+
+ if (!dbus_running) {
+ htsmsg_destroy(msg);
+ return;
+ }
+ ds->sig_name = strdup(sig_name);
+ ds->msg = msg;
+ pthread_mutex_lock(&dbus_lock);
+ TAILQ_INSERT_TAIL(&dbus_signals, ds, link);
+ pthread_mutex_unlock(&dbus_lock);
+ write(dbus_pipe.wr, "s", 1); /* do not wait here - no tvh_write() */
+}
+
+void
+dbus_emit_signal_str(const char *sig_name, const char *value)
+{
+ htsmsg_t *l = htsmsg_create_list();
+ htsmsg_add_str(l, NULL, value);
+ dbus_emit_signal(sig_name, l);
+}
+
+void
+dbus_emit_signal_s64(const char *sig_name, int64_t value)
+{
+ htsmsg_t *l = htsmsg_create_list();
+ htsmsg_add_s64(l, NULL, value);
+ dbus_emit_signal(sig_name, l);
+}
+
+/**
+ *
+ */
+static void
+dbus_from_htsmsg(htsmsg_t *msg, DBusMessageIter *args)
+{
+ htsmsg_field_t *f;
+
+ TAILQ_FOREACH(f, &msg->hm_fields, hmf_link) {
+ switch(f->hmf_type) {
+ case HMF_STR:
+ dbus_message_iter_append_basic(args, DBUS_TYPE_STRING, &f->hmf_str);
+ break;
+ case HMF_S64:
+ dbus_message_iter_append_basic(args, DBUS_TYPE_INT64, &f->hmf_s64);
+ break;
+ case HMF_BOOL:
+ dbus_message_iter_append_basic(args, DBUS_TYPE_BOOLEAN, &f->hmf_bool);
+ break;
+ case HMF_DBL:
+ dbus_message_iter_append_basic(args, DBUS_TYPE_DOUBLE, &f->hmf_dbl);
+ break;
+ default:
+ assert(0);
+ }
+ }
+}
+
+/**
+ *
+ */
+static DBusConnection *
+dbus_create_session(const char *name)
+{
+ DBusConnection *conn;
+ DBusError err;
+ int ret;
+
+ dbus_error_init(&err);
+
+ conn = dbus_bus_get_private(DBUS_BUS_SESSION, &err);
+ if (dbus_error_is_set(&err)) {
+ tvherror("dbus", "Connection error: %s", err.message);
+ dbus_error_free(&err);
+ return NULL;
+ }
+
+ ret = dbus_bus_request_name(conn, name, DBUS_NAME_FLAG_REPLACE_EXISTING, &err);
+ if (dbus_error_is_set(&err)) {
+ tvherror("dbus", "Name error: %s", err.message);
+ dbus_error_free(&err);
+ dbus_connection_unref(conn);
+ return NULL;
+ }
+ if (DBUS_REQUEST_NAME_REPLY_PRIMARY_OWNER != ret) {
+ tvherror("dbus", "Not primary owner");
+ dbus_connection_unref(conn);
+ return NULL;
+ }
+ return conn;
+}
+
+/**
+ * Send a signal
+ */
+static int
+dbus_send_signal(DBusConnection *conn, const char *obj_name,
+ const char *if_name, const char *sig_name,
+ htsmsg_t *value)
+{
+ DBusMessage *msg;
+ DBusMessageIter args;
+
+ msg = dbus_message_new_signal(obj_name, if_name, sig_name);
+ if (msg == NULL) {
+ tvherror("dbus", "Unable to create signal %s %s %s",
+ obj_name, if_name, sig_name);
+ dbus_connection_unref(conn);
+ return -1;
+ }
+ dbus_message_iter_init_append(msg, &args);
+ dbus_from_htsmsg(value, &args);
+ if (!dbus_connection_send(conn, msg, NULL)) {
+ tvherror("dbus", "Unable to send signal %s %s %s",
+ obj_name, if_name, sig_name);
+ dbus_message_unref(msg);
+ dbus_connection_unref(conn);
+ return -1;
+ }
+ dbus_connection_flush(conn);
+ dbus_message_unref(msg);
+ return 0;
+}
+
+/**
+ * Simple ping (alive) RPC, just return the string
+ */
+static void
+dbus_reply_to_ping(DBusMessage *msg, DBusConnection *conn)
+{
+ DBusMessageIter args;
+ DBusMessage *reply;
+ char *param;
+
+ if (!dbus_message_iter_init(msg, &args))
+ return;
+ if (DBUS_TYPE_STRING != dbus_message_iter_get_arg_type(&args))
+ return;
+ dbus_message_iter_get_basic(&args, ¶m);
+ reply = dbus_message_new_method_return(msg);
+ dbus_message_iter_init_append(reply, &args);
+ dbus_message_iter_append_basic(&args, DBUS_TYPE_STRING, ¶m);
+ dbus_connection_send(conn, reply, NULL);
+ dbus_connection_flush(conn);
+ dbus_message_unref(reply);
+}
+
+/**
+ *
+ */
+#if 0
+static void
+dbus_server_close_hack(DBusConnection *conn,
+ DBusDispatchStatus new_status,
+ void *data)
+{
+ /* buggy refcounting in libdbus */
+ dbus_connection_unref(conn);
+}
+#endif
+
+static void
+dbus_connection_safe_close(DBusConnection *conn)
+{
+ dbus_connection_flush(conn);
+
+//#if ENABLE_TRACE /* a little bit wrong condition */
+ /* ugly bug here, note that the fixed version of dbus will broke this */
+// dbus_connection_set_dispatch_status_function(conn, dbus_server_close_hack, NULL, NULL);
+//#endif
+
+ dbus_connection_close(conn);
+ dbus_connection_unref(conn);
+}
+
+/**
+ *
+ */
+static void
+dbus_flush_queue(DBusConnection *conn)
+{
+ dbus_sig_t *ds;
+
+ while (1) {
+ pthread_mutex_lock(&dbus_lock);
+ ds = TAILQ_FIRST(&dbus_signals);
+ if (ds)
+ TAILQ_REMOVE(&dbus_signals, ds, link);
+ pthread_mutex_unlock(&dbus_lock);
+
+ if (ds == NULL)
+ break;
+
+ if (conn)
+ dbus_send_signal(conn, "/", "org.tvheadend.notify", ds->sig_name, ds->msg);
+
+ htsmsg_destroy(ds->msg);
+ free(ds->sig_name);
+ free(ds);
+ }
+ if (conn)
+ dbus_connection_flush(conn);
+}
+
+/**
+ * Listen for remote requests
+ */
+static void *
+dbus_server_thread(void *aux)
+{
+ DBusMessage *msg;
+ DBusConnection *conn, *notify;
+ tvhpoll_t *poll;
+ tvhpoll_event_t ev;
+ int n;
+ uint8_t c;
+
+ conn = dbus_create_session("org.tvheadend.server");
+ if (conn == NULL)
+ return NULL;
+
+ notify = dbus_create_session("org.tvheadend.notify");
+ if (notify == NULL) {
+ dbus_connection_safe_close(conn);
+ return NULL;
+ }
+
+ poll = tvhpoll_create(2);
+ memset(&ev, 0, sizeof(ev));
+ ev.fd = dbus_pipe.rd;
+ ev.events = TVHPOLL_IN;
+ ev.data.ptr = &dbus_pipe;
+ tvhpoll_add(poll, &ev, 1);
+ memset(&ev, 0, sizeof(ev));
+ if (!dbus_connection_get_unix_fd(conn, &ev.fd)) {
+ tvhpoll_destroy(poll);
+ dbus_connection_safe_close(notify);
+ dbus_connection_safe_close(conn);
+ return NULL;
+ }
+ ev.events = TVHPOLL_IN;
+ ev.data.ptr = conn;
+ tvhpoll_add(poll, &ev, 1);
+
+ while (dbus_running) {
+
+ n = tvhpoll_wait(poll, &ev, 1, -1);
+ if (n < 0) {
+ if (dbus_running && !ERRNO_AGAIN(errno))
+ tvherror("dbus", "tvhpoll_wait() error");
+ } else if (n == 0) {
+ continue;
+ }
+
+ if (ev.data.ptr == &dbus_pipe) {
+ if (read(dbus_pipe.rd, &c, 1) == 1) {
+ if (c == 's')
+ dbus_flush_queue(notify);
+ else
+ break; /* end-of-task */
+ }
+ continue;
+ }
+
+ dbus_connection_read_write(conn, 0);
+ msg = dbus_connection_pop_message(conn);
+ if (msg == NULL)
+ continue;
+
+ if (dbus_message_is_method_call(msg, "org.tvheadend", "ping"))
+ dbus_reply_to_ping(msg, conn);
+
+ dbus_message_unref(msg);
+ }
+
+ dbus_connection_safe_close(conn);
+ dbus_flush_queue(notify);
+ dbus_connection_safe_close(notify);
+ tvhpoll_destroy(poll);
+ return NULL;
+}
+
+/**
+ *
+ */
+pthread_t dbus_tid;
+
+void
+dbus_server_init(void)
+{
+ pthread_mutex_init(&dbus_lock, NULL);
+ TAILQ_INIT(&dbus_signals);
+ tvh_pipe(O_NONBLOCK, &dbus_pipe);
+ dbus_threads_init_default();
+ dbus_running = 1;
+ dbus_emit_signal_str("start", tvheadend_version);
+}
+
+void
+dbus_server_start(void)
+{
+ tvhthread_create(&dbus_tid, NULL, dbus_server_thread, NULL);
+}
+
+void
+dbus_server_done(void)
+{
+ dbus_emit_signal_str("stop", "bye");
+ dbus_running = 0;
+ tvh_write(dbus_pipe.wr, "", 1);
+ pthread_kill(dbus_tid, SIGTERM);
+ pthread_join(dbus_tid, NULL);
+ dbus_flush_queue(NULL);
+}
projects / thirdparty / tvheadend.git / commitdiff
