Implement unshipped messages in message port.

cobalt/web/BUILD.gn

@@ -185,6 +185,7 @@ target(gtest_target_type, "web_test") {
     "event_target_test.cc",
     "event_test.cc",
     "message_event_test.cc",
+    "message_port_test.cc",
     "url_test.cc",
     "url_utils_test.cc",
     "window_timers_test.cc",

cobalt/web/event_target.cc

@@ -282,9 +282,22 @@ void EventTarget::TraceMembers(script::Tracer* tracer) {
   // instead.
 }
 
+void EventTarget::AddEventListenerRegistrationCallback(
+    void* object, base::Token token, base::OnceClosure callback) {
+  base::AutoLock lock(event_listener_registration_mutex_);
+  event_listener_registration_callbacks_[token][object] = std::move(callback);
+}
+
+void EventTarget::RemoveEventListenerRegistrationCallbacks(void* object) {
+  base::AutoLock lock(event_listener_registration_mutex_);
+  for (auto& token : event_listener_registration_callbacks_)
+    token.second.erase(object);
+}
+
 void EventTarget::AddEventListenerInternal(
     std::unique_ptr<EventTargetListenerInfo> listener_info) {
   DCHECK_CALLED_ON_VALID_THREAD(thread_checker_);
+  DCHECK(listener_info);
 
   // Remove existing attribute listener of the same type.
   if (listener_info->is_attribute()) {
@@ -314,7 +327,21 @@ void EventTarget::AddEventListenerInternal(
   debugger_hooks().AsyncTaskScheduled(
       listener_info->task(), listener_info->type().c_str(),
       base::DebuggerHooks::AsyncTaskFrequency::kRecurring);
+
+  base::Token type = listener_info->type();
   event_listener_infos_.push_back(std::move(listener_info));
+
+  {
+    // Call the event listener registration callback.
+    base::AutoLock lock(event_listener_registration_mutex_);
+    auto callbacks = event_listener_registration_callbacks_.find(type);
+    if (callbacks != event_listener_registration_callbacks_.end()) {
+      for (auto& object : callbacks->second) {
+        std::move(object.second).Run();
+      }
+      event_listener_registration_callbacks_.erase(type);
+    }
+  }
 }
 
 bool EventTarget::HasEventListener(base::Token type) {

cobalt/web/event_target.h

@@ -15,6 +15,7 @@
 #ifndef COBALT_WEB_EVENT_TARGET_H_
 #define COBALT_WEB_EVENT_TARGET_H_
 
+#include <map>
 #include <memory>
 #include <set>
 #include <string>
@@ -536,7 +537,6 @@ class EventTarget : public script::Wrappable,
   web::EnvironmentSettings* environment_settings() const {
     return environment_settings_;
   }
-
   std::set<base::Token>& event_listener_event_types() const {
     static std::set<base::Token> event_listener_event_types;
     for (auto& event_listener_info : event_listener_infos_) {
@@ -545,6 +545,12 @@ class EventTarget : public script::Wrappable,
     return event_listener_event_types;
   }
 
+  // Register a callback to be called when an event listener is added for the
+  // given type.
+  void AddEventListenerRegistrationCallback(void* object, base::Token type,
+                                            base::OnceClosure callback);
+  void RemoveEventListenerRegistrationCallbacks(void* object);
+
  protected:
   virtual ~EventTarget() { environment_settings_ = nullptr; }
 
@@ -569,8 +575,12 @@ class EventTarget : public script::Wrappable,
   // the special case of window.onerror handling.
   bool unpack_onerror_events_;
 
-  // Thread checker ensures all calls to the EventTarget are made from the same
-  // thread that it is created in.
+  base::Lock event_listener_registration_mutex_;
+  std::map<base::Token, std::map<void*, base::OnceClosure>>
+      event_listener_registration_callbacks_;
+
+  // Thread checker ensures all calls to the EventTarget are made from the
+  // same thread that it is created in.
   THREAD_CHECKER(thread_checker_);
 };
 

cobalt/web/message_port.cc

@@ -23,7 +23,6 @@
 #include "base/logging.h"
 #include "base/memory/ptr_util.h"
 #include "base/memory/ref_counted.h"
-#include "base/message_loop/message_loop.h"
 #include "base/task_runner.h"
 #include "cobalt/script/environment_settings.h"
 #include "cobalt/web/context.h"
@@ -34,64 +33,105 @@
 namespace cobalt {
 namespace web {
 
-MessagePort::MessagePort(web::EventTarget* event_target)
-    : event_target_(event_target) {
-  if (!event_target_) {
-    return;
+void MessagePort::EntangleWithEventTarget(web::EventTarget* event_target) {
+  DCHECK(!event_target_);
+  {
+    base::AutoLock lock(mutex_);
+    event_target_ = event_target;
+    if (!event_target_) {
+      enabled_ = false;
+      return;
+    }
   }
-  Context* context = event_target_->environment_settings()->context();
-  base::MessageLoop* message_loop = context->message_loop();
-  if (!message_loop) {
-    return;
-  }
-  message_loop->task_runner()->PostTask(
+  target_task_runner()->PostTask(
       FROM_HERE,
       base::BindOnce(&Context::AddEnvironmentSettingsChangeObserver,
-                     base::Unretained(context), base::Unretained(this)));
+                     base::Unretained(context()), base::Unretained(this)));
   remove_environment_settings_change_observer_ =
       base::BindOnce(&Context::RemoveEnvironmentSettingsChangeObserver,
-                     base::Unretained(context), base::Unretained(this));
+                     base::Unretained(context()), base::Unretained(this));
+
+  target_task_runner()->PostTask(
+      FROM_HERE,
+      base::Bind(
+          [](MessagePort* message_port,
+             web::EventTarget*
+                 event_target) {  // The first time a MessagePort object's
+                                  // onmessage IDL attribute is set, the
+            // port's port message queue must be enabled, as if the start()
+            // method had been called.
+            //   https://html.spec.whatwg.org/commit-snapshots/465a6b672c703054de278b0f8133eb3ad33d93f4/#messageport
+            if (event_target->HasEventListener(base::Tokens::message())) {
+              message_port->Start();
+            } else {
+              event_target->AddEventListenerRegistrationCallback(
+                  message_port, base::Tokens::message(),
+                  base::BindOnce(&MessagePort::Start,
+                                 base::Unretained(message_port)));
+            }
+          },
+          base::Unretained(this), base::Unretained(event_target)));
 }
 
-MessagePort::~MessagePort() { Close(); }
+void MessagePort::Start() {
+  // The start() method steps are to enable this's port message queue, if it is
+  // not already enabled.
+  //   https://html.spec.whatwg.org/commit-snapshots/465a6b672c703054de278b0f8133eb3ad33d93f4/#dom-messageport-start
+  base::AutoLock lock(mutex_);
+  if (!event_target_) {
+    return;
+  }
+  enabled_ = true;
+  for (auto& message : unshipped_messages_) {
+    PostMessageSerializedLocked(std::move(message));
+  }
+  unshipped_messages_.clear();
+}
 
 void MessagePort::Close() {
+  base::AutoLock lock(mutex_);
+  unshipped_messages_.clear();
   if (!event_target_) {
     return;
   }
+  event_target_->RemoveEventListenerRegistrationCallbacks(this);
   if (remove_environment_settings_change_observer_) {
     std::move(remove_environment_settings_change_observer_).Run();
   }
   event_target_ = nullptr;
 }
 
 void MessagePort::PostMessage(const script::ValueHandleHolder& message) {
-  PostMessageSerialized(std::make_unique<script::StructuredClone>(message));
+  auto structured_clone = std::make_unique<script::StructuredClone>(message);
+  {
+    base::AutoLock lock(mutex_);
+    if (!(event_target_ && enabled_)) {
+      unshipped_messages_.push_back(std::move(structured_clone));
+      return;
+    }
+    PostMessageSerializedLocked(std::move(structured_clone));
+  }
 }
 
-void MessagePort::PostMessageSerialized(
+void MessagePort::PostMessageSerializedLocked(
     std::unique_ptr<script::StructuredClone> structured_clone) {
-  if (!event_target_ || !structured_clone) {
-    return;
-  }
-  // TODO: Forward the location of the origating API call to the PostTask call.
-  base::MessageLoop* message_loop =
-      event_target_->environment_settings()->context()->message_loop();
-  if (!message_loop) {
+  if (!structured_clone || !event_target_ || !enabled_) {
     return;
   }
+  // TODO: Forward the location of the origating API call to the PostTask
+  // call.
   //   https://html.spec.whatwg.org/multipage/workers.html#handler-worker-onmessage
   // TODO: Update MessageEvent to support more types. (b/227665847)
   // TODO: Remove dependency of MessageEvent on net iobuffer (b/227665847)
-  message_loop->task_runner()->PostTask(
-      FROM_HERE, base::BindOnce(&MessagePort::DispatchMessage, AsWeakPtr(),
-                                std::move(structured_clone)));
-}
-
-void MessagePort::DispatchMessage(
-    std::unique_ptr<script::StructuredClone> structured_clone) {
-  event_target_->DispatchEvent(new web::MessageEvent(
-      base::Tokens::message(), std::move(structured_clone)));
+  target_task_runner()->PostTask(
+      FROM_HERE,
+      base::BindOnce(
+          [](base::WeakPtr<web::EventTarget> event_target,
+             std::unique_ptr<script::StructuredClone> structured_clone) {
+            event_target->DispatchEvent(new web::MessageEvent(
+                base::Tokens::message(), std::move(structured_clone)));
+          },
+          event_target_->AsWeakPtr(), std::move(structured_clone)));
 }
 
 }  // namespace web

cobalt/web/message_port.h

@@ -17,6 +17,8 @@
 
 #include <memory>
 #include <string>
+#include <utility>
+#include <vector>
 
 #include "base/callback_forward.h"
 #include "base/memory/weak_ptr.h"
@@ -34,14 +36,16 @@ namespace cobalt {
 namespace web {
 
 class MessagePort : public script::Wrappable,
-                    public base::SupportsWeakPtr<MessagePort>,
                     public Context::EnvironmentSettingsChangeObserver {
  public:
-  explicit MessagePort(web::EventTarget* event_target);
-  ~MessagePort();
+  MessagePort() = default;
+  ~MessagePort() { Close(); }
+
   MessagePort(const MessagePort&) = delete;
   MessagePort& operator=(const MessagePort&) = delete;
 
+  void EntangleWithEventTarget(web::EventTarget* event_target);
+
   void OnEnvironmentSettingsChanged(bool context_valid) override {
     if (!context_valid) {
       Close();
@@ -53,10 +57,8 @@ class MessagePort : public script::Wrappable,
   // -> void PostMessage(const script::ValueHandleHolder& message,
   //                     script::Sequence<script::ValueHandle*> transfer) {}
   void PostMessage(const script::ValueHandleHolder& message);
-  void PostMessageSerialized(
-      std::unique_ptr<script::StructuredClone> structured_clone);
 
-  void Start() {}
+  void Start();
   void Close();
 
   const web::EventTargetListenerInfo::EventListenerScriptValue* onmessage()
@@ -91,15 +93,32 @@ class MessagePort : public script::Wrappable,
                                              event_listener);
   }
 
-  web::EventTarget* event_target() { return event_target_; }
+  EventTarget* event_target() const { return event_target_; }
+  Context* context() const {
+    return event_target_ ? event_target_->environment_settings()->context()
+                         : nullptr;
+  }
+  base::TaskRunner* target_task_runner() const {
+    return event_target_ ? event_target_->environment_settings()
+                               ->context()
+                               ->message_loop()
+                               ->task_runner()
+                         : nullptr;
+  }
 
   DEFINE_WRAPPABLE_TYPE(MessagePort);
 
  private:
-  void DispatchMessage(
+  void PostMessageSerializedLocked(
       std::unique_ptr<script::StructuredClone> structured_clone);
 
-  // The event target to dispatch events to.
+  base::Lock mutex_;
+  std::vector<std::unique_ptr<script::StructuredClone>> unshipped_messages_;
+
+  // A port message queue can be enabled or disabled, and is initially disabled.
+  //   https://html.spec.whatwg.org/commit-snapshots/465a6b672c703054de278b0f8133eb3ad33d93f4/#message-ports
+  bool enabled_ = false;
+
   web::EventTarget* event_target_ = nullptr;
   base::OnceClosure remove_environment_settings_change_observer_;
 };

cobalt/web/message_port_test.cc

@@ -0,0 +1,45 @@
+// Copyright 2023 The Cobalt Authors. All Rights Reserved.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+#include "cobalt/web/message_port.h"
+
+#include "base/strings/string_util.h"
+#include "cobalt/web/testing/test_with_javascript.h"
+
+
+#define EXPECT_SUBSTRING(needle, haystack) \
+  EXPECT_PRED_FORMAT2(::testing::IsSubstring, (needle), (haystack))
+
+namespace cobalt {
+namespace web {
+
+namespace {
+class MessagePortTestWithJavaScript : public testing::TestWebWithJavaScript {};
+}  // namespace
+
+TEST_P(MessagePortTestWithJavaScript, MessagePortIsNotConstructible) {
+  std::string result;
+  EXPECT_FALSE(EvaluateScript("var event = new MessagePort();", &result))
+      << "Failed to evaluate script.";
+  EXPECT_SUBSTRING("TypeError: MessagePort is not constructible", result)
+      << result;
+}
+
+INSTANTIATE_TEST_CASE_P(
+    MessagePortTestsWithJavaScript, MessagePortTestWithJavaScript,
+    ::testing::ValuesIn(testing::TestWebWithJavaScript::GetWebTypes()),
+    testing::TestWebWithJavaScript::GetTypeName);
+
+}  // namespace web
+}  // namespace cobalt

cobalt/worker/client.cc

@@ -22,12 +22,12 @@
 
 namespace cobalt {
 namespace worker {
-Client::Client(web::EnvironmentSettings* client)
-    : MessagePort(client->context()
-                      ->GetWindowOrWorkerGlobalScope()
-                      ->navigator_base()
-                      ->service_worker()) {
+Client::Client(web::EnvironmentSettings* client) {
   DCHECK(client);
+  EntangleWithEventTarget(client->context()
+                              ->GetWindowOrWorkerGlobalScope()
+                              ->navigator_base()
+                              ->service_worker());
   // Algorithm for Create Client:
   //   https://www.w3.org/TR/2022/CRD-service-workers-20220712/#create-client
   // 1. Let clientObject be a new Client object.

cobalt/worker/dedicated_worker.cc

@@ -85,7 +85,8 @@ void DedicatedWorker::Initialize(script::ExceptionState* exception_state) {
   // 6. Let worker be a new Worker object.
   // 7. Let outside port be a new MessagePort in outside settings's Realm.
   // 8. Associate the outside port with worker.
-  outside_port_ = new web::MessagePort(this);
+  outside_port_ = new web::MessagePort();
+  outside_port_->EntangleWithEventTarget(this);
   // 9. Run this step in parallel:
   //    1. Run a worker given worker, worker URL, outside settings, outside
   //    port, and options.