Simplify interface for media/cast: CastTransportSenderImpl

media/cast/net/cast_transport_sender_impl.cc

 // Copyright 2014 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/net/cast_transport_sender_impl.h"
 
 #include <stddef.h>
 #include <algorithm>
 #include <string>
 #include <utility>
@@ skipping 43 matching lines @@
                               kMaxBurstSize) *
       kMaxIpPacketSize;
   int32_t min_send_buffer_size =
       LookupOptionWithDefault(options, kOptionSendBufferMinSize, 0);
   return std::max(max_burst_size, min_send_buffer_size);
 }
 
 }  // namespace
 
 scoped_ptr<CastTransportSender> CastTransportSender::Create(
-    net::NetLog* net_log,
-    base::TickClock* clock,
-    const net::IPEndPoint& local_end_point,
-    const net::IPEndPoint& remote_end_point,
+    base::TickClock* clock,  // Owned by the caller.
+    const UdpTransportParams& upd_transport_params,
+    base::TimeDelta logging_flush_interval,
     scoped_ptr<base::DictionaryValue> options,
-    const CastTransportStatusCallback& status_callback,
-    const BulkRawEventsCallback& raw_events_callback,
-    base::TimeDelta raw_events_callback_interval,
-    const PacketReceiverCallback& packet_callback,
+    scoped_ptr<Client> client,
     const scoped_refptr<base::SingleThreadTaskRunner>& transport_task_runner) {
   return scoped_ptr<CastTransportSender>(new CastTransportSenderImpl(
-      net_log, clock, local_end_point, remote_end_point, std::move(options),
-      status_callback, raw_events_callback, raw_events_callback_interval,
-      transport_task_runner.get(), packet_callback, NULL));
+      clock, upd_transport_params, logging_flush_interval, std::move(options),
+      std::move(client), transport_task_runner.get(), nullptr));
 }
 
 PacketReceiverCallback CastTransportSender::PacketReceiverForTesting() {
   return PacketReceiverCallback();
 }
 
 CastTransportSenderImpl::CastTransportSenderImpl(
-    net::NetLog* net_log,
     base::TickClock* clock,
-    const net::IPEndPoint& local_end_point,
-    const net::IPEndPoint& remote_end_point,
+    const UdpTransportParams& upd_transport_params,
+    base::TimeDelta logging_flush_interval,
     scoped_ptr<base::DictionaryValue> options,
-    const CastTransportStatusCallback& status_callback,
-    const BulkRawEventsCallback& raw_events_callback,
-    base::TimeDelta raw_events_callback_interval,
+    scoped_ptr<Client> client,
     const scoped_refptr<base::SingleThreadTaskRunner>& transport_task_runner,
-    const PacketReceiverCallback& packet_callback,
     PacketSender* external_transport)
-    : clock_(clock),
-      status_callback_(status_callback),
+    : transport_client_(std::move(client)),
+      clock_(clock),
       transport_task_runner_(transport_task_runner),
-      transport_(external_transport
-                     ? nullptr
-                     : new UdpTransport(net_log,
-                                        transport_task_runner,
-                                        local_end_point,
-                                        remote_end_point,
-                                        GetTransportSendBufferSize(*options),
-                                        status_callback)),
-      pacer_(LookupOptionWithDefault(*options,
-                                     kOptionPacerTargetBurstSize,
-                                     kTargetBurstSize),
-             LookupOptionWithDefault(*options,
-                                     kOptionPacerMaxBurstSize,
-                                     kMaxBurstSize),
-             clock,
-             raw_events_callback.is_null() ? nullptr : &recent_packet_events_,
-             external_transport ? external_transport : transport_.get(),
-             transport_task_runner),
-      raw_events_callback_(raw_events_callback),
-      raw_events_callback_interval_(raw_events_callback_interval),
+      logging_flush_interval_(logging_flush_interval),
       last_byte_acked_for_audio_(0),
-      packet_callback_(packet_callback),
       weak_factory_(this) {
-  DCHECK(clock_);
-  if (!raw_events_callback_.is_null()) {
-    DCHECK(raw_events_callback_interval > base::TimeDelta());
-    transport_task_runner->PostDelayedTask(
-        FROM_HERE,
-        base::Bind(&CastTransportSenderImpl::SendRawEvents,
-                   weak_factory_.GetWeakPtr()),
-        raw_events_callback_interval);
+  DCHECK(clock);
+  transport_.reset(
+      external_transport
+          ? nullptr
+          : new UdpTransport(
+                upd_transport_params.net_log, transport_task_runner_,
+                upd_transport_params.local_end_point,
+                upd_transport_params.remote_end_point,
+                GetTransportSendBufferSize(*options),
+                base::Bind(&CastTransportSenderImpl::OnStatusChanged,
+                           weak_factory_.GetWeakPtr())));
+  pacer_.reset(new PacedSender(
+      LookupOptionWithDefault(*options, kOptionPacerTargetBurstSize,
+                              kTargetBurstSize),
+      LookupOptionWithDefault(*options, kOptionPacerMaxBurstSize,
+                              kMaxBurstSize),
+      clock, logging_flush_interval > base::TimeDelta() ? &recent_packet_events_
+                                                        : nullptr,
+      external_transport ? external_transport : transport_.get(),
+      transport_task_runner));
+  if (logging_flush_interval_ > base::TimeDelta()) {
+    transport_task_runner_->PostDelayedTask(
+        FROM_HERE, base::Bind(&CastTransportSenderImpl::SendRawEvents,
+                              weak_factory_.GetWeakPtr()),
+        logging_flush_interval_);
   }
   if (transport_) {
     if (options->HasKey(kOptionDscp)) {
       // The default DSCP value for cast is AF41. Which gives it a higher
       // priority over other traffic.
       transport_->SetDscp(net::DSCP_AF41);
     }
 #if defined(OS_WIN)
     if (!options->HasKey(kOptionDisableNonBlockingIO)) {
       transport_->UseNonBlockingIO();
@@ skipping 21 matching lines @@
   }
 }
 
 void CastTransportSenderImpl::InitializeAudio(
     const CastTransportRtpConfig& config,
     const RtcpCastMessageCallback& cast_message_cb,
     const RtcpRttCallback& rtt_cb) {
   LOG_IF(WARNING, config.aes_key.empty() || config.aes_iv_mask.empty())
       << "Unsafe to send audio with encryption DISABLED.";
   if (!audio_encryptor_.Initialize(config.aes_key, config.aes_iv_mask)) {
-    status_callback_.Run(TRANSPORT_AUDIO_UNINITIALIZED);
+    transport_client_->OnStatusChanged(TRANSPORT_AUDIO_UNINITIALIZED);
     return;
   }
 
-  audio_sender_.reset(new RtpSender(transport_task_runner_, &pacer_));
+  audio_sender_.reset(new RtpSender(transport_task_runner_, pacer_.get()));
   if (audio_sender_->Initialize(config)) {
     // Audio packets have a higher priority.
-    pacer_.RegisterAudioSsrc(config.ssrc);
-    pacer_.RegisterPrioritySsrc(config.ssrc);
-    status_callback_.Run(TRANSPORT_AUDIO_INITIALIZED);
+    pacer_->RegisterAudioSsrc(config.ssrc);
+    pacer_->RegisterPrioritySsrc(config.ssrc);
+    transport_client_->OnStatusChanged(TRANSPORT_AUDIO_INITIALIZED);
   } else {
     audio_sender_.reset();
-    status_callback_.Run(TRANSPORT_AUDIO_UNINITIALIZED);
+    transport_client_->OnStatusChanged(TRANSPORT_AUDIO_UNINITIALIZED);
     return;
   }
 
   audio_rtcp_session_.reset(new SenderRtcpSession(
       base::Bind(&CastTransportSenderImpl::OnReceivedCastMessage,
                  weak_factory_.GetWeakPtr(), config.ssrc, cast_message_cb),
       rtt_cb, base::Bind(&CastTransportSenderImpl::OnReceivedLogMessage,
                          weak_factory_.GetWeakPtr(), AUDIO_EVENT),
-      clock_, &pacer_, config.ssrc, config.feedback_ssrc));
-  pacer_.RegisterAudioSsrc(config.ssrc);
+      clock_, pacer_.get(), config.ssrc, config.feedback_ssrc));
+  pacer_->RegisterAudioSsrc(config.ssrc);
   AddValidSsrc(config.feedback_ssrc);
-  status_callback_.Run(TRANSPORT_AUDIO_INITIALIZED);
+  transport_client_->OnStatusChanged(TRANSPORT_AUDIO_INITIALIZED);
 }
 
 void CastTransportSenderImpl::InitializeVideo(
     const CastTransportRtpConfig& config,
     const RtcpCastMessageCallback& cast_message_cb,
     const RtcpRttCallback& rtt_cb) {
   LOG_IF(WARNING, config.aes_key.empty() || config.aes_iv_mask.empty())
       << "Unsafe to send video with encryption DISABLED.";
   if (!video_encryptor_.Initialize(config.aes_key, config.aes_iv_mask)) {
-    status_callback_.Run(TRANSPORT_VIDEO_UNINITIALIZED);
+    transport_client_->OnStatusChanged(TRANSPORT_VIDEO_UNINITIALIZED);
     return;
   }
 
-  video_sender_.reset(new RtpSender(transport_task_runner_, &pacer_));
+  video_sender_.reset(new RtpSender(transport_task_runner_, pacer_.get()));
   if (!video_sender_->Initialize(config)) {
     video_sender_.reset();
-    status_callback_.Run(TRANSPORT_VIDEO_UNINITIALIZED);
+    transport_client_->OnStatusChanged(TRANSPORT_VIDEO_UNINITIALIZED);
     return;
   }
 
   video_rtcp_session_.reset(new SenderRtcpSession(
       base::Bind(&CastTransportSenderImpl::OnReceivedCastMessage,
                  weak_factory_.GetWeakPtr(), config.ssrc, cast_message_cb),
       rtt_cb, base::Bind(&CastTransportSenderImpl::OnReceivedLogMessage,
                          weak_factory_.GetWeakPtr(), VIDEO_EVENT),
-      clock_, &pacer_, config.ssrc, config.feedback_ssrc));
-  pacer_.RegisterVideoSsrc(config.ssrc);
+      clock_, pacer_.get(), config.ssrc, config.feedback_ssrc));
+  pacer_->RegisterVideoSsrc(config.ssrc);
   AddValidSsrc(config.feedback_ssrc);
-  status_callback_.Run(TRANSPORT_VIDEO_INITIALIZED);
+  transport_client_->OnStatusChanged(TRANSPORT_VIDEO_INITIALIZED);
 }
 
 namespace {
 void EncryptAndSendFrame(const EncodedFrame& frame,
                          TransportEncryptionHandler* encryptor,
                          RtpSender* sender) {
   // TODO(miu): We probably shouldn't attempt to send an empty frame, but this
   // issue is still under investigation.  http://crbug.com/519022
   if (encryptor->is_activated() && !frame.data.empty()) {
     EncodedFrame encrypted_frame;
@@ skipping 85 matching lines @@
   }
 }
 
 PacketReceiverCallback CastTransportSenderImpl::PacketReceiverForTesting() {
   return base::Bind(
       base::IgnoreResult(&CastTransportSenderImpl::OnReceivedPacket),
       weak_factory_.GetWeakPtr());
 }
 
 void CastTransportSenderImpl::SendRawEvents() {
-  DCHECK(!raw_events_callback_.is_null());
+  DCHECK(logging_flush_interval_ > base::TimeDelta());
 
   if (!recent_frame_events_.empty() || !recent_packet_events_.empty()) {
     scoped_ptr<std::vector<FrameEvent>> frame_events(
         new std::vector<FrameEvent>());
     frame_events->swap(recent_frame_events_);
     scoped_ptr<std::vector<PacketEvent>> packet_events(
         new std::vector<PacketEvent>());
     packet_events->swap(recent_packet_events_);
-    raw_events_callback_.Run(std::move(frame_events), std::move(packet_events));
+    transport_client_->OnLoggingEventsReceived(std::move(frame_events),
+                                               std::move(packet_events));
   }
 
   transport_task_runner_->PostDelayedTask(
-      FROM_HERE,
-      base::Bind(&CastTransportSenderImpl::SendRawEvents,
-                 weak_factory_.GetWeakPtr()),
-      raw_events_callback_interval_);
+      FROM_HERE, base::Bind(&CastTransportSenderImpl::SendRawEvents,
+                            weak_factory_.GetWeakPtr()),
+      logging_flush_interval_);
 }
 
 bool CastTransportSenderImpl::OnReceivedPacket(scoped_ptr<Packet> packet) {
   const uint8_t* const data = &packet->front();
   const size_t length = packet->size();
   uint32_t ssrc;
   if (IsRtcpPacket(data, length)) {
     ssrc = GetSsrcOfSender(data, length);
   } else if (!RtpParser::ParseSsrc(data, length, &ssrc)) {
     VLOG(1) << "Invalid RTP packet.";
     return false;
   }
   if (valid_ssrcs_.find(ssrc) == valid_ssrcs_.end()) {
     VLOG(1) << "Stale packet received.";
     return false;
   }
 
   if (audio_rtcp_session_ &&
       audio_rtcp_session_->IncomingRtcpPacket(data, length)) {
     return true;
   }
   if (video_rtcp_session_ &&
       video_rtcp_session_->IncomingRtcpPacket(data, length)) {
     return true;
   }
-  if (packet_callback_.is_null()) {
-    VLOG(1) << "Stale packet received.";
-    return false;
-  }
-  packet_callback_.Run(std::move(packet));
+  transport_client_->OnPacketsReceived(std::move(packet));
   return true;
 }
 
 void CastTransportSenderImpl::OnReceivedLogMessage(
     EventMediaType media_type,
     const RtcpReceiverLogMessage& log) {
-  if (raw_events_callback_.is_null())
+  if (logging_flush_interval_ <= base::TimeDelta())
     return;
 
   // Add received log messages into our log system.
   for (const RtcpReceiverFrameLogMessage& frame_log_message : log) {
     for (const RtcpReceiverEventLogMessage& event_log_message :
          frame_log_message.event_log_messages_) {
       switch (event_log_message.type) {
         case PACKET_RECEIVED: {
           recent_packet_events_.push_back(PacketEvent());
           PacketEvent& receive_event = recent_packet_events_.back();
@@ skipping 69 matching lines @@
 // TODO(isheriff): This interface needs clean up.
 // https://crbug.com/569259
 void CastTransportSenderImpl::SendRtcpFromRtpReceiver(
     uint32_t ssrc,
     uint32_t sender_ssrc,
     const RtcpTimeData& time_data,
     const RtcpCastMessage* cast_message,
     base::TimeDelta target_delay,
     const ReceiverRtcpEventSubscriber::RtcpEvents* rtcp_events,
     const RtpReceiverStatistics* rtp_receiver_statistics) {
-  const ReceiverRtcpSession rtcp(clock_, &pacer_, ssrc, sender_ssrc);
+  const ReceiverRtcpSession rtcp(clock_, pacer_.get(), ssrc, sender_ssrc);
   rtcp.SendRtcpReport(time_data, cast_message, target_delay, rtcp_events,
                       rtp_receiver_statistics);
 }
 
+void CastTransportSenderImpl::OnStatusChanged(CastTransportStatus status) {
+  transport_client_->OnStatusChanged(status);
+}
+
 }  // namespace cast
 }  // namespace media