This source file includes following definitions.
- GetAbsoluteSignalName
- EmptyResponseCallbackBody
- ignore_service_unknown_errors_
- CallMethodAndBlock
- CallMethod
- CallMethodWithErrorCallback
- ConnectToSignal
- SetNameOwnerChangedCallback
- WaitForServiceToBeAvailable
- Detach
- EmptyResponseCallback
- start_time
- StartAsyncMethodCall
- OnPendingCallIsComplete
- RunResponseCallback
- OnPendingCallIsCompleteThunk
- ConnectToNameOwnerChangedSignal
- ConnectToSignalInternal
- WaitForServiceToBeAvailableInternal
- HandleMessage
- RunMethod
- HandleMessageThunk
- LogMethodCallFailure
- OnCallMethodError
- AddMatchRuleWithCallback
- AddMatchRuleWithoutCallback
- UpdateNameOwnerAndBlock
- HandleNameOwnerChanged
- RunNameOwnerChangedCallback
- RunWaitForServiceToBeAvailableCallbacks
#include "dbus/bus.h"
#include "base/bind.h"
#include "base/logging.h"
#include "base/message_loop/message_loop.h"
#include "base/metrics/histogram.h"
#include "base/strings/string_piece.h"
#include "base/strings/stringprintf.h"
#include "base/task_runner_util.h"
#include "base/threading/thread.h"
#include "base/threading/thread_restrictions.h"
#include "dbus/dbus_statistics.h"
#include "dbus/message.h"
#include "dbus/object_path.h"
#include "dbus/object_proxy.h"
#include "dbus/scoped_dbus_error.h"
namespace dbus {
namespace {
const char kErrorServiceUnknown[] = "org.freedesktop.DBus.Error.ServiceUnknown";
const int kSuccessRatioHistogramMaxValue = 2;
const char kDBusSystemObjectPath[] = "/org/freedesktop/DBus";
const char kDBusSystemObjectInterface[] = "org.freedesktop.DBus";
const char kDBusSystemObjectAddress[] = "org.freedesktop.DBus";
const char kNameOwnerChangedMember[] = "NameOwnerChanged";
std::string GetAbsoluteSignalName(
const std::string& interface_name,
const std::string& signal_name) {
return interface_name + "." + signal_name;
}
void EmptyResponseCallbackBody(Response* ) {
}
}
ObjectProxy::ObjectProxy(Bus* bus,
const std::string& service_name,
const ObjectPath& object_path,
int options)
: bus_(bus),
service_name_(service_name),
object_path_(object_path),
filter_added_(false),
ignore_service_unknown_errors_(
options & IGNORE_SERVICE_UNKNOWN_ERRORS) {
}
ObjectProxy::~ObjectProxy() {
}
scoped_ptr<Response> ObjectProxy::CallMethodAndBlock(MethodCall* method_call,
int timeout_ms) {
bus_->AssertOnDBusThread();
if (!bus_->Connect() ||
!method_call->SetDestination(service_name_) ||
!method_call->SetPath(object_path_))
return scoped_ptr<Response>();
DBusMessage* request_message = method_call->raw_message();
ScopedDBusError error;
const base::TimeTicks start_time = base::TimeTicks::Now();
DBusMessage* response_message =
bus_->SendWithReplyAndBlock(request_message, timeout_ms, error.get());
UMA_HISTOGRAM_ENUMERATION("DBus.SyncMethodCallSuccess",
response_message ? 1 : 0,
kSuccessRatioHistogramMaxValue);
statistics::AddBlockingSentMethodCall(service_name_,
method_call->GetInterface(),
method_call->GetMember());
if (!response_message) {
LogMethodCallFailure(method_call->GetInterface(),
method_call->GetMember(),
error.is_set() ? error.name() : "unknown error type",
error.is_set() ? error.message() : "");
return scoped_ptr<Response>();
}
UMA_HISTOGRAM_TIMES("DBus.SyncMethodCallTime",
base::TimeTicks::Now() - start_time);
return Response::FromRawMessage(response_message);
}
void ObjectProxy::CallMethod(MethodCall* method_call,
int timeout_ms,
ResponseCallback callback) {
CallMethodWithErrorCallback(method_call, timeout_ms, callback,
base::Bind(&ObjectProxy::OnCallMethodError,
this,
method_call->GetInterface(),
method_call->GetMember(),
callback));
}
void ObjectProxy::CallMethodWithErrorCallback(MethodCall* method_call,
int timeout_ms,
ResponseCallback callback,
ErrorCallback error_callback) {
bus_->AssertOnOriginThread();
const base::TimeTicks start_time = base::TimeTicks::Now();
if (!method_call->SetDestination(service_name_) ||
!method_call->SetPath(object_path_)) {
DBusMessage* response_message = NULL;
base::Closure task = base::Bind(&ObjectProxy::RunResponseCallback,
this,
callback,
error_callback,
start_time,
response_message);
bus_->GetOriginTaskRunner()->PostTask(FROM_HERE, task);
return;
}
DBusMessage* request_message = method_call->raw_message();
dbus_message_ref(request_message);
base::Closure task = base::Bind(&ObjectProxy::StartAsyncMethodCall,
this,
timeout_ms,
request_message,
callback,
error_callback,
start_time);
statistics::AddSentMethodCall(service_name_,
method_call->GetInterface(),
method_call->GetMember());
bus_->GetDBusTaskRunner()->PostTask(FROM_HERE, task);
}
void ObjectProxy::ConnectToSignal(const std::string& interface_name,
const std::string& signal_name,
SignalCallback signal_callback,
OnConnectedCallback on_connected_callback) {
bus_->AssertOnOriginThread();
base::PostTaskAndReplyWithResult(
bus_->GetDBusTaskRunner(),
FROM_HERE,
base::Bind(&ObjectProxy::ConnectToSignalInternal,
this,
interface_name,
signal_name,
signal_callback),
base::Bind(on_connected_callback,
interface_name,
signal_name));
}
void ObjectProxy::SetNameOwnerChangedCallback(
NameOwnerChangedCallback callback) {
bus_->AssertOnOriginThread();
name_owner_changed_callback_ = callback;
}
void ObjectProxy::WaitForServiceToBeAvailable(
WaitForServiceToBeAvailableCallback callback) {
bus_->AssertOnOriginThread();
wait_for_service_to_be_available_callbacks_.push_back(callback);
bus_->GetDBusTaskRunner()->PostTask(
FROM_HERE,
base::Bind(&ObjectProxy::WaitForServiceToBeAvailableInternal, this));
}
void ObjectProxy::Detach() {
bus_->AssertOnDBusThread();
if (filter_added_) {
if (!bus_->RemoveFilterFunction(&ObjectProxy::HandleMessageThunk, this)) {
LOG(ERROR) << "Failed to remove filter function";
}
}
for (std::set<std::string>::iterator iter = match_rules_.begin();
iter != match_rules_.end(); ++iter) {
ScopedDBusError error;
bus_->RemoveMatch(*iter, error.get());
if (error.is_set()) {
LOG(ERROR) << "Failed to remove match rule: " << *iter;
}
}
match_rules_.clear();
}
ObjectProxy::ResponseCallback ObjectProxy::EmptyResponseCallback() {
return base::Bind(&EmptyResponseCallbackBody);
}
ObjectProxy::OnPendingCallIsCompleteData::OnPendingCallIsCompleteData(
ObjectProxy* in_object_proxy,
ResponseCallback in_response_callback,
ErrorCallback in_error_callback,
base::TimeTicks in_start_time)
: object_proxy(in_object_proxy),
response_callback(in_response_callback),
error_callback(in_error_callback),
start_time(in_start_time) {
}
ObjectProxy::OnPendingCallIsCompleteData::~OnPendingCallIsCompleteData() {
}
void ObjectProxy::StartAsyncMethodCall(int timeout_ms,
DBusMessage* request_message,
ResponseCallback response_callback,
ErrorCallback error_callback,
base::TimeTicks start_time) {
bus_->AssertOnDBusThread();
if (!bus_->Connect() || !bus_->SetUpAsyncOperations()) {
DBusMessage* response_message = NULL;
base::Closure task = base::Bind(&ObjectProxy::RunResponseCallback,
this,
response_callback,
error_callback,
start_time,
response_message);
bus_->GetOriginTaskRunner()->PostTask(FROM_HERE, task);
dbus_message_unref(request_message);
return;
}
DBusPendingCall* pending_call = NULL;
bus_->SendWithReply(request_message, &pending_call, timeout_ms);
OnPendingCallIsCompleteData* data =
new OnPendingCallIsCompleteData(this, response_callback, error_callback,
start_time);
const bool success = dbus_pending_call_set_notify(
pending_call,
&ObjectProxy::OnPendingCallIsCompleteThunk,
data,
NULL);
CHECK(success) << "Unable to allocate memory";
dbus_pending_call_unref(pending_call);
dbus_message_unref(request_message);
}
void ObjectProxy::OnPendingCallIsComplete(DBusPendingCall* pending_call,
ResponseCallback response_callback,
ErrorCallback error_callback,
base::TimeTicks start_time) {
bus_->AssertOnDBusThread();
DBusMessage* response_message = dbus_pending_call_steal_reply(pending_call);
base::Closure task = base::Bind(&ObjectProxy::RunResponseCallback,
this,
response_callback,
error_callback,
start_time,
response_message);
bus_->GetOriginTaskRunner()->PostTask(FROM_HERE, task);
}
void ObjectProxy::RunResponseCallback(ResponseCallback response_callback,
ErrorCallback error_callback,
base::TimeTicks start_time,
DBusMessage* response_message) {
bus_->AssertOnOriginThread();
bool method_call_successful = false;
if (!response_message) {
error_callback.Run(NULL);
} else if (dbus_message_get_type(response_message) ==
DBUS_MESSAGE_TYPE_ERROR) {
scoped_ptr<ErrorResponse> error_response(
ErrorResponse::FromRawMessage(response_message));
error_callback.Run(error_response.get());
bus_->GetDBusTaskRunner()->PostTask(
FROM_HERE,
base::Bind(&base::DeletePointer<ErrorResponse>,
error_response.release()));
} else {
scoped_ptr<Response> response(Response::FromRawMessage(response_message));
response_callback.Run(response.get());
bus_->GetDBusTaskRunner()->PostTask(
FROM_HERE,
base::Bind(&base::DeletePointer<Response>, response.release()));
method_call_successful = true;
UMA_HISTOGRAM_TIMES("DBus.AsyncMethodCallTime",
base::TimeTicks::Now() - start_time);
}
UMA_HISTOGRAM_ENUMERATION("DBus.AsyncMethodCallSuccess",
method_call_successful,
kSuccessRatioHistogramMaxValue);
}
void ObjectProxy::OnPendingCallIsCompleteThunk(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->error_callback,
data->start_time);
delete data;
}
bool ObjectProxy::ConnectToNameOwnerChangedSignal() {
bus_->AssertOnDBusThread();
if (!bus_->Connect() || !bus_->SetUpAsyncOperations())
return false;
if (!filter_added_) {
if (bus_->AddFilterFunction(&ObjectProxy::HandleMessageThunk, this)) {
filter_added_ = true;
} else {
LOG(ERROR) << "Failed to add filter function";
}
}
const std::string name_owner_changed_match_rule =
base::StringPrintf(
"type='signal',interface='org.freedesktop.DBus',"
"member='NameOwnerChanged',path='/org/freedesktop/DBus',"
"sender='org.freedesktop.DBus',arg0='%s'",
service_name_.c_str());
const bool success =
AddMatchRuleWithoutCallback(name_owner_changed_match_rule,
"org.freedesktop.DBus.NameOwnerChanged");
UpdateNameOwnerAndBlock();
return success;
}
bool ObjectProxy::ConnectToSignalInternal(const std::string& interface_name,
const std::string& signal_name,
SignalCallback signal_callback) {
bus_->AssertOnDBusThread();
if (!ConnectToNameOwnerChangedSignal())
return false;
const std::string absolute_signal_name =
GetAbsoluteSignalName(interface_name, signal_name);
const std::string match_rule =
base::StringPrintf("type='signal', interface='%s', path='%s'",
interface_name.c_str(),
object_path_.value().c_str());
return AddMatchRuleWithCallback(match_rule,
absolute_signal_name,
signal_callback);
}
void ObjectProxy::WaitForServiceToBeAvailableInternal() {
bus_->AssertOnDBusThread();
if (!ConnectToNameOwnerChangedSignal()) {
const bool service_is_ready = false;
bus_->GetOriginTaskRunner()->PostTask(
FROM_HERE,
base::Bind(&ObjectProxy::RunWaitForServiceToBeAvailableCallbacks,
this, service_is_ready));
return;
}
const bool service_is_available = !service_name_owner_.empty();
if (service_is_available) {
bus_->GetOriginTaskRunner()->PostTask(
FROM_HERE,
base::Bind(&ObjectProxy::RunWaitForServiceToBeAvailableCallbacks,
this, service_is_available));
return;
}
}
DBusHandlerResult ObjectProxy::HandleMessage(
DBusConnection* connection,
DBusMessage* raw_message) {
bus_->AssertOnDBusThread();
if (dbus_message_get_type(raw_message) != DBUS_MESSAGE_TYPE_SIGNAL)
return DBUS_HANDLER_RESULT_NOT_YET_HANDLED;
dbus_message_ref(raw_message);
scoped_ptr<Signal> signal(
Signal::FromRawMessage(raw_message));
const ObjectPath path = signal->GetPath();
if (path != object_path_) {
if (path.value() == kDBusSystemObjectPath &&
signal->GetMember() == kNameOwnerChangedMember) {
return HandleNameOwnerChanged(signal.Pass());
}
return DBUS_HANDLER_RESULT_NOT_YET_HANDLED;
}
const std::string interface = signal->GetInterface();
const std::string member = signal->GetMember();
statistics::AddReceivedSignal(service_name_, interface, member);
const std::string absolute_signal_name = GetAbsoluteSignalName(
interface, member);
MethodTable::const_iterator iter = method_table_.find(absolute_signal_name);
if (iter == method_table_.end()) {
return DBUS_HANDLER_RESULT_NOT_YET_HANDLED;
}
VLOG(1) << "Signal received: " << signal->ToString();
std::string sender = signal->GetSender();
if (service_name_owner_ != sender) {
LOG(ERROR) << "Rejecting a message from a wrong sender.";
UMA_HISTOGRAM_COUNTS("DBus.RejectedSignalCount", 1);
return DBUS_HANDLER_RESULT_NOT_YET_HANDLED;
}
const base::TimeTicks start_time = base::TimeTicks::Now();
if (bus_->HasDBusThread()) {
Signal* released_signal = signal.release();
bus_->GetOriginTaskRunner()->PostTask(FROM_HERE,
base::Bind(&ObjectProxy::RunMethod,
this,
start_time,
iter->second,
released_signal));
} else {
const base::TimeTicks start_time = base::TimeTicks::Now();
Signal* released_signal = signal.release();
RunMethod(start_time, iter->second, released_signal);
}
return DBUS_HANDLER_RESULT_NOT_YET_HANDLED;
}
void ObjectProxy::RunMethod(base::TimeTicks start_time,
std::vector<SignalCallback> signal_callbacks,
Signal* signal) {
bus_->AssertOnOriginThread();
for (std::vector<SignalCallback>::iterator iter = signal_callbacks.begin();
iter != signal_callbacks.end(); ++iter)
iter->Run(signal);
bus_->GetDBusTaskRunner()->PostTask(
FROM_HERE,
base::Bind(&base::DeletePointer<Signal>, signal));
UMA_HISTOGRAM_TIMES("DBus.SignalHandleTime",
base::TimeTicks::Now() - start_time);
}
DBusHandlerResult ObjectProxy::HandleMessageThunk(
DBusConnection* connection,
DBusMessage* raw_message,
void* user_data) {
ObjectProxy* self = reinterpret_cast<ObjectProxy*>(user_data);
return self->HandleMessage(connection, raw_message);
}
void ObjectProxy::LogMethodCallFailure(
const base::StringPiece& interface_name,
const base::StringPiece& method_name,
const base::StringPiece& error_name,
const base::StringPiece& error_message) const {
if (ignore_service_unknown_errors_ && error_name == kErrorServiceUnknown)
return;
LOG(ERROR) << "Failed to call method: "
<< interface_name << "." << method_name
<< ": object_path= " << object_path_.value()
<< ": " << error_name << ": " << error_message;
}
void ObjectProxy::OnCallMethodError(const std::string& interface_name,
const std::string& method_name,
ResponseCallback response_callback,
ErrorResponse* error_response) {
if (error_response) {
MessageReader reader(error_response);
std::string error_message;
reader.PopString(&error_message);
LogMethodCallFailure(interface_name,
method_name,
error_response->GetErrorName(),
error_message);
}
response_callback.Run(NULL);
}
bool ObjectProxy::AddMatchRuleWithCallback(
const std::string& match_rule,
const std::string& absolute_signal_name,
SignalCallback signal_callback) {
DCHECK(!match_rule.empty());
DCHECK(!absolute_signal_name.empty());
bus_->AssertOnDBusThread();
if (match_rules_.find(match_rule) == match_rules_.end()) {
ScopedDBusError error;
bus_->AddMatch(match_rule, error.get());
if (error.is_set()) {
LOG(ERROR) << "Failed to add match rule \"" << match_rule << "\". Got "
<< error.name() << ": " << error.message();
return false;
} else {
match_rules_.insert(match_rule);
method_table_[absolute_signal_name].push_back(signal_callback);
return true;
}
} else {
method_table_[absolute_signal_name].push_back(signal_callback);
return true;
}
}
bool ObjectProxy::AddMatchRuleWithoutCallback(
const std::string& match_rule,
const std::string& absolute_signal_name) {
DCHECK(!match_rule.empty());
DCHECK(!absolute_signal_name.empty());
bus_->AssertOnDBusThread();
if (match_rules_.find(match_rule) != match_rules_.end())
return true;
ScopedDBusError error;
bus_->AddMatch(match_rule, error.get());
if (error.is_set()) {
LOG(ERROR) << "Failed to add match rule \"" << match_rule << "\". Got "
<< error.name() << ": " << error.message();
return false;
}
match_rules_.insert(match_rule);
return true;
}
void ObjectProxy::UpdateNameOwnerAndBlock() {
bus_->AssertOnDBusThread();
service_name_owner_ =
bus_->GetServiceOwnerAndBlock(service_name_, Bus::SUPPRESS_ERRORS);
}
DBusHandlerResult ObjectProxy::HandleNameOwnerChanged(
scoped_ptr<Signal> signal) {
DCHECK(signal);
bus_->AssertOnDBusThread();
if (signal->GetMember() == kNameOwnerChangedMember &&
signal->GetInterface() == kDBusSystemObjectInterface &&
signal->GetSender() == kDBusSystemObjectAddress) {
MessageReader reader(signal.get());
std::string name, old_owner, new_owner;
if (reader.PopString(&name) &&
reader.PopString(&old_owner) &&
reader.PopString(&new_owner) &&
name == service_name_) {
service_name_owner_ = new_owner;
bus_->GetOriginTaskRunner()->PostTask(
FROM_HERE,
base::Bind(&ObjectProxy::RunNameOwnerChangedCallback,
this, old_owner, new_owner));
const bool service_is_available = !service_name_owner_.empty();
if (service_is_available) {
bus_->GetOriginTaskRunner()->PostTask(
FROM_HERE,
base::Bind(&ObjectProxy::RunWaitForServiceToBeAvailableCallbacks,
this, service_is_available));
}
}
}
return DBUS_HANDLER_RESULT_NOT_YET_HANDLED;
}
void ObjectProxy::RunNameOwnerChangedCallback(const std::string& old_owner,
const std::string& new_owner) {
bus_->AssertOnOriginThread();
if (!name_owner_changed_callback_.is_null())
name_owner_changed_callback_.Run(old_owner, new_owner);
}
void ObjectProxy::RunWaitForServiceToBeAvailableCallbacks(
bool service_is_available) {
bus_->AssertOnOriginThread();
std::vector<WaitForServiceToBeAvailableCallback> callbacks;
callbacks.swap(wait_for_service_to_be_available_callbacks_);
for (size_t i = 0; i < callbacks.size(); ++i)
callbacks[i].Run(service_is_available);
}
}