make Warning, Error and Update send unicast dbus messages

Signed-off-by: Denys Vlasenko <vda.linux@googlemail.com>

10 files changed, 101 insertions, 91 deletions

diff --git a/lib/CommLayer/CommLayerInner.cpp b/lib/CommLayer/CommLayerInner.cpp
index ec4dc904..5450b41c 100644
--- a/lib/CommLayer/CommLayerInner.cpp
+++ b/lib/CommLayer/CommLayerInner.cpp
@@ -1,50 +1,63 @@
-#include <pthread.h> /* pthread_self() */
+#include <pthread.h>
+#include <map>
 #include "abrtlib.h"
 #include "CommLayerInner.h"
 
 static CObserver *s_pObs;
-static pthread_t s_main_id;
+
+typedef std::map<uint64_t, std::string> map_uint_str_t;
+static map_uint_str_t s_mapClientID;
+static pthread_mutex_t s_map_mutex;
+static bool s_map_mutex_inited;
 
 void init_daemon_logging(CObserver *pObs)
 {
     s_pObs = pObs;
-    s_main_id = pthread_self();
+    if (!s_map_mutex_inited)
+    {
+        pthread_mutex_init(&s_map_mutex, NULL);
+        s_map_mutex_inited = true;
+    }
+}
+
+void set_client_name(const char* name)
+{
+    uint64_t key = uint64_t(pthread_self());
+
+    pthread_mutex_lock(&s_map_mutex);
+    if (!name)
+        s_mapClientID.erase(key);
+    else
+        s_mapClientID[key] = name;
+    pthread_mutex_unlock(&s_map_mutex);
 }
 
 void warn_client(const std::string& pMessage)
 {
     if (!s_pObs)
         return;
-    pthread_t self = pthread_self();
-    if (self != s_main_id)
-    {
-        s_pObs->Warning(pMessage,(uint64_t)self);
-//log("w: '%s'", s.c_str());
-    }
-    else
-    {
-        s_pObs->Warning(pMessage);
-// debug: this should not happen - if it is, we are trying to log to a client
-// but we have no job id!
-log("W: '%s'", pMessage.c_str());
-    }
+
+    uint64_t key = uint64_t(pthread_self());
+
+    pthread_mutex_lock(&s_map_mutex);
+    map_uint_str_t::const_iterator ki = s_mapClientID.find(key);
+    const char* peer = (ki != s_mapClientID.end() ? ki->second.c_str() : NULL);
+    pthread_mutex_unlock(&s_map_mutex);
+
+    s_pObs->Warning(pMessage, peer, key);
 }
 
 void update_client(const std::string& pMessage)
 {
     if (!s_pObs)
         return;
-    pthread_t self = pthread_self();
-    if (self != s_main_id)
-    {
-        s_pObs->Status(pMessage, (uint64_t)self);
-//log("u: '%s'", s.c_str());
-    }
-    else
-    {
-        s_pObs->Status(pMessage);
-// debug: this should not happen - if it is, we are trying to log to a client
-// but we have no job id!
-log("U: '%s'", pMessage.c_str());
-    }
+
+    uint64_t key = uint64_t(pthread_self());
+
+    pthread_mutex_lock(&s_map_mutex);
+    map_uint_str_t::const_iterator ki = s_mapClientID.find(key);
+    const char* peer = (ki != s_mapClientID.end() ? ki->second.c_str() : NULL);
+    pthread_mutex_unlock(&s_map_mutex);
+
+    s_pObs->Status(pMessage, peer, key);
 }
diff --git a/lib/CommLayer/CommLayerInner.h b/lib/CommLayer/CommLayerInner.h
index 2b4f63a3..d161cfc7 100644
--- a/lib/CommLayer/CommLayerInner.h
+++ b/lib/CommLayer/CommLayerInner.h
@@ -5,6 +5,10 @@
 
 void init_daemon_logging(CObserver *pObs);
 
+/*
+ * Set client's name (dbus ID). NULL unsets it.
+ */
+void set_client_name(const char* name);
 /* Ask a client to warn the user about a non-fatal, but unexpected condition.
  * In GUI, it will usually be presented as a popup message.
  */
diff --git a/lib/CommLayer/Observer.h b/lib/CommLayer/Observer.h
index 5c983949..421dc0cc 100644
--- a/lib/CommLayer/Observer.h
+++ b/lib/CommLayer/Observer.h
@@ -9,8 +9,8 @@
 class CObserver {
     public:
         virtual ~CObserver() {}
-        virtual void Status(const std::string& pMessage, uint64_t pDest=0) = 0;
-        virtual void Warning(const std::string& pMessage, uint64_t pDest=0) = 0;
+        virtual void Status(const std::string& pMessage, const char* peer, uint64_t pDest) = 0;
+        virtual void Warning(const std::string& pMessage, const char* peer, uint64_t pDest) = 0;
 };
 
 #endif
diff --git a/src/Daemon/CommLayerServer.h b/src/Daemon/CommLayerServer.h
index d2d2c4ec..0b1027ac 100644
--- a/src/Daemon/CommLayerServer.h
+++ b/src/Daemon/CommLayerServer.h
@@ -17,10 +17,9 @@ class CCommLayerServer {
         virtual void JobDone(const char* pDest, const char* pUUID) = 0;
         virtual void JobStarted(const char* pDest) {};
 
-        virtual void Error(const std::string& arg1) {}
-        virtual void Update(const std::string& pMessage, uint64_t pJobID) {};
-        virtual void Warning(const std::string& pMessage) {};
-        virtual void Warning(const std::string& pMessage, uint64_t pJobID) {};
+        virtual void Error(const std::string& pMessage, const char* peer) {}
+        virtual void Update(const std::string& pMessage, const char* peer, uint64_t pJobID) {};
+        virtual void Warning(const std::string& pMessage, const char* peer, uint64_t pJobID) {};
 };
 
 #endif
diff --git a/src/Daemon/CommLayerServerDBus.cpp b/src/Daemon/CommLayerServerDBus.cpp
index d426338b..c6908f5f 100644
--- a/src/Daemon/CommLayerServerDBus.cpp
+++ b/src/Daemon/CommLayerServerDBus.cpp
@@ -336,12 +336,15 @@ static inline int load_val(DBusMessageIter* iter, std::map<K,V>& val)  { return
  */
 
 /* helpers */
-static DBusMessage* new_signal_msg(const char* member)
+static DBusMessage* new_signal_msg(const char* member, const char* peer = NULL)
 {
     /* path, interface, member name */
     DBusMessage* msg = dbus_message_new_signal(CC_DBUS_PATH, CC_DBUS_IFACE, member);
     if (!msg)
         die_out_of_memory();
+    /* Send unicast dbus signal if peer is known */
+    if (peer && !dbus_message_set_destination(msg, peer))
+        die_out_of_memory();
     return msg;
 }
 static void send_flush_and_unref(DBusMessage* msg)
@@ -367,74 +370,57 @@ void CCommLayerServerDBus::Crash(const std::string& progname, const std::string&
     send_flush_and_unref(msg);
 }
 
-void CCommLayerServerDBus::JobStarted(const char* pDest)
+void CCommLayerServerDBus::JobStarted(const char* peer)
 {
-    DBusMessage* msg = new_signal_msg("JobStarted");
-    /* send unicast dbus signal */
-    if (!dbus_message_set_destination(msg, pDest))
-        die_out_of_memory();
+    DBusMessage* msg = new_signal_msg("JobStarted", peer);
     uint64_t nJobID = uint64_t(pthread_self());
     dbus_message_append_args(msg,
-            DBUS_TYPE_STRING, &pDest, /* TODO: redundant parameter, remove from API */
+            DBUS_TYPE_STRING, &peer, /* TODO: redundant parameter, remove from API */
             DBUS_TYPE_UINT64, &nJobID, /* TODO: redundant parameter, remove from API */
             DBUS_TYPE_INVALID);
-    VERB2 log("Sending signal JobStarted('%s',%llx)", pDest, (unsigned long long)nJobID);
+    VERB2 log("Sending signal JobStarted('%s',%llx)", peer, (unsigned long long)nJobID);
     send_flush_and_unref(msg);
 }
 
-void CCommLayerServerDBus::JobDone(const char* pDest, const char* pUUID)
+void CCommLayerServerDBus::JobDone(const char* peer, const char* pUUID)
 {
-    DBusMessage* msg = new_signal_msg("JobDone");
-    /* send unicast dbus signal */
-    if (!dbus_message_set_destination(msg, pDest))
-        die_out_of_memory();
+    DBusMessage* msg = new_signal_msg("JobDone", peer);
     dbus_message_append_args(msg,
-            DBUS_TYPE_STRING, &pDest, /* TODO: redundant parameter, remove from API */
+            DBUS_TYPE_STRING, &peer, /* TODO: redundant parameter, remove from API */
             DBUS_TYPE_STRING, &pUUID, /* TODO: redundant parameter, remove from API */
             DBUS_TYPE_INVALID);
-    VERB2 log("Sending signal JobDone('%s','%s')", pDest, pUUID);
+    VERB2 log("Sending signal JobDone('%s','%s')", peer, pUUID);
     send_flush_and_unref(msg);
 }
 
-void CCommLayerServerDBus::Error(const std::string& arg1)
+void CCommLayerServerDBus::Error(const std::string& pMessage, const char* peer)
 {
-    DBusMessage* msg = new_signal_msg("Error");
-    const char* c_arg1 = arg1.c_str();
-    dbus_message_append_args(msg,
-            DBUS_TYPE_STRING, &c_arg1,
-            DBUS_TYPE_INVALID);
-    send_flush_and_unref(msg);
-}
-
-void CCommLayerServerDBus::Update(const std::string& pMessage, uint64_t job_id)
-{
-    DBusMessage* msg = new_signal_msg("Update");
+    DBusMessage* msg = new_signal_msg("Error", peer);
     const char* c_message = pMessage.c_str();
     dbus_message_append_args(msg,
             DBUS_TYPE_STRING, &c_message,
-            DBUS_TYPE_UINT64, &job_id,
             DBUS_TYPE_INVALID);
     send_flush_and_unref(msg);
 }
 
-/* TODO: one Warning()? */
-void CCommLayerServerDBus::Warning(const std::string& pMessage)
+void CCommLayerServerDBus::Update(const std::string& pMessage, const char* peer, uint64_t job_id)
 {
-    DBusMessage* msg = new_signal_msg("Warning");
+    DBusMessage* msg = new_signal_msg("Update", peer);
     const char* c_message = pMessage.c_str();
     dbus_message_append_args(msg,
             DBUS_TYPE_STRING, &c_message,
+            DBUS_TYPE_UINT64, &job_id, /* TODO: redundant parameter, remove from API */
             DBUS_TYPE_INVALID);
     send_flush_and_unref(msg);
 }
 
-void CCommLayerServerDBus::Warning(const std::string& pMessage, uint64_t job_id)
+void CCommLayerServerDBus::Warning(const std::string& pMessage, const char* peer, uint64_t job_id)
 {
-    DBusMessage* msg = new_signal_msg("Warning");
+    DBusMessage* msg = new_signal_msg("Warning", peer);
     const char* c_message = pMessage.c_str();
     dbus_message_append_args(msg,
             DBUS_TYPE_STRING, &c_message,
-            DBUS_TYPE_UINT64, &job_id,
+            DBUS_TYPE_UINT64, &job_id, /* TODO: redundant parameter, remove from API */
             DBUS_TYPE_INVALID);
     send_flush_and_unref(msg);
 }
@@ -885,6 +871,8 @@ static DBusHandlerResult message_received(DBusConnection *conn, DBusMessage *msg
     const char* member = dbus_message_get_member(msg);
     log("%s(method:'%s')", __func__, member);
 
+    set_client_name(dbus_message_get_sender(msg));
+
     DBusMessage* reply = dbus_message_new_method_return(msg);
     int r = -1;
     if (strcmp(member, "GetCrashInfos") == 0)
@@ -931,6 +919,8 @@ static DBusHandlerResult message_received(DBusConnection *conn, DBusMessage *msg
         }
     }
 
+    set_client_name(NULL);
+
     return DBUS_HANDLER_RESULT_HANDLED;
 }
 /* Callback: "DBusObjectPathVTable is unregistered (or its connection is freed)" */
diff --git a/src/Daemon/CommLayerServerDBus.h b/src/Daemon/CommLayerServerDBus.h
index 979823d3..3a8822de 100644
--- a/src/Daemon/CommLayerServerDBus.h
+++ b/src/Daemon/CommLayerServerDBus.h
@@ -15,10 +15,9 @@ class CCommLayerServerDBus
         virtual void JobStarted(const char* pDest);
         virtual void JobDone(const char* pDest, const char* pUUID);
 
-        virtual void Error(const std::string& arg1);
-        virtual void Update(const std::string& pMessage, uint64_t pJobID);
-        virtual void Warning(const std::string& pMessage);
-        virtual void Warning(const std::string& pMessage, uint64_t pJobID);
+        virtual void Error(const std::string& pMessage, const char* peer);
+        virtual void Update(const std::string& pMessage, const char* peer, uint64_t pJobID);
+        virtual void Warning(const std::string& pMessage, const char* peer, uint64_t pJobID);
 };
 
 #endif
diff --git a/src/Daemon/CommLayerServerSocket.cpp b/src/Daemon/CommLayerServerSocket.cpp
index a3240a8d..8f1fd4fe 100644
--- a/src/Daemon/CommLayerServerSocket.cpp
+++ b/src/Daemon/CommLayerServerSocket.cpp
@@ -246,7 +246,7 @@ void CCommLayerServerSocket::Crash(const std::string& arg1)
     //Send("(CRASH)New Crash Detected: " + arg1);
 }
 
-void CCommLayerServerSocket::Error(const std::string& arg1)
+void CCommLayerServerSocket::Error(const std::string& arg1, const char* peer)
 {
     //Send("(ERROR)Error: " + arg1);
 }
diff --git a/src/Daemon/CommLayerServerSocket.h b/src/Daemon/CommLayerServerSocket.h
index 9d4f21cf..0ee47014 100644
--- a/src/Daemon/CommLayerServerSocket.h
+++ b/src/Daemon/CommLayerServerSocket.h
@@ -33,5 +33,5 @@ class CCommLayerServerSocket : public CCommLayerServer
         virtual void Crash(const std::string& arg1);
         virtual void JobStarted(const char* pDest) {};
 
-        virtual void Error(const std::string& arg1);
+        virtual void Error(const std::string& arg1, const char* peer);
 };
diff --git a/src/Daemon/CrashWatcher.cpp b/src/Daemon/CrashWatcher.cpp
index 5ad419b4..6c4d6545 100644
--- a/src/Daemon/CrashWatcher.cpp
+++ b/src/Daemon/CrashWatcher.cpp
@@ -24,19 +24,18 @@
 #include "ABRTException.h"
 #include "CrashWatcher.h"
 
-void CCrashWatcher::Status(const std::string& pMessage, uint64_t pJobID)
+void CCrashWatcher::Status(const std::string& pMessage, const char* peer, uint64_t pJobID)
 {
-    log("Update: %s", pMessage.c_str());
-    //FIXME: send updates only to job owner
+    VERB1 log("Update('%s'): %s", peer, pMessage.c_str());
     if (g_pCommLayer != NULL)
-        g_pCommLayer->Update(pMessage, pJobID);
+        g_pCommLayer->Update(pMessage, peer, pJobID);
 }
 
-void CCrashWatcher::Warning(const std::string& pMessage, uint64_t pJobID)
+void CCrashWatcher::Warning(const std::string& pMessage, const char* peer, uint64_t pJobID)
 {
-    log("Warning: %s", pMessage.c_str());
+    VERB1 log("Warning('%s'): %s", peer, pMessage.c_str());
     if (g_pCommLayer != NULL)
-        g_pCommLayer->Warning(pMessage, pJobID);
+        g_pCommLayer->Warning(pMessage, peer, pJobID);
 }
 
 CCrashWatcher::CCrashWatcher()
@@ -147,38 +146,44 @@ typedef struct thread_data_t {
     pthread_t thread_id;
     char* UUID;
     char* UID;
-    char* dest;
+    char* peer;
 } thread_data_t;
 static void* create_report(void* arg)
 {
     thread_data_t *thread_data = (thread_data_t *) arg;
 
-    g_pCommLayer->JobStarted(thread_data->dest);
+    g_pCommLayer->JobStarted(thread_data->peer);
+
+    /* Client name is per-thread, need to set it */
+    set_client_name(thread_data->peer);
 
     try
     {
         /* "GetJobResult" is a bit of a misnomer */
         log("Creating report...");
         map_crash_info_t crashReport = GetJobResult(thread_data->UUID, thread_data->UID);
-        g_pCommLayer->JobDone(thread_data->dest, thread_data->UUID);
+        g_pCommLayer->JobDone(thread_data->peer, thread_data->UUID);
     }
     catch (CABRTException& e)
     {
         if (e.type() == EXCEP_FATAL)
         {
+	    set_client_name(NULL);
             /* free strduped strings */
             free(thread_data->UUID);
             free(thread_data->UID);
-            free(thread_data->dest);
+            free(thread_data->peer);
             free(thread_data);
             throw e;
         }
         warn_client(e.what());
     }
+    set_client_name(NULL);
+
     /* free strduped strings */
     free(thread_data->UUID);
     free(thread_data->UID);
-    free(thread_data->dest);
+    free(thread_data->peer);
     free(thread_data);
 
     /* Bogus value. pthreads require us to return void* */
@@ -189,7 +194,7 @@ int CreateReportThread(const char* pUUID, const char* pUID, const char* pSender)
     thread_data_t *thread_data = (thread_data_t *)xzalloc(sizeof(thread_data_t));
     thread_data->UUID = xstrdup(pUUID);
     thread_data->UID = xstrdup(pUID);
-    thread_data->dest = xstrdup(pSender);
+    thread_data->peer = xstrdup(pSender);
 //TODO: do we need this?
 //pthread_attr_t attr;
 //pthread_attr_init(&attr);
@@ -199,7 +204,7 @@ int CreateReportThread(const char* pUUID, const char* pUID, const char* pSender)
     {
         free(thread_data->UUID);
         free(thread_data->UID);
-        free(thread_data->dest);
+        free(thread_data->peer);
         free(thread_data);
         /* The only reason this may happen is system-wide resource starvation,
          * or ulimit is exceeded (someone floods us with CreateReport() dbus calls?)
diff --git a/src/Daemon/CrashWatcher.h b/src/Daemon/CrashWatcher.h
index bd98d826..00bb4d51 100644
--- a/src/Daemon/CrashWatcher.h
+++ b/src/Daemon/CrashWatcher.h
@@ -44,8 +44,8 @@ class CCrashWatcher
 
     public:
         /* Observer methods */
-        virtual void Status(const std::string& pMessage, uint64_t pJobID=0);
-        virtual void Warning(const std::string& pMessage, uint64_t pJobID=0);
+        virtual void Status(const std::string& pMessage, const char* peer, uint64_t pJobID);
+        virtual void Warning(const std::string& pMessage, const char* peer, uint64_t pJobID);
 };
 
 vector_crash_infos_t GetCrashInfos(const std::string &pUID);