cast/video_receiver/video_receiver.cc

// Copyright 2013 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/cast/video_receiver/video_receiver.h"

#include <algorithm>

#include "base/bind.h"
#include "base/logging.h"
#include "base/message_loop/message_loop.h"
#include "crypto/encryptor.h"
#include "crypto/symmetric_key.h"
#include "media/cast/cast_defines.h"
#include "media/cast/framer/framer.h"
#include "media/cast/video_receiver/video_decoder.h"

namespace media {
namespace cast {

const int64 kMinSchedulingDelayMs = 1;

static const int64 kMinTimeBetweenOffsetUpdatesMs = 2000;
static const int kTimeOffsetFilter = 8;
static const int64_t kMinProcessIntervalMs = 5;

// Local implementation of RtpData (defined in rtp_rtcp_defines.h).
// Used to pass payload data into the video receiver.
class LocalRtpVideoData : public RtpData {
 public:
  explicit LocalRtpVideoData(VideoReceiver* video_receiver)
      : video_receiver_(video_receiver) {}

  virtual ~LocalRtpVideoData() {}

  virtual void OnReceivedPayloadData(const uint8* payload_data,
                                     size_t payload_size,
                                     const RtpCastHeader* rtp_header) OVERRIDE {
    video_receiver_->IncomingParsedRtpPacket(payload_data, payload_size,
                                             *rtp_header);
  }

 private:
  VideoReceiver* video_receiver_;
};

// Local implementation of RtpPayloadFeedback (defined in rtp_defines.h)
// Used to convey cast-specific feedback from receiver to sender.
// Callback triggered by the Framer (cast message builder).
class LocalRtpVideoFeedback : public RtpPayloadFeedback {
 public:
  explicit LocalRtpVideoFeedback(VideoReceiver* video_receiver)
      : video_receiver_(video_receiver) {
  }

  virtual void CastFeedback(const RtcpCastMessage& cast_message) OVERRIDE {
    video_receiver_->CastFeedback(cast_message);
  }

 private:
  VideoReceiver* video_receiver_;
};

// Local implementation of RtpReceiverStatistics (defined by rtcp.h).
// Used to pass statistics data from the RTP module to the RTCP module.
class LocalRtpReceiverStatistics : public RtpReceiverStatistics {
 public:
  explicit LocalRtpReceiverStatistics(RtpReceiver* rtp_receiver)
     : rtp_receiver_(rtp_receiver) {
  }

  virtual void GetStatistics(uint8* fraction_lost,
                             uint32* cumulative_lost,  // 24 bits valid.
                             uint32* extended_high_sequence_number,
                             uint32* jitter) OVERRIDE {
    rtp_receiver_->GetStatistics(fraction_lost,
                                 cumulative_lost,
                                 extended_high_sequence_number,
                                 jitter);
  }

 private:
  RtpReceiver* rtp_receiver_;
};

VideoReceiver::VideoReceiver(scoped_refptr<CastEnvironment> cast_environment,
                             const VideoReceiverConfig& video_config,
                             PacedPacketSender* const packet_sender)
      : cast_environment_(cast_environment),
        codec_(video_config.codec),
        target_delay_delta_(
            base::TimeDelta::FromMilliseconds(video_config.rtp_max_delay_ms)),
        frame_delay_(base::TimeDelta::FromMilliseconds(
            1000 / video_config.max_frame_rate)),
        incoming_payload_callback_(new LocalRtpVideoData(this)),
        incoming_payload_feedback_(new LocalRtpVideoFeedback(this)),
        rtp_receiver_(cast_environment_->Clock(), NULL, &video_config,
                      incoming_payload_callback_.get()),
        rtp_video_receiver_statistics_(
            new LocalRtpReceiverStatistics(&rtp_receiver_)),
        time_incoming_packet_updated_(false),
        incoming_rtp_timestamp_(0),
        weak_factory_(this) {
  int max_unacked_frames = video_config.rtp_max_delay_ms *
      video_config.max_frame_rate / 1000;
  DCHECK(max_unacked_frames) << "Invalid argument";

  if (video_config.aes_iv_mask.size() == kAesKeySize &&
      video_config.aes_key.size() == kAesKeySize) {
    iv_mask_ = video_config.aes_iv_mask;
    crypto::SymmetricKey* key = crypto::SymmetricKey::Import(
        crypto::SymmetricKey::AES, video_config.aes_key);
    decryptor_.reset(new crypto::Encryptor());
    decryptor_->Init(key, crypto::Encryptor::CTR, std::string());
  } else if (video_config.aes_iv_mask.size() != 0 ||
             video_config.aes_key.size() != 0) {
    DCHECK(false) << "Invalid crypto configuration";
  }

  framer_.reset(new Framer(cast_environment->Clock(),
                           incoming_payload_feedback_.get(),
                           video_config.incoming_ssrc,
                           video_config.decoder_faster_than_max_frame_rate,
                           max_unacked_frames));
  if (!video_config.use_external_decoder) {
    video_decoder_.reset(new VideoDecoder(video_config, cast_environment));
  }

  rtcp_.reset(
      new Rtcp(cast_environment_,
               NULL,
               packet_sender,
               NULL,
               rtp_video_receiver_statistics_.get(),
               video_config.rtcp_mode,
               base::TimeDelta::FromMilliseconds(video_config.rtcp_interval),
               video_config.feedback_ssrc,
               video_config.incoming_ssrc,
               video_config.rtcp_c_name));
}

VideoReceiver::~VideoReceiver() {}

void VideoReceiver::InitializeTimers() {
  DCHECK(cast_environment_->CurrentlyOn(CastEnvironment::MAIN));
  ScheduleNextRtcpReport();
  ScheduleNextCastMessage();
}

void VideoReceiver::GetRawVideoFrame(
    const VideoFrameDecodedCallback& callback) {
  DCHECK(cast_environment_->CurrentlyOn(CastEnvironment::MAIN));
  GetEncodedVideoFrame(base::Bind(&VideoReceiver::DecodeVideoFrame,
                                  base::Unretained(this), callback));
}

// Called when we have a frame to decode.
void VideoReceiver::DecodeVideoFrame(
    const VideoFrameDecodedCallback& callback,
    scoped_ptr<EncodedVideoFrame> encoded_frame,
    const base::TimeTicks& render_time) {
  DCHECK(cast_environment_->CurrentlyOn(CastEnvironment::MAIN));
  // Hand the ownership of the encoded frame to the decode thread.
  cast_environment_->PostTask(CastEnvironment::VIDEO_DECODER, FROM_HERE,
      base::Bind(&VideoReceiver::DecodeVideoFrameThread, base::Unretained(this),
                 base::Passed(&encoded_frame), render_time, callback));
}

// Utility function to run the decoder on a designated decoding thread.
void VideoReceiver::DecodeVideoFrameThread(
    scoped_ptr<EncodedVideoFrame> encoded_frame,
    const base::TimeTicks render_time,
    const VideoFrameDecodedCallback& frame_decoded_callback) {
  DCHECK(cast_environment_->CurrentlyOn(CastEnvironment::VIDEO_DECODER));
  DCHECK(video_decoder_);

  if (!(video_decoder_->DecodeVideoFrame(encoded_frame.get(), render_time,
        frame_decoded_callback))) {
    // This will happen if we decide to decode but not show a frame.
    cast_environment_->PostTask(CastEnvironment::MAIN, FROM_HERE,
        base::Bind(&VideoReceiver::GetRawVideoFrame, base::Unretained(this),
                   frame_decoded_callback));
  }
}

bool VideoReceiver::DecryptVideoFrame(
    scoped_ptr<EncodedVideoFrame>* video_frame) {
  DCHECK(cast_environment_->CurrentlyOn(CastEnvironment::MAIN));
  DCHECK(decryptor_) << "Invalid state";

  if (!decryptor_->SetCounter(GetAesNonce((*video_frame)->frame_id,
                                          iv_mask_))) {
    NOTREACHED() << "Failed to set counter";
    return false;
  }
  std::string decrypted_video_data;
  if (!decryptor_->Decrypt((*video_frame)->data, &decrypted_video_data)) {
    VLOG(1) << "Decryption error";
    // Give up on this frame, release it from jitter buffer.
    framer_->ReleaseFrame((*video_frame)->frame_id);
    return false;
  }
  (*video_frame)->data.swap(decrypted_video_data);
  return true;
}

// Called from the main cast thread.
void VideoReceiver::GetEncodedVideoFrame(
    const VideoFrameEncodedCallback& callback) {
  DCHECK(cast_environment_->CurrentlyOn(CastEnvironment::MAIN));
  scoped_ptr<EncodedVideoFrame> encoded_frame(new EncodedVideoFrame());
  uint32 rtp_timestamp = 0;
  bool next_frame = false;

  if (!framer_->GetEncodedVideoFrame(encoded_frame.get(), &rtp_timestamp,
                                     &next_frame)) {
    // We have no video frames. Wait for new packet(s).
    queued_encoded_callbacks_.push_back(callback);
    return;
  }

  if (decryptor_ && !DecryptVideoFrame(&encoded_frame)) {
    // Logging already done.
    queued_encoded_callbacks_.push_back(callback);
    return;
  }

  base::TimeTicks render_time;
  if (PullEncodedVideoFrame(rtp_timestamp, next_frame, &encoded_frame,
                            &render_time)) {
    cast_environment_->PostTask(CastEnvironment::MAIN, FROM_HERE,
        base::Bind(callback, base::Passed(&encoded_frame), render_time));
  } else {
    // We have a video frame; however we are missing packets and we have time
    // to wait for new packet(s).
    queued_encoded_callbacks_.push_back(callback);
  }
}

// Should we pull the encoded video frame from the framer? decided by if this is
// the next frame or we are running out of time and have to pull the following
// frame.
// If the frame is too old to be rendered we set the don't show flag in the
// video bitstream where possible.
bool VideoReceiver::PullEncodedVideoFrame(uint32 rtp_timestamp,
    bool next_frame, scoped_ptr<EncodedVideoFrame>* encoded_frame,
    base::TimeTicks* render_time) {
  DCHECK(cast_environment_->CurrentlyOn(CastEnvironment::MAIN));
  base::TimeTicks now = cast_environment_->Clock()->NowTicks();
  *render_time = GetRenderTime(now, rtp_timestamp);

  // TODO(mikhal): Store actual render time and not diff.
  cast_environment_->Logging()->InsertFrameEventWithDelay(kVideoRenderDelay,
      rtp_timestamp, (*encoded_frame)->frame_id, now - *render_time);

  // Minimum time before a frame is due to be rendered before we pull it for
  // decode.
  base::TimeDelta min_wait_delta = frame_delay_;
  base::TimeDelta time_until_render = *render_time - now;
  if (!next_frame && (time_until_render > min_wait_delta)) {
    // Example:
    // We have decoded frame 1 and we have received the complete frame 3, but
    // not frame 2. If we still have time before frame 3 should be rendered we
    // will wait for 2 to arrive, however if 2 never show up this timer will hit
    // and we will pull out frame 3 for decoding and rendering.
    base::TimeDelta time_until_release = time_until_render - min_wait_delta;
    cast_environment_->PostDelayedTask(CastEnvironment::MAIN, FROM_HERE,
        base::Bind(&VideoReceiver::PlayoutTimeout, weak_factory_.GetWeakPtr()),
        time_until_release);
    VLOG(1) << "Wait before releasing frame "
            << static_cast<int>((*encoded_frame)->frame_id)
            << " time " << time_until_release.InMilliseconds();
    return false;
  }

  base::TimeDelta dont_show_timeout_delta =
      base::TimeDelta::FromMilliseconds(-kDontShowTimeoutMs);
  if (codec_ == kVp8 && time_until_render < dont_show_timeout_delta) {
    (*encoded_frame)->data[0] &= 0xef;
    VLOG(1) << "Don't show frame "
            << static_cast<int>((*encoded_frame)->frame_id)
            << " time_until_render:" << time_until_render.InMilliseconds();
  } else {
    VLOG(1) << "Show frame "
            << static_cast<int>((*encoded_frame)->frame_id)
            << " time_until_render:" << time_until_render.InMilliseconds();
  }
  // We have a copy of the frame, release this one.
  framer_->ReleaseFrame((*encoded_frame)->frame_id);
  (*encoded_frame)->codec = codec_;
  return true;
}

void VideoReceiver::PlayoutTimeout() {
  DCHECK(cast_environment_->CurrentlyOn(CastEnvironment::MAIN));
  if (queued_encoded_callbacks_.empty()) return;

  uint32 rtp_timestamp = 0;
  bool next_frame = false;
  scoped_ptr<EncodedVideoFrame> encoded_frame(new EncodedVideoFrame());

  if (!framer_->GetEncodedVideoFrame(encoded_frame.get(), &rtp_timestamp,
                                     &next_frame)) {
    // We have no video frames. Wait for new packet(s).
    // Since the application can post multiple VideoFrameEncodedCallback and
    // we only check the next frame to play out we might have multiple timeout
    // events firing after each other; however this should be a rare event.
    VLOG(1) << "Failed to retrieved a complete frame at this point in time";
    return;
  }
  VLOG(1) << "PlayoutTimeout retrieved frame "
          << static_cast<int>(encoded_frame->frame_id);

  if (decryptor_ && !DecryptVideoFrame(&encoded_frame)) {
    // Logging already done.
    return;
  }

  base::TimeTicks render_time;
  if (PullEncodedVideoFrame(rtp_timestamp, next_frame, &encoded_frame,
                            &render_time)) {
    if (!queued_encoded_callbacks_.empty()) {
      VideoFrameEncodedCallback callback = queued_encoded_callbacks_.front();
      queued_encoded_callbacks_.pop_front();
      cast_environment_->PostTask(CastEnvironment::MAIN, FROM_HERE,
          base::Bind(callback, base::Passed(&encoded_frame), render_time));
    }
  } else {
    // We have a video frame; however we are missing packets and we have time
    // to wait for new packet(s).
  }
}

base::TimeTicks VideoReceiver::GetRenderTime(base::TimeTicks now,
                                             uint32 rtp_timestamp) {
  DCHECK(cast_environment_->CurrentlyOn(CastEnvironment::MAIN));
  // Senders time in ms when this frame was captured.
  // Note: the senders clock and our local clock might not be synced.
  base::TimeTicks rtp_timestamp_in_ticks;

  if (time_offset_.InMilliseconds() == 0) {
    if (!rtcp_->RtpTimestampInSenderTime(kVideoFrequency,
                                         incoming_rtp_timestamp_,
                                         &rtp_timestamp_in_ticks)) {
      // We have not received any RTCP to sync the stream play it out as soon as
      // possible.
      return now;
    }
    time_offset_ = time_incoming_packet_ - rtp_timestamp_in_ticks;
  } else if (time_incoming_packet_updated_) {
    if (rtcp_->RtpTimestampInSenderTime(kVideoFrequency,
                                        incoming_rtp_timestamp_,
                                        &rtp_timestamp_in_ticks)) {
      // Time to update the time_offset.
      base::TimeDelta time_offset =
          time_incoming_packet_ - rtp_timestamp_in_ticks;
      time_offset_ = ((kTimeOffsetFilter - 1) * time_offset_ + time_offset)
          / kTimeOffsetFilter;
    }
  }
  // Reset |time_incoming_packet_updated_| to enable a future measurement.
  time_incoming_packet_updated_ = false;
  if (!rtcp_->RtpTimestampInSenderTime(kVideoFrequency,
                                       rtp_timestamp,
                                       &rtp_timestamp_in_ticks)) {
    // This can fail if we have not received any RTCP packets in a long time.
    return now;
  }
  return (rtp_timestamp_in_ticks + time_offset_ + target_delay_delta_);
}

void VideoReceiver::IncomingPacket(const uint8* packet, size_t length,
                                   const base::Closure callback) {
  DCHECK(cast_environment_->CurrentlyOn(CastEnvironment::MAIN));
  if (Rtcp::IsRtcpPacket(packet, length)) {
    rtcp_->IncomingRtcpPacket(packet, length);
  } else {
    rtp_receiver_.ReceivedPacket(packet, length);
  }
  cast_environment_->PostTask(CastEnvironment::MAIN, FROM_HERE, callback);
}

void VideoReceiver::IncomingParsedRtpPacket(const uint8* payload_data,
                                            size_t payload_size,
                                            const RtpCastHeader& rtp_header) {
  DCHECK(cast_environment_->CurrentlyOn(CastEnvironment::MAIN));

  base::TimeTicks now = cast_environment_->Clock()->NowTicks();
  if (time_incoming_packet_.is_null() || now - time_incoming_packet_ >
      base::TimeDelta::FromMilliseconds(kMinTimeBetweenOffsetUpdatesMs)) {
    if (time_incoming_packet_.is_null()) InitializeTimers();
    incoming_rtp_timestamp_ = rtp_header.webrtc.header.timestamp;
    time_incoming_packet_ = now;
    time_incoming_packet_updated_ = true;
  }

  cast_environment_->Logging()->InsertPacketEvent(kPacketReceived,
      rtp_header.webrtc.header.timestamp, rtp_header.frame_id,
      rtp_header.packet_id, rtp_header.max_packet_id, payload_size);

  bool complete = framer_->InsertPacket(payload_data, payload_size, rtp_header);

  if (!complete) return;  // Video frame not complete; wait for more packets.
  if (queued_encoded_callbacks_.empty()) return;  // No pending callback.

  VideoFrameEncodedCallback callback = queued_encoded_callbacks_.front();
  queued_encoded_callbacks_.pop_front();
  cast_environment_->PostTask(CastEnvironment::MAIN, FROM_HERE,
      base::Bind(&VideoReceiver::GetEncodedVideoFrame,
          weak_factory_.GetWeakPtr(), callback));
}

// Send a cast feedback message. Actual message created in the framer (cast
// message builder).
void VideoReceiver::CastFeedback(const RtcpCastMessage& cast_message) {
  DCHECK(cast_environment_->CurrentlyOn(CastEnvironment::MAIN));
  // TODO(pwestin): wire up log messages.
  rtcp_->SendRtcpFromRtpReceiver(&cast_message, NULL);
  time_last_sent_cast_message_= cast_environment_->Clock()->NowTicks();
}

// Cast messages should be sent within a maximum interval. Schedule a call
// if not triggered elsewhere, e.g. by the cast message_builder.
void VideoReceiver::ScheduleNextCastMessage() {
  DCHECK(cast_environment_->CurrentlyOn(CastEnvironment::MAIN));
  base::TimeTicks send_time;
  framer_->TimeToSendNextCastMessage(&send_time);

  base::TimeDelta time_to_send = send_time -
      cast_environment_->Clock()->NowTicks();
  time_to_send = std::max(time_to_send,
      base::TimeDelta::FromMilliseconds(kMinSchedulingDelayMs));
  cast_environment_->PostDelayedTask(CastEnvironment::MAIN, FROM_HERE,
      base::Bind(&VideoReceiver::SendNextCastMessage,
                 weak_factory_.GetWeakPtr()), time_to_send);
}

void VideoReceiver::SendNextCastMessage() {
  DCHECK(cast_environment_->CurrentlyOn(CastEnvironment::MAIN));
  framer_->SendCastMessage();  // Will only send a message if it is time.
  ScheduleNextCastMessage();
}

// Schedule the next RTCP report to be sent back to the sender.
void VideoReceiver::ScheduleNextRtcpReport() {
  DCHECK(cast_environment_->CurrentlyOn(CastEnvironment::MAIN));
  base::TimeDelta time_to_next = rtcp_->TimeToSendNextRtcpReport() -
      cast_environment_->Clock()->NowTicks();

  time_to_next = std::max(time_to_next,
      base::TimeDelta::FromMilliseconds(kMinSchedulingDelayMs));

  cast_environment_->PostDelayedTask(CastEnvironment::MAIN, FROM_HERE,
      base::Bind(&VideoReceiver::SendNextRtcpReport,
                weak_factory_.GetWeakPtr()), time_to_next);
}

void VideoReceiver::SendNextRtcpReport() {
  DCHECK(cast_environment_->CurrentlyOn(CastEnvironment::MAIN));
  rtcp_->SendRtcpFromRtpReceiver(NULL, NULL);
  ScheduleNextRtcpReport();
}

}  // namespace cast
}  // namespace media