1 files changed, 195 insertions, 0 deletions
diff --git a/ipc/glue/MessageLink.cpp b/ipc/glue/MessageLink.cpp
new file mode 100644
index 0000000000..872ff48b27
--- /dev/null
+++ b/ipc/glue/MessageLink.cpp
@@ -0,0 +1,195 @@
+/* -*- Mode: C++; tab-width: 4; indent-tabs-mode: nil; c-basic-offset: 2 -*-
+ * vim: sw=2 ts=4 et :
+ */
+/* This Source Code Form is subject to the terms of the Mozilla Public
+ * License, v. 2.0. If a copy of the MPL was not distributed with this
+ * file, You can obtain one at http://mozilla.org/MPL/2.0/. */
+
+#include "mozilla/ipc/MessageLink.h"
+#include "mojo/core/ports/event.h"
+#include "mojo/core/ports/node.h"
+#include "mozilla/ipc/MessageChannel.h"
+#include "mozilla/ipc/BrowserProcessSubThread.h"
+#include "mozilla/ipc/ProtocolUtils.h"
+#include "mozilla/ipc/NodeController.h"
+#include "chrome/common/ipc_channel.h"
+#include "base/task.h"
+
+#include "mozilla/Assertions.h"
+#include "mozilla/DebugOnly.h"
+#include "nsDebug.h"
+#include "nsExceptionHandler.h"
+#include "nsISupportsImpl.h"
+#include "nsPrintfCString.h"
+#include "nsXULAppAPI.h"
+
+using namespace mozilla;
+
+namespace mozilla {
+namespace ipc {
+
+MessageLink::MessageLink(MessageChannel* aChan) : mChan(aChan) {}
+
+MessageLink::~MessageLink() {
+#ifdef DEBUG
+  mChan = nullptr;
+#endif
+}
+
+class PortLink::PortObserverThunk : public NodeController::PortObserver {
+ public:
+  PortObserverThunk(RefCountedMonitor* aMonitor, PortLink* aLink)
+      : mMonitor(aMonitor), mLink(aLink) {}
+
+  void OnPortStatusChanged() override {
+    MonitorAutoLock lock(*mMonitor);
+    if (mLink) {
+      mLink->OnPortStatusChanged();
+    }
+  }
+
+ private:
+  friend class PortLink;
+
+  // The monitor from our PortLink's MessageChannel. Guards access to `mLink`.
+  RefPtr<RefCountedMonitor> mMonitor;
+
+  // Cleared by `PortLink` in `PortLink::Clear()`.
+  PortLink* MOZ_NON_OWNING_REF mLink;
+};
+
+PortLink::PortLink(MessageChannel* aChan, ScopedPort aPort)
+    : MessageLink(aChan), mNode(aPort.Controller()), mPort(aPort.Release()) {
+  mChan->mMonitor->AssertCurrentThreadOwns();
+
+  mObserver = new PortObserverThunk(mChan->mMonitor, this);
+  mNode->SetPortObserver(mPort, mObserver);
+
+  // Dispatch an event to the IO loop to trigger an initial
+  // `OnPortStatusChanged` to deliver any pending messages. This needs to be run
+  // asynchronously from a different thread (or in the case of a same-thread
+  // channel, from the current thread), for now due to assertions in
+  // `MessageChannel`.
+  nsCOMPtr<nsIRunnable> openRunnable = NewRunnableMethod(
+      "PortLink::Open", mObserver, &PortObserverThunk::OnPortStatusChanged);
+  if (aChan->mIsSameThreadChannel) {
+    aChan->mWorkerThread->Dispatch(openRunnable.forget());
+  } else {
+    XRE_GetIOMessageLoop()->PostTask(openRunnable.forget());
+  }
+}
+
+PortLink::~PortLink() {
+  MOZ_RELEASE_ASSERT(!mObserver, "PortLink destroyed without being closed!");
+}
+
+void PortLink::SendMessage(UniquePtr<Message> aMessage) {
+  mChan->mMonitor->AssertCurrentThreadOwns();
+
+  if (aMessage->size() > IPC::Channel::kMaximumMessageSize) {
+    CrashReporter::AnnotateCrashReport(
+        CrashReporter::Annotation::IPCMessageName,
+        nsDependentCString(aMessage->name()));
+    CrashReporter::AnnotateCrashReport(
+        CrashReporter::Annotation::IPCMessageSize,
+        static_cast<unsigned int>(aMessage->size()));
+    MOZ_CRASH("IPC message size is too large");
+  }
+  aMessage->AssertAsLargeAsHeader();
+
+  RefPtr<PortObserverThunk> observer = mObserver;
+  if (!observer) {
+    NS_WARNING("Ignoring message to closed PortLink");
+    return;
+  }
+
+  // Make local copies of relevant member variables, so we can unlock the
+  // monitor for the rest of this function. This protects us in case `this` is
+  // deleted during the call (although that shouldn't happen in practice).
+  //
+  // We don't want the monitor to be held when calling into ports, as we may be
+  // re-entrantly called by our `PortObserverThunk` which will attempt to
+  // acquire the monitor.
+  RefPtr<RefCountedMonitor> monitor = mChan->mMonitor;
+  RefPtr<NodeController> node = mNode;
+  PortRef port = mPort;
+
+  bool ok = false;
+  monitor->AssertCurrentThreadOwns();
+  {
+    MonitorAutoUnlock guard(*monitor);
+    ok = node->SendUserMessage(port, std::move(aMessage));
+  }
+  if (!ok) {
+    // The send failed, but double-check that we weren't closed racily while
+    // sending, which could lead to an invalid state error.
+    if (observer->mLink) {
+      MOZ_CRASH("Invalid argument to SendUserMessage");
+    }
+    NS_WARNING("Message dropped as PortLink was closed");
+  }
+}
+
+void PortLink::Close() {
+  mChan->mMonitor->AssertCurrentThreadOwns();
+
+  if (!mObserver) {
+    // We're already being closed.
+    return;
+  }
+
+  Clear();
+}
+
+void PortLink::Clear() {
+  mChan->mMonitor->AssertCurrentThreadOwns();
+
+  // NOTE: We're calling into `ports` with our monitor held! Usually, this could
+  // lead to deadlocks due to the PortObserverThunk acquiring the lock
+  // re-entrantly, but is OK here as we're immediately clearing the port's
+  // observer. We shouldn't have issues with any re-entrant calls on this thread
+  // acquiring this MessageChannel's monitor.
+  //
+  // We also clear out the reference in `mObserver` back to this type so that
+  // notifications from other threads won't try to call us again once we release
+  // the monitor.
+  mNode->SetPortObserver(mPort, nullptr);
+  mObserver->mLink = nullptr;
+  mObserver = nullptr;
+  mNode->ClosePort(mPort);
+}
+
+void PortLink::OnPortStatusChanged() {
+  mChan->mMonitor->AssertCurrentThreadOwns();
+
+  // Check if the port's remoteness status has updated, and tell our channel if
+  // it has.
+  if (Maybe<PortStatus> status = mNode->GetStatus(mPort);
+      status && status->peer_remote != mChan->IsCrossProcess()) {
+    mChan->SetIsCrossProcess(status->peer_remote);
+  }
+
+  while (mObserver) {
+    UniquePtr<IPC::Message> message;
+    if (!mNode->GetMessage(mPort, &message)) {
+      Clear();
+      mChan->OnChannelErrorFromLink();
+      return;
+    }
+    if (!message) {
+      return;
+    }
+
+    mChan->OnMessageReceivedFromLink(std::move(message));
+  }
+}
+
+bool PortLink::IsClosed() const {
+  if (Maybe<PortStatus> status = mNode->GetStatus(mPort)) {
+    return !(status->has_messages || status->receiving_messages);
+  }
+  return true;
+}
+
+}  // namespace ipc
+}  // namespace mozilla