chromecast/external_mojo/public/cpp/external_mojo_broker.cc - chromium/src - Git at Google

 // 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/public/cpp/external_mojo_broker.h"

 #include <map>
 #include <set>
 #include <utility>

 #include "base/containers/contains.h"
 #include "build/build_config.h"

 #if BUILDFLAG(IS_LINUX) || BUILDFLAG(IS_CHROMEOS)
 #include <sys/stat.h>
 #endif

 #include "base/functional/bind.h"
 #include "base/functional/callback_helpers.h"
 #include "base/location.h"
 #include "base/logging.h"
 #include "base/message_loop/message_pump_for_io.h"
 #include "base/task/current_thread.h"
 #include "base/token.h"
 #include "base/trace_event/trace_event.h"
 #include "chromecast/chromecast_buildflags.h"
 #include "chromecast/external_mojo/public/cpp/common.h"
 #include "chromecast/external_mojo/public/mojom/connector.mojom.h"
 #include "mojo/public/cpp/bindings/pending_receiver.h"
 #include "mojo/public/cpp/bindings/receiver_set.h"
 #include "mojo/public/cpp/bindings/remote.h"
 #include "mojo/public/cpp/platform/named_platform_channel.h"
 #include "mojo/public/cpp/platform/platform_channel_endpoint.h"
 #include "mojo/public/cpp/platform/platform_handle.h"
 #include "mojo/public/cpp/platform/socket_utils_posix.h"
 #include "mojo/public/cpp/system/invitation.h"
 #include "services/service_manager/public/cpp/connector.h"
 #include "services/service_manager/public/cpp/constants.h"
 #include "services/service_manager/public/cpp/identity.h"
 #include "services/service_manager/public/cpp/service.h"
 #include "services/service_manager/public/cpp/service_filter.h"
 #include "services/service_manager/public/cpp/service_receiver.h"
 #include "services/service_manager/public/mojom/connector.mojom.h"
 #include "services/service_manager/public/mojom/service.mojom.h"
 #include "third_party/abseil-cpp/absl/types/optional.h"

 #if BUILDFLAG(IS_ANDROID)
 #include "base/android/bundle_utils.h"
 #endif

 namespace chromecast {
 namespace external_mojo {

 namespace {

 void OnRegisterServiceResult(const std::string& service_name,
                              service_manager::mojom::ConnectResult result) {
   // RegisterServiceInstance() currently returns INVALID_ARGUMENT on success.
   if (result == service_manager::mojom::ConnectResult::ACCESS_DENIED) {
     LOG(WARNING) << "Failed to register external service proxy for "
                  << service_name;
   }
 }

 void OnInternalBindResult(
     const std::string& service_name,
     const std::string& interface_name,
     service_manager::mojom::ConnectResult result,
     const absl::optional<service_manager::Identity>& identity) {
   if (result != service_manager::mojom::ConnectResult::SUCCEEDED) {
     LOG(ERROR) << "Failed to bind " << service_name << ":" << interface_name
                << ", result = " << result;
   }
 }

 }  // namespace

 class ExternalMojoBroker::ConnectorImpl : public mojom::ExternalConnector {
  public:
   ConnectorImpl() : connector_facade_(this) {}

   ConnectorImpl(const ConnectorImpl&) = delete;
   ConnectorImpl& operator=(const ConnectorImpl&) = delete;

   void InitializeChromium(
       std::unique_ptr<service_manager::Connector> connector,
       const std::vector<std::string>& external_services_to_proxy) {
     DCHECK(connector);
     connector_ = std::move(connector);
     RegisterExternalServices(external_services_to_proxy);
   }

   void AddReceiver(mojo::PendingReceiver<mojom::ExternalConnector> receiver) {
     receivers_.Add(this, std::move(receiver));
   }

  private:
   class ExternalServiceProxy : public ::service_manager::Service {
    public:
     ExternalServiceProxy(
         ConnectorImpl* connector,
         std::string service_name,
         mojo::PendingReceiver<::service_manager::mojom::Service> receiver)
         : connector_(connector),
           service_name_(std::move(service_name)),
           service_receiver_(this, std::move(receiver)) {
       DCHECK(connector_);
     }

     ExternalServiceProxy(const ExternalServiceProxy&) = delete;
     ExternalServiceProxy& operator=(const ExternalServiceProxy&) = delete;

    private:
     void OnBindInterface(
         const service_manager::BindSourceInfo& source,
         const std::string& interface_name,
         mojo::ScopedMessagePipeHandle interface_pipe) override {
       connector_->BindExternalInterface(service_name_, interface_name,
                                         std::move(interface_pipe));
     }

     ConnectorImpl* const connector_;
     const std::string service_name_;
     service_manager::ServiceReceiver service_receiver_;
   };

   class ServiceManagerConnectorFacade
       : public service_manager::mojom::Connector {
    public:
     explicit ServiceManagerConnectorFacade(
         ExternalMojoBroker::ConnectorImpl* connector)
         : connector_(connector) {
       DCHECK(connector_);
     }

     void AddReceiver(
         mojo::PendingReceiver<service_manager::mojom::Connector> receiver) {
       receivers_.Add(this, std::move(receiver));
     }

    private:
     void BindInterface(const ::service_manager::ServiceFilter& filter,
                        const std::string& interface_name,
                        mojo::ScopedMessagePipeHandle interface_pipe,
                        service_manager::mojom::BindInterfacePriority priority,
                        BindInterfaceCallback callback) override {
       connector_->BindInterface(filter.service_name(), interface_name,
                                 std::move(interface_pipe));
       std::move(callback).Run(service_manager::mojom::ConnectResult::SUCCEEDED,
                               absl::nullopt);
     }

     void QueryService(const std::string& service_name,
                       QueryServiceCallback callback) override {
       // TODO(kmackay) Could add a wrapper as needed.
       NOTIMPLEMENTED();
     }

     void WarmService(const ::service_manager::ServiceFilter& filter,
                      WarmServiceCallback callback) override {
       std::move(callback).Run(service_manager::mojom::ConnectResult::SUCCEEDED,
                               absl::nullopt);
     }

     void RegisterServiceInstance(
         const ::service_manager::Identity& identity,
         mojo::ScopedMessagePipeHandle service,
         mojo::PendingReceiver<service_manager::mojom::ProcessMetadata>
             metadata_receiver,
         RegisterServiceInstanceCallback callback) override {
       // TODO(kmackay) Could add a wrapper as needed.
       NOTIMPLEMENTED();
     }

     void Clone(mojo::PendingReceiver<service_manager::mojom::Connector>
                    receiver) override {
       AddReceiver(std::move(receiver));
     }

     ExternalMojoBroker::ConnectorImpl* const connector_;

     mojo::ReceiverSet<service_manager::mojom::Connector> receivers_;
   };

   struct PendingBindRequest {
     PendingBindRequest(const std::string& interface_name,
                        mojo::ScopedMessagePipeHandle interface_pipe)
         : interface_name(interface_name),
           interface_pipe(std::move(interface_pipe)) {}

     const std::string interface_name;
     mojo::ScopedMessagePipeHandle interface_pipe;
   };

   void RegisterExternalServices(
       const std::vector<std::string>& external_services_to_proxy) {
     if (external_services_to_proxy.empty()) {
       return;
     }

     external_services_to_proxy_.insert(external_services_to_proxy.begin(),
                                        external_services_to_proxy.end());

     for (const auto& service_name : external_services_to_proxy) {
       LOG(INFO) << "Register proxy for external " << service_name;
       mojo::PendingRemote<service_manager::mojom::Service> service_remote;
       registered_external_services_[service_name] =
           std::make_unique<ExternalServiceProxy>(
               this, service_name,
               service_remote.InitWithNewPipeAndPassReceiver());

       connector_->RegisterServiceInstance(
           service_manager::Identity(service_name,
                                     service_manager::kSystemInstanceGroup,
                                     base::Token{}, base::Token::CreateRandom()),
           std::move(service_remote),
           mojo::NullReceiver() /* metadata_receiver */,
           base::BindOnce(&OnRegisterServiceResult, service_name));
     }
   }

   // Helper for ExternalServiceProxy.
   void BindExternalInterface(const std::string& service_name,
                              const std::string& interface_name,
                              mojo::ScopedMessagePipeHandle interface_pipe) {
     LOG(INFO) << "Internal request for " << service_name << ":"
               << interface_name;
     auto it = services_.find(service_name);
     if (it != services_.end()) {
       it->second->OnBindInterface(interface_name, std::move(interface_pipe));
       return;
     }

     ServiceNotFound(service_name, interface_name, std::move(interface_pipe));
   }

   void RegisterServiceInstance(
       const std::string& service_name,
       mojo::PendingRemote<mojom::ExternalService> service_remote) {
     if (base::Contains(services_, service_name)) {
       LOG(ERROR) << "Duplicate service " << service_name;
       return;
     }
     TRACE_EVENT_INSTANT1("mojom", "RegisterService", TRACE_EVENT_SCOPE_THREAD,
                          "service", service_name);
     LOG(INFO) << "Register service " << service_name;
     mojo::Remote<mojom::ExternalService> service(std::move(service_remote));
     service.set_disconnect_handler(base::BindOnce(
         &ConnectorImpl::OnServiceLost, base::Unretained(this), service_name));
     auto it = services_.emplace(service_name, std::move(service)).first;

     auto p = pending_bind_requests_.find(service_name);
     if (p != pending_bind_requests_.end()) {
       for (auto& request : p->second) {
         it->second->OnBindInterface(request.interface_name,
                                     std::move(request.interface_pipe));
       }
       pending_bind_requests_.erase(p);
     }

     auto& info_entry = services_info_[service_name];
     info_entry.name = service_name;
     info_entry.connect_time = base::TimeTicks::Now();
     info_entry.disconnect_time = base::TimeTicks();
   }

   // standalone::mojom::Connector implementation:
   void RegisterServiceInstances(
       std::vector<chromecast::external_mojo::mojom::ServiceInstanceInfoPtr>
           service_instances_info) override {
     for (auto& instance_info_ptr : service_instances_info) {
       RegisterServiceInstance(instance_info_ptr->service_name,
                               std::move(instance_info_ptr->service_remote));
     }
   }

   void BindInterface(const std::string& service_name,
                      const std::string& interface_name,
                      mojo::ScopedMessagePipeHandle interface_pipe) override {
     LOG(INFO) << "Request for " << service_name << ":" << interface_name;
     TRACE_EVENT_INSTANT1("mojom", "BindToService", TRACE_EVENT_SCOPE_THREAD,
                          "service", service_name);
     auto it = services_.find(service_name);
     if (it != services_.end()) {
       LOG(INFO) << "Found externally-registered " << service_name;
       it->second->OnBindInterface(interface_name, std::move(interface_pipe));
       return;
     }

     auto service_proxy_it = external_services_to_proxy_.find(service_name);

     if (!connector_ || service_proxy_it != external_services_to_proxy_.end()) {
       ServiceNotFound(service_name, interface_name, std::move(interface_pipe));
       return;
     }

     connector_->QueryService(
         service_name,
         base::BindOnce(&ConnectorImpl::OnQueryResult, base::Unretained(this),
                        service_name, interface_name,
                        std::move(interface_pipe)));
   }

   void Clone(
       mojo::PendingReceiver<mojom::ExternalConnector> receiver) override {
     AddReceiver(std::move(receiver));
   }

   void BindChromiumConnector(
       mojo::ScopedMessagePipeHandle interface_pipe) override {
     if (!connector_) {
       connector_facade_.AddReceiver(
           mojo::PendingReceiver<service_manager::mojom::Connector>(
               std::move(interface_pipe)));
       return;
     }

     connector_->BindConnectorReceiver(
         mojo::PendingReceiver<service_manager::mojom::Connector>(
             std::move(interface_pipe)));
   }

   void QueryServiceList(QueryServiceListCallback callback) override {
     std::vector<chromecast::external_mojo::mojom::ExternalServiceInfoPtr> infos;
     for (const auto& it : services_info_) {
       infos.emplace_back(it.second.Clone());
     }
     std::move(callback).Run(std::move(infos));
   }

   void OnQueryResult(const std::string& service_name,
                      const std::string& interface_name,
                      mojo::ScopedMessagePipeHandle interface_pipe,
                      service_manager::mojom::ServiceInfoPtr info) {
     if (!info) {
       ServiceNotFound(service_name, interface_name, std::move(interface_pipe));
       return;
     }

     LOG(INFO) << "Found internal " << service_name;
     connector_->BindInterface(
         service_manager::ServiceFilter::ByName(service_name), interface_name,
         std::move(interface_pipe),
         service_manager::mojom::BindInterfacePriority::kImportant,
         base::BindOnce(&OnInternalBindResult, service_name, interface_name));
   }

   void ServiceNotFound(const std::string& service_name,
                        const std::string& interface_name,
                        mojo::ScopedMessagePipeHandle interface_pipe) {
     LOG(INFO) << service_name << " not found";
     // Assume the service will be registered later, and wait until then.
     pending_bind_requests_[service_name].emplace_back(
         interface_name, std::move(interface_pipe));
   }

   void OnServiceLost(const std::string& service_name) {
     LOG(INFO) << service_name << " disconnected";
     TRACE_EVENT_INSTANT1("mojom", "ServiceDisconnected",
                          TRACE_EVENT_SCOPE_THREAD, "service", service_name);
     services_.erase(service_name);
     services_info_[service_name].disconnect_time = base::TimeTicks::Now();
   }

   ServiceManagerConnectorFacade connector_facade_;
   std::unique_ptr<service_manager::Connector> connector_;

   mojo::ReceiverSet<mojom::ExternalConnector> receivers_;
   std::set<std::string> external_services_to_proxy_;
   std::map<std::string, std::unique_ptr<ExternalServiceProxy>>
       registered_external_services_;

   std::map<std::string, mojo::Remote<mojom::ExternalService>> services_;
   std::map<std::string, std::vector<PendingBindRequest>> pending_bind_requests_;
   std::map<std::string, mojom::ExternalServiceInfo> services_info_;
 };

 class ExternalMojoBroker::ReadWatcher
     : public base::MessagePumpForIO::FdWatcher {
  public:
   ReadWatcher(ConnectorImpl* connector, mojo::PlatformHandle listen_handle)
       : connector_(connector),
         listen_handle_(std::move(listen_handle)),
         watch_controller_(FROM_HERE) {
     DCHECK(listen_handle_.is_valid());
     base::CurrentIOThread::Get().WatchFileDescriptor(
         listen_handle_.GetFD().get(), true /* persistent */,
         base::MessagePumpForIO::WATCH_READ, &watch_controller_, this);
   }

   ReadWatcher(const ReadWatcher&) = delete;
   ReadWatcher& operator=(const ReadWatcher&) = delete;

   // base::MessagePumpForIO::FdWatcher implementation:
   void OnFileCanReadWithoutBlocking(int fd) override {
     base::ScopedFD accepted_fd;
     if (mojo::AcceptSocketConnection(fd, &accepted_fd,
                                      false /* check_peer_user */) &&
         accepted_fd.is_valid()) {
       mojo::OutgoingInvitation invitation;
       auto pipe = invitation.AttachMessagePipe(0);
       mojo::OutgoingInvitation::Send(
           std::move(invitation), base::kNullProcessHandle,
           mojo::PlatformChannelEndpoint(
               mojo::PlatformHandle(std::move(accepted_fd))));
       connector_->AddReceiver(
           mojo::PendingReceiver<mojom::ExternalConnector>(std::move(pipe)));
     }
   }

   void OnFileCanWriteWithoutBlocking(int /* fd */) override {}

  private:
   ConnectorImpl* const connector_;
   const mojo::PlatformHandle listen_handle_;
   base::MessagePumpForIO::FdWatchController watch_controller_;
 };

 ExternalMojoBroker::ExternalMojoBroker(const std::string& broker_path) {
   connector_ = std::make_unique<ConnectorImpl>();

   // For external service support, we expose a channel endpoint on the
   // |broker_path|. Otherwise, only services in the same process network can
   // make use of the broker.
 #if BUILDFLAG(ENABLE_EXTERNAL_MOJO_SERVICES)
 #if BUILDFLAG(IS_ANDROID)
   // Monolithic MediaShell can just access the service broker directly in the
   // same process, so there's no need to stand up a server.
   if (!base::android::BundleUtils::IsBundle()) {
     return;
   }
   // On Android, use the abstract namespace to avoid filesystem access.
   bool use_abstract_namespace = true;
 #else
   bool use_abstract_namespace = false;
 #endif  // BUILDFLAG(IS_ANDROID)

   LOG(INFO) << "Initializing external mojo broker at: " << broker_path;

   mojo::NamedPlatformChannel::Options channel_options;
   channel_options.server_name = broker_path;
   channel_options.use_abstract_namespace = use_abstract_namespace;
   mojo::NamedPlatformChannel named_channel(channel_options);

   mojo::PlatformChannelServerEndpoint server_endpoint =
       named_channel.TakeServerEndpoint();
   DCHECK(server_endpoint.is_valid());

 #if BUILDFLAG(IS_LINUX) || BUILDFLAG(IS_CHROMEOS)
   chmod(broker_path.c_str(), 0770);
 #endif

   read_watcher_ = std::make_unique<ReadWatcher>(
       connector_.get(), server_endpoint.TakePlatformHandle());
 #endif  // BUILDFLAG(ENABLE_EXTERNAL_MOJO_SERVICES)
 }

 void ExternalMojoBroker::InitializeChromium(
     std::unique_ptr<service_manager::Connector> connector,
     const std::vector<std::string>& external_services_to_proxy) {
   connector_->InitializeChromium(std::move(connector),
                                  external_services_to_proxy);
 }

 mojo::PendingRemote<mojom::ExternalConnector>
 ExternalMojoBroker::CreateConnector() {
   mojo::PendingRemote<mojom::ExternalConnector> remote;
   connector_->AddReceiver(remote.InitWithNewPipeAndPassReceiver());
   return remote;
 }

 void ExternalMojoBroker::BindConnector(
     mojo::PendingReceiver<mojom::ExternalConnector> receiver) {
   connector_->AddReceiver(std::move(receiver));
 }

 ExternalMojoBroker::~ExternalMojoBroker() = default;

 }  // namespace external_mojo
 }  // namespace chromecast
	// 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/public/cpp/external_mojo_broker.h"

	#include <map>
	#include <set>
	#include <utility>

	#include "base/containers/contains.h"
	#include "build/build_config.h"

	#if BUILDFLAG(IS_LINUX) \|\| BUILDFLAG(IS_CHROMEOS)
	#include <sys/stat.h>
	#endif

	#include "base/functional/bind.h"
	#include "base/functional/callback_helpers.h"
	#include "base/location.h"
	#include "base/logging.h"
	#include "base/message_loop/message_pump_for_io.h"
	#include "base/task/current_thread.h"
	#include "base/token.h"
	#include "base/trace_event/trace_event.h"
	#include "chromecast/chromecast_buildflags.h"
	#include "chromecast/external_mojo/public/cpp/common.h"
	#include "chromecast/external_mojo/public/mojom/connector.mojom.h"
	#include "mojo/public/cpp/bindings/pending_receiver.h"
	#include "mojo/public/cpp/bindings/receiver_set.h"
	#include "mojo/public/cpp/bindings/remote.h"
	#include "mojo/public/cpp/platform/named_platform_channel.h"
	#include "mojo/public/cpp/platform/platform_channel_endpoint.h"
	#include "mojo/public/cpp/platform/platform_handle.h"
	#include "mojo/public/cpp/platform/socket_utils_posix.h"
	#include "mojo/public/cpp/system/invitation.h"
	#include "services/service_manager/public/cpp/connector.h"
	#include "services/service_manager/public/cpp/constants.h"
	#include "services/service_manager/public/cpp/identity.h"
	#include "services/service_manager/public/cpp/service.h"
	#include "services/service_manager/public/cpp/service_filter.h"
	#include "services/service_manager/public/cpp/service_receiver.h"
	#include "services/service_manager/public/mojom/connector.mojom.h"
	#include "services/service_manager/public/mojom/service.mojom.h"
	#include "third_party/abseil-cpp/absl/types/optional.h"

	#if BUILDFLAG(IS_ANDROID)
	#include "base/android/bundle_utils.h"
	#endif

	namespace chromecast {
	namespace external_mojo {

	namespace {

	void OnRegisterServiceResult(const std::string& service_name,
	service_manager::mojom::ConnectResult result) {
	// RegisterServiceInstance() currently returns INVALID_ARGUMENT on success.
	if (result == service_manager::mojom::ConnectResult::ACCESS_DENIED) {
	LOG(WARNING) << "Failed to register external service proxy for "
	<< service_name;
	}
	}

	void OnInternalBindResult(
	const std::string& service_name,
	const std::string& interface_name,
	service_manager::mojom::ConnectResult result,
	const absl::optional<service_manager::Identity>& identity) {
	if (result != service_manager::mojom::ConnectResult::SUCCEEDED) {
	LOG(ERROR) << "Failed to bind " << service_name << ":" << interface_name
	<< ", result = " << result;
	}
	}

	} // namespace

	class ExternalMojoBroker::ConnectorImpl : public mojom::ExternalConnector {
	public:
	ConnectorImpl() : connector_facade_(this) {}

	ConnectorImpl(const ConnectorImpl&) = delete;
	ConnectorImpl& operator=(const ConnectorImpl&) = delete;

	void InitializeChromium(
	std::unique_ptr<service_manager::Connector> connector,
	const std::vector<std::string>& external_services_to_proxy) {
	DCHECK(connector);
	connector_ = std::move(connector);
	RegisterExternalServices(external_services_to_proxy);
	}

	void AddReceiver(mojo::PendingReceiver<mojom::ExternalConnector> receiver) {
	receivers_.Add(this, std::move(receiver));
	}

	private:
	class ExternalServiceProxy : public ::service_manager::Service {
	public:
	ExternalServiceProxy(
	ConnectorImpl* connector,
	std::string service_name,
	mojo::PendingReceiver<::service_manager::mojom::Service> receiver)
	: connector_(connector),
	service_name_(std::move(service_name)),
	service_receiver_(this, std::move(receiver)) {
	DCHECK(connector_);
	}

	ExternalServiceProxy(const ExternalServiceProxy&) = delete;
	ExternalServiceProxy& operator=(const ExternalServiceProxy&) = delete;

	private:
	void OnBindInterface(
	const service_manager::BindSourceInfo& source,
	const std::string& interface_name,
	mojo::ScopedMessagePipeHandle interface_pipe) override {
	connector_->BindExternalInterface(service_name_, interface_name,
	std::move(interface_pipe));
	}

	ConnectorImpl* const connector_;
	const std::string service_name_;
	service_manager::ServiceReceiver service_receiver_;
	};

	class ServiceManagerConnectorFacade
	: public service_manager::mojom::Connector {
	public:
	explicit ServiceManagerConnectorFacade(
	ExternalMojoBroker::ConnectorImpl* connector)
	: connector_(connector) {
	DCHECK(connector_);
	}

	void AddReceiver(
	mojo::PendingReceiver<service_manager::mojom::Connector> receiver) {
	receivers_.Add(this, std::move(receiver));
	}

	private:
	void BindInterface(const ::service_manager::ServiceFilter& filter,
	const std::string& interface_name,
	mojo::ScopedMessagePipeHandle interface_pipe,
	service_manager::mojom::BindInterfacePriority priority,
	BindInterfaceCallback callback) override {
	connector_->BindInterface(filter.service_name(), interface_name,
	std::move(interface_pipe));
	std::move(callback).Run(service_manager::mojom::ConnectResult::SUCCEEDED,
	absl::nullopt);
	}

	void QueryService(const std::string& service_name,
	QueryServiceCallback callback) override {
	// TODO(kmackay) Could add a wrapper as needed.
	NOTIMPLEMENTED();
	}

	void WarmService(const ::service_manager::ServiceFilter& filter,
	WarmServiceCallback callback) override {
	std::move(callback).Run(service_manager::mojom::ConnectResult::SUCCEEDED,
	absl::nullopt);
	}

	void RegisterServiceInstance(
	const ::service_manager::Identity& identity,
	mojo::ScopedMessagePipeHandle service,
	mojo::PendingReceiver<service_manager::mojom::ProcessMetadata>
	metadata_receiver,
	RegisterServiceInstanceCallback callback) override {
	// TODO(kmackay) Could add a wrapper as needed.
	NOTIMPLEMENTED();
	}

	void Clone(mojo::PendingReceiver<service_manager::mojom::Connector>
	receiver) override {
	AddReceiver(std::move(receiver));
	}

	ExternalMojoBroker::ConnectorImpl* const connector_;

	mojo::ReceiverSet<service_manager::mojom::Connector> receivers_;
	};

	struct PendingBindRequest {
	PendingBindRequest(const std::string& interface_name,
	mojo::ScopedMessagePipeHandle interface_pipe)
	: interface_name(interface_name),
	interface_pipe(std::move(interface_pipe)) {}

	const std::string interface_name;
	mojo::ScopedMessagePipeHandle interface_pipe;
	};

	void RegisterExternalServices(
	const std::vector<std::string>& external_services_to_proxy) {
	if (external_services_to_proxy.empty()) {
	return;
	}

	external_services_to_proxy_.insert(external_services_to_proxy.begin(),
	external_services_to_proxy.end());

	for (const auto& service_name : external_services_to_proxy) {
	LOG(INFO) << "Register proxy for external " << service_name;
	mojo::PendingRemote<service_manager::mojom::Service> service_remote;
	registered_external_services_[service_name] =
	std::make_unique<ExternalServiceProxy>(
	this, service_name,
	service_remote.InitWithNewPipeAndPassReceiver());

	connector_->RegisterServiceInstance(
	service_manager::Identity(service_name,
	service_manager::kSystemInstanceGroup,
	base::Token{}, base::Token::CreateRandom()),
	std::move(service_remote),
	mojo::NullReceiver() /* metadata_receiver */,
	base::BindOnce(&OnRegisterServiceResult, service_name));
	}
	}

	// Helper for ExternalServiceProxy.
	void BindExternalInterface(const std::string& service_name,
	const std::string& interface_name,
	mojo::ScopedMessagePipeHandle interface_pipe) {
	LOG(INFO) << "Internal request for " << service_name << ":"
	<< interface_name;
	auto it = services_.find(service_name);
	if (it != services_.end()) {
	it->second->OnBindInterface(interface_name, std::move(interface_pipe));
	return;
	}

	ServiceNotFound(service_name, interface_name, std::move(interface_pipe));
	}

	void RegisterServiceInstance(
	const std::string& service_name,
	mojo::PendingRemote<mojom::ExternalService> service_remote) {
	if (base::Contains(services_, service_name)) {
	LOG(ERROR) << "Duplicate service " << service_name;
	return;
	}
	TRACE_EVENT_INSTANT1("mojom", "RegisterService", TRACE_EVENT_SCOPE_THREAD,
	"service", service_name);
	LOG(INFO) << "Register service " << service_name;
	mojo::Remote<mojom::ExternalService> service(std::move(service_remote));
	service.set_disconnect_handler(base::BindOnce(
	&ConnectorImpl::OnServiceLost, base::Unretained(this), service_name));
	auto it = services_.emplace(service_name, std::move(service)).first;

	auto p = pending_bind_requests_.find(service_name);
	if (p != pending_bind_requests_.end()) {
	for (auto& request : p->second) {
	it->second->OnBindInterface(request.interface_name,
	std::move(request.interface_pipe));
	}
	pending_bind_requests_.erase(p);
	}

	auto& info_entry = services_info_[service_name];
	info_entry.name = service_name;
	info_entry.connect_time = base::TimeTicks::Now();
	info_entry.disconnect_time = base::TimeTicks();
	}

	// standalone::mojom::Connector implementation:
	void RegisterServiceInstances(
	std::vector<chromecast::external_mojo::mojom::ServiceInstanceInfoPtr>
	service_instances_info) override {
	for (auto& instance_info_ptr : service_instances_info) {
	RegisterServiceInstance(instance_info_ptr->service_name,
	std::move(instance_info_ptr->service_remote));
	}
	}

	void BindInterface(const std::string& service_name,
	const std::string& interface_name,
	mojo::ScopedMessagePipeHandle interface_pipe) override {
	LOG(INFO) << "Request for " << service_name << ":" << interface_name;
	TRACE_EVENT_INSTANT1("mojom", "BindToService", TRACE_EVENT_SCOPE_THREAD,
	"service", service_name);
	auto it = services_.find(service_name);
	if (it != services_.end()) {
	LOG(INFO) << "Found externally-registered " << service_name;
	it->second->OnBindInterface(interface_name, std::move(interface_pipe));
	return;
	}

	auto service_proxy_it = external_services_to_proxy_.find(service_name);

	if (!connector_ \|\| service_proxy_it != external_services_to_proxy_.end()) {
	ServiceNotFound(service_name, interface_name, std::move(interface_pipe));
	return;
	}

	connector_->QueryService(
	service_name,
	base::BindOnce(&ConnectorImpl::OnQueryResult, base::Unretained(this),
	service_name, interface_name,
	std::move(interface_pipe)));
	}

	void Clone(
	mojo::PendingReceiver<mojom::ExternalConnector> receiver) override {
	AddReceiver(std::move(receiver));
	}

	void BindChromiumConnector(
	mojo::ScopedMessagePipeHandle interface_pipe) override {
	if (!connector_) {
	connector_facade_.AddReceiver(
	mojo::PendingReceiver<service_manager::mojom::Connector>(
	std::move(interface_pipe)));
	return;
	}

	connector_->BindConnectorReceiver(
	mojo::PendingReceiver<service_manager::mojom::Connector>(
	std::move(interface_pipe)));
	}

	void QueryServiceList(QueryServiceListCallback callback) override {
	std::vector<chromecast::external_mojo::mojom::ExternalServiceInfoPtr> infos;
	for (const auto& it : services_info_) {
	infos.emplace_back(it.second.Clone());
	}
	std::move(callback).Run(std::move(infos));
	}

	void OnQueryResult(const std::string& service_name,
	const std::string& interface_name,
	mojo::ScopedMessagePipeHandle interface_pipe,
	service_manager::mojom::ServiceInfoPtr info) {
	if (!info) {
	ServiceNotFound(service_name, interface_name, std::move(interface_pipe));
	return;
	}

	LOG(INFO) << "Found internal " << service_name;
	connector_->BindInterface(
	service_manager::ServiceFilter::ByName(service_name), interface_name,
	std::move(interface_pipe),
	service_manager::mojom::BindInterfacePriority::kImportant,
	base::BindOnce(&OnInternalBindResult, service_name, interface_name));
	}

	void ServiceNotFound(const std::string& service_name,
	const std::string& interface_name,
	mojo::ScopedMessagePipeHandle interface_pipe) {
	LOG(INFO) << service_name << " not found";
	// Assume the service will be registered later, and wait until then.
	pending_bind_requests_[service_name].emplace_back(
	interface_name, std::move(interface_pipe));
	}

	void OnServiceLost(const std::string& service_name) {
	LOG(INFO) << service_name << " disconnected";
	TRACE_EVENT_INSTANT1("mojom", "ServiceDisconnected",
	TRACE_EVENT_SCOPE_THREAD, "service", service_name);
	services_.erase(service_name);
	services_info_[service_name].disconnect_time = base::TimeTicks::Now();
	}

	ServiceManagerConnectorFacade connector_facade_;
	std::unique_ptr<service_manager::Connector> connector_;

	mojo::ReceiverSet<mojom::ExternalConnector> receivers_;
	std::set<std::string> external_services_to_proxy_;
	std::map<std::string, std::unique_ptr<ExternalServiceProxy>>
	registered_external_services_;

	std::map<std::string, mojo::Remote<mojom::ExternalService>> services_;
	std::map<std::string, std::vector<PendingBindRequest>> pending_bind_requests_;
	std::map<std::string, mojom::ExternalServiceInfo> services_info_;
	};

	class ExternalMojoBroker::ReadWatcher
	: public base::MessagePumpForIO::FdWatcher {
	public:
	ReadWatcher(ConnectorImpl* connector, mojo::PlatformHandle listen_handle)
	: connector_(connector),
	listen_handle_(std::move(listen_handle)),
	watch_controller_(FROM_HERE) {
	DCHECK(listen_handle_.is_valid());
	base::CurrentIOThread::Get().WatchFileDescriptor(
	listen_handle_.GetFD().get(), true /* persistent */,
	base::MessagePumpForIO::WATCH_READ, &watch_controller_, this);
	}

	ReadWatcher(const ReadWatcher&) = delete;
	ReadWatcher& operator=(const ReadWatcher&) = delete;

	// base::MessagePumpForIO::FdWatcher implementation:
	void OnFileCanReadWithoutBlocking(int fd) override {
	base::ScopedFD accepted_fd;
	if (mojo::AcceptSocketConnection(fd, &accepted_fd,
	false /* check_peer_user */) &&
	accepted_fd.is_valid()) {
	mojo::OutgoingInvitation invitation;
	auto pipe = invitation.AttachMessagePipe(0);
	mojo::OutgoingInvitation::Send(
	std::move(invitation), base::kNullProcessHandle,
	mojo::PlatformChannelEndpoint(
	mojo::PlatformHandle(std::move(accepted_fd))));
	connector_->AddReceiver(
	mojo::PendingReceiver<mojom::ExternalConnector>(std::move(pipe)));
	}
	}

	void OnFileCanWriteWithoutBlocking(int /* fd */) override {}

	private:
	ConnectorImpl* const connector_;
	const mojo::PlatformHandle listen_handle_;
	base::MessagePumpForIO::FdWatchController watch_controller_;
	};

	ExternalMojoBroker::ExternalMojoBroker(const std::string& broker_path) {
	connector_ = std::make_unique<ConnectorImpl>();

	// For external service support, we expose a channel endpoint on the
	// \|broker_path\|. Otherwise, only services in the same process network can
	// make use of the broker.
	#if BUILDFLAG(ENABLE_EXTERNAL_MOJO_SERVICES)
	#if BUILDFLAG(IS_ANDROID)
	// Monolithic MediaShell can just access the service broker directly in the
	// same process, so there's no need to stand up a server.
	if (!base::android::BundleUtils::IsBundle()) {
	return;
	}
	// On Android, use the abstract namespace to avoid filesystem access.
	bool use_abstract_namespace = true;
	#else
	bool use_abstract_namespace = false;
	#endif // BUILDFLAG(IS_ANDROID)

	LOG(INFO) << "Initializing external mojo broker at: " << broker_path;

	mojo::NamedPlatformChannel::Options channel_options;
	channel_options.server_name = broker_path;
	channel_options.use_abstract_namespace = use_abstract_namespace;
	mojo::NamedPlatformChannel named_channel(channel_options);

	mojo::PlatformChannelServerEndpoint server_endpoint =
	named_channel.TakeServerEndpoint();
	DCHECK(server_endpoint.is_valid());

	#if BUILDFLAG(IS_LINUX) \|\| BUILDFLAG(IS_CHROMEOS)
	chmod(broker_path.c_str(), 0770);
	#endif

	read_watcher_ = std::make_unique<ReadWatcher>(
	connector_.get(), server_endpoint.TakePlatformHandle());
	#endif // BUILDFLAG(ENABLE_EXTERNAL_MOJO_SERVICES)
	}

	void ExternalMojoBroker::InitializeChromium(
	std::unique_ptr<service_manager::Connector> connector,
	const std::vector<std::string>& external_services_to_proxy) {
	connector_->InitializeChromium(std::move(connector),
	external_services_to_proxy);
	}

	mojo::PendingRemote<mojom::ExternalConnector>
	ExternalMojoBroker::CreateConnector() {
	mojo::PendingRemote<mojom::ExternalConnector> remote;
	connector_->AddReceiver(remote.InitWithNewPipeAndPassReceiver());
	return remote;
	}

	void ExternalMojoBroker::BindConnector(
	mojo::PendingReceiver<mojom::ExternalConnector> receiver) {
	connector_->AddReceiver(std::move(receiver));
	}

	ExternalMojoBroker::~ExternalMojoBroker() = default;

	} // namespace external_mojo
	} // namespace chromecast