// 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/audio_receiver/audio_receiver.h"
#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/audio_receiver/audio_decoder.h"
#include "media/cast/framer/framer.h"
#include "media/cast/rtcp/rtcp.h"
#include "media/cast/rtp_receiver/rtp_receiver.h"
// Max time we wait until an audio frame is due to be played out is released.
static const int64 kMaxAudioFrameWaitMs = 20;
static const int64 kMinSchedulingDelayMs = 1;
namespace media {
namespace cast {
DecodedAudioCallbackData::DecodedAudioCallbackData()
: number_of_10ms_blocks(0),
desired_frequency(0),
callback() {}
DecodedAudioCallbackData::~DecodedAudioCallbackData() {}
// Local implementation of RtpData (defined in rtp_rtcp_defines.h).
// Used to pass payload data into the audio receiver.
class LocalRtpAudioData : public RtpData {
public:
explicit LocalRtpAudioData(AudioReceiver* audio_receiver)
: audio_receiver_(audio_receiver) {}
virtual void OnReceivedPayloadData(
const uint8* payload_data,
size_t payload_size,
const RtpCastHeader* rtp_header) OVERRIDE {
audio_receiver_->IncomingParsedRtpPacket(payload_data, payload_size,
*rtp_header);
}
private:
AudioReceiver* audio_receiver_;
};
// Local implementation of RtpPayloadFeedback (defined in rtp_defines.h)
// Used to convey cast-specific feedback from receiver to sender.
class LocalRtpAudioFeedback : public RtpPayloadFeedback {
public:
explicit LocalRtpAudioFeedback(AudioReceiver* audio_receiver)
: audio_receiver_(audio_receiver) {
}
virtual void CastFeedback(const RtcpCastMessage& cast_message) OVERRIDE {
audio_receiver_->CastFeedback(cast_message);
}
private:
AudioReceiver* audio_receiver_;
};
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_;
};
AudioReceiver::AudioReceiver(scoped_refptr<CastEnvironment> cast_environment,
const AudioReceiverConfig& audio_config,
PacedPacketSender* const packet_sender)
: cast_environment_(cast_environment),
codec_(audio_config.codec),
frequency_(audio_config.frequency),
audio_buffer_(),
audio_decoder_(),
time_offset_(),
weak_factory_(this) {
target_delay_delta_ =
base::TimeDelta::FromMilliseconds(audio_config.rtp_max_delay_ms);
incoming_payload_callback_.reset(new LocalRtpAudioData(this));
incoming_payload_feedback_.reset(new LocalRtpAudioFeedback(this));
if (audio_config.use_external_decoder) {
audio_buffer_.reset(new Framer(cast_environment->Clock(),
incoming_payload_feedback_.get(),
audio_config.incoming_ssrc,
true,
0));
} else {
audio_decoder_.reset(new AudioDecoder(cast_environment,
audio_config,
incoming_payload_feedback_.get()));
}
if (audio_config.aes_iv_mask.size() == kAesKeySize &&
audio_config.aes_key.size() == kAesKeySize) {
iv_mask_ = audio_config.aes_iv_mask;
crypto::SymmetricKey* key = crypto::SymmetricKey::Import(
crypto::SymmetricKey::AES, audio_config.aes_key);
decryptor_.reset(new crypto::Encryptor());
decryptor_->Init(key, crypto::Encryptor::CTR, std::string());
} else if (audio_config.aes_iv_mask.size() != 0 ||
audio_config.aes_key.size() != 0) {
DCHECK(false) << "Invalid crypto configuration";
}
rtp_receiver_.reset(new RtpReceiver(cast_environment->Clock(),
&audio_config,
NULL,
incoming_payload_callback_.get()));
rtp_audio_receiver_statistics_.reset(
new LocalRtpReceiverStatistics(rtp_receiver_.get()));
base::TimeDelta rtcp_interval_delta =
base::TimeDelta::FromMilliseconds(audio_config.rtcp_interval);
rtcp_.reset(new Rtcp(cast_environment,
NULL,
packet_sender,
NULL,
rtp_audio_receiver_statistics_.get(),
audio_config.rtcp_mode,
rtcp_interval_delta,
audio_config.feedback_ssrc,
audio_config.incoming_ssrc,
audio_config.rtcp_c_name));
}
AudioReceiver::~AudioReceiver() {}
void AudioReceiver::InitializeTimers() {
DCHECK(cast_environment_->CurrentlyOn(CastEnvironment::MAIN));
ScheduleNextRtcpReport();
ScheduleNextCastMessage();
}
void AudioReceiver::IncomingParsedRtpPacket(const uint8* payload_data,
size_t payload_size,
const RtpCastHeader& rtp_header) {
DCHECK(cast_environment_->CurrentlyOn(CastEnvironment::MAIN));
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);
// TODO(pwestin): update this as video to refresh over time.
DCHECK(cast_environment_->CurrentlyOn(CastEnvironment::MAIN));
if (time_first_incoming_packet_.is_null()) {
InitializeTimers();
first_incoming_rtp_timestamp_ = rtp_header.webrtc.header.timestamp;
time_first_incoming_packet_ = cast_environment_->Clock()->NowTicks();
}
if (audio_decoder_) {
DCHECK(!audio_buffer_) << "Invalid internal state";
std::string plaintext(reinterpret_cast<const char*>(payload_data),
payload_size);
if (decryptor_) {
plaintext.clear();
if (!decryptor_->SetCounter(GetAesNonce(rtp_header.frame_id, iv_mask_))) {
NOTREACHED() << "Failed to set counter";
return;
}
if (!decryptor_->Decrypt(base::StringPiece(reinterpret_cast<const char*>(
payload_data), payload_size), &plaintext)) {
VLOG(0) << "Decryption error";
return;
}
}
audio_decoder_->IncomingParsedRtpPacket(
reinterpret_cast<const uint8*>(plaintext.data()), plaintext.size(),
rtp_header);
if (!queued_decoded_callbacks_.empty()) {
DecodedAudioCallbackData decoded_data = queued_decoded_callbacks_.front();
queued_decoded_callbacks_.pop_front();
cast_environment_->PostTask(CastEnvironment::AUDIO_DECODER, FROM_HERE,
base::Bind(&AudioReceiver::DecodeAudioFrameThread,
base::Unretained(this),
decoded_data.number_of_10ms_blocks,
decoded_data.desired_frequency,
decoded_data.callback));
}
return;
}
DCHECK(audio_buffer_) << "Invalid internal state";
DCHECK(!audio_decoder_) << "Invalid internal state";
bool complete = audio_buffer_->InsertPacket(payload_data, payload_size,
rtp_header);
if (!complete) return; // Audio frame not complete; wait for more packets.
if (queued_encoded_callbacks_.empty()) return;
AudioFrameEncodedCallback callback = queued_encoded_callbacks_.front();
queued_encoded_callbacks_.pop_front();
cast_environment_->PostTask(CastEnvironment::MAIN, FROM_HERE,
base::Bind(&AudioReceiver::GetEncodedAudioFrame,
weak_factory_.GetWeakPtr(), callback));
}
void AudioReceiver::GetRawAudioFrame(int number_of_10ms_blocks,
int desired_frequency, const AudioFrameDecodedCallback& callback) {
DCHECK(cast_environment_->CurrentlyOn(CastEnvironment::MAIN));
DCHECK(audio_decoder_) << "Invalid function call in this configuration";
// TODO(pwestin): we can skip this function by posting direct to the decoder.
cast_environment_->PostTask(CastEnvironment::AUDIO_DECODER, FROM_HERE,
base::Bind(&AudioReceiver::DecodeAudioFrameThread,
base::Unretained(this),
number_of_10ms_blocks,
desired_frequency,
callback));
}
void AudioReceiver::DecodeAudioFrameThread(
int number_of_10ms_blocks,
int desired_frequency,
const AudioFrameDecodedCallback callback) {
DCHECK(cast_environment_->CurrentlyOn(CastEnvironment::AUDIO_DECODER));
// TODO(mikhal): Allow the application to allocate this memory.
scoped_ptr<PcmAudioFrame> audio_frame(new PcmAudioFrame());
uint32 rtp_timestamp = 0;
if (!audio_decoder_->GetRawAudioFrame(number_of_10ms_blocks,
desired_frequency,
audio_frame.get(),
&rtp_timestamp)) {
DecodedAudioCallbackData callback_data;
callback_data.number_of_10ms_blocks = number_of_10ms_blocks;
callback_data.desired_frequency = desired_frequency;
callback_data.callback = callback;
queued_decoded_callbacks_.push_back(callback_data);
return;
}
base::TimeTicks now = cast_environment_->Clock()->NowTicks();
cast_environment_->PostTask(CastEnvironment::MAIN, FROM_HERE,
base::Bind(&AudioReceiver::ReturnDecodedFrameWithPlayoutDelay,
base::Unretained(this), base::Passed(&audio_frame), rtp_timestamp,
callback));
}
void AudioReceiver::ReturnDecodedFrameWithPlayoutDelay(
scoped_ptr<PcmAudioFrame> audio_frame, uint32 rtp_timestamp,
const AudioFrameDecodedCallback callback) {
DCHECK(cast_environment_->CurrentlyOn(CastEnvironment::MAIN));
cast_environment_->Logging()->InsertFrameEvent(kAudioFrameDecoded,
rtp_timestamp, kFrameIdUnknown);
base::TimeTicks now = cast_environment_->Clock()->NowTicks();
base::TimeTicks playout_time = GetPlayoutTime(now, rtp_timestamp);
cast_environment_->Logging()->InsertFrameEventWithDelay(kAudioPlayoutDelay,
rtp_timestamp, kFrameIdUnknown, playout_time - now);
// Frame is ready - Send back to the caller.
cast_environment_->PostTask(CastEnvironment::MAIN, FROM_HERE,
base::Bind(callback, base::Passed(&audio_frame), playout_time));
}
void AudioReceiver::PlayoutTimeout() {
DCHECK(cast_environment_->CurrentlyOn(CastEnvironment::MAIN));
DCHECK(audio_buffer_) << "Invalid function call in this configuration";
if (queued_encoded_callbacks_.empty()) {
// Already released by incoming packet.
return;
}
uint32 rtp_timestamp = 0;
bool next_frame = false;
scoped_ptr<EncodedAudioFrame> encoded_frame(new EncodedAudioFrame());
if (!audio_buffer_->GetEncodedAudioFrame(encoded_frame.get(),
&rtp_timestamp, &next_frame)) {
// We have no audio frames. Wait for new packet(s).
// Since the application can post multiple AudioFrameEncodedCallback 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;
}
if (decryptor_ && !DecryptAudioFrame(&encoded_frame)) {
// Logging already done.
return;
}
if (PostEncodedAudioFrame(queued_encoded_callbacks_.front(), rtp_timestamp,
next_frame, &encoded_frame)) {
// Call succeed remove callback from list.
queued_encoded_callbacks_.pop_front();
}
}
void AudioReceiver::GetEncodedAudioFrame(
const AudioFrameEncodedCallback& callback) {
DCHECK(cast_environment_->CurrentlyOn(CastEnvironment::MAIN));
DCHECK(audio_buffer_) << "Invalid function call in this configuration";
uint32 rtp_timestamp = 0;
bool next_frame = false;
scoped_ptr<EncodedAudioFrame> encoded_frame(new EncodedAudioFrame());
if (!audio_buffer_->GetEncodedAudioFrame(encoded_frame.get(),
&rtp_timestamp, &next_frame)) {
// We have no audio frames. Wait for new packet(s).
VLOG(1) << "Wait for more audio packets in frame";
queued_encoded_callbacks_.push_back(callback);
return;
}
if (decryptor_ && !DecryptAudioFrame(&encoded_frame)) {
// Logging already done.
queued_encoded_callbacks_.push_back(callback);
return;
}
if (!PostEncodedAudioFrame(callback, rtp_timestamp, next_frame,
&encoded_frame)) {
// We have an audio frame; however we are missing packets and we have time
// to wait for new packet(s).
queued_encoded_callbacks_.push_back(callback);
}
}
bool AudioReceiver::PostEncodedAudioFrame(
const AudioFrameEncodedCallback& callback,
uint32 rtp_timestamp,
bool next_frame,
scoped_ptr<EncodedAudioFrame>* encoded_frame) {
DCHECK(cast_environment_->CurrentlyOn(CastEnvironment::MAIN));
DCHECK(audio_buffer_) << "Invalid function call in this configuration";
base::TimeTicks now = cast_environment_->Clock()->NowTicks();
base::TimeTicks playout_time = GetPlayoutTime(now, rtp_timestamp);
base::TimeDelta time_until_playout = playout_time - now;
base::TimeDelta min_wait_delta =
base::TimeDelta::FromMilliseconds(kMaxAudioFrameWaitMs);
if (!next_frame && (time_until_playout > min_wait_delta)) {
base::TimeDelta time_until_release = time_until_playout - min_wait_delta;
cast_environment_->PostDelayedTask(CastEnvironment::MAIN, FROM_HERE,
base::Bind(&AudioReceiver::PlayoutTimeout, weak_factory_.GetWeakPtr()),
time_until_release);
VLOG(1) << "Wait until time to playout:"
<< time_until_release.InMilliseconds();
return false;
}
(*encoded_frame)->codec = codec_;
audio_buffer_->ReleaseFrame((*encoded_frame)->frame_id);
cast_environment_->PostTask(CastEnvironment::MAIN, FROM_HERE,
base::Bind(callback, base::Passed(encoded_frame), playout_time));
return true;
}
void AudioReceiver::IncomingPacket(const uint8* packet, size_t length,
const base::Closure callback) {
DCHECK(cast_environment_->CurrentlyOn(CastEnvironment::MAIN));
bool rtcp_packet = Rtcp::IsRtcpPacket(packet, length);
if (!rtcp_packet) {
rtp_receiver_->ReceivedPacket(packet, length);
} else {
rtcp_->IncomingRtcpPacket(packet, length);
}
cast_environment_->PostTask(CastEnvironment::MAIN, FROM_HERE, callback);
}
void AudioReceiver::CastFeedback(const RtcpCastMessage& cast_message) {
// TODO(pwestin): add logging.
rtcp_->SendRtcpFromRtpReceiver(&cast_message, NULL);
}
base::TimeTicks AudioReceiver::GetPlayoutTime(base::TimeTicks now,
uint32 rtp_timestamp) {
DCHECK(cast_environment_->CurrentlyOn(CastEnvironment::MAIN));
// Senders time in ms when this frame was recorded.
// Note: the senders clock and our local clock might not be synced.
base::TimeTicks rtp_timestamp_in_ticks;
if (time_offset_ == base::TimeDelta()) {
if (rtcp_->RtpTimestampInSenderTime(frequency_,
first_incoming_rtp_timestamp_,
&rtp_timestamp_in_ticks)) {
time_offset_ = time_first_incoming_packet_ - rtp_timestamp_in_ticks;
} else {
// We have not received any RTCP to sync the stream play it out as soon as
// possible.
uint32 rtp_timestamp_diff = rtp_timestamp - first_incoming_rtp_timestamp_;
int frequency_khz = frequency_ / 1000;
base::TimeDelta rtp_time_diff_delta =
base::TimeDelta::FromMilliseconds(rtp_timestamp_diff / frequency_khz);
base::TimeDelta time_diff_delta = now - time_first_incoming_packet_;
return now + std::max(rtp_time_diff_delta - time_diff_delta,
base::TimeDelta());
}
}
// This can fail if we have not received any RTCP packets in a long time.
return rtcp_->RtpTimestampInSenderTime(frequency_, rtp_timestamp,
&rtp_timestamp_in_ticks) ?
rtp_timestamp_in_ticks + time_offset_ + target_delay_delta_ :
now;
}
bool AudioReceiver::DecryptAudioFrame(
scoped_ptr<EncodedAudioFrame>* audio_frame) {
DCHECK(decryptor_) << "Invalid state";
if (!decryptor_->SetCounter(GetAesNonce((*audio_frame)->frame_id,
iv_mask_))) {
NOTREACHED() << "Failed to set counter";
return false;
}
std::string decrypted_audio_data;
if (!decryptor_->Decrypt((*audio_frame)->data, &decrypted_audio_data)) {
VLOG(0) << "Decryption error";
// Give up on this frame, release it from jitter buffer.
audio_buffer_->ReleaseFrame((*audio_frame)->frame_id);
return false;
}
(*audio_frame)->data.swap(decrypted_audio_data);
return true;
}
void AudioReceiver::ScheduleNextRtcpReport() {
DCHECK(cast_environment_->CurrentlyOn(CastEnvironment::MAIN));
base::TimeDelta time_to_send = rtcp_->TimeToSendNextRtcpReport() -
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(&AudioReceiver::SendNextRtcpReport,
weak_factory_.GetWeakPtr()), time_to_send);
}
void AudioReceiver::SendNextRtcpReport() {
DCHECK(cast_environment_->CurrentlyOn(CastEnvironment::MAIN));
// TODO(pwestin): add logging.
rtcp_->SendRtcpFromRtpReceiver(NULL, NULL);
ScheduleNextRtcpReport();
}
// Cast messages should be sent within a maximum interval. Schedule a call
// if not triggered elsewhere, e.g. by the cast message_builder.
void AudioReceiver::ScheduleNextCastMessage() {
DCHECK(cast_environment_->CurrentlyOn(CastEnvironment::MAIN));
base::TimeTicks send_time;
if (audio_buffer_) {
audio_buffer_->TimeToSendNextCastMessage(&send_time);
} else if (audio_decoder_) {
audio_decoder_->TimeToSendNextCastMessage(&send_time);
} else {
NOTREACHED();
}
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(&AudioReceiver::SendNextCastMessage,
weak_factory_.GetWeakPtr()), time_to_send);
}
void AudioReceiver::SendNextCastMessage() {
DCHECK(cast_environment_->CurrentlyOn(CastEnvironment::MAIN));
if (audio_buffer_) {
// Will only send a message if it is time.
audio_buffer_->SendCastMessage();
}
if (audio_decoder_) {
// Will only send a message if it is time.
audio_decoder_->SendCastMessage();
}
ScheduleNextCastMessage();
}
} // namespace cast
} // namespace media