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