Add EventEmitter C++ bridging type (#44808)

Summary:
Pull Request resolved: #44808

Adds an `AsyncEventEmitter` class which can be used as a property of currently C++ only Turbo Modules to send type safe data back to JavaScript.

Adding support for ObjC / Java Turbo Modules is possible, straight forward and can be added as an afterthought.

It implements this interface
```
export type EventEmitter<T> = {
  addListener(handler: (T) => mixed): EventSubscription,
};
```

## Hybrid
It is a 'hybrid' object.

1.) You `addListener(handler: (T) => mixed)` in JavaScript for emitted events (coming from C++, native code)
2.) You `emit(...Arg)` events in C++, native code (getting sent to JavaScript)

## Changelog:

[General] [Added] - Add EventEmitter C++ bridging type

## Facebook:
Apps usually create custom functionality to achieve this kind of behavior - e.g. https://www.internalfb.com/code/fbsource/[e72bd42a028a]/arvr/js/apps/RemoteDesktopCompanion/shared/turbo_modules/TMSubscription.h

Reviewed By: javache

Differential Revision: D57424391

fbshipit-source-id: 4999cafe9daeac125712a4bb7679d7acb9a6c389

packages/react-native/Libraries/Types/CodegenTypes.js

@@ -10,6 +10,7 @@
 
 'use strict';
 
+import type {EventSubscription} from '../vendor/emitter/EventEmitter';
 import type {SyntheticEvent} from './CoreEventTypes';
 
 // Event types
@@ -40,3 +41,7 @@ type DefaultTypes = number | boolean | string | $ReadOnlyArray<string>;
 //
 // eslint-disable-next-line no-unused-vars
 export type WithDefault<Type: DefaultTypes, Value: ?Type | string> = ?Type;
+
+export type EventEmitter<T> = {
+  addListener(handler: (T) => mixed): EventSubscription,
+};

packages/react-native/Libraries/__tests__/__snapshots__/public-api-test.js.snap

@@ -8133,6 +8133,9 @@ export type UnsafeObject = $FlowFixMe;
 export type UnsafeMixed = mixed;
 type DefaultTypes = number | boolean | string | $ReadOnlyArray<string>;
 export type WithDefault<Type: DefaultTypes, Value: ?Type | string> = ?Type;
+export type EventEmitter<T> = {
+  addListener(handler: (T) => mixed): EventSubscription,
+};
 "
 `;
 

packages/react-native/ReactCommon/react/bridging/Bridging.h

@@ -13,6 +13,7 @@
 #include <react/bridging/Class.h>
 #include <react/bridging/Dynamic.h>
 #include <react/bridging/Error.h>
+#include <react/bridging/EventEmitter.h>
 #include <react/bridging/Function.h>
 #include <react/bridging/Number.h>
 #include <react/bridging/Object.h>

packages/react-native/ReactCommon/react/bridging/EventEmitter.h

@@ -0,0 +1,128 @@
+/*
+ * Copyright (c) Meta Platforms, Inc. and affiliates.
+ *
+ * This source code is licensed under the MIT license found in the
+ * LICENSE file in the root directory of this source tree.
+ */
+
+#pragma once
+
+#include <react/bridging/Function.h>
+#include <functional>
+#include <memory>
+#include <mutex>
+#include <unordered_map>
+
+#define FRIEND_TEST(test_case_name, test_name) \
+  friend class test_case_name##_##test_name##_Test
+
+namespace facebook::react {
+
+class EventSubscription {
+ public:
+  explicit EventSubscription(std::function<void()> remove)
+      : remove_(std::move(remove)) {}
+  ~EventSubscription() = default;
+  EventSubscription(EventSubscription&&) noexcept = default;
+  EventSubscription& operator=(EventSubscription&&) noexcept = default;
+  EventSubscription(const EventSubscription&) = delete;
+  EventSubscription& operator=(const EventSubscription&) = delete;
+
+ private:
+  friend Bridging<EventSubscription>;
+
+  std::function<void()> remove_;
+};
+
+template <>
+struct Bridging<EventSubscription> {
+  static jsi::Object toJs(
+      jsi::Runtime& rt,
+      const EventSubscription& eventSubscription,
+      const std::shared_ptr<CallInvoker>& jsInvoker) {
+    auto result = jsi::Object(rt);
+    result.setProperty(
+        rt, "remove", bridging::toJs(rt, eventSubscription.remove_, jsInvoker));
+    return result;
+  }
+};
+
+class IAsyncEventEmitter {
+ public:
+  IAsyncEventEmitter() noexcept = default;
+  virtual ~IAsyncEventEmitter() noexcept = default;
+  IAsyncEventEmitter(IAsyncEventEmitter&&) noexcept = default;
+  IAsyncEventEmitter& operator=(IAsyncEventEmitter&&) noexcept = default;
+  IAsyncEventEmitter(const IAsyncEventEmitter&) = delete;
+  IAsyncEventEmitter& operator=(const IAsyncEventEmitter&) = delete;
+
+  virtual jsi::Object get(
+      jsi::Runtime& rt,
+      const std::shared_ptr<CallInvoker>& jsInvoker) const = 0;
+};
+
+template <typename... Args>
+class AsyncEventEmitter : public IAsyncEventEmitter {
+  static_assert(
+      sizeof...(Args) <= 1,
+      "AsyncEventEmitter must have at most one argument");
+
+ public:
+  AsyncEventEmitter() : state_(std::make_shared<SharedState>()) {
+    listen_ = [state = state_](AsyncCallback<Args...> listener) {
+      std::lock_guard<std::mutex> lock(state->mutex);
+      auto listenerId = state->listenerId++;
+      state->listeners.emplace(listenerId, std::move(listener));
+      return EventSubscription([state, listenerId]() {
+        std::lock_guard<std::mutex> innerLock(state->mutex);
+        state->listeners.erase(listenerId);
+      });
+    };
+  }
+  ~AsyncEventEmitter() override = default;
+  AsyncEventEmitter(AsyncEventEmitter&&) noexcept = default;
+  AsyncEventEmitter& operator=(AsyncEventEmitter&&) noexcept = default;
+  AsyncEventEmitter(const AsyncEventEmitter&) = delete;
+  AsyncEventEmitter& operator=(const AsyncEventEmitter&) = delete;
+
+  void emit(Args... value) {
+    std::lock_guard<std::mutex> lock(state_->mutex);
+    for (const auto& [_, listener] : state_->listeners) {
+      listener.call(static_cast<Args>(value)...);
+    }
+  }
+
+  jsi::Object get(
+      jsi::Runtime& rt,
+      const std::shared_ptr<CallInvoker>& jsInvoker) const override {
+    auto result = jsi::Object(rt);
+    result.setProperty(
+        rt, "addListener", bridging::toJs(rt, listen_, jsInvoker));
+    return result;
+  }
+
+ private:
+  friend Bridging<AsyncEventEmitter>;
+  FRIEND_TEST(BridgingTest, eventEmitterTest);
+
+  struct SharedState {
+    std::mutex mutex;
+    std::unordered_map<size_t, AsyncCallback<Args...>> listeners;
+    size_t listenerId{};
+  };
+
+  std::function<EventSubscription(AsyncCallback<Args...>)> listen_;
+  std::shared_ptr<SharedState> state_;
+};
+
+template <typename... Args>
+struct Bridging<AsyncEventEmitter<Args...>> {
+  static jsi::Object toJs(
+      jsi::Runtime& rt,
+      const AsyncEventEmitter<Args...>& eventEmitter,
+      const std::shared_ptr<CallInvoker>& jsInvoker) {
+    return eventEmitter.get(rt, jsInvoker);
+  }
+};
+
+} // namespace facebook::react

packages/react-native/ReactCommon/react/bridging/tests/BridgingTest.cpp

@@ -123,7 +123,7 @@ TEST_F(BridgingTest, hostObjectTest) {
   struct TestHostObject : public jsi::HostObject {
     jsi::Value get(jsi::Runtime& rt, const jsi::PropNameID& name) override {
       if (name.utf8(rt) == "test") {
-        return jsi::Value(1);
+        return {1};
       }
       return jsi::Value::undefined();
     }
@@ -426,6 +426,110 @@ TEST_F(BridgingTest, promiseTest) {
   EXPECT_NO_THROW(promise.reject("ignored"));
 }
 
+using EventType = std::vector<std::string>;
+using EventSubscriptionsWithLastEvent =
+    std::vector<std::pair<jsi::Object, std::shared_ptr<EventType>>>;
+
+namespace {
+
+template <typename EventType>
+void addEventSubscription(
+    jsi::Runtime& rt,
+    const AsyncEventEmitter<EventType>& eventEmitter,
+    EventSubscriptionsWithLastEvent& eventSubscriptionsWithListener,
+    const std::shared_ptr<TestCallInvoker>& invoker) {
+  auto eventEmitterJs = bridging::toJs(rt, eventEmitter, invoker);
+  auto lastEvent = std::make_shared<EventType>();
+  auto listenJs = bridging::toJs(
+      rt,
+      [lastEvent = lastEvent](const EventType& event) { *lastEvent = event; },
+      invoker);
+  eventSubscriptionsWithListener.emplace_back(std::make_pair(
+      jsi::Object(eventEmitterJs.getPropertyAsFunction(rt, "addListener")
+                      .callWithThis(rt, eventEmitterJs, listenJs)
+                      .asObject(rt)),
+      std::move(lastEvent)));
+}
+
+} // namespace
+
+TEST_F(BridgingTest, eventEmitterTest) {
+  EventSubscriptionsWithLastEvent eventSubscriptionsWithListener;
+
+  AsyncEventEmitter<EventType> eventEmitter;
+  EXPECT_NO_THROW(eventEmitter.emit({"one", "two", "three"}));
+  EXPECT_EQ(0, eventSubscriptionsWithListener.size());
+
+  // register 3 JavaScript listeners to the event emitter
+  for (int i = 0; i < 3; ++i) {
+    addEventSubscription<EventType>(
+        rt, eventEmitter, eventSubscriptionsWithListener, invoker);
+  }
+
+  EXPECT_TRUE(eventEmitter.state_->listeners.contains(0));
+  EXPECT_TRUE(eventEmitter.state_->listeners.contains(1));
+  EXPECT_TRUE(eventEmitter.state_->listeners.contains(2));
+
+  EXPECT_NO_THROW(eventEmitter.emit({"four", "five", "six"}));
+  flushQueue();
+
+  // verify all listeners received the event
+  for (const auto& [_, lastEvent] : eventSubscriptionsWithListener) {
+    EXPECT_EQ(3, lastEvent->size());
+    EXPECT_EQ("four", lastEvent->at(0));
+    EXPECT_EQ("five", lastEvent->at(1));
+    EXPECT_EQ("six", lastEvent->at(2));
+  }
+
+  // Remove 2nd eventSubscriptions
+  eventSubscriptionsWithListener[1]
+      .first.getPropertyAsFunction(rt, "remove")
+      .callWithThis(rt, eventSubscriptionsWithListener[1].first);
+  eventSubscriptionsWithListener.erase(
+      eventSubscriptionsWithListener.begin() + 1);
+
+  // Add 4th and 5th eventSubscriptions
+  addEventSubscription<EventType>(
+      rt, eventEmitter, eventSubscriptionsWithListener, invoker);
+  addEventSubscription<EventType>(
+      rt, eventEmitter, eventSubscriptionsWithListener, invoker);
+
+  EXPECT_TRUE(eventEmitter.state_->listeners.contains(0));
+  EXPECT_FALSE(eventEmitter.state_->listeners.contains(1));
+  EXPECT_TRUE(eventEmitter.state_->listeners.contains(2));
+  EXPECT_TRUE(eventEmitter.state_->listeners.contains(3));
+  EXPECT_TRUE(eventEmitter.state_->listeners.contains(4));
+
+  // Emit more events
+  EXPECT_NO_THROW(eventEmitter.emit({"seven", "eight", "nine"}));
+  flushQueue();
+
+  for (const auto& [_, lastEvent] : eventSubscriptionsWithListener) {
+    EXPECT_EQ(3, lastEvent->size());
+    EXPECT_EQ("seven", lastEvent->at(0));
+    EXPECT_EQ("eight", lastEvent->at(1));
+    EXPECT_EQ("nine", lastEvent->at(2));
+  }
+
+  // clean-up the event subscriptions
+  for (const auto& [eventSubscription, _] : eventSubscriptionsWithListener) {
+    eventSubscription.getPropertyAsFunction(rt, "remove")
+        .callWithThis(rt, eventSubscription);
+  }
+  flushQueue();
+
+  EXPECT_NO_THROW(eventEmitter.emit({"ten", "eleven", "twelve"}));
+  flushQueue();
+
+  // no new data as listeners had been removed
+  for (const auto& [_, lastEvent] : eventSubscriptionsWithListener) {
+    EXPECT_EQ(3, lastEvent->size());
+    EXPECT_EQ("seven", lastEvent->at(0));
+    EXPECT_EQ("eight", lastEvent->at(1));
+    EXPECT_EQ("nine", lastEvent->at(2));
+  }
+}
+
 TEST_F(BridgingTest, optionalTest) {
   EXPECT_EQ(
       1, bridging::fromJs<std::optional<int>>(rt, jsi::Value(1), invoker));