Direct Sockets: Implements error handling for TCPSocket.

Below documents are from Eric Willigers <ericwilligers@chromium.org>.
Explainer: https://github.com/WICG/raw-sockets/blob/master/docs/explainer.md

Design doc:
https://docs.google.com/document/d/1Xa5nFkIWxkL3hZHvDYWPhT8sZvNeFpCUKNuqIwZHxnE/edit?usp=sharing

Bug: 905818
Change-Id: I3d956d6310e865b603a25122d937d70d6dcf2383
Reviewed-on: https://chromium-review.googlesource.com/c/chromium/src/+/2608092Reviewed-by: Glen Robertson <glenrob@chromium.org>
Reviewed-by: Eric Willigers <ericwilligers@chromium.org>
Commit-Queue: Ke He <kehe@chromium.org>
Cr-Commit-Position: refs/heads/master@{#840064}

third_party/blink/renderer/modules/direct_sockets/tcp_socket.cc

@@ -76,35 +76,7 @@ void TCPSocket::Init(int32_t result,
 }
 
 ScriptPromise TCPSocket::close(ScriptState* script_state, ExceptionState&) {
-  local_addr_ = base::nullopt;
-  peer_addr_ = base::nullopt;
-  tcp_socket_.reset();
-  socket_observer_receiver_.reset();
-  feature_handle_for_scheduler_.reset();
-  if (resolver_) {
-    resolver_->Reject(MakeGarbageCollected<DOMException>(
-        DOMExceptionCode::kAbortError, "The request was aborted locally"));
-    resolver_ = nullptr;
-
-    DCHECK(!tcp_readable_stream_wrapper_);
-    DCHECK(!tcp_writable_stream_wrapper_);
-
-    return ScriptPromise::CastUndefined(script_state);
-  }
-
-  if (tcp_readable_stream_wrapper_ &&
-      tcp_readable_stream_wrapper_->GetState() ==
-          TCPReadableStreamWrapper::State::kOpen) {
-    tcp_readable_stream_wrapper_->Reset();
-  }
-  tcp_readable_stream_wrapper_ = nullptr;
-
-  if (tcp_writable_stream_wrapper_ &&
-      tcp_writable_stream_wrapper_->GetState() ==
-          TCPWritableStreamWrapper::State::kOpen) {
-    tcp_writable_stream_wrapper_->Reset();
-  }
-  tcp_writable_stream_wrapper_ = nullptr;
+  DoClose(/*is_local_close=*/true);
 
   return ScriptPromise::CastUndefined(script_state);
 }
@@ -130,13 +102,19 @@ uint16_t TCPSocket::remotePort() const {
 }
 
 void TCPSocket::OnReadError(int32_t net_error) {
-  // TODO(crbug.com/905818): Implement error handling.
-  NOTIMPLEMENTED();
+  if (net_error > 0 || net_error == net::Error::ERR_IO_PENDING) {
+    return;
+  }
+
+  ResetReadableStream();
 }
 
 void TCPSocket::OnWriteError(int32_t net_error) {
-  // TODO(crbug.com/905818): Implement error handling.
-  NOTIMPLEMENTED();
+  if (net_error > 0 || net_error == net::Error::ERR_IO_PENDING) {
+    return;
+  }
+
+  ResetWritableStream();
 }
 
 void TCPSocket::Trace(Visitor* visitor) const {
@@ -147,24 +125,65 @@ void TCPSocket::Trace(Visitor* visitor) const {
 }
 
 void TCPSocket::OnSocketObserverConnectionError() {
-  // TODO(crbug.com/905818): Implement error handling.
-  NOTIMPLEMENTED();
+  DoClose(/*is_local_close=*/false);
 }
 
 void TCPSocket::OnReadableStreamAbort() {
-  if (tcp_writable_stream_wrapper_->GetState() ==
-      TCPWritableStreamWrapper::State::kAborted) {
+  ResetWritableStream();
+}
+
+void TCPSocket::OnWritableStreamAbort() {
+  ResetReadableStream();
+}
+
+void TCPSocket::DoClose(bool is_local_close) {
+  local_addr_ = base::nullopt;
+  peer_addr_ = base::nullopt;
+  tcp_socket_.reset();
+  socket_observer_receiver_.reset();
+  feature_handle_for_scheduler_.reset();
+
+  if (resolver_) {
+    DOMExceptionCode code = is_local_close ? DOMExceptionCode::kAbortError
+                                           : DOMExceptionCode::kNetworkError;
+    String message =
+        String::Format("The request was aborted %s",
+                       is_local_close ? "locally" : "due to connection error");
+    resolver_->Reject(MakeGarbageCollected<DOMException>(code, message));
+    resolver_ = nullptr;
+
+    DCHECK(!tcp_readable_stream_wrapper_);
+    DCHECK(!tcp_writable_stream_wrapper_);
+
     return;
   }
-  tcp_writable_stream_wrapper_->Reset();
+
+  ResetReadableStream();
+  ResetWritableStream();
 }
 
-void TCPSocket::OnWritableStreamAbort() {
+void TCPSocket::ResetReadableStream() {
+  if (!tcp_readable_stream_wrapper_)
+    return;
+
   if (tcp_readable_stream_wrapper_->GetState() ==
       TCPReadableStreamWrapper::State::kAborted) {
     return;
   }
   tcp_readable_stream_wrapper_->Reset();
+  tcp_readable_stream_wrapper_ = nullptr;
+}
+
+void TCPSocket::ResetWritableStream() {
+  if (!tcp_writable_stream_wrapper_)
+    return;
+
+  if (tcp_writable_stream_wrapper_->GetState() ==
+      TCPWritableStreamWrapper::State::kAborted) {
+    return;
+  }
+  tcp_writable_stream_wrapper_->Reset();
+  tcp_writable_stream_wrapper_ = nullptr;
 }
 
 }  // namespace blink

third_party/blink/renderer/modules/direct_sockets/tcp_socket.h

@@ -71,6 +71,10 @@ class MODULES_EXPORT TCPSocket final
   void OnReadableStreamAbort();
   void OnWritableStreamAbort();
 
+  void DoClose(bool is_local_close);
+  void ResetReadableStream();
+  void ResetWritableStream();
+
   Member<ScriptPromiseResolver> resolver_;
   FrameOrWorkerScheduler::SchedulingAffectingFeatureHandle
       feature_handle_for_scheduler_;

third_party/blink/renderer/modules/direct_sockets/tcp_socket_unittest.cc

@@ -9,6 +9,7 @@
 #include "third_party/blink/renderer/bindings/core/v8/script_promise.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/core/v8/v8_dom_exception.h"
 #include "third_party/blink/renderer/platform/heap/heap.h"
 
 namespace blink {
@@ -70,6 +71,13 @@ TEST(TCPSocketTest, CloseBeforeInit) {
 
   create_tester.WaitUntilSettled();
   ASSERT_TRUE(create_tester.IsRejected());
+
+  DOMException* create_exception = V8DOMException::ToImplWithTypeCheck(
+      scope.GetIsolate(), create_tester.Value().V8Value());
+  ASSERT_TRUE(create_exception);
+  EXPECT_EQ(create_exception->name(), "AbortError");
+  EXPECT_EQ(create_exception->message(), "The request was aborted locally");
+
   close_tester.WaitUntilSettled();
   ASSERT_TRUE(close_tester.IsFulfilled());
 }
@@ -95,6 +103,13 @@ TEST(TCPSocketTest, CloseAfterInitWithoutResultOK) {
 
   create_tester.WaitUntilSettled();
   ASSERT_TRUE(create_tester.IsRejected());
+
+  DOMException* create_exception = V8DOMException::ToImplWithTypeCheck(
+      scope.GetIsolate(), create_tester.Value().V8Value());
+  ASSERT_TRUE(create_exception);
+  EXPECT_EQ(create_exception->name(), "NotAllowedError");
+  EXPECT_EQ(create_exception->message(), "Permission denied");
+
   close_tester.WaitUntilSettled();
   ASSERT_TRUE(close_tester.IsFulfilled());
 }
@@ -126,6 +141,31 @@ TEST(TCPSocketTest, CloseAfterInitWithResultOK) {
   ASSERT_TRUE(close_tester.IsFulfilled());
 }
 
+TEST(TCPSocketTest, OnSocketObserverConnectionError) {
+  V8TestingScope scope;
+  TCPSocketCreator tcp_socket_creator;
+
+  auto* tcp_socket = tcp_socket_creator.Create(scope);
+  auto* script_state = scope.GetScriptState();
+  auto create_promise = tcp_socket_creator.GetSciptPromise();
+  ScriptPromiseTester create_tester(script_state, create_promise);
+  ASSERT_FALSE(create_tester.IsRejected());
+
+  // Trigger OnSocketObserverConnectionError().
+  auto observer = tcp_socket->GetTCPSocketObserver();
+  observer.reset();
+
+  create_tester.WaitUntilSettled();
+  ASSERT_TRUE(create_tester.IsRejected());
+
+  DOMException* create_exception = V8DOMException::ToImplWithTypeCheck(
+      scope.GetIsolate(), create_tester.Value().V8Value());
+  ASSERT_TRUE(create_exception);
+  EXPECT_EQ(create_exception->name(), "NetworkError");
+  EXPECT_EQ(create_exception->message(),
+            "The request was aborted due to connection error");
+}
+
 }  // namespace
 
 }  // namespace blink