Mojo Video Capture service in media/capture

media/capture/service/stream_impl.cc

Index: media/capture/service/stream_impl.cc
diff --git a/media/capture/service/stream_impl.cc b/media/capture/service/stream_impl.cc
new file mode 100644
index 0000000000000000000000000000000000000000..f0b5d1a4c67ce68124a686ccb42c238d5c675d72
--- /dev/null
+++ b/media/capture/service/stream_impl.cc
@@ -0,0 +1,132 @@
+// 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 "media/capture/service/stream_impl.h"
+
+#include "base/bind.h"
+#include "base/callback.h"
+#include "base/thread_task_runner_handle.h"
+#include "base/time/time.h"
+#include "media/base/bind_to_current_loop.h"
+#include "media/capture/interfaces/video_capture.mojom.h"
+#include "media/capture/service/mojo_video_frame.h"
+#include "mojo/converters/geometry/geometry_type_converters.h"
+
+namespace media {
+
+StreamImpl::StreamImpl(mojo::InterfaceRequest<mojom::Stream> request,
+                       const base::Closure& start_callback,
+                       const base::Closure& close_callback,
+                       const base::Closure& error_callback)
+    : binding_(this, std::move(request)),
+      start_callback_(start_callback),
+      close_callback_(close_callback),
+      error_callback_(error_callback),
+      task_runner_(base::ThreadTaskRunnerHandle::Get()),
+      weak_factory_(this) {
+  binding_.set_connection_error_handler(error_callback_);
+}
+
+StreamImpl::~StreamImpl() {
+  DCHECK(task_runner_->BelongsToCurrentThread());
+}
+
+void StreamImpl::Start(mojom::StreamClientPtr client) {
+  DCHECK(task_runner_->BelongsToCurrentThread());
+  DCHECK(!client_) << "Start() should not be called several times.";
+  DCHECK(!paused_);
+  client_ = std::move(client);
+  client_.set_connection_error_handler(error_callback_);
+  start_callback_.Run();
+}
+
+void StreamImpl::Pause() {
+  DCHECK(task_runner_->BelongsToCurrentThread());
+  DCHECK(client_);
+  paused_ = true;
+}
+
+void StreamImpl::Resume() {
+  DCHECK(task_runner_->BelongsToCurrentThread());
+  DCHECK(client_);
+  paused_ = false;
+}
+
+void StreamImpl::Stop() {
+  DCHECK(task_runner_->BelongsToCurrentThread());
+  DCHECK(client_);
+  close_callback_.Run();
+}
+
+void StreamImpl::OnFrame(const scoped_refptr<MojoVideoFrame>& frame,
+                         const base::TimeTicks& timestamp) {
+  task_runner_->PostTask(
+      FROM_HERE, base::Bind(&StreamImpl::NotifyFrameAvailable,
+                            weak_factory_.GetWeakPtr(), frame, timestamp));
+}
+
+void StreamImpl::OnError(const std::string& reason) {
+  DLOG(ERROR) << __FUNCTION__ << " " << reason;
+  task_runner_->PostTask(
+      FROM_HERE,
+      base::Bind(&StreamImpl::NotifyError, weak_factory_.GetWeakPtr(), reason));
+}
+
+void StreamImpl::NotifyFrameAvailable(
+    const scoped_refptr<MojoVideoFrame>& frame,
+    const base::TimeTicks& timestamp) {
+  DCHECK(task_runner_->BelongsToCurrentThread());
+  DCHECK(client_);
+
+  if (paused_)  // Drop the frame immediately if we aren't playing.
+    return;
+
+  const int32_t buffer_id = running_buffer_id_++;
+
+  mojom::FrameInfoPtr frame_info = mojom::FrameInfo::New();
+
+  // We duplicate the handle immediately to avoid passing around the primordial
+  // handle. While it's OK for someone to continue using the buffer after the
+  // pool drops ownership, it's not OK to dupe the primordial handle once it's
+  // been invalidated.
+  mojo::ScopedSharedBufferHandle handle;
+  MojoResult rv = mojo::DuplicateBuffer(frame->handle(), nullptr, &handle);
+  DLOG_IF(ERROR, rv != MOJO_RESULT_OK) << "Error duplicating handle";
+  if (rv != MOJO_RESULT_OK)
+    return;
+  frame_info->storage_handle = std::move(handle);
+  frame_info->storage_size = frame->mapped_size();
+
+  frame_info->pixel_format =
+      static_cast<media::interfaces::VideoFormat>(frame->format());
+  frame_info->coded_size = mojo::Size::From(frame->coded_size());
+  frame_info->visible_rect = mojo::Rect::From(frame->visible_rect());
+  frame_info->timestamp = timestamp.ToInternalValue();
+
+  in_use_video_frames_.insert(std::make_pair(buffer_id, frame));
+
+  client_->FrameAvailable(std::move(frame_info),
+                          base::Bind(&StreamImpl::OnFrameConsumed,
+                                     weak_factory_.GetWeakPtr(), buffer_id));
+}
+
+void StreamImpl::NotifyError(const mojo::String& error) {
+  DCHECK(task_runner_->BelongsToCurrentThread());
+  // A device may notify of error(s) before Start() registers a |client_|.
+  if (client_)
+    client_->Error(error);
+  error_callback_.Run();
+}
+
+void StreamImpl::OnFrameConsumed(int32_t buffer_id) {
+  DCHECK(task_runner_->BelongsToCurrentThread());
+  const auto video_frame = in_use_video_frames_.find(buffer_id);
+  if (video_frame == in_use_video_frames_.end()) {
+    DLOG(ERROR) << "Buffer id " << buffer_id << " not found!";
+    return;
+  }
+  in_use_video_frames_.erase(video_frame);
+}
+
+}  // namespace media