media/cast/net/cast_transport_sender_impl.cc

   1 // Copyright 2014 The Chromium Authors. All rights reserved.
   2 // Use of this source code is governed by a BSD-style license that can be
   3 // found in the LICENSE file.
   4
   5 #include "media/cast/net/cast_transport_sender_impl.h"
   6
   7 #include "base/single_thread_task_runner.h"
   8 #include "base/values.h"
   9 #include "media/cast/net/cast_transport_config.h"
  10 #include "media/cast/net/cast_transport_defines.h"
  11 #include "media/cast/net/udp_transport.h"
  12 #include "net/base/net_errors.h"
  13 #include "net/base/net_util.h"
  14
  15 namespace media {
  16 namespace cast {
  17
  18 namespace {
  19
  20 // See header file for what these mean.
  21 const char kOptionDscp[] = "DSCP";
  22 #if defined(OS_WIN)
  23 const char kOptionNonBlockingIO[] = "non_blocking_io";
  24 #endif
  25 const char kOptionPacerTargetBurstSize[] = "pacer_target_burst_size";
  26 const char kOptionPacerMaxBurstSize[] = "pacer_max_burst_size";
  27 const char kOptionSendBufferMinSize[] = "send_buffer_min_size";
  28 const char kOptionWifiDisableScan[] = "disable_wifi_scan";
  29 const char kOptionWifiMediaStreamingMode[] = "media_streaming_mode";
  30
  31 int LookupOptionWithDefault(const base::DictionaryValue& options,
  32                             const std::string& path,
  33                             int default_value) {
  34   int ret;
  35   if (options.GetInteger(path, &ret)) {
  36     return ret;
  37   } else {
  38     return default_value;
  39   }
  40 };
  41
  42 int32 GetTransportSendBufferSize(const base::DictionaryValue& options) {
  43   // Socket send buffer size needs to be at least greater than one burst
  44   // size.
  45   int32 max_burst_size =
  46       LookupOptionWithDefault(options, kOptionPacerMaxBurstSize,
  47                               kMaxBurstSize) * kMaxIpPacketSize;
  48   int32 min_send_buffer_size =
  49       LookupOptionWithDefault(options, kOptionSendBufferMinSize, 0);
  50   return std::max(max_burst_size, min_send_buffer_size);
  51 }
  52
  53 }  // namespace
  54
  55 scoped_ptr<CastTransportSender> CastTransportSender::Create(
  56     net::NetLog* net_log,
  57     base::TickClock* clock,
  58     const net::IPEndPoint& local_end_point,
  59     const net::IPEndPoint& remote_end_point,
  60     scoped_ptr<base::DictionaryValue> options,
  61     const CastTransportStatusCallback& status_callback,
  62     const BulkRawEventsCallback& raw_events_callback,
  63     base::TimeDelta raw_events_callback_interval,
  64     const PacketReceiverCallback& packet_callback,
  65     const scoped_refptr<base::SingleThreadTaskRunner>& transport_task_runner) {
  66   return scoped_ptr<CastTransportSender>(
  67       new CastTransportSenderImpl(net_log,
  68                                   clock,
  69                                   local_end_point,
  70                                   remote_end_point,
  71                                   options.Pass(),
  72                                   status_callback,
  73                                   raw_events_callback,
  74                                   raw_events_callback_interval,
  75                                   transport_task_runner.get(),
  76                                   packet_callback,
  77                                   NULL));
  78 }
  79
  80 PacketReceiverCallback CastTransportSender::PacketReceiverForTesting() {
  81   return PacketReceiverCallback();
  82 }
  83
  84 CastTransportSenderImpl::CastTransportSenderImpl(
  85     net::NetLog* net_log,
  86     base::TickClock* clock,
  87     const net::IPEndPoint& local_end_point,
  88     const net::IPEndPoint& remote_end_point,
  89     scoped_ptr<base::DictionaryValue> options,
  90     const CastTransportStatusCallback& status_callback,
  91     const BulkRawEventsCallback& raw_events_callback,
  92     base::TimeDelta raw_events_callback_interval,
  93     const scoped_refptr<base::SingleThreadTaskRunner>& transport_task_runner,
  94     const PacketReceiverCallback& packet_callback,
  95     PacketSender* external_transport)
  96     : clock_(clock),
  97       status_callback_(status_callback),
  98       transport_task_runner_(transport_task_runner),
  99       transport_(
 100           external_transport ?
 101               NULL :
 102               new UdpTransport(net_log,
 103                                transport_task_runner,
 104                                local_end_point,
 105                                remote_end_point,
 106                                GetTransportSendBufferSize(*options),
 107                                status_callback)),
 108       pacer_(LookupOptionWithDefault(*options,
 109                                      kOptionPacerTargetBurstSize,
 110                                      kTargetBurstSize),
 111              LookupOptionWithDefault(*options,
 112                                      kOptionPacerMaxBurstSize,
 113                                      kMaxBurstSize),
 114              clock,
 115              &logging_,
 116              external_transport ? external_transport : transport_.get(),
 117              transport_task_runner),
 118       raw_events_callback_(raw_events_callback),
 119       raw_events_callback_interval_(raw_events_callback_interval),
 120       last_byte_acked_for_audio_(0),
 121       packet_callback_(packet_callback),
 122       weak_factory_(this) {
 123   DCHECK(clock_);
 124   if (!raw_events_callback_.is_null()) {
 125     DCHECK(raw_events_callback_interval > base::TimeDelta());
 126     event_subscriber_.reset(new SimpleEventSubscriber);
 127     logging_.AddRawEventSubscriber(event_subscriber_.get());
 128     transport_task_runner->PostDelayedTask(
 129         FROM_HERE,
 130         base::Bind(&CastTransportSenderImpl::SendRawEvents,
 131                    weak_factory_.GetWeakPtr()),
 132         raw_events_callback_interval);
 133   }
 134   if (transport_) {
 135     if (options->HasKey(kOptionDscp)) {
 136       // The default DSCP value for cast is AF41. Which gives it a higher
 137       // priority over other traffic.
 138       transport_->SetDscp(net::DSCP_AF41);
 139     }
 140 #if defined(OS_WIN)
 141     if (options->HasKey(kOptionNonBlockingIO)) {
 142       transport_->UseNonBlockingIO();
 143     }
 144 #endif
 145     transport_->StartReceiving(
 146         base::Bind(&CastTransportSenderImpl::OnReceivedPacket,
 147                    base::Unretained(this)));
 148     int wifi_options = 0;
 149     if (options->HasKey(kOptionWifiDisableScan)) {
 150       wifi_options |= net::WIFI_OPTIONS_DISABLE_SCAN;
 151     }
 152     if (options->HasKey(kOptionWifiMediaStreamingMode)) {
 153       wifi_options |= net::WIFI_OPTIONS_MEDIA_STREAMING_MODE;
 154     }
 155     if (wifi_options) {
 156       wifi_options_autoreset_ = net::SetWifiOptions(wifi_options);
 157     }
 158   }
 159 }
 160
 161 CastTransportSenderImpl::~CastTransportSenderImpl() {
 162   if (transport_) {
 163     transport_->StopReceiving();
 164   }
 165   if (event_subscriber_.get())
 166     logging_.RemoveRawEventSubscriber(event_subscriber_.get());
 167 }
 168
 169 void CastTransportSenderImpl::InitializeAudio(
 170     const CastTransportRtpConfig& config,
 171     const RtcpCastMessageCallback& cast_message_cb,
 172     const RtcpRttCallback& rtt_cb) {
 173   LOG_IF(WARNING, config.aes_key.empty() || config.aes_iv_mask.empty())
 174       << "Unsafe to send audio with encryption DISABLED.";
 175   if (!audio_encryptor_.Initialize(config.aes_key, config.aes_iv_mask)) {
 176     status_callback_.Run(TRANSPORT_AUDIO_UNINITIALIZED);
 177     return;
 178   }
 179
 180   audio_sender_.reset(new RtpSender(transport_task_runner_, &pacer_));
 181   if (audio_sender_->Initialize(config)) {
 182     // Audio packets have a higher priority.
 183     pacer_.RegisterAudioSsrc(config.ssrc);
 184     pacer_.RegisterPrioritySsrc(config.ssrc);
 185     status_callback_.Run(TRANSPORT_AUDIO_INITIALIZED);
 186   } else {
 187     audio_sender_.reset();
 188     status_callback_.Run(TRANSPORT_AUDIO_UNINITIALIZED);
 189     return;
 190   }
 191
 192   audio_rtcp_session_.reset(
 193       new Rtcp(base::Bind(&CastTransportSenderImpl::OnReceivedCastMessage,
 194                           weak_factory_.GetWeakPtr(), config.ssrc,
 195                           cast_message_cb),
 196                rtt_cb,
 197                base::Bind(&CastTransportSenderImpl::OnReceivedLogMessage,
 198                           weak_factory_.GetWeakPtr(), AUDIO_EVENT),
 199                clock_,
 200                &pacer_,
 201                config.ssrc,
 202                config.feedback_ssrc));
 203   pacer_.RegisterAudioSsrc(config.ssrc);
 204   AddValidSsrc(config.feedback_ssrc);
 205   status_callback_.Run(TRANSPORT_AUDIO_INITIALIZED);
 206 }
 207
 208 void CastTransportSenderImpl::InitializeVideo(
 209     const CastTransportRtpConfig& config,
 210     const RtcpCastMessageCallback& cast_message_cb,
 211     const RtcpRttCallback& rtt_cb) {
 212   LOG_IF(WARNING, config.aes_key.empty() || config.aes_iv_mask.empty())
 213       << "Unsafe to send video with encryption DISABLED.";
 214   if (!video_encryptor_.Initialize(config.aes_key, config.aes_iv_mask)) {
 215     status_callback_.Run(TRANSPORT_VIDEO_UNINITIALIZED);
 216     return;
 217   }
 218
 219   video_sender_.reset(new RtpSender(transport_task_runner_, &pacer_));
 220   if (!video_sender_->Initialize(config)) {
 221     video_sender_.reset();
 222     status_callback_.Run(TRANSPORT_VIDEO_UNINITIALIZED);
 223     return;
 224   }
 225
 226   video_rtcp_session_.reset(
 227       new Rtcp(base::Bind(&CastTransportSenderImpl::OnReceivedCastMessage,
 228                           weak_factory_.GetWeakPtr(), config.ssrc,
 229                           cast_message_cb),
 230                rtt_cb,
 231                base::Bind(&CastTransportSenderImpl::OnReceivedLogMessage,
 232                           weak_factory_.GetWeakPtr(), VIDEO_EVENT),
 233                clock_,
 234                &pacer_,
 235                config.ssrc,
 236                config.feedback_ssrc));
 237   pacer_.RegisterVideoSsrc(config.ssrc);
 238   AddValidSsrc(config.feedback_ssrc);
 239   status_callback_.Run(TRANSPORT_VIDEO_INITIALIZED);
 240 }
 241
 242 namespace {
 243 void EncryptAndSendFrame(const EncodedFrame& frame,
 244                          TransportEncryptionHandler* encryptor,
 245                          RtpSender* sender) {
 246   if (encryptor->is_activated()) {
 247     EncodedFrame encrypted_frame;
 248     frame.CopyMetadataTo(&encrypted_frame);
 249     if (encryptor->Encrypt(frame.frame_id, frame.data, &encrypted_frame.data)) {
 250       sender->SendFrame(encrypted_frame);
 251     } else {
 252       LOG(ERROR) << "Encryption failed.  Not sending frame with ID "
 253                  << frame.frame_id;
 254     }
 255   } else {
 256     sender->SendFrame(frame);
 257   }
 258 }
 259 }  // namespace
 260
 261 void CastTransportSenderImpl::InsertFrame(uint32 ssrc,
 262                                           const EncodedFrame& frame) {
 263   if (audio_sender_ && ssrc == audio_sender_->ssrc()) {
 264     EncryptAndSendFrame(frame, &audio_encryptor_, audio_sender_.get());
 265   } else if (video_sender_ && ssrc == video_sender_->ssrc()) {
 266     EncryptAndSendFrame(frame, &video_encryptor_, video_sender_.get());
 267   } else {
 268     NOTREACHED() << "Invalid InsertFrame call.";
 269   }
 270 }
 271
 272 void CastTransportSenderImpl::SendSenderReport(
 273     uint32 ssrc,
 274     base::TimeTicks current_time,
 275     uint32 current_time_as_rtp_timestamp) {
 276   if (audio_sender_ && ssrc == audio_sender_->ssrc()) {
 277     audio_rtcp_session_->SendRtcpFromRtpSender(
 278         current_time, current_time_as_rtp_timestamp,
 279         audio_sender_->send_packet_count(), audio_sender_->send_octet_count());
 280   } else if (video_sender_ && ssrc == video_sender_->ssrc()) {
 281     video_rtcp_session_->SendRtcpFromRtpSender(
 282         current_time, current_time_as_rtp_timestamp,
 283         video_sender_->send_packet_count(), video_sender_->send_octet_count());
 284   } else {
 285     NOTREACHED() << "Invalid request for sending RTCP packet.";
 286   }
 287 }
 288
 289 void CastTransportSenderImpl::CancelSendingFrames(
 290     uint32 ssrc,
 291     const std::vector<uint32>& frame_ids) {
 292   if (audio_sender_ && ssrc == audio_sender_->ssrc()) {
 293     audio_sender_->CancelSendingFrames(frame_ids);
 294   } else if (video_sender_ && ssrc == video_sender_->ssrc()) {
 295     video_sender_->CancelSendingFrames(frame_ids);
 296   } else {
 297     NOTREACHED() << "Invalid request for cancel sending.";
 298   }
 299 }
 300
 301 void CastTransportSenderImpl::ResendFrameForKickstart(uint32 ssrc,
 302                                                       uint32 frame_id) {
 303   if (audio_sender_ && ssrc == audio_sender_->ssrc()) {
 304     DCHECK(audio_rtcp_session_);
 305     audio_sender_->ResendFrameForKickstart(
 306         frame_id,
 307         audio_rtcp_session_->current_round_trip_time());
 308   } else if (video_sender_ && ssrc == video_sender_->ssrc()) {
 309     DCHECK(video_rtcp_session_);
 310     video_sender_->ResendFrameForKickstart(
 311         frame_id,
 312         video_rtcp_session_->current_round_trip_time());
 313   } else {
 314     NOTREACHED() << "Invalid request for kickstart.";
 315   }
 316 }
 317
 318 void CastTransportSenderImpl::ResendPackets(
 319     uint32 ssrc,
 320     const MissingFramesAndPacketsMap& missing_packets,
 321     bool cancel_rtx_if_not_in_list,
 322     const DedupInfo& dedup_info) {
 323   if (audio_sender_ && ssrc == audio_sender_->ssrc()) {
 324     audio_sender_->ResendPackets(missing_packets,
 325                                  cancel_rtx_if_not_in_list,
 326                                  dedup_info);
 327   } else if (video_sender_ && ssrc == video_sender_->ssrc()) {
 328     video_sender_->ResendPackets(missing_packets,
 329                                  cancel_rtx_if_not_in_list,
 330                                  dedup_info);
 331   } else {
 332     NOTREACHED() << "Invalid request for retransmission.";
 333   }
 334 }
 335
 336 PacketReceiverCallback CastTransportSenderImpl::PacketReceiverForTesting() {
 337   return base::Bind(
 338       base::IgnoreResult(&CastTransportSenderImpl::OnReceivedPacket),
 339       weak_factory_.GetWeakPtr());
 340 }
 341
 342 void CastTransportSenderImpl::SendRawEvents() {
 343   DCHECK(event_subscriber_.get());
 344   DCHECK(!raw_events_callback_.is_null());
 345   std::vector<PacketEvent> packet_events;
 346   std::vector<FrameEvent> frame_events;
 347   event_subscriber_->GetPacketEventsAndReset(&packet_events);
 348   event_subscriber_->GetFrameEventsAndReset(&frame_events);
 349   raw_events_callback_.Run(packet_events, frame_events);
 350
 351   transport_task_runner_->PostDelayedTask(
 352       FROM_HERE,
 353       base::Bind(&CastTransportSenderImpl::SendRawEvents,
 354                  weak_factory_.GetWeakPtr()),
 355       raw_events_callback_interval_);
 356 }
 357
 358 bool CastTransportSenderImpl::OnReceivedPacket(scoped_ptr<Packet> packet) {
 359   const uint8_t* const data = &packet->front();
 360   const size_t length = packet->size();
 361   uint32 ssrc;
 362   if (Rtcp::IsRtcpPacket(data, length)) {
 363     ssrc = Rtcp::GetSsrcOfSender(data, length);
 364   } else if (!RtpParser::ParseSsrc(data, length, &ssrc)) {
 365     VLOG(1) << "Invalid RTP packet.";
 366     return false;
 367   }
 368   if (valid_ssrcs_.find(ssrc) == valid_ssrcs_.end()) {
 369     VLOG(1) << "Stale packet received.";
 370     return false;
 371   }
 372
 373   if (audio_rtcp_session_ &&
 374       audio_rtcp_session_->IncomingRtcpPacket(data, length)) {
 375     return true;
 376   }
 377   if (video_rtcp_session_ &&
 378       video_rtcp_session_->IncomingRtcpPacket(data, length)) {
 379     return true;
 380   }
 381   if (packet_callback_.is_null()) {
 382     VLOG(1) << "Stale packet received.";
 383     return false;
 384   }
 385   packet_callback_.Run(packet.Pass());
 386   return true;
 387 }
 388
 389 void CastTransportSenderImpl::OnReceivedLogMessage(
 390     EventMediaType media_type,
 391     const RtcpReceiverLogMessage& log) {
 392   // Add received log messages into our log system.
 393   RtcpReceiverLogMessage::const_iterator it = log.begin();
 394   for (; it != log.end(); ++it) {
 395     uint32 rtp_timestamp = it->rtp_timestamp_;
 396
 397     RtcpReceiverEventLogMessages::const_iterator event_it =
 398         it->event_log_messages_.begin();
 399     for (; event_it != it->event_log_messages_.end(); ++event_it) {
 400       switch (event_it->type) {
 401         case PACKET_RECEIVED:
 402           logging_.InsertPacketEvent(
 403               event_it->event_timestamp, event_it->type,
 404               media_type, rtp_timestamp,
 405               kFrameIdUnknown, event_it->packet_id, 0, 0);
 406           break;
 407         case FRAME_ACK_SENT:
 408         case FRAME_DECODED:
 409           logging_.InsertFrameEvent(
 410               event_it->event_timestamp, event_it->type, media_type,
 411               rtp_timestamp, kFrameIdUnknown);
 412           break;
 413         case FRAME_PLAYOUT:
 414           logging_.InsertFrameEventWithDelay(
 415               event_it->event_timestamp, event_it->type, media_type,
 416               rtp_timestamp, kFrameIdUnknown, event_it->delay_delta);
 417           break;
 418         default:
 419           VLOG(2) << "Received log message via RTCP that we did not expect: "
 420                   << static_cast<int>(event_it->type);
 421           break;
 422       }
 423     }
 424   }
 425 }
 426
 427 void CastTransportSenderImpl::OnReceivedCastMessage(
 428     uint32 ssrc,
 429     const RtcpCastMessageCallback& cast_message_cb,
 430     const RtcpCastMessage& cast_message) {
 431   if (!cast_message_cb.is_null())
 432     cast_message_cb.Run(cast_message);
 433
 434   DedupInfo dedup_info;
 435   if (audio_sender_ && audio_sender_->ssrc() == ssrc) {
 436     const int64 acked_bytes =
 437         audio_sender_->GetLastByteSentForFrame(cast_message.ack_frame_id);
 438     last_byte_acked_for_audio_ =
 439         std::max(acked_bytes, last_byte_acked_for_audio_);
 440   } else if (video_sender_ && video_sender_->ssrc() == ssrc) {
 441     dedup_info.resend_interval = video_rtcp_session_->current_round_trip_time();
 442
 443     // Only use audio stream to dedup if there is one.
 444     if (audio_sender_) {
 445       dedup_info.last_byte_acked_for_audio = last_byte_acked_for_audio_;
 446     }
 447   }
 448
 449   if (cast_message.missing_frames_and_packets.empty())
 450     return;
 451
 452   // This call does two things.
 453   // 1. Specifies that retransmissions for packets not listed in the set are
 454   //    cancelled.
 455   // 2. Specifies a deduplication window. For video this would be the most
 456   //    recent RTT. For audio there is no deduplication.
 457   ResendPackets(ssrc,
 458                 cast_message.missing_frames_and_packets,
 459                 true,
 460                 dedup_info);
 461 }
 462
 463 void CastTransportSenderImpl::AddValidSsrc(uint32 ssrc) {
 464   valid_ssrcs_.insert(ssrc);
 465 }
 466
 467 void CastTransportSenderImpl::SendRtcpFromRtpReceiver(
 468     uint32 ssrc,
 469     uint32 sender_ssrc,
 470     const RtcpTimeData& time_data,
 471     const RtcpCastMessage* cast_message,
 472     base::TimeDelta target_delay,
 473     const ReceiverRtcpEventSubscriber::RtcpEvents* rtcp_events,
 474     const RtpReceiverStatistics* rtp_receiver_statistics) {
 475   const Rtcp rtcp(RtcpCastMessageCallback(),
 476                   RtcpRttCallback(),
 477                   RtcpLogMessageCallback(),
 478                   clock_,
 479                   &pacer_,
 480                   ssrc,
 481                   sender_ssrc);
 482   rtcp.SendRtcpFromRtpReceiver(time_data,
 483                                cast_message,
 484                                target_delay,
 485                                rtcp_events,
 486                                rtp_receiver_statistics);
 487 }
 488
 489 }  // namespace cast
 490 }  // namespace media