dbus: initial support

Makefile

@@ -303,6 +303,9 @@ endif
 # libaesdec
 SRCS-${CONFIG_SSL} += src/descrambler/libaesdec/libaesdec.c
 
+# DBUS
+SRCS-${CONFIG_DBUS_1}  += src/dbus.c
+
 # File bundles
 SRCS-${CONFIG_BUNDLE}     += bundle.c
 BUNDLES-yes               += docs/html docs/docresources src/webui/static

configure

@@ -38,6 +38,7 @@ OPTIONS=(
   "bundle:no"
   "dvbcsa:no"
   "kqueue:no"
+  "dbus_1:auto"
 )
 
 #
@@ -344,6 +345,18 @@ if enabled v4l; then
   enable mpegps
 fi
 
+#
+# DBus
+#
+if enabled_or_auto dbus_1; then
+  if check_pkg dbus-1; then
+    enable dbus_1
+  elif enabled dbus-1; then
+    die "DBus-1 development support not found (use --disable-dbus-1)"
+  fi
+fi
+
+
 # ###########################################################################
 # Write config
 # ###########################################################################

src/dbus.c

@@ -0,0 +1,365 @@
+/*
+ *  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, &param);
+  reply = dbus_message_new_method_return(msg);
+  dbus_message_iter_init_append(reply, &args);
+  dbus_message_iter_append_basic(&args, DBUS_TYPE_STRING, &param);
+  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);
+}

src/dbus.h

@@ -0,0 +1,52 @@
+/*
+ *  tvheadend, UPnP 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/>.
+ */
+
+#ifndef DBUS_H_
+#define DBUS_H_
+
+#include "build.h"
+#include "htsmsg.h"
+
+#if ENABLE_DBUS_1
+
+void
+dbus_emit_signal(const char *sig_name, htsmsg_t *msg);
+void
+dbus_emit_signal_str(const char *sig_name, const char *value);
+void
+dbus_emit_signal_s64(const char *sig_name, int64_t value);
+
+void dbus_server_init(void);
+void dbus_server_start(void);
+void dbus_server_done(void);
+
+#else
+
+static inline void
+dbus_emit_signal(const char *sig_name, htsmsg_t *msg) { htsmsg_destroy(msg); }
+static inline void
+dbus_emit_signal_str(const char *sig_name, const char *value) { }
+static inline void
+dbus_emit_signal_s64(const char *sig_name, int64_t value) { }
+
+static inline void dbus_server_init(void) { }
+static inline void dbus_server_done(void) { }
+
+#endif
+
+#endif /* DBUS_H_ */

src/main.c

@@ -64,6 +64,7 @@
 #include "lang_codes.h"
 #include "esfilter.h"
 #include "intlconv.h"
+#include "dbus.h"
 #if ENABLE_LIBAV
 #include "libav.h"
 #include "plumbing/transcoding.h"
@@ -767,6 +768,8 @@ main(int argc, char **argv)
    * Initialize subsystems
    */
 
+  dbus_server_init();
+
   intlconv_init();
   
   api_init();
@@ -815,6 +818,8 @@ main(int argc, char **argv)
 
   dvr_init();
 
+  dbus_server_start();
+
   htsp_init(opt_bindaddr);
 
 
@@ -852,6 +857,9 @@ main(int argc, char **argv)
 
   mainloop();
 
+#if ENABLE_DBUS_1
+  tvhftrace("main", dbus_server_done);
+#endif
 #if ENABLE_UPNP
   tvhftrace("main", upnp_server_done);
 #endif
@@ -921,6 +929,10 @@ main(int argc, char **argv)
   }
   /* end of OpenSSL cleanup code */
 
+#if ENABLE_DBUS_1
+  extern void dbus_shutdown(void);
+  dbus_shutdown();
+#endif
   return 0;
 }