media::mojom::Remoter and CastRemotingConnector/Sender

chrome/browser/media/cast_remoting_sender.cc

Index: chrome/browser/media/cast_remoting_sender.cc
diff --git a/chrome/browser/media/cast_remoting_sender.cc b/chrome/browser/media/cast_remoting_sender.cc
new file mode 100644
index 0000000000000000000000000000000000000000..b6afe2521b6e9cf14a71f80f70882312b52a516b
--- /dev/null
+++ b/chrome/browser/media/cast_remoting_sender.cc
@@ -0,0 +1,131 @@
+// Copyright 2016 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/browser/media/cast_remoting_sender.h"
+
+#include <map>
+
+#include "base/bind.h"
+#include "base/bind_helpers.h"
+#include "base/callback.h"
+#include "base/lazy_instance.h"
+#include "content/public/browser/browser_thread.h"
+#include "media/base/bind_to_current_loop.h"
+
+using content::BrowserThread;
+
+namespace {
+
+// Global map for looking-up CastRemotingSender instances by their
+// |rtp_stream_id|.
+using CastRemotingSenderMap = std::map<int32_t, CastRemotingSender*>;
+base::LazyInstance<CastRemotingSenderMap>::Leaky g_sender_map =
+    LAZY_INSTANCE_INITIALIZER;
+
+}  // namespace
+
+CastRemotingSender::CastRemotingSender(int32_t rtp_stream_id)
+    : rtp_stream_id_(rtp_stream_id), binding_(this) {
+  DCHECK_CURRENTLY_ON(BrowserThread::IO);
+  CastRemotingSender*& pointer_in_map = g_sender_map.Get()[rtp_stream_id_];
+  DCHECK(!pointer_in_map);
+  pointer_in_map = this;
+}
+
+CastRemotingSender::~CastRemotingSender() {
+  DCHECK_CURRENTLY_ON(BrowserThread::IO);
+  g_sender_map.Pointer()->erase(rtp_stream_id_);
+}
+
+// static
+void CastRemotingSender::FindAndBind(
+    int32_t rtp_stream_id,
+    mojo::ScopedDataPipeConsumerHandle pipe,
+    media::mojom::RemotingDataStreamSenderRequest request,
+    const base::Closure& error_callback) {
+  // CastRemotingSender lives entirely on the IO thread, so trampoline if
+  // necessary.
+  if (!BrowserThread::CurrentlyOn(BrowserThread::IO)) {
+    BrowserThread::PostTask(
+        BrowserThread::IO, FROM_HERE,
+        base::Bind(&CastRemotingSender::FindAndBind, rtp_stream_id,
+                   base::Passed(&pipe), base::Passed(&request),
+                   // Using media::BindToCurrentLoop() so the |error_callback|
+                   // is trampolined back to the original thread.
+                   media::BindToCurrentLoop(error_callback)));
+    return;
+  }
+
+  DCHECK(!error_callback.is_null());
+
+  // Look-up the CastRemotingSender instance by its |rtp_stream_id|.
+  const auto it = g_sender_map.Pointer()->find(rtp_stream_id);
+  if (it == g_sender_map.Pointer()->end()) {
+    DLOG(ERROR) << "Cannot find CastRemotingSender instance by ID: "
+                << rtp_stream_id;
+    error_callback.Run();
+    return;
+  }
+
+  // Confirm that the CastRemotingSender isn't already bound to a message pipe.
+  CastRemotingSender* const sender = it->second;
+  if (sender->binding_.is_bound()) {
+    DLOG(ERROR) << "Attempt to bind to CastRemotingSender a second time (id="
+                << rtp_stream_id << ")!";
+    error_callback.Run();
+    return;
+  }
+
+  DCHECK(sender->error_callback_.is_null());
+  sender->error_callback_ = error_callback;
+
+  sender->pipe_ = std::move(pipe);
+  sender->binding_.Bind(std::move(request));
+  sender->binding_.set_connection_error_handler(
+      base::Bind(&base::Closure::Run,
+                 base::Unretained(&sender->error_callback_)));
+}
+
+void CastRemotingSender::ConsumeDataChunk(uint32_t offset, uint32_t size,
+                                          uint32_t total_payload_size) {
+  DCHECK_CURRENTLY_ON(BrowserThread::IO);
+
+  const void* buffer;
+  uint32_t buffer_num_bytes = size;
+  if (offset + size > total_payload_size || !pipe_.is_valid() ||
+      mojo::BeginReadDataRaw(pipe_.get(), &buffer, &buffer_num_bytes,
+                             MOJO_READ_DATA_FLAG_ALL_OR_NONE) !=
+          MOJO_RESULT_OK ||
+      buffer_num_bytes != size) {
+    pipe_.reset();
+    binding_.Close();
+    error_callback_.Run();
+    return;
+  }
+  // If |total_payload_size| has changed, resize the data string. If it has not
+  // changed, the following statement will be a no-op.
+  next_frame_data_.resize(total_payload_size, '\0');
+  memcpy(&next_frame_data_.front() + offset, buffer, buffer_num_bytes);
+  mojo::EndReadDataRaw(pipe_.get(), buffer_num_bytes);
+}
+
+void CastRemotingSender::SendFrame() {
+  DCHECK_CURRENTLY_ON(BrowserThread::IO);
+
+  // TODO(miu): Merge with xjz's recent change, which implements this
+  // functionality.
+  NOTIMPLEMENTED();  // encoder_frame.data.swap(next_frame_data_)
+}
+
+void CastRemotingSender::CancelInFlightData() {
+  DCHECK_CURRENTLY_ON(BrowserThread::IO);
+
+  // Note: Not calling clear(), in order to force a significant amount of memory
+  // to be freed.
+  std::string().swap(next_frame_data_);
+
+  // TODO(miu): Merge with xjz's recent change, which implements this
+  // functionality.
+  NOTIMPLEMENTED();  // transport_->CancelSendingFrames(...);
+}