CommonAPI-D-Bus 3.1.53.1.5

6 files changed, 89 insertions, 23 deletions

diff --git a/src/CommonAPI/DBus/DBusConnection.cpp b/src/CommonAPI/DBus/DBusConnection.cpp
index c5ab406..47751cc 100644
--- a/src/CommonAPI/DBus/DBusConnection.cpp
+++ b/src/CommonAPI/DBus/DBusConnection.cpp
@@ -53,7 +53,8 @@ void DBusConnection::dispatch() {
     loop_->run();
 }
 
-DBusConnection::DBusConnection(DBusType_t busType) :
+DBusConnection::DBusConnection(DBusType_t busType,
+                               const ConnectionId_t& _connectionId) :
                 dispatchThread_(NULL),
                 stopDispatching_(false),
                 dispatchSource_(),
@@ -66,12 +67,14 @@ DBusConnection::DBusConnection(DBusType_t busType) :
                 dbusObjectMessageHandler_(),
                 connectionNameCount_(),
                 enforcerThread_(NULL),
-                enforcerThreadCancelled_(false) {
+                enforcerThreadCancelled_(false),
+                connectionId_(_connectionId) {
 
     dbus_threads_init_default();
 }
 
-DBusConnection::DBusConnection(::DBusConnection *_connection) :
+DBusConnection::DBusConnection(::DBusConnection *_connection,
+                               const ConnectionId_t& _connectionId) :
                 dispatchThread_(NULL),
                 stopDispatching_(false),
                 dispatchSource_(),
@@ -84,7 +87,8 @@ DBusConnection::DBusConnection(::DBusConnection *_connection) :
                 dbusObjectMessageHandler_(),
                 connectionNameCount_(),
                 enforcerThread_(NULL),
-                enforcerThreadCancelled_(false) {
+                enforcerThreadCancelled_(false),
+                connectionId_(_connectionId) {
 
     dbus_threads_init_default();
 }
@@ -805,6 +809,20 @@ bool DBusConnection::hasDispatchThread() {
     return (dispatchThread_ != NULL);
 }
 
+const ConnectionId_t& DBusConnection::getConnectionId() const {
+    return connectionId_;
+}
+
+bool DBusConnection::sendPendingSelectiveSubscription(DBusProxy* proxy, std::string methodName) {
+    bool subscriptionAccepted;
+    CommonAPI::CallStatus callStatus;
+        DBusProxyHelper<CommonAPI::DBus::DBusSerializableArguments<>,
+                        CommonAPI::DBus::DBusSerializableArguments<bool>>::callMethodWithReply(
+                                *proxy, methodName.c_str(), "", &CommonAPI::DBus::defaultCallInfo, callStatus, subscriptionAccepted);
+
+    return subscriptionAccepted;
+}
+
 DBusProxyConnection::DBusSignalHandlerToken DBusConnection::subscribeForSelectiveBroadcast(
                     bool& subscriptionAccepted,
                     const std::string& objectPath,
@@ -824,7 +842,7 @@ DBusProxyConnection::DBusSignalHandlerToken DBusConnection::subscribeForSelectiv
                     *callingProxy, methodName.c_str(), "", &CommonAPI::DBus::defaultCallInfo, callStatus, subscriptionAccepted);
 
     DBusProxyConnection::DBusSignalHandlerToken subscriptionToken;
-    if (callStatus == CommonAPI::CallStatus::SUCCESS && subscriptionAccepted) {
+    if ((callStatus == CommonAPI::CallStatus::SUCCESS && subscriptionAccepted) || !callingProxy->isAvailable()) {
         subscriptionToken = addSignalMemberHandler(
                                 objectPath,
                                 interfaceName,
@@ -1386,12 +1404,12 @@ void notifyDBusOMSignalHandlers(DBusSignalHandlersTable& dbusSignalHandlerstable
     return dbusConnection->onLibdbusObjectPathMessage(libdbusMessage);
 }
 
-std::shared_ptr<DBusConnection> DBusConnection::getBus(const DBusType_t &_type) {
-    return std::make_shared<DBusConnection>(_type);
+std::shared_ptr<DBusConnection> DBusConnection::getBus(const DBusType_t &_type, const ConnectionId_t& _connectionId) {
+    return std::make_shared<DBusConnection>(_type, _connectionId);
 }
 
-std::shared_ptr<DBusConnection> DBusConnection::wrap(::DBusConnection *_connection) {
-    return std::make_shared<DBusConnection>(_connection);
+std::shared_ptr<DBusConnection> DBusConnection::wrap(::DBusConnection *_connection, const ConnectionId_t& _connectionId) {
+    return std::make_shared<DBusConnection>(_connection, _connectionId);
 }
 
 } // namespace DBus
diff --git a/src/CommonAPI/DBus/DBusFactory.cpp b/src/CommonAPI/DBus/DBusFactory.cpp
index 6371da9..40da41f 100644
--- a/src/CommonAPI/DBus/DBusFactory.cpp
+++ b/src/CommonAPI/DBus/DBusFactory.cpp
@@ -230,7 +230,7 @@ Factory::getConnection(const ConnectionId_t &_connectionId) {
     // No connection found, lets create and initialize one
     DBusType_t dbusType = DBusAddressTranslator::get()->getDBusBusType(_connectionId);
     std::shared_ptr<DBusConnection> itsConnection
-        = std::make_shared<DBusConnection>(dbusType);
+        = std::make_shared<DBusConnection>(dbusType, _connectionId);
     connections_.insert({ _connectionId, itsConnection });
 
     itsConnection->connect(true);
@@ -249,7 +249,7 @@ Factory::getConnection(std::shared_ptr<MainLoopContext> _context) {
 
     // No connection found, lets create and initialize one
     std::shared_ptr<DBusConnection> itsConnection
-        = std::make_shared<DBusConnection>(DBusType_t::SESSION);
+        = std::make_shared<DBusConnection>(DBusType_t::SESSION, _context->getName());
     contextConnections_.insert({ _context.get(), itsConnection } );
 
     itsConnection->connect(false);
diff --git a/src/CommonAPI/DBus/DBusMainLoop.cpp b/src/CommonAPI/DBus/DBusMainLoop.cpp
index f674387..a46072f 100755
--- a/src/CommonAPI/DBus/DBusMainLoop.cpp
+++ b/src/CommonAPI/DBus/DBusMainLoop.cpp
@@ -6,6 +6,7 @@
 #ifdef WIN32
 #include <WinSock2.h>
 #include <ws2tcpip.h>
+#include <atomic>
 #define DEFAULT_BUFLEN 512
 #else
 #include <poll.h>
@@ -558,7 +559,11 @@ void DBusMainLoop::wakeup() {
 
     int iResult = send(sendFd_.fd, sendbuf, (int)strlen(sendbuf), 0);
     if (iResult == SOCKET_ERROR) {
-        printf("send failed with error: %d\n", WSAGetLastError());
+        int error = WSAGetLastError();
+
+        if (error != WSANOTINITIALISED) {
+            printf("send failed with error: %d\n", error);
+        }
     }
 #else
     int64_t wake = 1;
@@ -644,7 +649,9 @@ void DBusMainLoop::registerWatch(Watch* watch,
 
     registerFileDescriptor(fdToRegister);
     std::mutex* mtx = new std::mutex;
-#ifndef WIN32
+#ifdef WIN32
+    std::atomic_signal_fence(std::memory_order_acq_rel);
+#else
     asm volatile ("":::"memory");
 #endif
     WatchToDispatchStruct* watchStruct = new WatchToDispatchStruct(fdToRegister.fd, watch, mtx, false, false);
diff --git a/src/CommonAPI/DBus/DBusProxy.cpp b/src/CommonAPI/DBus/DBusProxy.cpp
index 19f2eb0..e92d90c 100644
--- a/src/CommonAPI/DBus/DBusProxy.cpp
+++ b/src/CommonAPI/DBus/DBusProxy.cpp
@@ -131,6 +131,16 @@ void DBusProxy::onDBusServiceInstanceStatus(const AvailabilityStatus& availabili
                             &signalMemberHandlerInfo_);
                 }
             }
+            {
+                std::lock_guard < std::mutex > queueLock(selectiveBroadcastHandlersMutex_);
+                for (auto selectiveBroadcasts : selectiveBroadcastHandlers) {
+                    std::string methodName = "subscribeFor" + selectiveBroadcasts.first + "Selective";
+                    bool subscriptionAccepted = connection_->sendPendingSelectiveSubscription(this, methodName);
+                    if (!subscriptionAccepted) {
+                        selectiveBroadcasts.second->onError(CommonAPI::CallStatus::SUBSCRIPTION_REFUSED);
+                    }
+                }
+            }
         } else {
             std::lock_guard < std::mutex > queueLock(signalMemberHandlerQueueMutex_);
 
@@ -163,7 +173,8 @@ DBusProxyConnection::DBusSignalHandlerToken DBusProxy::subscribeForSelectiveBroa
                                                       const std::string& interfaceMemberSignature,
                                                       DBusProxyConnection::DBusSignalHandler* dbusSignalHandler) {
 
-    return getDBusConnection()->subscribeForSelectiveBroadcast(
+    DBusProxyConnection::DBusSignalHandlerToken token =
+            getDBusConnection()->subscribeForSelectiveBroadcast(
                     subscriptionAccepted,
                     objectPath,
                     interfaceName,
@@ -171,12 +182,30 @@ DBusProxyConnection::DBusSignalHandlerToken DBusProxy::subscribeForSelectiveBroa
                     interfaceMemberSignature,
                     dbusSignalHandler,
                     this);
+
+    if (!isAvailable()) {
+        subscriptionAccepted = true;
+    }
+    if (subscriptionAccepted) {
+        std::lock_guard < std::mutex > queueLock(selectiveBroadcastHandlersMutex_);
+        selectiveBroadcastHandlers[interfaceMemberName] = dbusSignalHandler;
+    }
+
+    return token;
 }
 
 void DBusProxy::unsubscribeFromSelectiveBroadcast(const std::string& eventName,
                                                  DBusProxyConnection::DBusSignalHandlerToken subscription,
                                                  const DBusProxyConnection::DBusSignalHandler* dbusSignalHandler) {
+
     getDBusConnection()->unsubscribeFromSelectiveBroadcast(eventName, subscription, this, dbusSignalHandler);
+
+    std::lock_guard < std::mutex > queueLock(selectiveBroadcastHandlersMutex_);
+    std::string interfaceMemberName = std::get<2>(subscription);
+    auto its_handler = selectiveBroadcastHandlers.find(interfaceMemberName);
+    if (its_handler != selectiveBroadcastHandlers.end()) {
+        selectiveBroadcastHandlers.erase(its_handler);
+    }
 }
 
 DBusProxyConnection::DBusSignalHandlerToken DBusProxy::addSignalMemberHandler(
diff --git a/src/CommonAPI/DBus/DBusProxyManager.cpp b/src/CommonAPI/DBus/DBusProxyManager.cpp
index 6a48994..8ea30e3 100644
--- a/src/CommonAPI/DBus/DBusProxyManager.cpp
+++ b/src/CommonAPI/DBus/DBusProxyManager.cpp
@@ -33,7 +33,9 @@ DBusProxyManager::getInterface() const {
 
 const ConnectionId_t &
 DBusProxyManager::getConnectionId() const {
-    return connectionId_;
+    // Every DBusProxyConnection is created as DBusConnection
+    // in Factory::getConnection and is only stored as DBusProxyConnection
+    return std::static_pointer_cast<DBusConnection>(proxy_.getDBusConnection())->getConnectionId();
 }
 
 void
diff --git a/src/CommonAPI/DBus/DBusServiceRegistry.cpp b/src/CommonAPI/DBus/DBusServiceRegistry.cpp
index 55e0239..0ee7705 100644
--- a/src/CommonAPI/DBus/DBusServiceRegistry.cpp
+++ b/src/CommonAPI/DBus/DBusServiceRegistry.cpp
@@ -126,7 +126,21 @@ DBusServiceRegistry::subscribeAvailabilityListener(
         availabilityStatus = AvailabilityStatus::NOT_AVAILABLE;
     } else if (dbusServiceListenersRecord.uniqueBusNameState != DBusRecordState::RESOLVING &&
                dbusInterfaceNameListenersRecord.state == DBusRecordState::UNKNOWN) {
-        dbusInterfaceNameListenersRecord.state = resolveDBusInterfaceNameState(dbusAddress, dbusServiceListenersRecord);
+        if(dbusPredefinedServices_.find(dbusAddress.getService()) != dbusPredefinedServices_.end()) {
+            //service is predefined -> notify service listeners about availability
+            auto dbusServiceNameMapIterator = dbusServiceNameMap_.find(dbusAddress.getService());
+            if(dbusServiceNameMapIterator != dbusServiceNameMap_.end()) {
+                std::unordered_set<std::string> dbusInterfaceNames;
+                for(auto dbusInterfaceNameListenerRecordIterator = dbusInterfaceNameListenersMap.begin();
+                        dbusInterfaceNameListenerRecordIterator != dbusInterfaceNameListenersMap.end();
+                        ++dbusInterfaceNameListenerRecordIterator) {
+                    dbusInterfaceNames.insert(dbusInterfaceNameListenerRecordIterator->first);
+                }
+                notifyDBusServiceListeners(*dbusServiceNameMapIterator->second, dbusAddress.getObjectPath(), dbusInterfaceNames, DBusRecordState::AVAILABLE);
+            }
+        } else {
+            dbusInterfaceNameListenersRecord.state = resolveDBusInterfaceNameState(dbusAddress, dbusServiceListenersRecord);
+        }
     }
 
     if(availabilityStatus == AvailabilityStatus::UNKNOWN) {
@@ -1142,17 +1156,13 @@ void DBusServiceRegistry::onDBusServiceAvailable(const std::string& dbusServiceN
             }
         } else {
             //service is predefined -> notify service listeners about availability
+            std::unordered_set<std::string> dbusInterfaceNames;
             for(auto dbusInterfaceNameListenerRecordIterator = dbusInterfaceNameListenersMap.begin();
                     dbusInterfaceNameListenerRecordIterator != dbusInterfaceNameListenersMap.end();
                     ++dbusInterfaceNameListenerRecordIterator) {
-                auto& dbusInterfaceNameListenerRecord = dbusInterfaceNameListenerRecordIterator->second;
-                dbusInterfaceNameListenerRecord.state = DBusRecordState::RESOLVED;
-                for(auto dbusServiceListenerIterator = dbusInterfaceNameListenerRecord.listenerList.begin();
-                        dbusServiceListenerIterator != dbusInterfaceNameListenerRecord.listenerList.end();
-                        ++dbusServiceListenerIterator) {
-                    (*dbusServiceListenerIterator)(AvailabilityStatus::AVAILABLE);
-                }
+                dbusInterfaceNames.insert(dbusInterfaceNameListenerRecordIterator->first);
             }
+            notifyDBusServiceListeners(*dbusUniqueNameRecord, listenersDBusObjectPath, dbusInterfaceNames, DBusRecordState::AVAILABLE);
         }
 
         if (dbusInterfaceNameListenersMap.empty()) {