// Copyright 2019 The Chromium Authors
// 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/websockets/websocket_stream.h"

#include <ios>
#include <memory>
#include <ostream>
#include <string>
#include <utility>

#include "third_party/blink/renderer/bindings/core/v8/capture_source_location.h"
#include "third_party/blink/renderer/bindings/core/v8/idl_types.h"
#include "third_party/blink/renderer/bindings/core/v8/native_value_traits_impl.h"
#include "third_party/blink/renderer/bindings/core/v8/script_promise.h"
#include "third_party/blink/renderer/bindings/core/v8/script_promise_resolver.h"
#include "third_party/blink/renderer/bindings/core/v8/v8_binding_for_core.h"
#include "third_party/blink/renderer/bindings/core/v8/v8_throw_dom_exception.h"
#include "third_party/blink/renderer/bindings/modules/v8/v8_websocket_close_info.h"
#include "third_party/blink/renderer/bindings/modules/v8/v8_websocket_error.h"
#include "third_party/blink/renderer/bindings/modules/v8/v8_websocket_open_info.h"
#include "third_party/blink/renderer/bindings/modules/v8/v8_websocket_stream_options.h"
#include "third_party/blink/renderer/core/dom/abort_signal.h"
#include "third_party/blink/renderer/core/execution_context/execution_context.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/core/streams/underlying_sink_base.h"
#include "third_party/blink/renderer/core/streams/underlying_source_base.h"
#include "third_party/blink/renderer/core/streams/writable_stream.h"
#include "third_party/blink/renderer/core/streams/writable_stream_default_controller.h"
#include "third_party/blink/renderer/core/typed_arrays/dom_array_buffer.h"
#include "third_party/blink/renderer/core/typed_arrays/dom_array_buffer_view.h"
#include "third_party/blink/renderer/core/typed_arrays/dom_typed_array.h"
#include "third_party/blink/renderer/modules/websockets/websocket_channel_impl.h"
#include "third_party/blink/renderer/modules/websockets/websocket_error.h"
#include "third_party/blink/renderer/platform/bindings/exception_code.h"
#include "third_party/blink/renderer/platform/bindings/exception_state.h"
#include "third_party/blink/renderer/platform/bindings/script_state.h"
#include "third_party/blink/renderer/platform/bindings/v8_binding.h"
#include "third_party/blink/renderer/platform/heap/garbage_collected.h"
#include "third_party/blink/renderer/platform/runtime_enabled_features.h"

namespace blink {

namespace {

// This is used in several places, so use a constant to avoid typos.
constexpr char kWebSocketNotCleanlyClosedErrorMessage[] =
    "WebSocket was not cleanly closed.";

constexpr char kClosedWebSocketStreamMessage[] =
    "Cannot write to a closed WebSocketStream.";

}  // namespace

class WebSocketStream::UnderlyingSource final : public UnderlyingSourceBase {
 public:
  UnderlyingSource(ScriptState* script_state, WebSocketStream* creator)
      : UnderlyingSourceBase(script_state), creator_(creator) {}

  // UnderlyingSourceBase implementation.
  ScriptPromise<IDLUndefined> Pull(ScriptState*, ExceptionState&) override;
  ScriptPromise<IDLUndefined> Cancel(ScriptState*,
                                     ScriptValue reason,
                                     ExceptionState&) override;

  // API for WebSocketStream.
  void DidReceiveTextMessage(const String&);
  void DidReceiveBinaryMessage(const Vector<base::span<const uint8_t>>&);
  void DidStartClosingHandshake();
  void DidCloseCleanly(uint16_t code, const String& reason);
  void CloseWithError(v8::Local<v8::Value> error);

  void Trace(Visitor* visitor) const override {
    visitor->Trace(creator_);
    UnderlyingSourceBase::Trace(visitor);
  }

 private:
  Member<WebSocketStream> creator_;
  bool closed_ = false;
};

class WebSocketStream::UnderlyingSink final : public UnderlyingSinkBase {
 public:
  explicit UnderlyingSink(WebSocketStream* creator) : creator_(creator) {}

  // UnderlyingSinkBase implementation.
  ScriptPromise<IDLUndefined> start(ScriptState*,
                                    WritableStreamDefaultController*,
                                    ExceptionState&) override;
  ScriptPromise<IDLUndefined> write(ScriptState*,
                                    ScriptValue chunk,
                                    WritableStreamDefaultController*,
                                    ExceptionState&) override;
  ScriptPromise<IDLUndefined> close(ScriptState*, ExceptionState&) override;
  ScriptPromise<IDLUndefined> abort(ScriptState*,
                                    ScriptValue reason,
                                    ExceptionState&) override;

  // API for WebSocketStream.
  void DidStartClosingHandshake();
  void DidCloseCleanly(uint16_t code, const String& reason);
  void CloseWithError(v8::Local<v8::Value> error);
  bool AllDataHasBeenConsumed() { return !is_writing_; }

  void Trace(Visitor* visitor) const override {
    visitor->Trace(creator_);
    visitor->Trace(close_resolver_);
    visitor->Trace(error_);
    UnderlyingSinkBase::Trace(visitor);
  }

 private:
  friend SendCompletionWatcherImpl;

  void ErrorControllerBecauseClosed();
  void ErrorControllerIfNeeded(v8::Local<v8::Value> with_error);
  void ResolveClose(bool was_clean);
  void SendArrayBuffer(ScriptState*,
                       DOMArrayBuffer*,
                       size_t offset,
                       size_t length,
                       std::unique_ptr<SendCompletionWatcherImpl> watcher);
  void SendString(ScriptState*,
                  v8::Local<v8::Value> v8chunk,
                  ScriptPromiseResolver<IDLUndefined>*,
                  std::unique_ptr<SendCompletionWatcherImpl> watcher);

  Member<WebSocketStream> creator_;
  Member<ScriptPromiseResolver<IDLUndefined>> close_resolver_;

  // If the stream is errored, `error_` contains the error we errored it with.
  ScriptValue error_;
  bool closed_ = false;
  bool is_writing_ = false;
};

class WebSocketStream::SendCompletionWatcherImpl final
    : public WebSocketChannel::SendCompletionWatcher {
 public:
  SendCompletionWatcherImpl(UnderlyingSink* sink,
                            ScriptPromiseResolver<IDLUndefined>* resolver)
      : sink_(sink), resolver_(resolver) {}

  ~SendCompletionWatcherImpl() override {
    if (!resolver_) {
      // Message was sent.
      return;
    }

    CHECK(sink_);
    resolver_->Reject(sink_->error_);
  }

  void OnMessageSent(bool synchronously) override {
    DVLOG(1) << "SendCompletionWatcherImpl::OnMessageSent " << this
             << " OnMessageSent(); synchronously = " << std::boolalpha
             << synchronously;
    CHECK(resolver_) << "OnMessageSent() should only be called once";
    resolver_->Resolve();
    resolver_ = nullptr;
    sink_->is_writing_ = false;
    sink_ = nullptr;
  }

 private:
  Persistent<UnderlyingSink> sink_;
  Persistent<ScriptPromiseResolver<IDLUndefined>> resolver_;
};

ScriptPromise<IDLUndefined> WebSocketStream::UnderlyingSource::Pull(
    ScriptState* script_state,
    ExceptionState&) {
  DVLOG(1) << "WebSocketStream::UnderlyingSource " << this << " Pull()";
  creator_->channel_->RemoveBackpressure();
  return ToResolvedUndefinedPromise(script_state);
}

ScriptPromise<IDLUndefined> WebSocketStream::UnderlyingSource::Cancel(
    ScriptState* script_state,
    ScriptValue reason,
    ExceptionState& exception_state) {
  DVLOG(1) << "WebSocketStream::UnderlyingSource " << this << " Cancel()";
  closed_ = true;
  creator_->CloseMaybeWithReason(reason, exception_state);
  return ToResolvedUndefinedPromise(script_state);
}

void WebSocketStream::UnderlyingSource::DidReceiveTextMessage(
    const String& string) {
  DVLOG(1) << "WebSocketStream::UnderlyingSource " << this
           << " DidReceiveTextMessage() string=" << string;

  DCHECK(!closed_);
  Controller()->Enqueue(
      V8String(creator_->script_state_->GetIsolate(), string));
  creator_->channel_->ApplyBackpressure();
}

void WebSocketStream::UnderlyingSource::DidReceiveBinaryMessage(
    const Vector<base::span<const uint8_t>>& data) {
  DVLOG(1) << "WebSocketStream::UnderlyingSource " << this
           << " DidReceiveBinaryMessage()";

  DCHECK(!closed_);

  if (RuntimeEnabledFeatures::WebSocketStreamStandardBinaryChunkTypeEnabled()) {
    auto* uint8Array = DOMUint8Array::Create(data);
    Controller()->Enqueue(uint8Array);
  } else {
    auto* buffer = DOMArrayBuffer::Create(data);
    Controller()->Enqueue(buffer);
  }

  creator_->channel_->ApplyBackpressure();
}

void WebSocketStream::UnderlyingSource::DidStartClosingHandshake() {
  DVLOG(1) << "WebSocketStream::UnderlyingSource " << this
           << " DidStartClosingHandshake()";

  DCHECK(!closed_);
  Controller()->Close();
  closed_ = true;
}

void WebSocketStream::UnderlyingSource::DidCloseCleanly(uint16_t code,
                                                        const String& reason) {
  DVLOG(1) << "WebSocketStream::UnderlyingSource " << this
           << " DidCloseCleanly() code=" << code << " reason=" << reason;

  if (closed_)
    return;

  closed_ = true;
  Controller()->Close();
}

void WebSocketStream::UnderlyingSource::CloseWithError(
    v8::Local<v8::Value> error) {
  DVLOG(1) << "WebSocketStream::UnderlyingSource::CloseWithError";
  if (closed_) {
    return;
  }

  closed_ = true;

  Controller()->Error(error);
}

ScriptPromise<IDLUndefined> WebSocketStream::UnderlyingSink::start(
    ScriptState* script_state,
    WritableStreamDefaultController*,
    ExceptionState& exception_state) {
  DVLOG(1) << "WebSocketStream::UnderlyingSink " << this << " start()";
  return ToResolvedUndefinedPromise(script_state);
}

ScriptPromise<IDLUndefined> WebSocketStream::UnderlyingSink::write(
    ScriptState* script_state,
    ScriptValue chunk,
    WritableStreamDefaultController*,
    ExceptionState& exception_state) {
  DVLOG(1) << "WebSocketStream::UnderlyingSink " << this << " write()";
  is_writing_ = true;

  v8::Local<v8::Value> v8chunk = chunk.V8Value();
  auto* isolate = script_state->GetIsolate();
  DOMArrayBuffer* data = nullptr;
  size_t offset = 0;
  size_t length = 0;
  if (v8chunk->IsArrayBuffer()) {
    data = NativeValueTraits<DOMArrayBuffer>::NativeValue(isolate, v8chunk,
                                                          exception_state);
    if (exception_state.HadException()) {
      closed_ = true;
      is_writing_ = false;
      return EmptyPromise();
    }
    length = data->ByteLength();
  } else if (v8chunk->IsArrayBufferView()) {
    NotShared<DOMArrayBufferView> data_view =
        NativeValueTraits<NotShared<DOMArrayBufferView>>::NativeValue(
            isolate, v8chunk, exception_state);
    if (exception_state.HadException()) {
      closed_ = true;
      is_writing_ = false;
      return EmptyPromise();
    }
    data = data_view->buffer();
    offset = data_view->byteOffset();
    length = data_view->byteLength();
  }

  auto* resolver = MakeGarbageCollected<ScriptPromiseResolver<IDLUndefined>>(
      script_state, exception_state.GetContext());
  auto result = resolver->Promise();
  auto watcher = std::make_unique<SendCompletionWatcherImpl>(this, resolver);
  if (data) {
    SendArrayBuffer(script_state, data, offset, length, std::move(watcher));
  } else {
    SendString(script_state, v8chunk, resolver, std::move(watcher));
  }
  return result;
}

ScriptPromise<IDLUndefined> WebSocketStream::UnderlyingSink::close(
    ScriptState* script_state,
    ExceptionState& exception_state) {
  DVLOG(1) << "WebSocketStream::UnderlyingSink " << this << " close()";
  closed_ = true;
  creator_->CloseWithUnspecifiedCode(exception_state);
  DCHECK(!close_resolver_);
  close_resolver_ = MakeGarbageCollected<ScriptPromiseResolver<IDLUndefined>>(
      script_state, exception_state.GetContext());
  return close_resolver_->Promise();
}

ScriptPromise<IDLUndefined> WebSocketStream::UnderlyingSink::abort(
    ScriptState* script_state,
    ScriptValue reason,
    ExceptionState& exception_state) {
  DVLOG(1) << "WebSocketStream::UnderlyingSink " << this << " abort()";

  closed_ = true;
  creator_->CloseMaybeWithReason(reason, exception_state);
  return ToResolvedUndefinedPromise(script_state);
}

void WebSocketStream::UnderlyingSink::DidStartClosingHandshake() {
  DVLOG(1) << "WebSocketStream::UnderlyingSink " << this
           << " DidStartClosingHandshake()";

  if (closed_)
    return;
  closed_ = true;

  ErrorControllerBecauseClosed();
}

void WebSocketStream::UnderlyingSink::DidCloseCleanly(uint16_t code,
                                                      const String& reason) {
  DVLOG(1) << "WebSocketStream::UnderlyingSink " << this
           << " DidCloseCleanly() code=" << code << " reason=" << reason;

  if (close_resolver_) {
    ResolveClose(/*was_clean=*/true);
  }

  if (closed_)
    return;
  closed_ = true;

  ErrorControllerBecauseClosed();
}

void WebSocketStream::UnderlyingSink::CloseWithError(
    v8::Local<v8::Value> error) {
  if (close_resolver_) {
    ResolveClose(/*was_clean=*/false);
  }

  if (closed_) {
    return;
  }
  closed_ = true;

  ErrorControllerIfNeeded(error);
}

void WebSocketStream::UnderlyingSink::ErrorControllerBecauseClosed() {
  ScriptState* script_state = creator_->script_state_;
  ErrorControllerIfNeeded(V8ThrowDOMException::CreateOrEmpty(
      script_state->GetIsolate(), DOMExceptionCode::kInvalidStateError,
      kClosedWebSocketStreamMessage));
}

void WebSocketStream::UnderlyingSink::ErrorControllerIfNeeded(
    v8::Local<v8::Value> with_error) {
  if (error_.IsEmpty()) {
    ScriptState* script_state = creator_->script_state_;
    error_ = ScriptValue(script_state->GetIsolate(), with_error);
    Controller()->error(script_state, error_);
  }
}

void WebSocketStream::UnderlyingSink::ResolveClose(bool was_clean) {
  DCHECK(close_resolver_);

  if (was_clean) {
    close_resolver_->Resolve();
    return;
  }

  close_resolver_->Reject(
      creator_->CreateWebSocketError(kWebSocketNotCleanlyClosedErrorMessage));
}

void WebSocketStream::UnderlyingSink::SendArrayBuffer(
    ScriptState* script_state,
    DOMArrayBuffer* buffer,
    size_t offset,
    size_t length,
    std::unique_ptr<SendCompletionWatcherImpl> watcher) {
  DVLOG(1) << "WebSocketStream::UnderlyingSink " << this
           << " SendArrayBuffer() buffer = " << buffer << " offset = " << offset
           << " length = " << length;

  creator_->channel_->Send(*buffer, offset, length, std::move(watcher));
}

void WebSocketStream::UnderlyingSink::SendString(
    ScriptState* script_state,
    v8::Local<v8::Value> v8chunk,
    ScriptPromiseResolver<IDLUndefined>* resolver,
    std::unique_ptr<SendCompletionWatcherImpl> watcher) {
  DVLOG(1) << "WebSocketStream::UnderlyingSink " << this << " SendString()";
  auto* isolate = script_state->GetIsolate();
  v8::TryCatch try_catch(isolate);
  v8::Local<v8::String> string_chunk;
  if (!v8chunk->ToString(script_state->GetContext()).ToLocal(&string_chunk)) {
    closed_ = true;
    resolver->Reject(try_catch.Exception());
    is_writing_ = false;
    return;
  }
  // Skip one string copy by using v8::String UTF8 conversion instead of going
  // via blink::String.
  size_t utf8_length = string_chunk->Utf8LengthV2(isolate);
  std::string message(utf8_length, '\0');
  size_t written_length =
      string_chunk->WriteUtf8V2(isolate, message.data(), utf8_length,
                                v8::String::WriteFlags::kReplaceInvalidUtf8);
  DCHECK_EQ(utf8_length, written_length);
  creator_->channel_->Send(message, std::move(watcher));
}

WebSocketStream* WebSocketStream::Create(ScriptState* script_state,
                                         const String& url,
                                         WebSocketStreamOptions* options,
                                         ExceptionState& exception_state) {
  DVLOG(1) << "WebSocketStream Create() url=" << url << " options=" << options;
  return CreateInternal(script_state, url, options, nullptr, exception_state);
}

WebSocketStream* WebSocketStream::CreateForTesting(
    ScriptState* script_state,
    const String& url,
    WebSocketStreamOptions* options,
    WebSocketChannel* channel,
    ExceptionState& exception_state) {
  DVLOG(1) << "WebSocketStream CreateForTesting() url=" << url
           << " options=" << options << " channel=" << channel;

  DCHECK(channel) << "Don't use a real channel when testing";
  return CreateInternal(script_state, url, options, channel, exception_state);
}

WebSocketStream* WebSocketStream::CreateInternal(
    ScriptState* script_state,
    const String& url,
    WebSocketStreamOptions* options,
    WebSocketChannel* channel,
    ExceptionState& exception_state) {
  DVLOG(1) << "WebSocketStream CreateInternal() url=" << url
           << " options=" << options << " channel=" << channel;

  auto* execution_context = ExecutionContext::From(script_state);
  auto* stream =
      MakeGarbageCollected<WebSocketStream>(execution_context, script_state);
  if (channel) {
    stream->channel_ = channel;
  } else {
    stream->channel_ = WebSocketChannelImpl::Create(
        execution_context, stream, CaptureSourceLocation(execution_context));
  }
  stream->Connect(script_state, url, options, exception_state);
  if (exception_state.HadException()) {
    return nullptr;
  }

  return stream;
}

WebSocketStream::WebSocketStream(ExecutionContext* execution_context,
                                 ScriptState* script_state)
    : ActiveScriptWrappable<WebSocketStream>({}),
      ExecutionContextLifecycleObserver(execution_context),
      script_state_(script_state),
      opened_(MakeGarbageCollected<
              ScriptPromiseProperty<WebSocketOpenInfo, IDLAny>>(
          GetExecutionContext())),
      closed_(MakeGarbageCollected<
              ScriptPromiseProperty<WebSocketCloseInfo, IDLAny>>(
          GetExecutionContext())) {}

WebSocketStream::~WebSocketStream() = default;

ScriptPromise<WebSocketOpenInfo> WebSocketStream::opened(
    ScriptState* script_state) const {
  return opened_->Promise(script_state->World());
}

ScriptPromise<WebSocketCloseInfo> WebSocketStream::closed(
    ScriptState* script_state) const {
  return closed_->Promise(script_state->World());
}

void WebSocketStream::close(WebSocketCloseInfo* info,
                            ExceptionState& exception_state) {
  DVLOG(1) << "WebSocketStream " << this << " close() info=" << info;
  CloseInternal(info->hasCloseCode() ? std::make_optional(info->closeCode())
                                     : std::nullopt,
                info->reason(), exception_state);
}

void WebSocketStream::DidConnect(const String& subprotocol,
                                 const String& extensions) {
  DVLOG(1) << "WebSocketStream " << this
           << " DidConnect() subprotocol=" << subprotocol
           << " extensions=" << extensions;

  if (!channel_) {
    return;
  }

  ScriptState::Scope scope(script_state_);
  if (common_.GetState() != WebSocketCommon::kConnecting) {
    return;
  }
  common_.SetState(WebSocketCommon::kOpen);
  was_ever_connected_ = true;
  auto* open_info = MakeGarbageCollected<WebSocketOpenInfo>();
  open_info->setProtocol(subprotocol);
  open_info->setExtensions(extensions);
  source_ = MakeGarbageCollected<UnderlyingSource>(script_state_, this);
  auto* readable = ReadableStream::CreateWithCountQueueingStrategy(
      script_state_, source_, 1);
  sink_ = MakeGarbageCollected<UnderlyingSink>(this);
  auto* writable =
      WritableStream::CreateWithCountQueueingStrategy(script_state_, sink_, 1);
  open_info->setReadable(readable);
  open_info->setWritable(writable);
  opened_->Resolve(open_info);
  abort_handle_.Clear();
}

void WebSocketStream::DidReceiveTextMessage(const String& string) {
  DVLOG(1) << "WebSocketStream " << this
           << " DidReceiveTextMessage() string=" << string;

  if (!channel_) {
    return;
  }

  ScriptState::Scope scope(script_state_);
  source_->DidReceiveTextMessage(string);
}

void WebSocketStream::DidReceiveBinaryMessage(
    const Vector<base::span<const uint8_t>>& data) {
  DVLOG(1) << "WebSocketStream " << this << " DidReceiveBinaryMessage()";
  if (!channel_) {
    return;
  }

  ScriptState::Scope scope(script_state_);
  source_->DidReceiveBinaryMessage(data);
}

void WebSocketStream::DidError() {
  // This is not useful as it is always followed by a call to DidClose().
}

void WebSocketStream::DidConsumeBufferedAmount(uint64_t consumed) {
  // This is only relevant to DOMWebSocket.
}

void WebSocketStream::DidStartClosingHandshake() {
  DVLOG(1) << "WebSocketStream " << this << " DidStartClosingHandshake()";
  if (!channel_) {
    return;
  }

  ScriptState::Scope scope(script_state_);
  common_.SetState(WebSocketCommon::kClosing);
  source_->DidStartClosingHandshake();
  sink_->DidStartClosingHandshake();
}

void WebSocketStream::DidClose(
    ClosingHandshakeCompletionStatus closing_handshake_completion,
    uint16_t code,
    const String& reason) {
  DVLOG(1) << "WebSocketStream " << this
           << " DidClose() closing_handshake_completion="
           << closing_handshake_completion << " code=" << code
           << " reason=" << reason;

  if (!channel_) {
    return;
  }

  ScriptState::Scope scope(script_state_);
  if (!was_ever_connected_) {
    opened_->Reject(ScriptValue(
        script_state_->GetIsolate(),
        CreateWebSocketError("WebSocket closed before handshake complete.")));
  }
  bool all_data_was_consumed = sink_ ? sink_->AllDataHasBeenConsumed() : true;
  bool was_clean = common_.GetState() == WebSocketCommon::kClosing &&
                   all_data_was_consumed &&
                   closing_handshake_completion == kClosingHandshakeComplete &&
                   code != WebSocketChannel::kCloseEventCodeAbnormalClosure;
  common_.SetState(WebSocketCommon::kClosed);

  channel_->Disconnect();
  abort_handle_.Clear();
  if (was_clean) {
    if (source_) {
      source_->DidCloseCleanly(code, reason);
    }
    if (sink_) {
      sink_->DidCloseCleanly(code, reason);
    }
    closed_->Resolve(MakeCloseInfo(code, reason));
  } else {
    auto error = CreateWebSocketError(kWebSocketNotCleanlyClosedErrorMessage,
                                      code, reason);
    if (source_) {
      source_->CloseWithError(error);
    }
    if (sink_) {
      sink_->CloseWithError(error);
    }
    closed_->Reject(ScriptValue(script_state_->GetIsolate(), error));
  }
  // By freeing the WebSocketChannel last, we ensure that any unsent messages
  // are able to see the error stored in the sink.
  channel_ = nullptr;
}

void WebSocketStream::ContextDestroyed() {
  DVLOG(1) << "WebSocketStream " << this << " ContextDestroyed()";
  if (channel_) {
    channel_ = nullptr;
  }
  if (common_.GetState() != WebSocketCommon::kClosed) {
    common_.SetState(WebSocketCommon::kClosed);
  }
  abort_handle_.Clear();
}

bool WebSocketStream::HasPendingActivity() const {
  return channel_ != nullptr;
}

void WebSocketStream::Trace(Visitor* visitor) const {
  visitor->Trace(script_state_);
  visitor->Trace(opened_);
  visitor->Trace(closed_);
  visitor->Trace(channel_);
  visitor->Trace(source_);
  visitor->Trace(sink_);
  visitor->Trace(abort_handle_);
  ScriptWrappable::Trace(visitor);
  ExecutionContextLifecycleObserver::Trace(visitor);
  WebSocketChannelClient::Trace(visitor);
}

void WebSocketStream::Connect(ScriptState* script_state,
                              const String& url,
                              WebSocketStreamOptions* options,
                              ExceptionState& exception_state) {
  DVLOG(1) << "WebSocketStream " << this << " Connect() url=" << url
           << " options=" << options;

  // Don't read all of a huge initial message before read() has been called.
  channel_->ApplyBackpressure();

  if (options->hasSignal()) {
    auto* signal = options->signal();
    if (signal->aborted()) {
      auto exception = V8ThrowDOMException::CreateOrEmpty(
          script_state->GetIsolate(), DOMExceptionCode::kAbortError,
          "WebSocket handshake was aborted");
      opened_->Reject(ScriptValue(script_state_->GetIsolate(), exception));
      closed_->Reject(ScriptValue(script_state_->GetIsolate(), exception));
      return;
    }

    abort_handle_ = signal->AddAlgorithm(
        BindOnce(&WebSocketStream::OnAbort, WrapWeakPersistent(this)));
  }

  auto result = common_.Connect(
      ExecutionContext::From(script_state), url,
      options->hasProtocols() ? options->protocols() : Vector<String>(),
      channel_, exception_state);

  switch (result) {
    case WebSocketCommon::ConnectResult::kSuccess:
      DCHECK(!exception_state.HadException());
      return;

    case WebSocketCommon::ConnectResult::kException:
      DCHECK(exception_state.HadException());
      channel_ = nullptr;
      return;

    case WebSocketCommon::ConnectResult::kAsyncError:
      DCHECK(!exception_state.HadException());
      auto exception = V8ThrowDOMException::CreateOrEmpty(
          script_state->GetIsolate(), DOMExceptionCode::kSecurityError,
          "An attempt was made to break through the security policy of the "
          "user agent.",
          "WebSocket mixed content check failed.");
      opened_->Reject(ScriptValue(script_state_->GetIsolate(), exception));
      closed_->Reject(ScriptValue(script_state_->GetIsolate(), exception));
      return;
  }
}

// If |maybe_reason| contains a valid code and reason, then closes with it,
// otherwise closes with unspecified code and reason.
void WebSocketStream::CloseMaybeWithReason(ScriptValue maybe_reason,
                                           ExceptionState& exception_state) {
  DVLOG(1) << "WebSocketStream " << this << " CloseMaybeWithReason()";
  WebSocketError* info = V8WebSocketError::ToWrappable(
      script_state_->GetIsolate(), maybe_reason.V8Value());
  if (info) {
    CloseInternal(info->closeCode(), info->reason(), exception_state);
  } else {
    CloseWithUnspecifiedCode(exception_state);
  }
}

void WebSocketStream::CloseWithUnspecifiedCode(
    ExceptionState& exception_state) {
  DVLOG(1) << "WebSocketStream " << this << " CloseWithUnspecifiedCode()";
  CloseInternal(std::nullopt, String(), exception_state);
  DCHECK(!exception_state.HadException());
}

void WebSocketStream::CloseInternal(std::optional<uint16_t> code,
                                    const String& reason,
                                    ExceptionState& exception_state) {
  DVLOG(1) << "WebSocketStream " << this
           << " CloseInternal() code=" << (code ? code.value() : uint16_t{0})
           << " reason=" << reason;

  common_.CloseInternal(code, reason, channel_, exception_state);
}

v8::Local<v8::Value> WebSocketStream::CreateWebSocketError(
    String message,
    std::optional<uint16_t> close_code,
    String reason) {
  return WebSocketError::Create(script_state_->GetIsolate(), std::move(message),
                                close_code, std::move(reason));
}

void WebSocketStream::OnAbort() {
  DVLOG(1) << "WebSocketStream " << this << " OnAbort()";

  if (was_ever_connected_ || !channel_) {
    return;
  }

  channel_->CancelHandshake();
  channel_ = nullptr;

  auto exception = V8ThrowDOMException::CreateOrEmpty(
      script_state_->GetIsolate(), DOMExceptionCode::kAbortError,
      "WebSocket handshake was aborted");
  opened_->Reject(ScriptValue(script_state_->GetIsolate(), exception));
  closed_->Reject(ScriptValue(script_state_->GetIsolate(), exception));
  abort_handle_.Clear();
}

WebSocketCloseInfo* WebSocketStream::MakeCloseInfo(uint16_t close_code,
                                                   const String& reason) {
  DVLOG(1) << "WebSocketStream MakeCloseInfo() code=" << close_code
           << " reason=" << reason;

  auto* info = MakeGarbageCollected<WebSocketCloseInfo>();
  info->setCloseCode(close_code);
  info->setReason(reason);
  return info;
}

}  // namespace blink