// Copyright 2019 The Chromium Authors
// Use of this source code is governed by a BSD-style license that can be
// found in the LICENSE file.
#include "chromecast/external_mojo/external_service_support/external_connector_impl.h"
#include <utility>
#include "base/functional/bind.h"
#include "base/location.h"
#include "base/logging.h"
#include "base/memory/ref_counted.h"
#include "base/memory/scoped_refptr.h"
#include "base/synchronization/lock.h"
#include "base/task/sequenced_task_runner.h"
#include "base/thread_annotations.h"
#include "base/time/time.h"
#include "chromecast/external_mojo/broker_service/broker_service.h"
#include "chromecast/external_mojo/external_service_support/external_service.h"
#include "mojo/public/cpp/bindings/pending_remote.h"
#include "mojo/public/cpp/platform/named_platform_channel.h"
#include "mojo/public/cpp/platform/platform_channel_endpoint.h"
#include "mojo/public/cpp/system/invitation.h"
#include "services/service_manager/public/cpp/connector.h"
namespace chromecast {
namespace external_service_support {
namespace {
constexpr base::TimeDelta kConnectRetryDelay = base::Milliseconds(500);
} // namespace
// Since we are only allowed to make a single underlying connection to the
// broker, we share the underlying connection between all ExternalConnector
// instances. The ExternalConnectors use clones of the underlying connection.
//
// Since connection error callbacks are called in some arbitrary order, we need
// to be careful to handle disconnection correctly. Each underlying connection
// has a unique token (int64_t) associated with it, which is propagated to all
// clones. If any clone receives a disconnect callback, it tries to reconnect by
// calling ConnectClone(), passing in the previous token (associated with the
// broken connection). BrokerConnection then attempts to reconnect the
// underlying connection if the broken connection token matches the token for
// the current connection; if it doesn't match, the connection was already
// recreated, so nothing needs to be done.
class ExternalConnectorImpl::BrokerConnection
: public base::RefCountedThreadSafe<BrokerConnection> {
public:
explicit BrokerConnection(std::string broker_path)
: broker_path_(std::move(broker_path)),
task_runner_(base::SequencedTaskRunner::GetCurrentDefault()) {
Connect();
}
int64_t ConnectClone(
int64_t dead_connection_token,
mojo::PendingReceiver<external_mojo::mojom::ExternalConnector> receiver) {
int64_t token;
{
base::AutoLock lock(lock_);
if (dead_connection_token == connection_token_) {
task_runner_->PostTask(
FROM_HERE, base::BindOnce(&BrokerConnection::Connect, this));
++connection_token_;
}
token = connection_token_;
}
task_runner_->PostTask(FROM_HERE,
base::BindOnce(&BrokerConnection::AttachClone, this,
std::move(receiver)));
return token;
}
private:
friend class base::RefCountedThreadSafe<BrokerConnection>;
~BrokerConnection() = default;
void Connect() {
DCHECK(task_runner_->RunsTasksInCurrentSequence());
connector_.reset();
pending_receiver_ = connector_.BindNewPipeAndPassReceiver();
AttemptBrokerConnection();
}
void AttachClone(
mojo::PendingReceiver<external_mojo::mojom::ExternalConnector> receiver) {
DCHECK(task_runner_->RunsTasksInCurrentSequence());
connector_->Clone(std::move(receiver));
}
void AttemptBrokerConnection() {
mojo::NamedPlatformChannel::Options channel_options;
channel_options.server_name = broker_path_;
#if BUILDFLAG(IS_ANDROID)
// On Android, use the abstract namespace to avoid filesystem access.
channel_options.use_abstract_namespace = true;
#endif
mojo::PlatformChannelEndpoint endpoint =
mojo::NamedPlatformChannel::ConnectToServer(channel_options);
if (!endpoint.is_valid()) {
task_runner_->PostDelayedTask(
FROM_HERE,
base::BindOnce(&BrokerConnection::AttemptBrokerConnection,
weak_factory_.GetWeakPtr()),
kConnectRetryDelay);
return;
}
auto invitation = mojo::IncomingInvitation::Accept(std::move(endpoint));
auto remote_pipe = invitation.ExtractMessagePipe(0);
if (!remote_pipe) {
LOG(ERROR) << "Invalid message pipe";
task_runner_->PostDelayedTask(
FROM_HERE,
base::BindOnce(&BrokerConnection::AttemptBrokerConnection,
weak_factory_.GetWeakPtr()),
kConnectRetryDelay);
return;
}
mojo::FuseMessagePipes(pending_receiver_.PassPipe(),
std::move(remote_pipe));
}
const std::string broker_path_;
const scoped_refptr<base::SequencedTaskRunner> task_runner_;
mojo::Remote<external_mojo::mojom::ExternalConnector> connector_;
mojo::PendingReceiver<external_mojo::mojom::ExternalConnector>
pending_receiver_;
base::Lock lock_;
int64_t connection_token_ GUARDED_BY(lock_) = 1;
base::WeakPtrFactory<BrokerConnection> weak_factory_{this};
};
// static
void ExternalConnector::Connect(
const std::string& broker_path,
base::OnceCallback<void(std::unique_ptr<ExternalConnector>)> callback) {
DCHECK(callback);
std::move(callback).Run(Create(broker_path));
}
// static
std::unique_ptr<ExternalConnector> ExternalConnector::Create(
const std::string& broker_path) {
return std::make_unique<ExternalConnectorImpl>(broker_path);
}
// static
std::unique_ptr<ExternalConnector> ExternalConnector::Create(
mojo::PendingRemote<external_mojo::mojom::ExternalConnector> remote) {
return std::make_unique<ExternalConnectorImpl>(std::move(remote));
}
// static
std::unique_ptr<ExternalConnector> ExternalConnector::Create(
service_manager::Connector* connector) {
mojo::PendingRemote<external_mojo::mojom::ExternalConnector> pending_remote;
connector->BindInterface(external_mojo::BrokerService::kServiceName,
pending_remote.InitWithNewPipeAndPassReceiver());
return std::make_unique<ExternalConnectorImpl>(std::move(pending_remote));
}
ExternalConnectorImpl::ExternalConnectorImpl(const std::string& broker_path)
: broker_connection_(base::MakeRefCounted<BrokerConnection>(broker_path)) {
DETACH_FROM_SEQUENCE(sequence_checker_);
Connect();
}
ExternalConnectorImpl::ExternalConnectorImpl(
scoped_refptr<BrokerConnection> broker_connection)
: broker_connection_(broker_connection) {
DETACH_FROM_SEQUENCE(sequence_checker_);
Connect();
}
ExternalConnectorImpl::ExternalConnectorImpl(
mojo::PendingRemote<external_mojo::mojom::ExternalConnector> pending_remote)
: pending_remote_(std::move(pending_remote)) {
DETACH_FROM_SEQUENCE(sequence_checker_);
}
ExternalConnectorImpl::~ExternalConnectorImpl() = default;
base::CallbackListSubscription
ExternalConnectorImpl::AddConnectionErrorCallback(
base::RepeatingClosure callback) {
DCHECK_CALLED_ON_VALID_SEQUENCE(sequence_checker_);
return error_closures_.Add(std::move(callback));
}
void ExternalConnectorImpl::RegisterService(const std::string& service_name,
ExternalService* service) {
RegisterService(service_name, service->GetReceiver());
}
void ExternalConnectorImpl::RegisterService(
const std::string& service_name,
mojo::PendingRemote<external_mojo::mojom::ExternalService> service_remote) {
BindConnectorIfNecessary();
auto service_instance_info =
chromecast::external_mojo::mojom::ServiceInstanceInfo::New(
service_name, std::move(service_remote));
std::vector<chromecast::external_mojo::mojom::ServiceInstanceInfoPtr> v;
v.emplace_back(std::move(service_instance_info));
connector_->RegisterServiceInstances(std::move(v));
}
void ExternalConnectorImpl::RegisterServices(
const std::vector<std::string>& service_names,
const std::vector<ExternalService*>& services) {
CHECK(service_names.size() == services.size());
std::vector<chromecast::external_mojo::mojom::ServiceInstanceInfoPtr>
service_instances_info;
service_instances_info.reserve(services.size());
for (size_t i = 0; i < services.size(); ++i) {
service_instances_info.emplace_back(
chromecast::external_mojo::mojom::ServiceInstanceInfo::New(
service_names[i], services[i]->GetReceiver()));
}
RegisterServices(std::move(service_instances_info));
}
void ExternalConnectorImpl::RegisterServices(
std::vector<chromecast::external_mojo::mojom::ServiceInstanceInfoPtr>
service_instances_info) {
BindConnectorIfNecessary();
connector_->RegisterServiceInstances(std::move(service_instances_info));
}
void ExternalConnectorImpl::QueryServiceList(
base::OnceCallback<void(
std::vector<chromecast::external_mojo::mojom::ExternalServiceInfoPtr>)>
callback) {
BindConnectorIfNecessary();
connector_->QueryServiceList(std::move(callback));
}
void ExternalConnectorImpl::BindInterface(
const std::string& service_name,
const std::string& interface_name,
mojo::ScopedMessagePipeHandle interface_pipe,
bool async) {
DCHECK_CALLED_ON_VALID_SEQUENCE(sequence_checker_);
if (!async) {
BindInterfaceImmediately(service_name, interface_name,
std::move(interface_pipe));
return;
}
base::SequencedTaskRunner::GetCurrentDefault()->PostTask(
FROM_HERE,
base::BindOnce(&ExternalConnectorImpl::BindInterfaceImmediately,
weak_factory_.GetWeakPtr(), service_name, interface_name,
std::move(interface_pipe)));
}
void ExternalConnectorImpl::BindInterfaceImmediately(
const std::string& service_name,
const std::string& interface_name,
mojo::ScopedMessagePipeHandle interface_pipe) {
BindConnectorIfNecessary();
connector_->BindInterface(service_name, interface_name,
std::move(interface_pipe));
}
std::unique_ptr<ExternalConnector> ExternalConnectorImpl::Clone() {
if (broker_connection_) {
return std::make_unique<ExternalConnectorImpl>(broker_connection_);
}
// Bind to the current sequence since this is a public method.
BindConnectorIfNecessary();
return std::make_unique<ExternalConnectorImpl>(RequestConnector());
}
mojo::PendingRemote<external_mojo::mojom::ExternalConnector>
ExternalConnectorImpl::RequestConnector() {
// Bind to the current sequence since this is a public method.
BindConnectorIfNecessary();
mojo::PendingRemote<external_mojo::mojom::ExternalConnector> remote;
connector_->Clone(remote.InitWithNewPipeAndPassReceiver());
return remote;
}
void ExternalConnectorImpl::SendChromiumConnectorRequest(
mojo::ScopedMessagePipeHandle request) {
BindConnectorIfNecessary();
connector_->BindChromiumConnector(std::move(request));
}
void ExternalConnectorImpl::Connect() {
DCHECK(broker_connection_);
connection_token_ = broker_connection_->ConnectClone(
connection_token_, pending_remote_.InitWithNewPipeAndPassReceiver());
}
void ExternalConnectorImpl::OnMojoDisconnect() {
DCHECK_CALLED_ON_VALID_SEQUENCE(sequence_checker_);
connector_.reset();
pending_remote_.reset();
if (broker_connection_) {
Connect();
BindConnectorIfNecessary();
}
error_closures_.Notify();
}
void ExternalConnectorImpl::BindConnectorIfNecessary() {
// Bind the message pipe and SequenceChecker to the current thread the first
// time it is used to connect.
DCHECK_CALLED_ON_VALID_SEQUENCE(sequence_checker_);
if (connector_.is_bound()) {
return;
}
DCHECK(pending_remote_.is_valid());
connector_.Bind(std::move(pending_remote_));
connector_.set_disconnect_handler(base::BindOnce(
&ExternalConnectorImpl::OnMojoDisconnect, base::Unretained(this)));
}
} // namespace external_service_support
} // namespace chromecast
Codebase Browser
chromium