[RTCInsertableStreams] Add underlying source and sink for encoded video.

This CL introduces underlying source and sink classes for encoded video.
The source allows sending frames from WebRTC to a ReadableStream.
The sink allows sending frames from a WritableStream to WebRTC.

Bug: 1052765
Change-Id: I191c0deca35630a960e962913f0a42f89c306bb4
Reviewed-on: https://chromium-review.googlesource.com/c/chromium/src/+/2080475Reviewed-by: Adam Rice <ricea@chromium.org>
Reviewed-by: Kinuko Yasuda <kinuko@chromium.org>
Reviewed-by: Harald Alvestrand <hta@chromium.org>
Auto-Submit: Guido Urdaneta <guidou@chromium.org>
Commit-Queue: Kinuko Yasuda <kinuko@chromium.org>
Cr-Commit-Position: refs/heads/master@{#746280}

third_party/blink/renderer/modules/BUILD.gn

@@ -378,6 +378,8 @@ jumbo_source_set("unit_tests") {
     "peerconnection/peer_connection_dependency_factory_test.cc",
     "peerconnection/peer_connection_tracker_test.cc",
     "peerconnection/rtc_data_channel_test.cc",
+    "peerconnection/rtc_encoded_video_underlying_sink_test.cc",
+    "peerconnection/rtc_encoded_video_underlying_source_test.cc",
     "peerconnection/rtc_ice_transport_test.cc",
     "peerconnection/rtc_ice_transport_test.h",
     "peerconnection/rtc_peer_connection_handler_test.cc",

third_party/blink/renderer/modules/peerconnection/BUILD.gn

@@ -80,6 +80,10 @@ blink_modules_sources("peerconnection") {
     "rtc_encoded_audio_frame.h",
     "rtc_encoded_video_frame.cc",
     "rtc_encoded_video_frame.h",
+    "rtc_encoded_video_underlying_sink.cc",
+    "rtc_encoded_video_underlying_sink.h",
+    "rtc_encoded_video_underlying_source.cc",
+    "rtc_encoded_video_underlying_source.h",
     "rtc_error.cc",
     "rtc_error.h",
     "rtc_error_event.cc",

third_party/blink/renderer/modules/peerconnection/rtc_encoded_video_frame.h

@@ -10,6 +10,7 @@
 #include <memory>
 #include <vector>
 
+#include "third_party/blink/renderer/modules/modules_export.h"
 #include "third_party/blink/renderer/platform/bindings/script_wrappable.h"
 #include "third_party/blink/renderer/platform/wtf/text/wtf_string.h"
 
@@ -23,7 +24,7 @@ namespace blink {
 
 class DOMArrayBuffer;
 
-class RTCEncodedVideoFrame final : public ScriptWrappable {
+class MODULES_EXPORT RTCEncodedVideoFrame final : public ScriptWrappable {
   DEFINE_WRAPPERTYPEINFO();
 
  public:

third_party/blink/renderer/modules/peerconnection/rtc_encoded_video_underlying_sink.cc

+// Copyright 2020 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 "third_party/blink/renderer/modules/peerconnection/rtc_encoded_video_underlying_sink.h"
+
+#include "third_party/blink/renderer/bindings/modules/v8/v8_rtc_encoded_video_frame.h"
+#include "third_party/blink/renderer/core/dom/dom_exception.h"
+#include "third_party/blink/renderer/platform/peerconnection/rtc_encoded_video_stream_transformer.h"
+#include "third_party/webrtc/api/video/encoded_frame.h"
+
+namespace blink {
+
+RTCEncodedVideoUnderlyingSink::RTCEncodedVideoUnderlyingSink(
+    ScriptState* script_state,
+    TransformerCallback transformer_callback)
+    : transformer_callback_(std::move(transformer_callback)) {
+  DCHECK(transformer_callback_);
+}
+
+ScriptPromise RTCEncodedVideoUnderlyingSink::start(
+    ScriptState* script_state,
+    WritableStreamDefaultController* controller,
+    ExceptionState&) {
+  // No extra setup needed.
+  return ScriptPromise::CastUndefined(script_state);
+}
+
+ScriptPromise RTCEncodedVideoUnderlyingSink::write(
+    ScriptState* script_state,
+    ScriptValue chunk,
+    WritableStreamDefaultController* controller,
+    ExceptionState& exception_state) {
+  RTCEncodedVideoFrame* encoded_frame =
+      V8RTCEncodedVideoFrame::ToImplWithTypeCheck(script_state->GetIsolate(),
+                                                  chunk.V8Value());
+  if (!encoded_frame) {
+    exception_state.ThrowDOMException(DOMExceptionCode::kTypeMismatchError,
+                                      "Invalid frame");
+    return ScriptPromise();
+  }
+
+  // Get webrtc frame and send it to the decoder.
+  if (!transformer_callback_) {
+    exception_state.ThrowDOMException(DOMExceptionCode::kInvalidStateError,
+                                      "Stream closed");
+    return ScriptPromise();
+  }
+
+  transformer_callback_.Run()->SendFrameToSink(encoded_frame->PassDelegate());
+  return ScriptPromise::CastUndefined(script_state);
+}
+
+ScriptPromise RTCEncodedVideoUnderlyingSink::close(ScriptState* script_state,
+                                                   ExceptionState&) {
+  // Disconnect from the transformer if the sink is closed.
+  if (transformer_callback_)
+    transformer_callback_.Reset();
+  return ScriptPromise::CastUndefined(script_state);
+}
+
+ScriptPromise RTCEncodedVideoUnderlyingSink::abort(
+    ScriptState* script_state,
+    ScriptValue reason,
+    ExceptionState& exception_state) {
+  // It is not possible to cancel any frames already sent to the WebRTC sink,
+  // thus abort() has the same effect as close().
+  return close(script_state, exception_state);
+}
+
+void RTCEncodedVideoUnderlyingSink::Trace(Visitor* visitor) {
+  UnderlyingSinkBase::Trace(visitor);
+}
+
+}  // namespace blink

third_party/blink/renderer/modules/peerconnection/rtc_encoded_video_underlying_sink.h

+// Copyright 2020 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.
+
+#ifndef THIRD_PARTY_BLINK_RENDERER_MODULES_PEERCONNECTION_RTC_ENCODED_VIDEO_UNDERLYING_SINK_H_
+#define THIRD_PARTY_BLINK_RENDERER_MODULES_PEERCONNECTION_RTC_ENCODED_VIDEO_UNDERLYING_SINK_H_
+
+#include "third_party/blink/renderer/core/streams/underlying_sink_base.h"
+#include "third_party/blink/renderer/modules/modules_export.h"
+
+namespace blink {
+
+class ExceptionState;
+class RTCEncodedVideoStreamTransformer;
+
+class MODULES_EXPORT RTCEncodedVideoUnderlyingSink final
+    : public UnderlyingSinkBase {
+ public:
+  using TransformerCallback =
+      base::RepeatingCallback<RTCEncodedVideoStreamTransformer*()>;
+  RTCEncodedVideoUnderlyingSink(ScriptState*, TransformerCallback);
+
+  // UnderlyingSinkBase
+  ScriptPromise start(ScriptState*,
+                      WritableStreamDefaultController*,
+                      ExceptionState&) override;
+  ScriptPromise write(ScriptState*,
+                      ScriptValue chunk,
+                      WritableStreamDefaultController*,
+                      ExceptionState&) override;
+  ScriptPromise close(ScriptState*, ExceptionState&) override;
+  ScriptPromise abort(ScriptState*,
+                      ScriptValue reason,
+                      ExceptionState&) override;
+
+  void Trace(Visitor*) override;
+
+ private:
+  TransformerCallback transformer_callback_;
+};
+
+}  // namespace blink
+
+#endif  // THIRD_PARTY_BLINK_RENDERER_MODULES_PEERCONNECTION_RTC_ENCODED_VIDEO_UNDERLYING_SINK_H_

third_party/blink/renderer/modules/peerconnection/rtc_encoded_video_underlying_sink_test.cc

+// Copyright 2020 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 "third_party/blink/renderer/modules/peerconnection/rtc_encoded_video_underlying_sink.h"
+
+#include "base/memory/scoped_refptr.h"
+#include "testing/gmock/include/gmock/gmock.h"
+#include "testing/gtest/include/gtest/gtest.h"
+#include "third_party/blink/public/platform/scheduler/test/renderer_scheduler_test_support.h"
+#include "third_party/blink/renderer/bindings/core/v8/script_promise_tester.h"
+#include "third_party/blink/renderer/bindings/core/v8/v8_binding_for_testing.h"
+#include "third_party/blink/renderer/bindings/modules/v8/v8_rtc_encoded_video_frame.h"
+#include "third_party/blink/renderer/core/streams/writable_stream.h"
+#include "third_party/blink/renderer/core/streams/writable_stream_default_writer.h"
+#include "third_party/blink/renderer/platform/bindings/exception_state.h"
+#include "third_party/blink/renderer/platform/peerconnection/rtc_encoded_video_stream_transformer.h"
+#include "third_party/blink/renderer/platform/testing/testing_platform_support.h"
+#include "third_party/webrtc/api/frame_transformer_interface.h"
+#include "third_party/webrtc/api/scoped_refptr.h"
+#include "third_party/webrtc/rtc_base/ref_counted_object.h"
+
+using testing::_;
+
+namespace blink {
+
+namespace {
+
+class MockWebRtcTransformedFrameCallback
+    : public webrtc::TransformedFrameCallback {
+ public:
+  MOCK_METHOD1(OnTransformedFrame,
+               void(std::unique_ptr<webrtc::video_coding::EncodedFrame>));
+};
+
+}  // namespace
+
+class RTCEncodedVideoUnderlyingSinkTest : public testing::Test {
+ public:
+  RTCEncodedVideoUnderlyingSinkTest()
+      : main_task_runner_(
+            blink::scheduler::GetSingleThreadTaskRunnerForTesting()),
+        webrtc_callback_(
+            new rtc::RefCountedObject<MockWebRtcTransformedFrameCallback>()),
+        transformer_(main_task_runner_) {}
+
+  void SetUp() override {
+    EXPECT_FALSE(transformer_.HasTransformedFrameCallback());
+    transformer_.RegisterTransformedFrameCallback(webrtc_callback_);
+    EXPECT_TRUE(transformer_.HasTransformedFrameCallback());
+  }
+
+  void TearDown() override {
+    platform_->RunUntilIdle();
+    transformer_.UnregisterTransformedFrameCallback();
+    EXPECT_FALSE(transformer_.HasTransformedFrameCallback());
+  }
+
+  RTCEncodedVideoUnderlyingSink* CreateSink(ScriptState* script_state) {
+    return MakeGarbageCollected<RTCEncodedVideoUnderlyingSink>(
+        script_state,
+        WTF::BindRepeating(&RTCEncodedVideoUnderlyingSinkTest::GetTransformer,
+                           WTF::Unretained(this)));
+  }
+
+  RTCEncodedVideoStreamTransformer* GetTransformer() { return &transformer_; }
+
+  ScriptValue CreateEncodedVideoFrameChunk(ScriptState* script_state) {
+    RTCEncodedVideoFrame* frame = MakeGarbageCollected<RTCEncodedVideoFrame>(
+        /*frame_delegate=*/nullptr, std::vector<uint8_t>());
+    return ScriptValue(script_state->GetIsolate(),
+                       ToV8(frame, script_state->GetContext()->Global(),
+                            script_state->GetIsolate()));
+  }
+
+ protected:
+  ScopedTestingPlatformSupport<TestingPlatformSupport> platform_;
+  scoped_refptr<base::SingleThreadTaskRunner> main_task_runner_;
+  rtc::scoped_refptr<MockWebRtcTransformedFrameCallback> webrtc_callback_;
+  RTCEncodedVideoStreamTransformer transformer_;
+};
+
+TEST_F(RTCEncodedVideoUnderlyingSinkTest,
+       WriteToStreamForwardsToWebRtcCallback) {
+  V8TestingScope v8_scope;
+  ScriptState* script_state = v8_scope.GetScriptState();
+  auto* sink = CreateSink(script_state);
+  auto* stream =
+      WritableStream::CreateWithCountQueueingStrategy(script_state, sink, 1u);
+
+  NonThrowableExceptionState exception_state;
+  auto* writer = stream->getWriter(script_state, exception_state);
+
+  EXPECT_CALL(*webrtc_callback_, OnTransformedFrame(_));
+  ScriptPromiseTester write_tester(
+      script_state,
+      writer->write(script_state, CreateEncodedVideoFrameChunk(script_state),
+                    exception_state));
+  EXPECT_FALSE(write_tester.IsFulfilled());
+
+  writer->releaseLock(script_state);
+  ScriptPromiseTester close_tester(
+      script_state, stream->close(script_state, exception_state));
+  close_tester.WaitUntilSettled();
+
+  // Writing to the sink after the stream closes should fail.
+  DummyExceptionStateForTesting dummy_exception_state;
+  sink->write(script_state, CreateEncodedVideoFrameChunk(script_state), nullptr,
+              dummy_exception_state);
+  EXPECT_TRUE(dummy_exception_state.HadException());
+  EXPECT_EQ(dummy_exception_state.Code(),
+            static_cast<ExceptionCode>(DOMExceptionCode::kInvalidStateError));
+}
+
+TEST_F(RTCEncodedVideoUnderlyingSinkTest, WriteInvalidDataFails) {
+  V8TestingScope v8_scope;
+  ScriptState* script_state = v8_scope.GetScriptState();
+  auto* sink = CreateSink(script_state);
+  ScriptValue v8_integer = ScriptValue::From(script_state, 0);
+
+  // Writing something that is not an RTCEncodedVideoFrame integer to the sink
+  // should fail.
+  DummyExceptionStateForTesting dummy_exception_state;
+  sink->write(script_state, v8_integer, nullptr, dummy_exception_state);
+  EXPECT_TRUE(dummy_exception_state.HadException());
+}
+
+}  // namespace blink

third_party/blink/renderer/modules/peerconnection/rtc_encoded_video_underlying_source.cc

+// Copyright 2020 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 "third_party/blink/renderer/modules/peerconnection/rtc_encoded_video_underlying_source.h"
+
+#include "third_party/blink/renderer/core/streams/readable_stream_default_controller_with_script_scope.h"
+#include "third_party/blink/renderer/modules/peerconnection/rtc_encoded_video_frame.h"
+#include "third_party/webrtc/api/video/encoded_frame.h"
+
+#include "third_party/blink/renderer/bindings/core/v8/v8_throw_dom_exception.h"
+#include "third_party/blink/renderer/platform/bindings/exception_code.h"
+#include "third_party/blink/renderer/platform/bindings/exception_state.h"
+namespace blink {
+
+// Frames should not be queued at all. We allow queuing a few frames to deal
+// with transient slowdowns. Specified as a negative number of frames since
+// queuing is reported by the stream controller as a negative desired size.
+const int RTCEncodedVideoUnderlyingSource::kMinQueueDesiredSize = -60;
+
+RTCEncodedVideoUnderlyingSource::RTCEncodedVideoUnderlyingSource(
+    ScriptState* script_state,
+    base::OnceClosure disconnect_callback)
+    : UnderlyingSourceBase(script_state),
+      script_state_(script_state),
+      disconnect_callback_(std::move(disconnect_callback)) {
+  DCHECK(disconnect_callback_);
+}
+
+ScriptPromise RTCEncodedVideoUnderlyingSource::pull(ScriptState* script_state) {
+  DCHECK_CALLED_ON_VALID_THREAD(thread_checker_);
+  // WebRTC is a push source without backpressure support, so nothing to do
+  // here.
+  return ScriptPromise::CastUndefined(script_state);
+}
+
+ScriptPromise RTCEncodedVideoUnderlyingSource::Cancel(ScriptState* script_state,
+                                                      ScriptValue reason) {
+  DCHECK_CALLED_ON_VALID_THREAD(thread_checker_);
+  if (disconnect_callback_)
+    std::move(disconnect_callback_).Run();
+  return ScriptPromise::CastUndefined(script_state);
+}
+
+void RTCEncodedVideoUnderlyingSource::Trace(Visitor* visitor) {
+  visitor->Trace(script_state_);
+  UnderlyingSourceBase::Trace(visitor);
+}
+
+void RTCEncodedVideoUnderlyingSource::OnFrameFromSource(
+    std::unique_ptr<webrtc::video_coding::EncodedFrame> webrtc_frame,
+    std::vector<uint8_t> additional_data) {
+  DCHECK_CALLED_ON_VALID_THREAD(thread_checker_);
+  // If the source is canceled or there are too many queued frames,
+  // drop the new frame.
+  if (!disconnect_callback_ || !Controller() ||
+      Controller()->DesiredSize() <= kMinQueueDesiredSize) {
+    return;
+  }
+
+  RTCEncodedVideoFrame* encoded_frame =
+      MakeGarbageCollected<RTCEncodedVideoFrame>(std::move(webrtc_frame),
+                                                 std::move(additional_data));
+  Controller()->Enqueue(encoded_frame);
+}
+
+void RTCEncodedVideoUnderlyingSource::Close() {
+  DCHECK_CALLED_ON_VALID_THREAD(thread_checker_);
+  if (disconnect_callback_)
+    std::move(disconnect_callback_).Run();
+
+  if (Controller())
+    Controller()->Close();
+}
+
+}  // namespace blink

third_party/blink/renderer/modules/peerconnection/rtc_encoded_video_underlying_source.h

+// Copyright 2020 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.
+
+#ifndef THIRD_PARTY_BLINK_RENDERER_MODULES_PEERCONNECTION_RTC_ENCODED_VIDEO_UNDERLYING_SOURCE_H_
+#define THIRD_PARTY_BLINK_RENDERER_MODULES_PEERCONNECTION_RTC_ENCODED_VIDEO_UNDERLYING_SOURCE_H_
+
+#include "base/threading/thread_checker.h"
+#include "third_party/blink/renderer/core/streams/underlying_source_base.h"
+#include "third_party/blink/renderer/modules/modules_export.h"
+
+namespace webrtc {
+namespace video_coding {
+class EncodedFrame;
+}
+}  // namespace webrtc
+
+namespace blink {
+
+class MODULES_EXPORT RTCEncodedVideoUnderlyingSource
+    : public UnderlyingSourceBase {
+ public:
+  explicit RTCEncodedVideoUnderlyingSource(
+      ScriptState*,
+      base::OnceClosure disconnect_callback);
+
+  // UnderlyingSourceBase
+  ScriptPromise pull(ScriptState*) override;
+  ScriptPromise Cancel(ScriptState*, ScriptValue reason) override;
+
+  void OnFrameFromSource(std::unique_ptr<webrtc::video_coding::EncodedFrame>,
+                         std::vector<uint8_t> additional_data);
+  void Close();
+
+  void Trace(Visitor*) override;
+
+ private:
+  FRIEND_TEST_ALL_PREFIXES(RTCEncodedVideoUnderlyingSourceTest, QueuedFramesAreDroppedWhenOverflow);
+  static const int kMinQueueDesiredSize;
+
+  const Member<ScriptState> script_state_;
+  base::OnceClosure disconnect_callback_;
+  THREAD_CHECKER(thread_checker_);
+};
+
+}  // namespace blink
+
+#endif  // THIRD_PARTY_BLINK_RENDERER_MODULES_PEERCONNECTION_RTC_ENCODED_VIDEO_UNDERLYING_SOURCE_H_

third_party/blink/renderer/modules/peerconnection/rtc_encoded_video_underlying_source_test.cc

+// Copyright 2020 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 "third_party/blink/renderer/modules/peerconnection/rtc_encoded_video_underlying_source.h"
+
+#include "base/test/mock_callback.h"
+#include "testing/gmock/include/gmock/gmock.h"
+#include "testing/gtest/include/gtest/gtest.h"
+#include "third_party/blink/renderer/bindings/core/v8/script_promise_tester.h"
+#include "third_party/blink/renderer/bindings/core/v8/v8_binding_for_testing.h"
+#include "third_party/blink/renderer/core/streams/readable_stream.h"
+#include "third_party/blink/renderer/core/streams/readable_stream_default_controller_with_script_scope.h"
+#include "third_party/blink/renderer/platform/bindings/exception_state.h"
+#include "third_party/webrtc/api/video/encoded_frame.h"
+
+namespace blink {
+
+class RTCEncodedVideoUnderlyingSourceTest : public testing::Test {
+ public:
+  RTCEncodedVideoUnderlyingSource* CreateSource(ScriptState* script_state) {
+    return MakeGarbageCollected<RTCEncodedVideoUnderlyingSource>(
+        script_state, WTF::Bind(disconnect_callback_.Get()));
+  }
+
+ protected:
+  base::MockOnceClosure disconnect_callback_;
+};
+
+TEST_F(RTCEncodedVideoUnderlyingSourceTest,
+       SourceDataFlowsThroughStreamAndCloses) {
+  V8TestingScope v8_scope;
+  ScriptState* script_state = v8_scope.GetScriptState();
+  auto* source = CreateSource(script_state);
+  auto* stream =
+      ReadableStream::CreateWithCountQueueingStrategy(script_state, source, 0);
+
+  NonThrowableExceptionState exception_state;
+  auto* reader = stream->getReader(script_state, exception_state);
+
+  ScriptPromiseTester read_tester(script_state,
+                                  reader->read(script_state, exception_state));
+  EXPECT_FALSE(read_tester.IsFulfilled());
+  source->OnFrameFromSource(nullptr, std::vector<uint8_t>());
+  read_tester.WaitUntilSettled();
+  EXPECT_TRUE(read_tester.IsFulfilled());
+
+  EXPECT_CALL(disconnect_callback_, Run());
+  source->Close();
+}
+
+TEST_F(RTCEncodedVideoUnderlyingSourceTest, CancelStream) {
+  V8TestingScope v8_scope;
+  auto* source = CreateSource(v8_scope.GetScriptState());
+  auto* stream = ReadableStream::CreateWithCountQueueingStrategy(
+      v8_scope.GetScriptState(), source, 0);
+
+  EXPECT_CALL(disconnect_callback_, Run());
+  NonThrowableExceptionState exception_state;
+  stream->cancel(v8_scope.GetScriptState(), exception_state);
+}
+
+TEST_F(RTCEncodedVideoUnderlyingSourceTest, QueuedFramesAreDroppedWhenOverflow) {
+  V8TestingScope v8_scope;
+  ScriptState* script_state = v8_scope.GetScriptState();
+  auto* source = CreateSource(script_state);
+  // Create a stream, to ensure there is a controller associated to the source.
+  ReadableStream::CreateWithCountQueueingStrategy(v8_scope.GetScriptState(),
+                                                  source, 0);
+  for (int i = 0; i > RTCEncodedVideoUnderlyingSource::kMinQueueDesiredSize;
+       --i) {
+    EXPECT_EQ(source->Controller()->DesiredSize(), i);
+    source->OnFrameFromSource(nullptr, std::vector<uint8_t>());
+  }
+  EXPECT_EQ(source->Controller()->DesiredSize(),
+            RTCEncodedVideoUnderlyingSource::kMinQueueDesiredSize);
+
+  source->OnFrameFromSource(nullptr, std::vector<uint8_t>());
+  EXPECT_EQ(source->Controller()->DesiredSize(),
+            RTCEncodedVideoUnderlyingSource::kMinQueueDesiredSize);
+
+  EXPECT_CALL(disconnect_callback_, Run());
+  source->Close();
+}
+
+}  // namespace blink