1 files changed, 173 insertions, 0 deletions
diff --git a/ipc/glue/NodeController.h b/ipc/glue/NodeController.h
new file mode 100644
index 0000000000..4b0d305b72
--- /dev/null
+++ b/ipc/glue/NodeController.h
@@ -0,0 +1,173 @@
+/* -*- Mode: C++; tab-width: 8; indent-tabs-mode: nil; c-basic-offset: 2 -*- */
+/* vim: set ts=8 sts=2 et sw=2 tw=80: */
+/* 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/. */
+
+#ifndef mozilla_ipc_NodeController_h
+#define mozilla_ipc_NodeController_h
+
+#include "mojo/core/ports/event.h"
+#include "mojo/core/ports/name.h"
+#include "mojo/core/ports/node.h"
+#include "mojo/core/ports/node_delegate.h"
+#include "chrome/common/ipc_message.h"
+#include "mozilla/ipc/ProtocolUtils.h"
+#include "nsTHashMap.h"
+#include "mozilla/Queue.h"
+#include "mozilla/DataMutex.h"
+#include "mozilla/UniquePtr.h"
+#include "mozilla/ipc/NodeChannel.h"
+
+namespace mozilla::ipc {
+
+class NodeController final : public mojo::core::ports::NodeDelegate,
+                             public NodeChannel::Listener {
+  using NodeName = mojo::core::ports::NodeName;
+  using PortName = mojo::core::ports::PortName;
+  using PortRef = mojo::core::ports::PortRef;
+  using Event = mojo::core::ports::Event;
+  using Node = mojo::core::ports::Node;
+  using UserData = mojo::core::ports::UserData;
+  using PortStatus = mojo::core::ports::PortStatus;
+  using UserMessageEvent = mojo::core::ports::UserMessageEvent;
+  using UserMessage = mojo::core::ports::UserMessage;
+
+ public:
+  NS_INLINE_DECL_THREADSAFE_REFCOUNTING(NodeController, override)
+
+  // Return the global singleton instance. The returned value is only valid
+  // while the IO thread is alive.
+  static NodeController* GetSingleton();
+
+  class PortObserver : public UserData {
+   public:
+    virtual void OnPortStatusChanged() = 0;
+
+   protected:
+    ~PortObserver() override = default;
+  };
+
+  // NOTE: For now there will always be a single broker process, and all
+  // processes in the graph need to be able to talk to it (the parent process).
+  // Give it a fixed node name for now to simplify things.
+  //
+  // If we ever decide to have multiple node networks intercommunicating (e.g.
+  // multiple instances or background services), we may need to change this.
+  static constexpr NodeName kBrokerNodeName{0x1, 0x1};
+
+  bool IsBroker() const { return mName == kBrokerNodeName; }
+
+  // Mint a new connected pair of ports within the current process.
+  std::pair<ScopedPort, ScopedPort> CreatePortPair();
+
+  // Get a reference to the port with the given name. Returns an invalid
+  // `PortRef` if the name wasn't found.
+  PortRef GetPort(const PortName& aName);
+
+  // Set the observer for the given port. This observer will be notified when
+  // the status of the port changes.
+  void SetPortObserver(const PortRef& aPort, PortObserver* aObserver);
+
+  // See `mojo::core::ports::Node::GetStatus`
+  Maybe<PortStatus> GetStatus(const PortRef& aPort);
+
+  // See `mojo::core::ports::Node::ClosePort`
+  void ClosePort(const PortRef& aPort);
+
+  // Send a message to the the port's connected peer.
+  bool SendUserMessage(const PortRef& aPort, UniquePtr<IPC::Message> aMessage);
+
+  // Get the next message from the port's message queue.
+  // Will set `*aMessage` to the found message, or `nullptr`.
+  // Returns `false` and sets `*aMessage` to `nullptr` if no further messages
+  // will be delivered to this port as its peer has been closed.
+  bool GetMessage(const PortRef& aPort, UniquePtr<IPC::Message>* aMessage);
+
+  // Called in the broker process from GeckoChildProcessHost to introduce a new
+  // child process into the network. Returns a `PortRef` which can be used to
+  // communicate with the `PortRef` returned from `InitChildProcess`, and a
+  // reference to the `NodeChannel` created for the new process. The port can
+  // immediately have messages sent to it.
+  std::tuple<ScopedPort, RefPtr<NodeChannel>> InviteChildProcess(
+      UniquePtr<IPC::Channel> aChannel);
+
+  // Called as the IO thread is started in the parent process.
+  static void InitBrokerProcess();
+
+  // Called as the IO thread is started in a child process.
+  static ScopedPort InitChildProcess(UniquePtr<IPC::Channel> aChannel,
+                                     int32_t aParentPid = -1);
+
+  // Called when the IO thread is torn down.
+  static void CleanUp();
+
+ private:
+  explicit NodeController(const NodeName& aName);
+  ~NodeController();
+
+  UniquePtr<IPC::Message> SerializeEventMessage(
+      UniquePtr<Event> aEvent, const NodeName* aRelayTarget = nullptr,
+      uint32_t aType = EVENT_MESSAGE_TYPE);
+  UniquePtr<Event> DeserializeEventMessage(UniquePtr<IPC::Message> aMessage,
+                                           NodeName* aRelayTarget = nullptr);
+
+  // Get the `NodeChannel` for the named node.
+  already_AddRefed<NodeChannel> GetNodeChannel(const NodeName& aName);
+
+  // Stop communicating with this peer. Must be called on the IO thread.
+  void DropPeer(NodeName aNodeName);
+
+  // Message Handlers
+  void OnEventMessage(const NodeName& aFromNode,
+                      UniquePtr<IPC::Message> aMessage) override;
+  void OnBroadcast(const NodeName& aFromNode,
+                   UniquePtr<IPC::Message> aMessage) override;
+  void OnIntroduce(const NodeName& aFromNode,
+                   NodeChannel::Introduction aIntroduction) override;
+  void OnRequestIntroduction(const NodeName& aFromNode,
+                             const NodeName& aName) override;
+  void OnAcceptInvite(const NodeName& aFromNode, const NodeName& aRealName,
+                      const PortName& aInitialPort) override;
+  void OnChannelError(const NodeName& aFromNode) override;
+
+  // NodeDelegate Implementation
+  void ForwardEvent(const NodeName& aNode, UniquePtr<Event> aEvent) override;
+  void BroadcastEvent(UniquePtr<Event> aEvent) override;
+  void PortStatusChanged(const PortRef& aPortRef) override;
+
+  const NodeName mName;
+  const UniquePtr<Node> mNode;
+
+  template <class T>
+  using NodeMap = nsTHashMap<NodeNameHashKey, T>;
+
+  struct Invite {
+    // The channel which is being invited. This will have a temporary name until
+    // the invite is completed.
+    RefPtr<NodeChannel> mChannel;
+    // The port which will be merged with the port information from the new
+    // child process when recieved.
+    PortRef mToMerge;
+  };
+
+  struct State {
+    // Channels for connecting to all known peers.
+    NodeMap<RefPtr<NodeChannel>> mPeers;
+
+    // Messages which are queued for peers which we been introduced to yet.
+    NodeMap<Queue<UniquePtr<IPC::Message>, 64>> mPendingMessages;
+
+    // Connections for peers being invited to the network.
+    NodeMap<Invite> mInvites;
+
+    // Ports which are waiting to be merged by a particular peer node.
+    NodeMap<nsTArray<PortRef>> mPendingMerges;
+  };
+
+  DataMutex<State> mState{"NodeController::mState"};
+};
+
+}  // namespace mozilla::ipc
+
+#endif