Implement Bus and ObjectProxy classes for our D-Bus library.

dbus/object_proxy.cc

Index: dbus/object_proxy.cc
diff --git a/dbus/object_proxy.cc b/dbus/object_proxy.cc
new file mode 100644
index 0000000000000000000000000000000000000000..50427a66e299e5be0540e73b3498b9b0d0280452
--- /dev/null
+++ b/dbus/object_proxy.cc
@@ -0,0 +1,207 @@
+// Copyright (c) 2011 The Chromium Authors. All rights reserved.
+// Use of this source code is governed by a BSD-style license that can be
+// found in the LICENSE file.
+
+#include "dbus/bus.h"
+
+#include "base/bind.h"
+#include "base/logging.h"
+#include "base/message_loop.h"
+#include "base/threading/thread.h"
+#include "base/threading/thread_restrictions.h"
+#include "dbus/message.h"
+#include "dbus/object_proxy.h"
+
+namespace dbus {
+
+ObjectProxy::ObjectProxy(Bus* bus,
+                         const std::string& service_name,
+                         const std::string& object_path)
+    : bus_(bus),
+      service_name_(service_name),
+      object_path_(object_path),
+      origin_thread_id_(base::PlatformThread::CurrentId()) {
+}
+
+ObjectProxy::~ObjectProxy() {
+}
+
+// Originally we tried to make |method_call| a const reference, but we
+// gave up as dbus_connection_send_with_reply_and_block() takes a
+// non-const pointer of DBusMessage as the second parameter.
+bool ObjectProxy::CallMethodAndWait(MethodCall* method_call,
+                                    Response* response,
+                                    int timeout_ms) {
+  base::ThreadRestrictions::AssertIOAllowed();
+
+  if (!bus_->Init())
+    return false;
+  method_call->SetServiceName(service_name_);
+  method_call->SetObjectPath(object_path_);
+  DBusMessage* request_message = method_call->raw_message();
+
+  DBusError error = {};
+  dbus_error_init(&error);
+
+  // Send the message synchronously.
+  DBusMessage* response_message = dbus_connection_send_with_reply_and_block(
+      bus_->connection(), request_message, timeout_ms, &error);
+
+  if (!response_message) {
+    if (dbus_error_is_set(&error)) {
+      LOG(ERROR) << error.message;
+    }
+    return false;
+  }
+  response->reset_raw_message(response_message);
+
+  return true;
+}
+
+void ObjectProxy::CallMethod(MethodCall* method_call,
+                             ResponseCallback callback,
+                             int timeout_ms) {
+  AssertOnOriginThread();
+
+  method_call->SetServiceName(service_name_);
+  method_call->SetObjectPath(object_path_);
+  // This is a bit tricky but increment the reference count of the request
+  // message here so that we can keep using the message even if the
+  // MethodCall object at the caller is gone before the method call is
+  // complete.
+  DBusMessage* request_message = method_call->raw_message();
+  dbus_message_ref(request_message);
+
+  // Bind() won't compile if we pass request_message as-is since
+  // DBusMessage is an opaque struct which Bind() does not like.
+  //
+  // /base/bind_helpers.h:145:
+  // error: invalid use of incomplete type 'struct DBusMessage'
+  //
+  // Hence we cast it to void* to workaround the issue.
+  base::Closure task = base::Bind(&ObjectProxy::StartAsyncMethodCall,
+                                  base::Unretained(this),
+                                  timeout_ms,
+                                  reinterpret_cast<void*>(request_message),
+                                  MessageLoop::current(),
+                                  callback);
+  if (bus_->use_io_thread()) {
+    bus_->PostTaskToIoThread(FROM_HERE, task);
+  } else {
+    MessageLoopForIO::current()->PostTask(FROM_HERE, task);
+  }
+}
+
+ObjectProxy::OnPendingCallIsCompleteData::OnPendingCallIsCompleteData(
+    ObjectProxy* in_object_proxy,
+    MessageLoop* in_origin_loop,
+    ResponseCallback in_response_callback)
+    : object_proxy(in_object_proxy),
+      origin_loop(in_origin_loop),
+      response_callback(in_response_callback) {
+}
+
+void ObjectProxy::StartAsyncMethodCall(int timeout_ms,
+                                       void* in_request_message,
+                                       MessageLoop* origin_loop,
+                                       ResponseCallback response_callback) {
+  AssertOnIoThreadIfConfigured();
+
+  DBusMessage* request_message =
+      reinterpret_cast<DBusMessage*>(in_request_message);
+  if (bus_->Init()) {
+    DBusPendingCall* pending_call = NULL;
+    // This returns false only when unable to allocate memory.
+    bool success = false;
+    success = dbus_connection_send_with_reply(
+        bus_->connection(), request_message, &pending_call, timeout_ms);
+    dbus_pending_call_ref(pending_call);
+    DCHECK(success) << "Unable to allocate memory";
+
+    // Prepare the data we'll be passing to OnPendingCallIsCompleteStub().
+    // The data will be deleted in OnPendingCallIsCompleteStub().
+    OnPendingCallIsCompleteData* data =
+        new OnPendingCallIsCompleteData(this,
+                                        origin_loop,
+                                        response_callback);
+
+    // This returns false only when unable to allocate memory.
+    success = dbus_pending_call_set_notify(
+        pending_call,
+        &ObjectProxy::OnPendingCallIsCompleteStub,
+        data,
+        NULL);
+    DCHECK(success) << "Unable to allocate memory";
+  } else {
+    Response* response = NULL;
+    base::Closure task = base::Bind(&ObjectProxy::RunResponseCallback,
+                                    base::Unretained(this),
+                                    response_callback,
+                                    response);
+    origin_loop->PostTask(FROM_HERE, task);
+  }
+  // It's now safe to unref the request message.
+  dbus_message_unref(request_message);
+}
+
+void ObjectProxy::RunResponseCallback(ResponseCallback response_callback,
+                                      Response* response) {
+  AssertOnOriginThread();
+
+  if (!response) {
+    response_callback.Run(NULL);
+  } else {
+    if (response->GetMessageType() == Message::MESSAGE_ERROR) {
+      // Error message should contain the error message as string.
+      LOG(ERROR) << "Method call failed: " << response->ToString();
+      // We don't give the error message to the callback.
+      response_callback.Run(NULL);
+    } else {
+      response_callback.Run(response);
+    }
+  }
+}
+
+void ObjectProxy::OnPendingCallIsComplete(DBusPendingCall* pending_call,
+                                          ResponseCallback response_callback,
+                                          MessageLoop* origin_loop) {
+  AssertOnIoThreadIfConfigured();
+
+  DBusMessage* response_message = dbus_pending_call_steal_reply(pending_call);
+  DCHECK(response_message);
+
+  // This will be deleted in the callback.
+  Response* response = new Response;
+  response->reset_raw_message(response_message);
+  base::Closure task = base::Bind(&ObjectProxy::RunResponseCallback,
+                                  base::Unretained(this),
+                                  response_callback,
+                                  response);
+  origin_loop->PostTask(FROM_HERE, task);
+}
+
+void ObjectProxy::AssertOnOriginThread() {
+  DCHECK_EQ(origin_thread_id_, base::PlatformThread::CurrentId());
+}
+
+void ObjectProxy::AssertOnIoThreadIfConfigured() {
+  if (bus_->use_io_thread()) {
+    DCHECK_EQ(bus_->io_thread()->thread_id(),
+              base::PlatformThread::CurrentId());
+  } else {
+    AssertOnOriginThread();
+  }
+}
+
+void ObjectProxy::OnPendingCallIsCompleteStub(DBusPendingCall* pending_call,
+                                              void* user_data) {
+  OnPendingCallIsCompleteData* data =
+      reinterpret_cast<OnPendingCallIsCompleteData*>(user_data);
+  ObjectProxy* self = data->object_proxy;
+  self->OnPendingCallIsComplete(pending_call,
+                                data->response_callback,
+                                data->origin_loop);
+  delete data;
+}
+
+}  // namespace dbus