chromium/src/third_party/blink/renderer/modules/direct_sockets/tcp_socket.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/direct_sockets/tcp_socket.h"

#include "base/macros.h"
#include "net/base/net_errors.h"
#include "third_party/blink/renderer/core/dom/dom_exception.h"
#include "third_party/blink/renderer/core/execution_context/execution_context.h"
#include "third_party/blink/renderer/platform/scheduler/public/scheduling_policy.h"
#include "third_party/blink/renderer/platform/wtf/functional.h"

namespace blink {

TCPSocket::TCPSocket(ScriptPromiseResolver& resolver)
    : resolver_(&resolver),
      feature_handle_for_scheduler_(
          ExecutionContext::From(resolver_->GetScriptState())
              ->GetScheduler()
              ->RegisterFeature(
                  SchedulingPolicy::Feature::
                      kOutstandingNetworkRequestDirectSocket,
                  {SchedulingPolicy::DisableBackForwardCache()})) {
  DCHECK(resolver_);
}

TCPSocket::~TCPSocket() = default;

mojo::PendingReceiver<network::mojom::blink::TCPConnectedSocket>
TCPSocket::GetTCPSocketReceiver() {
  DCHECK(resolver_);
  return tcp_socket_.BindNewPipeAndPassReceiver();
}

mojo::PendingRemote<network::mojom::blink::SocketObserver>
TCPSocket::GetTCPSocketObserver() {
  DCHECK(resolver_);
  auto result = socket_observer_receiver_.BindNewPipeAndPassRemote();

  socket_observer_receiver_.set_disconnect_handler(WTF::Bind(
      &TCPSocket::OnSocketObserverConnectionError, WrapPersistent(this)));

  return result;
}

void TCPSocket::Init(int32_t result,
                     const base::Optional<net::IPEndPoint>& local_addr,
                     const base::Optional<net::IPEndPoint>& peer_addr,
                     mojo::ScopedDataPipeConsumerHandle receive_stream,
                     mojo::ScopedDataPipeProducerHandle send_stream) {
  DCHECK(resolver_);
  DCHECK(!tcp_readable_stream_wrapper_);
  DCHECK(!tcp_writable_stream_wrapper_);
  if (result == net::Error::OK) {
    local_addr_ = local_addr;
    peer_addr_ = peer_addr;
    tcp_readable_stream_wrapper_ =
        MakeGarbageCollected<TCPReadableStreamWrapper>(
            resolver_->GetScriptState(),
            WTF::Bind(&TCPSocket::OnReadableStreamAbort,
                      WrapWeakPersistent(this)),
            std::move(receive_stream));
    tcp_writable_stream_wrapper_ =
        MakeGarbageCollected<TCPWritableStreamWrapper>(
            resolver_->GetScriptState(),
            WTF::Bind(&TCPSocket::OnWritableStreamAbort,
                      WrapWeakPersistent(this)),
            std::move(send_stream));
    resolver_->Resolve(this);
  } else {
    resolver_->Reject(MakeGarbageCollected<DOMException>(
        DOMExceptionCode::kNotAllowedError, "Permission denied"));
    socket_observer_receiver_.reset();
  }
  resolver_ = nullptr;
}

ScriptPromise TCPSocket::close(ScriptState* script_state, ExceptionState&) {
  DoClose(/*is_local_close=*/true);

  return ScriptPromise::CastUndefined(script_state);
}

ReadableStream* TCPSocket::readable() const {
  DCHECK(tcp_readable_stream_wrapper_);
  return tcp_readable_stream_wrapper_->Readable();
}

WritableStream* TCPSocket::writable() const {
  DCHECK(tcp_writable_stream_wrapper_);
  return tcp_writable_stream_wrapper_->Writable();
}

String TCPSocket::remoteAddress() const {
  DCHECK(peer_addr_);
  return String::FromUTF8(peer_addr_->ToStringWithoutPort());
}

uint16_t TCPSocket::remotePort() const {
  DCHECK(peer_addr_);
  return peer_addr_->port();
}

void TCPSocket::OnReadError(int32_t net_error) {
  if (net_error > 0 || net_error == net::Error::ERR_IO_PENDING) {
    return;
  }

  ResetReadableStream();
}

void TCPSocket::OnWriteError(int32_t net_error) {
  if (net_error > 0 || net_error == net::Error::ERR_IO_PENDING) {
    return;
  }

  ResetWritableStream();
}

void TCPSocket::Trace(Visitor* visitor) const {
  visitor->Trace(resolver_);
  visitor->Trace(tcp_readable_stream_wrapper_);
  visitor->Trace(tcp_writable_stream_wrapper_);
  ScriptWrappable::Trace(visitor);
}

void TCPSocket::OnSocketObserverConnectionError() {
  DoClose(/*is_local_close=*/false);
}

void TCPSocket::OnReadableStreamAbort() {
  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;
  }

  ResetReadableStream();
  ResetWritableStream();
}

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