src/Messaging/DBus/Connection.cpp

   1 //
   2 // Tizen Web Device API
   3 // Copyright (c) 2012 Samsung Electronics Co., Ltd.
   4 //
   5 // Licensed under the Apache License, Version 2.0 (the License);
   6 // you may not use this file except in compliance with the License.
   7 // You may obtain a copy of the License at
   8 //
   9 // http://www.apache.org/licenses/LICENSE-2.0
  10 //
  11 // Unless required by applicable law or agreed to in writing, software
  12 // distributed under the License is distributed on an "AS IS" BASIS,
  13 // WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  14 // See the License for the specific language governing permissions and
  15 // limitations under the License.
  16 //
  17
  18 #include "Connection.h"
  19 #include <cassert>
  20 #include <algorithm>
  21 #include <Commons/Exception.h>
  22 #include "Message.h"
  23 #include <Logger.h>
  24
  25 namespace DeviceAPI {
  26 namespace Messaging {
  27 namespace DBus {
  28 Connection::Connection() :
  29     m_connection(NULL),
  30     m_messageDispatcher(NULL),
  31     m_connected(false)
  32 {
  33     dbus_error_init(&m_error);
  34     m_messageDispatcher = new MessageDispatcher(this);
  35 }
  36
  37 Connection::~Connection()
  38 {
  39     if(m_connected)
  40     {
  41         LoggerE("DBus connection has not been closed.");
  42     }
  43     delete m_messageDispatcher;
  44     if (dbus_error_is_set(&m_error)) {
  45         dbus_error_free(&m_error);
  46     }
  47 }
  48
  49 void Connection::open(DBusBusType bus)
  50 {
  51     if (m_connected) { return; }
  52
  53     m_connection = dbus_bus_get_private(bus, &m_error);
  54     if (!m_connection || dbus_error_is_set(&m_error)) {
  55         std::string message = m_error.message;
  56         dbus_error_free(&m_error);
  57         ThrowMsg(WrtDeviceApis::Commons::InvalidArgumentException,
  58                  "Couldn't attach to DBus system bus: " << message);
  59     }
  60
  61     dbus_connection_set_exit_on_disconnect(m_connection, FALSE);
  62
  63     for (FilterSetIterator it = m_filters.begin(); it != m_filters.end();
  64          ++it) {
  65         addFilterInternal(*it);
  66     }
  67
  68     if (!dbus_connection_set_watch_functions(m_connection, addWatch,
  69                                              removeWatch, toggleWatch, this,
  70                                              NULL)) {
  71         LoggerD("Couldn't set-up watch functions.");
  72         ThrowMsg(WrtDeviceApis::Commons::PlatformException, "Couldn't set-up watch functions.");
  73     }
  74
  75     if (!dbus_connection_add_filter(m_connection, filterMessage, this, NULL)) {
  76         LoggerD("Couldn't set signal handler callback.");
  77         ThrowMsg(WrtDeviceApis::Commons::PlatformException, "Couldn't set signal handler callback.");
  78     }
  79
  80     m_connected = true;
  81 }
  82
  83 void Connection::close()
  84 {
  85     LoggerD("ENTER");
  86     if (!m_connected) { return; }
  87
  88     dbus_connection_remove_filter(m_connection, filterMessage, this);
  89
  90     // Calls removeWatch() automagically.
  91     if (!dbus_connection_set_watch_functions(m_connection, NULL, NULL, NULL,
  92                                              NULL, NULL)) {
  93         DPL::Mutex::ScopedLock lock(&m_watchesMutex);
  94         WatchSetIterator it = m_watches.begin();
  95         for (; it != m_watches.end(); ++it) {
  96             if (it->second->isEnabled()) {
  97                 m_messageDispatcher->removeDescriptor(it->second->getHandle(),
  98                                                       it->second->getType());
  99             }
 100         }
 101         m_watches.clear();
 102     }
 103
 104     // TODO: think how to handle exception on filter removal; 1 - do nothing (now),
 105     // 2 - remove the buggy one but throw an exception, 3 - remove the buggy one
 106     // and don't throw any exception
 107     for (FilterSetIterator it = m_filters.begin(); it != m_filters.end();
 108          ++it) {
 109         removeFilterInternal(*it);
 110     }
 111
 112     dbus_connection_close(m_connection);
 113     dbus_connection_unref(m_connection);
 114
 115     m_connected = false;
 116 }
 117
 118 void Connection::addFilter(const std::string& rule)
 119 {
 120     if (m_connected) {
 121         addFilterInternal(rule);
 122     }
 123     m_filters.insert(rule);
 124 }
 125
 126 void Connection::removeFilter(const std::string& rule)
 127 {
 128     if (m_connected) {
 129         removeFilterInternal(rule);
 130     }
 131     m_filters.erase(rule);
 132 }
 133
 134 void Connection::setWorkerThread(DPL::Thread* thread)
 135 {
 136     if(m_connected)
 137     {
 138         LoggerE("Connection has already been established.");
 139         ThrowMsg(WrtDeviceApis::Commons::PlatformException,     "Connection has already been established.");
 140     }
 141     m_messageDispatcher->SwitchToThread(thread);
 142 }
 143
 144 void Connection::addFilterInternal(const std::string& rule)
 145 {
 146     if(!m_connection)
 147     {
 148         LoggerE("Connection has to be established first.");
 149         ThrowMsg(WrtDeviceApis::Commons::PlatformException, "Connection has to be established first.");
 150     }
 151     dbus_bus_add_match(m_connection, rule.c_str(), &m_error);
 152     if (dbus_error_is_set(&m_error)) {
 153         std::string message = m_error.message;
 154         dbus_error_free(&m_error);
 155         ThrowMsg(WrtDeviceApis::Commons::InvalidArgumentException, "Invalid rule: " << message);
 156     }
 157 }
 158
 159 void Connection::removeFilterInternal(const std::string& rule)
 160 {
 161     if(!m_connection)
 162     {
 163         LoggerE("Connection has to be established first.");
 164     }
 165     else
 166     {
 167         dbus_bus_remove_match(m_connection, rule.c_str(), &m_error);
 168         if (dbus_error_is_set(&m_error)) {
 169             std::string message = m_error.message;
 170             dbus_error_free(&m_error);
 171             LoggerD("removeFilterInternal error : " << message);
 172         }
 173         }
 174 }
 175
 176 dbus_bool_t Connection::addWatch(DBusWatch* watch,
 177         void* data)
 178 {
 179     LoggerD("ENTER");
 180     if(!data)
 181     {
 182         LoggerE("Connection should be passed as user data.");
 183         ThrowMsg(WrtDeviceApis::Commons::PlatformException,     "Connection should be passed as user data.");
 184     }
 185     Connection* this_ = static_cast<Connection*>(data);
 186     WatchPtr wrapper(new Watch(watch));
 187     if (wrapper->isEnabled()) {
 188         this_->m_messageDispatcher->addDescriptor(wrapper->getHandle(),
 189                                                   wrapper->getType());
 190     }
 191     DPL::Mutex::ScopedLock lock(&this_->m_watchesMutex);
 192     this_->m_watches[wrapper->getHandle()] = wrapper;
 193     return TRUE;
 194 }
 195
 196 void Connection::removeWatch(DBusWatch* watch,
 197         void* data)
 198 {
 199     LoggerD("ENTER");
 200     if(!data)
 201     {
 202         LoggerE("Connection should be passed as user data.");
 203         ThrowMsg(WrtDeviceApis::Commons::PlatformException,     "Connection should be passed as user data.");
 204     }
 205     Connection* this_ = static_cast<Connection*>(data);
 206     Watch wrapper(watch);
 207     if (wrapper.isEnabled()) {
 208         this_->m_messageDispatcher->removeDescriptor(wrapper.getHandle(),
 209                                                      wrapper.getType());
 210     }
 211     DPL::Mutex::ScopedLock lock(&this_->m_watchesMutex);
 212     this_->m_watches.erase(wrapper.getHandle());
 213 }
 214
 215 void Connection::toggleWatch(DBusWatch* watch,
 216         void* data)
 217 {
 218     LoggerD("ENTER");
 219     if(!data)
 220     {
 221         LoggerE("Connection should be passed as user data.");
 222         ThrowMsg(WrtDeviceApis::Commons::PlatformException,     "Connection should be passed as user data.");
 223     }
 224     Connection* this_ = static_cast<Connection*>(data);
 225     Watch wrapper(watch);
 226     if (wrapper.isEnabled()) {
 227         this_->m_messageDispatcher->addDescriptor(wrapper.getHandle(),
 228                                                   wrapper.getType());
 229     } else {
 230         this_->m_messageDispatcher->removeDescriptor(wrapper.getHandle(),
 231                                                      wrapper.getType());
 232     }
 233 }
 234
 235 DBusHandlerResult Connection::filterMessage(DBusConnection* connection,
 236         DBusMessage* message,
 237         void* data)
 238 {
 239     LoggerD("ENTER");
 240     if (data) {
 241         Connection* this_ = static_cast<Connection*>(data);
 242         MessagePtr msg;
 243         if (message) {
 244             LoggerI(
 245                 "DBUS message received from interface: " <<
 246                 dbus_message_get_interface(message));
 247             msg.Reset(new Message(message));
 248         } else {
 249             LoggerE("DBus message not set, this should not happen!");
 250             return DBUS_HANDLER_RESULT_NOT_YET_HANDLED;
 251         }
 252         FilterSetIterator it = this_->m_filters.begin();
 253         for (; it != this_->m_filters.end(); ++it) {
 254             // TODO: extend following matching procedure to check not only
 255             // interface's name.
 256             if (!msg->getInterface().empty() && ((*it).find(msg->getInterface()) != std::string::npos))
 257                         {
 258                 MessageEvent event(msg);
 259                 this_->EmitEvent(event);
 260             }
 261         }
 262     } else {
 263         LoggerE("User-data not set, this should not happen!");
 264     }
 265     return DBUS_HANDLER_RESULT_NOT_YET_HANDLED;
 266 }
 267
 268 Connection::MessageDispatcher::MessageDispatcher(Connection* connection) :
 269     m_connection(connection)
 270 {
 271     if(!m_connection)
 272     {
 273         LoggerE("Connection cannot be NULL.");
 274         ThrowMsg(WrtDeviceApis::Commons::PlatformException,     "Connection cannot be NULL.");
 275     }
 276     DPL::Event::ControllerEventHandler<AddDescriptorSyncEvent>::Touch();
 277     DPL::Event::ControllerEventHandler<RemoveDescriptorSyncEvent>::Touch();
 278 }
 279
 280 Connection::MessageDispatcher::~MessageDispatcher()
 281 {
 282     SwitchToThread(NULL);
 283 }
 284
 285 void Connection::MessageDispatcher::addDescriptor(DPL::WaitableHandle handle,
 286         DPL::WaitMode::Type mode)
 287 {
 288     LoggerD("ENTER");
 289     DPL::WaitableEvent wait;
 290     AddDescriptorSyncEvent event(handle, mode, &wait);
 291     DPL::Event::ControllerEventHandler<AddDescriptorSyncEvent>::PostEvent(event);
 292     DPL::WaitForSingleHandle(wait.GetHandle());
 293 }
 294
 295 void Connection::MessageDispatcher::removeDescriptor(DPL::WaitableHandle handle,
 296         DPL::WaitMode::Type mode)
 297 {
 298     LoggerD("ENTER");
 299     DPL::WaitableEvent wait;
 300     RemoveDescriptorSyncEvent event(handle, mode, &wait);
 301     DPL::Event::ControllerEventHandler<RemoveDescriptorSyncEvent>::PostEvent(event);
 302     DPL::WaitForSingleHandle(wait.GetHandle());
 303 }
 304
 305 void Connection::MessageDispatcher::OnEventReceived(
 306         const AddDescriptorSyncEvent& event)
 307 {
 308     LoggerD("ENTER");
 309     DPL::WaitableHandleWatchSupport::InheritedContext()->AddWaitableHandleWatch(
 310         this,
 311         event.GetArg0(),
 312         event.GetArg1());
 313     event.GetArg2()->Signal();
 314 }
 315
 316 void Connection::MessageDispatcher::OnEventReceived(
 317         const RemoveDescriptorSyncEvent& event)
 318 {
 319     LoggerD("ENTER");
 320     DPL::WaitableHandleWatchSupport::InheritedContext()->
 321         RemoveWaitableHandleWatch(this, event.GetArg0(), event.GetArg1());
 322     event.GetArg2()->Signal();
 323 }
 324
 325 void Connection::MessageDispatcher::OnWaitableHandleEvent(
 326         DPL::WaitableHandle waitableHandle,
 327         DPL::WaitMode::Type mode)
 328 {
 329     LoggerD("ENTER");
 330     {
 331         DPL::Mutex::ScopedLock lock(&m_connection->m_watchesMutex);
 332         WatchSetIterator it = m_connection->m_watches.find(waitableHandle);
 333         if (it != m_connection->m_watches.end()) {
 334             unsigned int flags =
 335                 (mode == DPL::WaitMode::Write ? DBUS_WATCH_WRITABLE
 336                         : DBUS_WATCH_READABLE);
 337             while (!dbus_watch_handle(it->second->get(), flags)) {
 338                 LoggerE("Too few memory!");
 339                 // TODO: add some check-point to not loop infinitely
 340             }
 341         }
 342         // TODO: not sure if there is point in dispatching connection when watch not found?
 343     }
 344     dbus_connection_ref(m_connection->m_connection);
 345     while (dbus_connection_dispatch(m_connection->m_connection) ==
 346            DBUS_DISPATCH_DATA_REMAINS) {
 347     }
 348     dbus_connection_unref(m_connection->m_connection);
 349 }
 350
 351 Connection::Watch::Watch(DBusWatch* watch) : m_watch(watch)
 352 {
 353     if(!m_watch)
 354     {
 355         LoggerE("Watch cannot be NULL.");
 356     }
 357 }
 358
 359 DPL::WaitableHandle Connection::Watch::getHandle() const
 360 {
 361     return dbus_watch_get_unix_fd(m_watch);
 362 }
 363
 364 DPL::WaitMode::Type Connection::Watch::getType() const
 365 {
 366     if ((dbus_watch_get_flags(m_watch) & DBUS_WATCH_WRITABLE) != 0) {
 367         return DPL::WaitMode::Write;
 368     }
 369     return DPL::WaitMode::Read;
 370 }
 371
 372 bool Connection::Watch::isEnabled() const
 373 {
 374     return (TRUE == dbus_watch_get_enabled(m_watch));
 375 }
 376
 377 DBusWatch* Connection::Watch::get() const
 378 {
 379     return m_watch;
 380 }
 381 } // DBus
 382 } // Platform
 383 } // WrtPlugins