services/media/factory_service/media_sink_impl.cc

// 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 "base/logging.h"
#include "mojo/services/media/common/cpp/linear_transform.h"
#include "mojo/services/media/common/cpp/local_time.h"
#include "services/media/factory_service/media_sink_impl.h"
#include "services/media/framework/conversion_pipeline_builder.h"
#include "services/media/framework_mojo/mojo_type_conversions.h"

namespace mojo {
namespace media {

// static
std::shared_ptr<MediaSinkImpl> MediaSinkImpl::Create(
    const String& destination_url,
    MediaTypePtr media_type,
    InterfaceRequest<MediaSink> request,
    MediaFactoryService* owner) {
  return std::shared_ptr<MediaSinkImpl>(new MediaSinkImpl(
      destination_url,
      media_type.Pass(),
      request.Pass(),
      owner));
}

MediaSinkImpl::MediaSinkImpl(
    const String& destination_url,
    MediaTypePtr media_type,
    InterfaceRequest<MediaSink> request,
    MediaFactoryService* owner)
    : MediaFactoryService::Product(owner),
      binding_(this, request.Pass()),
      consumer_(MojoConsumer::Create()),
      producer_(MojoProducer::Create()) {
  DCHECK(destination_url);
  DCHECK(media_type);

  // Go away when the client is no longer connected.
  binding_.set_connection_error_handler([this]() {
    ReleaseFromOwner();
  });

  // TODO(dalesat): Support file and network urls.
  // TODO(dalesat): Support mojo services in a reasonable way.

  PartRef consumer_ref = graph_.Add(consumer_);
  PartRef producer_ref = graph_.Add(producer_);

  consumer_->SetPrimeRequestedCallback(
      [this](const MediaConsumer::PrimeCallback& callback) {
        ready_.When([this, callback]() {
          DCHECK(producer_);
          producer_->PrimeConnection(callback);
        });
      });
  consumer_->SetFlushRequestedCallback(
      [this, consumer_ref](const MediaConsumer::FlushCallback& callback) {
        ready_.When([this, consumer_ref, callback]() {
          DCHECK(producer_);
          graph_.FlushOutput(consumer_ref.output());
          producer_->FlushConnection(callback);
          flushed_ = true;
        });
      });

  producer_->SetStatusCallback(
      [this](MediaState state) {
        producer_state_ = state;
        StatusUpdated();
        if (state == MediaState::ENDED) {
          Pause();
        }
      });

  if (destination_url != "mojo:audio_server") {
    LOG(ERROR) << "mojo:audio_server is the only supported destination";
    if (binding_.is_bound()) {
      binding_.Close();
    }
    return;
  }

  // TODO(dalesat): Once we have c++14, get rid of this shared pointer hack.
  std::shared_ptr<StreamType>
      captured_stream_type(Convert(media_type).release());

  // An AudioTrackController knows how to talk to an audio track, interrogating
  // it for supported stream types and configuring it for the chosen stream
  // type.
  controller_.reset(new AudioTrackController(destination_url, app()));

  controller_->GetSupportedMediaTypes(
      [this, consumer_ref, producer_ref, captured_stream_type]
      (std::unique_ptr<std::vector<std::unique_ptr<StreamTypeSet>>>
          supported_stream_types) {
        std::unique_ptr<StreamType> producer_stream_type;

        // Add transforms to the pipeline to convert from stream_type to a
        // type supported by the track.
        OutputRef out = consumer_ref.output();
        bool result = BuildConversionPipeline(
            *captured_stream_type,
            *supported_stream_types,
            &graph_,
            &out,
            &producer_stream_type);
        if (!result) {
          // Failed to build conversion pipeline.
          producer_state_ = MediaState::FAULT;
          StatusUpdated();
          return;
        }

        graph_.ConnectOutputToPart(out, producer_ref);

        switch (producer_stream_type->scheme()) {
          case StreamType::Scheme::kLpcm:
            frames_per_second_ =
                producer_stream_type->lpcm()->frames_per_second();
            break;
          case StreamType::Scheme::kCompressedAudio:
            frames_per_second_ =
                producer_stream_type->compressed_audio()->frames_per_second();
            break;
          default:
            // Unsupported producer stream type.
            producer_state_ = MediaState::FAULT;
            StatusUpdated();
            return;
        }

        controller_->Configure(
            std::move(producer_stream_type),
            [this]
                (MediaConsumerPtr consumer, RateControlPtr rate_control) {
                  DCHECK(consumer);
                  DCHECK(rate_control);
                  rate_control_ = rate_control.Pass();
                  producer_->Connect(consumer.Pass(), [this]() {
                    graph_.Prepare();
                    ready_.Occur();
                    MaybeSetRate();
                  });
                });
      });
}

MediaSinkImpl::~MediaSinkImpl() {}

void MediaSinkImpl::GetClockDisposition(
    const GetClockDispositionCallback& callback) {
  callback.Run(ClockDisposition::PREFER_MASTER);
  // TODO(dalesat): Varies by destination type.
}

void MediaSinkImpl::GetMasterClock(InterfaceRequest<Clock> master_clock) {
  // TODO(dalesat): Produce master clock as appropriate.
}

void MediaSinkImpl::SetMasterClock(InterfaceHandle<Clock> master_clock) {
  // TODO(dalesat): Accept master clock and arrange for synchronization.
}

void MediaSinkImpl::GetConsumer(InterfaceRequest<MediaConsumer> consumer) {
  consumer_->AddBinding(consumer.Pass());
}

void MediaSinkImpl::GetStatus(
    uint64_t version_last_seen,
    const GetStatusCallback& callback) {
  if (version_last_seen < status_version_) {
    RunStatusCallback(callback);
  } else {
    pending_status_requests_.push_back(callback);
  }
}

void MediaSinkImpl::Play() {
  target_rate_ = 1.0;
  MaybeSetRate();
}

void MediaSinkImpl::Pause() {
  target_rate_ = 0.0;
  MaybeSetRate();
}

void MediaSinkImpl::StatusUpdated() {
  ++status_version_;
  while (!pending_status_requests_.empty()) {
    RunStatusCallback(pending_status_requests_.front());
    pending_status_requests_.pop_front();
  }
}

void MediaSinkImpl::RunStatusCallback(const GetStatusCallback& callback) const {
  MediaSinkStatusPtr status = MediaSinkStatus::New();
  status->state = (producer_state_ == MediaState::PAUSED && rate_ != 0.0) ?
      MediaState::PLAYING :
      producer_state_;
  status->timeline_transform = status_transform_.Clone();
  callback.Run(status_version_, status.Pass());
}

void MediaSinkImpl::MaybeSetRate() {
  if (producer_state_ < MediaState::PAUSED || rate_ == target_rate_) {
    return;
  }

  DCHECK(rate_control_);

  // Desired rate in frames per second.
  LinearTransform::Ratio rate_frames_per_second(
      static_cast<uint32_t>(frames_per_second_ * target_rate_), 1);

  // Local time rate in seconds_per_tick.
  LinearTransform::Ratio local_seconds_per_tick(
      LocalDuration::period::num,
      LocalDuration::period::den);

  // Desired rate in frames per local tick.
  LinearTransform::Ratio rate_frames_per_tick;
  bool success =
      LinearTransform::Ratio::Compose(
          local_seconds_per_tick,
          rate_frames_per_second,
          &rate_frames_per_tick);
  DCHECK(success)
      << "LinearTransform::Ratio::Compose reports loss of precision";

  // TODO(dalesat): start_local_time should be supplied via the mojo interface.
  // For now, it's hard-coded to be 30ms in the future.
  // The local time when we want the rate to change.
  int64_t start_local_time =
      (LocalClock::now().time_since_epoch() + std::chrono::milliseconds(30)).
          count();

  // The media time corresponding to start_local_time.
  int64_t start_media_time;
  if (flushed_ &&
      producer_->GetFirstPtsSinceFlush() != Packet::kUnknownPts) {
    // We're getting started initially or after a flush/prime, so the media
    // time corresponding to start_local_time should be the PTS of
    // the first packet.
    start_media_time = producer_->GetFirstPtsSinceFlush();
  } else {
    // We're resuming, so the media time corresponding to start_local_time can
    // be calculated using the existing transform.
    success =
        transform_.DoForwardTransform(start_local_time, &start_media_time);
    DCHECK(success)
        << "LinearTransform::DoForwardTransform reports loss of precision";
  }

  flushed_ = false;

  // Update the transform.
  transform_ =
      LinearTransform(start_local_time, rate_frames_per_tick, start_media_time);

  // Set the rate.
  TimelineQuadPtr rate_quad = TimelineQuad::New();
  rate_quad->reference_offset = start_media_time;
  rate_quad->target_offset = start_local_time;
  rate_quad->reference_delta = rate_frames_per_tick.numerator;
  rate_quad->target_delta = rate_frames_per_tick.denominator;

  rate_control_->SetCurrentQuad(rate_quad.Pass());

  // Get the frame rate in frames per local tick.
  LinearTransform::Ratio frame_rate_frames_per_second(frames_per_second_, 1);
  LinearTransform::Ratio frame_rate_frames_per_tick;
  success = LinearTransform::Ratio::Compose(
      local_seconds_per_tick,
      frame_rate_frames_per_second,
      &frame_rate_frames_per_tick);
  DCHECK(success)
      << "LinearTransform::Ratio::Compose reports loss of precision";

  // Create a LinearTransform to translate from presentation units to
  // local duration units.
  LinearTransform local_to_presentation(0, frame_rate_frames_per_tick, 0);

  status_transform_ = TimelineTransform::New();
  status_transform_->quad = TimelineQuad::New();

  // Translate the current transform quad so the presentation time units
  // are the same as the local time units.
  success = local_to_presentation.DoReverseTransform(
      start_media_time,
      &status_transform_->quad->reference_offset);
  DCHECK(success)
      << "LinearTransform::DoReverseTransform reports loss of precision";
  status_transform_->quad->target_offset = start_local_time;
  int64_t presentation_delta;
  success = local_to_presentation.DoReverseTransform(
      static_cast<int64_t>(rate_frames_per_tick.numerator),
      &presentation_delta);
  DCHECK(success)
      << "LinearTransform::DoReverseTransform reports loss of precision";
  status_transform_->quad->reference_delta =
      static_cast<int32_t>(presentation_delta);
  status_transform_->quad->target_delta = rate_frames_per_tick.denominator;
  LinearTransform::Ratio::Reduce(
      &status_transform_->quad->reference_delta,
      &status_transform_->quad->target_delta);

  rate_ = target_rate_;
  StatusUpdated();
}

} // namespace media
} // namespace mojo