Cast: IPC glue between cast library transport and encoders.

chrome/renderer/media/cast_transport_sender_ipc.cc

+// Copyright (c) 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 "chrome/renderer/media/cast_transport_sender_ipc.h"
+
+#include "base/callback.h"
+#include "base/id_map.h"
+#include "chrome/common/cast_messages.h"
+#include "ipc/ipc_channel_proxy.h"
+#include "media/cast/cast_sender.h"
+#include "media/cast/transport/cast_transport_sender.h"
+
+class CastIPCDispatcher;
+
+// This implementation of the CastTransportSender interface
+// communicates with the browser process over IPC and relays
+// all calls to/from the transport sender to the browser process.
+// The primary reason for this arrangement is to give the
+// renderer less direct control over the UDP sockets.
+class CastTransportSenderIPC
+    : public media::cast::transport::CastTransportSender {
+ public:
+  CastTransportSenderIPC(
+      const media::cast::transport::CastTransportConfig& config,
+      const media::cast::transport::CastTransportStatusCallback& status_cb);
+
+  virtual ~CastTransportSenderIPC();
+
+  // media::cast::transport::CastTransportSender implementation.
+  virtual void SetPacketReceiver(
+      const media::cast::transport::PacketReceiverCallback& packet_callback)
+      OVERRIDE {
+    packet_callback_ = packet_callback;
+  }
+
+  virtual void InsertCodedAudioFrame(
+      const media::cast::transport::EncodedAudioFrame* audio_frame,
+      const base::TimeTicks& recorded_time) OVERRIDE {
+    Send(new CastHostMsg_InsertCodedAudioFrame(channel_id_,
+                                               *audio_frame,
+                                               recorded_time));
+  }
+
+  virtual void InsertCodedVideoFrame(
+      const media::cast::transport::EncodedVideoFrame* video_frame,
+      const base::TimeTicks& capture_time) OVERRIDE {
+    Send(new CastHostMsg_InsertCodedVideoFrame(channel_id_,
+                                               *video_frame,
+                                               capture_time));
+  }
+
+  virtual void SendRtcpFromRtpSender(
+      uint32 packet_type_flags,
+      const media::cast::transport::RtcpSenderInfo& sender_info,
+      const media::cast::transport::RtcpDlrrReportBlock& dlrr,
+      const media::cast::transport::RtcpSenderLogMessage& sender_log,
+      uint32 sending_ssrc,
+      const std::string& c_name) OVERRIDE {
+    struct media::cast::transport::SendRtcpFromRtpSenderData data;
+    data.packet_type_flags = packet_type_flags;
+    data.sending_ssrc = sending_ssrc;
+    data.c_name = data.c_name;
+    Send(new CastHostMsg_SendRtcpFromRtpSender(
+        channel_id_,
+        data,
+        sender_info,
+        dlrr,
+        sender_log));
+  }
+
+  virtual void ResendPackets(
+      bool is_audio,
+      const media::cast::MissingFramesAndPacketsMap& missing_packets) OVERRIDE {
+      Send(new CastHostMsg_ResendPackets(channel_id_,
+                                         is_audio,
+                                         missing_packets));
+  }
+
+  virtual void SubscribeAudioRtpStatsCallback(
+      const media::cast::transport::CastTransportRtpStatistics& callback)
+      OVERRIDE {
+    audio_rtp_callback_ = callback;
+  }
+
+  virtual void SubscribeVideoRtpStatsCallback(
+      const media::cast::transport::CastTransportRtpStatistics& callback)
+      OVERRIDE {
+    video_rtp_callback_ = callback;
+  }
+
+ private:
+  friend class CastIPCDispatcher;
+  void Send(IPC::Message *message);
+
+  int32 channel_id_;
+  media::cast::transport::PacketReceiverCallback packet_callback_;
+  media::cast::transport::CastTransportStatusCallback status_callback_;
+  media::cast::transport::CastTransportRtpStatistics audio_rtp_callback_;
+  media::cast::transport::CastTransportRtpStatistics video_rtp_callback_;
+  DISALLOW_COPY_AND_ASSIGN(CastTransportSenderIPC);
+};
+
+// This dispatcher listens to incoming IPC messages and sends
+// the call to the correct CastTransportSenderIPC instance.
+class CastIPCDispatcher : public IPC::ChannelProxy::MessageFilter {
+ public:
+  CastIPCDispatcher(
+      const scoped_refptr<base::MessageLoopProxy>& io_message_loop)
+      : channel_(NULL),
+        io_message_loop_(io_message_loop) {
+    DCHECK(!global_instance_);
+    global_instance_ = this;
+  }
+
+  // IPC::ChannelProxy::MessageFilter implementation
+  virtual bool OnMessageReceived(const IPC::Message& message) OVERRIDE {
+    DCHECK(io_message_loop_->BelongsToCurrentThread());
+    bool handled = true;
+    IPC_BEGIN_MESSAGE_MAP(CastIPCDispatcher, message)
+        IPC_MESSAGE_HANDLER(CastMsg_ReceivedPacket, OnReceivedPacket)
+        IPC_MESSAGE_HANDLER(CastMsg_NotifyStatusChange, OnNotifyStatusChange)
+        IPC_MESSAGE_HANDLER(CastMsg_RtpStatistics, OnRtpStatistics)
+        IPC_MESSAGE_UNHANDLED(handled = false);
+    IPC_END_MESSAGE_MAP();
+    return handled;
+  }
+
+  virtual void OnFilterAdded(IPC::Channel* channel) OVERRIDE {
+    DCHECK(io_message_loop_->BelongsToCurrentThread());
+    channel_ = channel;
+  }
+
+  virtual void OnFilterRemoved() OVERRIDE {
+    DCHECK(io_message_loop_->BelongsToCurrentThread());
+    channel_ = NULL;
+  }
+
+  virtual void OnChannelClosing() OVERRIDE {
+    DCHECK(io_message_loop_->BelongsToCurrentThread());
+    channel_ = NULL;
+  }
+
+ protected:
+  virtual ~CastIPCDispatcher() {
+    global_instance_ = NULL;
+  }
+
+ private:
+  void OnReceivedPacket(int32 channel_id, const media::cast::Packet& packet) {
+    CastTransportSenderIPC* ptr = id_map_.Lookup(channel_id);
+    if (ptr) {
+      if (!ptr->packet_callback_.is_null()) {
+        // TODO(hubbe): Perhaps an non-ownership-transferring cb here?
+        scoped_ptr<media::cast::transport::Packet> packet_copy(
+            new media::cast::transport::Packet(packet));
+        ptr->packet_callback_.Run(packet_copy.Pass());
+      } else {
+        LOG(ERROR) << "CastIPCDispatcher::OnReceivedPacket "
+                   << "no packet callback yet.";
+      }
+    } else {
+      LOG(ERROR) << "CastIPCDispatcher::OnReceivedPacket "
+                 << "on non-existing channel.";
+    }
+  }
+
+  void OnNotifyStatusChange(
+      int32 channel_id,
+      media::cast::transport::CastTransportStatus status) {
+    CastTransportSenderIPC* ptr = id_map_.Lookup(channel_id);
+    if (ptr) {
+      ptr->status_callback_.Run(status);
+    } else {
+      LOG(ERROR)
+          << "CastIPCDispatcher::OnNotifystatusChange on non-existing channel.";
+    }
+  }
+  void OnRtpStatistics(
+      int32 channel_id,
+      bool audio,
+      const media::cast::transport::RtcpSenderInfo& sender_info,
+      base::TimeTicks time_sent,
+      uint32 rtp_timestamp) {
+    CastTransportSenderIPC* ptr = id_map_.Lookup(channel_id);
+    if (ptr) {
+      const media::cast::transport::CastTransportRtpStatistics& callback =
+          audio ? ptr->audio_rtp_callback_ : ptr->video_rtp_callback_;
+      callback.Run(sender_info, time_sent, rtp_timestamp);
+    } else {
+      LOG(ERROR)
+          << "CastIPCDispatcher::OnNotifystatusChange on non-existing channel.";
+    }
+  }
+
+  friend class CastTransportSenderIPC;
+  static CastIPCDispatcher* global_instance_;
+
+  // IPC channel for Send(); must only be accesed on |io_message_loop_|.
+  IPC::Channel* channel_;
+
+  // Message loop on which IPC calls are driven.
+  const scoped_refptr<base::MessageLoopProxy> io_message_loop_;
+
+  // A map of stream ids to delegates; must only be accessed on
+  // |io_message_loop_|.
+  IDMap<CastTransportSenderIPC> id_map_;
+  DISALLOW_COPY_AND_ASSIGN(CastIPCDispatcher);
+};
+
+CastIPCDispatcher* CastIPCDispatcher::global_instance_ = NULL;
+
+// These three methods can't be written inside the class becasue
+// they access CastIPCDispatcher, which isn't declared yet.
+CastTransportSenderIPC::CastTransportSenderIPC(
+    const media::cast::transport::CastTransportConfig& config,
+    const media::cast::transport::CastTransportStatusCallback& status_cb)
+    : status_callback_(status_cb) {
+  if (CastIPCDispatcher::global_instance_) {
+    channel_id_ = CastIPCDispatcher::global_instance_->id_map_.Add(this);
+  }
+  Send(new CastHostMsg_New(channel_id_, config));
+}
+
+CastTransportSenderIPC::~CastTransportSenderIPC() {
+  Send(new CastHostMsg_Delete(channel_id_));
+  if (CastIPCDispatcher::global_instance_) {
+    CastIPCDispatcher::global_instance_->id_map_.Remove(channel_id_);
+  }
+}
+
+void CastTransportSenderIPC::Send(IPC::Message* message) {
+  if (CastIPCDispatcher::global_instance_ &&
+      CastIPCDispatcher::global_instance_->channel_) {
+    DCHECK(CastIPCDispatcher::global_instance_->io_message_loop_->
+           BelongsToCurrentThread());
+    CastIPCDispatcher::global_instance_->channel_->Send(message);
+  } else {
+    delete message;
+  }
+}
+
+// Factory functions.
+media::cast::transport::CastTransportSender* NewCastTransportSenderIPC(
+    const media::cast::transport::CastTransportConfig& config,
+    const media::cast::transport::CastTransportStatusCallback& status_cb) {
+  return new CastTransportSenderIPC(config, status_cb);
+}
+
+scoped_refptr<IPC::ChannelProxy::MessageFilter> NewCastIPCDispatcher(
+    const scoped_refptr<base::MessageLoopProxy>& io_message_loop) {
+  return scoped_refptr<IPC::ChannelProxy::MessageFilter>(
+      new CastIPCDispatcher(io_message_loop));
+}