text: // Copyright 2016 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 "media/remoting/shared_session.h"
  
  #include "base/bind.h"
  #include "base/logging.h"
  #include "media/remoting/proto_utils.h"
  
  namespace media {
  namespace remoting {
  
  SharedSession::SharedSession(mojom::RemotingSourceRequest source_request,
                               mojom::RemoterPtr remoter)
      : rpc_broker_(base::Bind(&SharedSession::SendMessageToSink,
                               base::Unretained(this))),
        binding_(this, std::move(source_request)),
        remoter_(std::move(remoter)) {
    DCHECK(remoter_);
  }
  
  SharedSession::~SharedSession() {
    DCHECK(thread_checker_.CalledOnValidThread());
  
    if (!clients_.empty()) {
      Shutdown();
      clients_.clear();
    }
  }
  
  void SharedSession::OnSinkAvailable(
      mojom::RemotingSinkCapabilities capabilities) {
    DCHECK(thread_checker_.CalledOnValidThread());
  
    if (capabilities == mojom::RemotingSinkCapabilities::NONE) {
      OnSinkGone();
      return;
    }
    sink_capabilities_ = capabilities;
    if (state_ == SESSION_UNAVAILABLE)
      UpdateAndNotifyState(SESSION_CAN_START);
  }
  
  void SharedSession::OnSinkGone() {
    DCHECK(thread_checker_.CalledOnValidThread());
  
    sink_capabilities_ = mojom::RemotingSinkCapabilities::NONE;
  
    if (state_ == SESSION_PERMANENTLY_STOPPED)
      return;
    if (state_ == SESSION_CAN_START) {
      UpdateAndNotifyState(SESSION_UNAVAILABLE);
      return;
    }
    if (state_ == SESSION_STARTED || state_ == SESSION_STARTING) {
      VLOG(1) << "Sink is gone in a remoting session.";
      // Remoting is being stopped by Remoter.
      UpdateAndNotifyState(SESSION_STOPPING);
    }
  }
  
  void SharedSession::OnStarted() {
    DCHECK(thread_checker_.CalledOnValidThread());
  
    VLOG(1) << "Remoting started successively.";
    if (clients_.empty() || state_ == SESSION_PERMANENTLY_STOPPED ||
        state_ == SESSION_STOPPING) {
      for (Client* client : clients_)
        client->OnStarted(false);
      return;
    }
    for (Client* client : clients_)
      client->OnStarted(true);
    state_ = SESSION_STARTED;
  }
  
  void SharedSession::OnStartFailed(mojom::RemotingStartFailReason reason) {
    DCHECK(thread_checker_.CalledOnValidThread());
  
    VLOG(1) << "Failed to start remoting:" << reason;
    for (Client* client : clients_)
      client->OnStarted(false);
    if (state_ == SESSION_PERMANENTLY_STOPPED)
      return;
    state_ = SESSION_UNAVAILABLE;
  }
  
  void SharedSession::OnStopped(mojom::RemotingStopReason reason) {
    DCHECK(thread_checker_.CalledOnValidThread());
  
    VLOG(1) << "Remoting stopped: " << reason;
    if (state_ == SESSION_PERMANENTLY_STOPPED)
      return;
    UpdateAndNotifyState(SESSION_UNAVAILABLE);
  }
  
  void SharedSession::OnMessageFromSink(const std::vector<uint8_t>& message) {
    DCHECK(thread_checker_.CalledOnValidThread());
  
    std::unique_ptr<pb::RpcMessage> rpc(new pb::RpcMessage());
    if (!rpc->ParseFromArray(message.data(), message.size())) {
      VLOG(1) << "corrupted Rpc message";
      Shutdown();
      return;
    }
    rpc_broker_.ProcessMessageFromRemote(std::move(rpc));
  }
  
  void SharedSession::UpdateAndNotifyState(SessionState state) {
    DCHECK(thread_checker_.CalledOnValidThread());
  
    if (state_ == state)
      return;
    state_ = state;
    for (Client* client : clients_)
      client->OnSessionStateChanged();
  }
  
  void SharedSession::StartRemoting(Client* client) {
    DCHECK(std::find(clients_.begin(), clients_.end(), client) != clients_.end());
  
    switch (state_) {
      case SESSION_CAN_START:
        remoter_->Start();
        UpdateAndNotifyState(SESSION_STARTING);
        break;
      case SESSION_STARTING:
        break;
      case SESSION_STARTED:
        client->OnStarted(true);
        break;
      case SESSION_STOPPING:
      case SESSION_UNAVAILABLE:
      case SESSION_PERMANENTLY_STOPPED:
        client->OnStarted(false);
        break;
    }
  }
  
  void SharedSession::StopRemoting(Client* client) {
    DCHECK(std::find(clients_.begin(), clients_.end(), client) != clients_.end());
  
    VLOG(1) << "SharedSession::StopRemoting: " << state_;
  
    if (state_ != SESSION_STARTING && state_ != SESSION_STARTED)
      return;
  
    remoter_->Stop(mojom::RemotingStopReason::LOCAL_PLAYBACK);
    UpdateAndNotifyState(SESSION_STOPPING);
  }
  
  void SharedSession::AddClient(Client* client) {
    DCHECK(thread_checker_.CalledOnValidThread());
    DCHECK(std::find(clients_.begin(), clients_.end(), client) == clients_.end());
  
    clients_.push_back(client);
  }
  
  void SharedSession::RemoveClient(Client* client) {
    DCHECK(thread_checker_.CalledOnValidThread());
  
    auto it = std::find(clients_.begin(), clients_.end(), client);
    DCHECK(it != clients_.end());
  
    clients_.erase(it);
    if (clients_.empty() &&
        (state_ == SESSION_STARTED || state_ == SESSION_STARTING)) {
      remoter_->Stop(mojom::RemotingStopReason::SOURCE_GONE);
      state_ = SESSION_STOPPING;
    }
  }
  
  void SharedSession::Shutdown() {
    DCHECK(thread_checker_.CalledOnValidThread());
  
    if (state_ == SESSION_STARTED || state_ == SESSION_STARTING)
      remoter_->Stop(mojom::RemotingStopReason::UNEXPECTED_FAILURE);
    UpdateAndNotifyState(SESSION_PERMANENTLY_STOPPED);
  }
  
  void SharedSession::StartDataPipe(
      std::unique_ptr<mojo::DataPipe> audio_data_pipe,
      std::unique_ptr<mojo::DataPipe> video_data_pipe,
      const DataPipeStartCallback& done_callback) {
    DCHECK(thread_checker_.CalledOnValidThread());
    DCHECK(!done_callback.is_null());
  
    bool audio = audio_data_pipe != nullptr;
    bool video = video_data_pipe != nullptr;
    if (!audio && !video) {
      LOG(ERROR) << "No audio nor video to establish data pipe";
      done_callback.Run(mojom::RemotingDataStreamSenderPtrInfo(),
                        mojom::RemotingDataStreamSenderPtrInfo(),
                        mojo::ScopedDataPipeProducerHandle(),
                        mojo::ScopedDataPipeProducerHandle());
      return;
    }
    mojom::RemotingDataStreamSenderPtr audio_stream_sender;
    mojom::RemotingDataStreamSenderPtr video_stream_sender;
    remoter_->StartDataStreams(audio ? std::move(audio_data_pipe->consumer_handle)
                                     : mojo::ScopedDataPipeConsumerHandle(),
                               video ? std::move(video_data_pipe->consumer_handle)
                                     : mojo::ScopedDataPipeConsumerHandle(),
                               audio ? mojo::MakeRequest(&audio_stream_sender)
                                     : mojom::RemotingDataStreamSenderRequest(),
                               video ? mojo::MakeRequest(&video_stream_sender)
                                     : mojom::RemotingDataStreamSenderRequest());
    done_callback.Run(audio_stream_sender.PassInterface(),
                      video_stream_sender.PassInterface(),
                      audio ? std::move(audio_data_pipe->producer_handle)
                            : mojo::ScopedDataPipeProducerHandle(),
                      video ? std::move(video_data_pipe->producer_handle)
                            : mojo::ScopedDataPipeProducerHandle());
  }
  
  void SharedSession::SendMessageToSink(
      std::unique_ptr<std::vector<uint8_t>> message) {
    DCHECK(thread_checker_.CalledOnValidThread());
    remoter_->SendMessageToSink(*message);
  }
  
  }  // namespace remoting
  }  // namespace media
  