add WhenAny and WhenAll helpers

Author: beekld

libs/internal/include/launchdarkly/async/promise.hpp

@@ -1,5 +1,6 @@
 #pragma once
 
+#include <atomic>
 #include <chrono>
 #include <condition_variable>
 #include <functional>
@@ -418,4 +419,95 @@ class Future {
     std::shared_ptr<PromiseInternal<T>> internal_;
 };
 
+// WhenAll takes a variadic list of Futures (each with potentially different
+// value types) and returns a Future<std::monostate> that resolves once all
+// of the input futures have resolved. The result carries no value; callers
+// who need the individual results can read them from their original futures
+// after WhenAll resolves.
+//
+// If called with no arguments, the returned future is already resolved.
+//
+// Example:
+//
+//   Future<int> f1 = ...;
+//   Future<std::string> f2 = ...;
+//   WhenAll(f1, f2).Then(
+//       [&](std::monostate const&) {
+//           // f1 and f2 are both finished here.
+//           use(*f1.GetResult(), *f2.GetResult());
+//           return std::monostate{};
+//       },
+//       executor);
+template <typename... Ts>
+Future<std::monostate> WhenAll(Future<Ts>... futures) {
+    Promise<std::monostate> promise;
+    Future<std::monostate> result = promise.GetFuture();
+
+    if constexpr (sizeof...(Ts) == 0) {
+        promise.Resolve(std::monostate{});
+        return result;
+    }
+
+    auto shared_promise =
+        std::make_shared<Promise<std::monostate>>(std::move(promise));
+    auto count = std::make_shared<std::atomic<std::size_t>>(sizeof...(Ts));
+
+    auto attach = [&](auto future) {
+        future.Then(
+            [shared_promise, count](auto const&) -> std::monostate {
+                if (count->fetch_sub(1) == 1) {
+                    shared_promise->Resolve(std::monostate{});
+                }
+                return std::monostate{};
+            },
+            [](Continuation<void()> f) { f(); });
+    };
+
+    (attach(futures), ...);
+
+    return result;
+}
+
+// WhenAny takes a variadic list of Futures (each with potentially different
+// value types) and returns a Future<std::size_t> that resolves with the
+// 0-based index of whichever input future resolves first. The caller can use
+// the index to identify the winning future and read its result directly.
+//
+// If called with no arguments, the returned future never resolves.
+//
+// Example:
+//
+//   Future<int> f0 = ...;
+//   Future<std::string> f1 = ...;
+//   WhenAny(f0, f1).Then(
+//       [&](std::size_t const& index) {
+//           if (index == 0) use(*f0.GetResult());
+//           else            use(*f1.GetResult());
+//           return std::monostate{};
+//       },
+//       executor);
+template <typename... Ts>
+Future<std::size_t> WhenAny(Future<Ts>... futures) {
+    Promise<std::size_t> promise;
+    Future<std::size_t> result = promise.GetFuture();
+
+    auto shared_promise =
+        std::make_shared<Promise<std::size_t>>(std::move(promise));
+
+    std::size_t index = 0;
+    auto attach = [&](auto future) {
+        std::size_t i = index++;
+        future.Then(
+            [shared_promise, i](auto const&) -> std::monostate {
+                shared_promise->Resolve(i);
+                return std::monostate{};
+            },
+            [](Continuation<void()> f) { f(); });
+    };
+
+    (attach(futures), ...);
+
+    return result;
+}
+
 }  // namespace launchdarkly::async

libs/internal/tests/promise_test.cpp

@@ -9,6 +9,45 @@
 
 using namespace launchdarkly::async;
 
+TEST(Promise, SimplePromise) {
+    Promise<int> promise;
+    Future<int> future = promise.GetFuture();
+
+    Future<float> future2 = future.Then(
+        [](int const& inner) { return static_cast<float>(inner * 2.0); },
+        [](Continuation<void()> f) { f(); });
+
+    promise.Resolve(43);
+
+    auto& result = future2.WaitForResult(std::chrono::seconds(5));
+    ASSERT_TRUE(result.has_value());
+    EXPECT_FLOAT_EQ(*result, 86.0f);
+}
+
+TEST(Promise, ASIOTest) {
+    boost::asio::io_context ioc;
+    auto work = boost::asio::make_work_guard(ioc);
+    std::thread ioc_thread([&]() { ioc.run(); });
+
+    Promise<int> promise;
+    Future<int> future = promise.GetFuture();
+
+    Future<float> future2 = future.Then(
+        [](int const& inner) { return static_cast<float>(inner * 2.0); },
+        [&ioc](Continuation<void()> f) {
+            boost::asio::post(ioc, [f = std::move(f)]() mutable { f(); });
+        });
+
+    promise.Resolve(42);
+
+    auto& result = future2.WaitForResult(std::chrono::seconds(5));
+    ASSERT_TRUE(result.has_value());
+    EXPECT_FLOAT_EQ(*result, 84.0f);
+
+    work.reset();
+    ioc_thread.join();
+}
+
 TEST(Promise, GetResultNotFinished) {
     Promise<int> promise;
     Future<int> future = promise.GetFuture();
@@ -53,9 +92,9 @@ TEST(Promise, ContinueByReturningFuture) {
     Promise<int> promise2;
     Future<int> future2 = promise2.GetFuture();
 
-    Future<int> chained = promise1.GetFuture().Then(
-        [future2](int const&) { return future2; },
-        [](Continuation<void()> f) { f(); });
+    Future<int> chained =
+        promise1.GetFuture().Then([future2](int const&) { return future2; },
+                                  [](Continuation<void()> f) { f(); });
 
     promise1.Resolve(0);
     promise2.Resolve(42);
@@ -71,9 +110,8 @@ TEST(Promise, ResolvedBeforeContinuation) {
 
     promise.Resolve(21);
 
-    Future<int> future2 = future.Then(
-        [](int const& val) { return val * 2; },
-        [](Continuation<void()> f) { f(); });
+    Future<int> future2 = future.Then([](int const& val) { return val * 2; },
+                                      [](Continuation<void()> f) { f(); });
 
     auto& result = future2.WaitForResult(std::chrono::seconds(5));
     ASSERT_TRUE(result.has_value());
@@ -84,9 +122,8 @@ TEST(Promise, ResolvedAfterContinuation) {
     Promise<int> promise;
     Future<int> future = promise.GetFuture();
 
-    Future<int> future2 = future.Then(
-        [](int const& val) { return val * 2; },
-        [](Continuation<void()> f) { f(); });
+    Future<int> future2 = future.Then([](int const& val) { return val * 2; },
+                                      [](Continuation<void()> f) { f(); });
 
     promise.Resolve(21);
 
@@ -140,11 +177,10 @@ TEST(Promise, MoveOnlyCallback) {
     Future<int> future = promise.GetFuture();
 
     auto captured = std::make_unique<int>(2);
-    Future<int> future2 = future.Then(
-        [captured = std::move(captured)](int const& val) {
-            return val * *captured;
-        },
-        [](Continuation<void()> f) { f(); });
+    Future<int> future2 =
+        future.Then([captured = std::move(captured)](
+                        int const& val) { return val * *captured; },
+                    [](Continuation<void()> f) { f(); });
 
     promise.Resolve(21);
 
@@ -161,8 +197,15 @@ TEST(Promise, ResultMovedWhenPossible) {
         int value;
         int* copy_count;
         explicit Counted(int v, int* c) : value(v), copy_count(c) {}
-        Counted(Counted const& o) : value(o.value), copy_count(o.copy_count) { ++(*copy_count); }
-        Counted& operator=(Counted const& o) { value = o.value; copy_count = o.copy_count; ++(*copy_count); return *this; }
+        Counted(Counted const& o) : value(o.value), copy_count(o.copy_count) {
+            ++(*copy_count);
+        }
+        Counted& operator=(Counted const& o) {
+            value = o.value;
+            copy_count = o.copy_count;
+            ++(*copy_count);
+            return *this;
+        }
         Counted(Counted&&) noexcept = default;
         Counted& operator=(Counted&&) noexcept = default;
     };
@@ -186,8 +229,15 @@ TEST(Promise, CallbackMovedWhenPossible) {
         int value;
         int* copy_count;
         explicit Counted(int v, int* c) : value(v), copy_count(c) {}
-        Counted(Counted const& o) : value(o.value), copy_count(o.copy_count) { ++(*copy_count); }
-        Counted& operator=(Counted const& o) { value = o.value; copy_count = o.copy_count; ++(*copy_count); return *this; }
+        Counted(Counted const& o) : value(o.value), copy_count(o.copy_count) {
+            ++(*copy_count);
+        }
+        Counted& operator=(Counted const& o) {
+            value = o.value;
+            copy_count = o.copy_count;
+            ++(*copy_count);
+            return *this;
+        }
         Counted(Counted&&) noexcept = default;
         Counted& operator=(Counted&&) noexcept = default;
     };
@@ -198,19 +248,18 @@ TEST(Promise, CallbackMovedWhenPossible) {
     Counted multiplier{2, &copies};
     copies = 0;  // reset after construction
 
-    future.Then(
-        [multiplier = std::move(multiplier)](int const& val) mutable {
-            return val * multiplier.value;
-        },
-        [](Continuation<void()> f) { f(); });
+    future.Then([multiplier = std::move(multiplier)](
+                    int const& val) mutable { return val * multiplier.value; },
+                [](Continuation<void()> f) { f(); });
 
     EXPECT_EQ(copies, 0);
 
     promise.Resolve(21);
 }
 
-// Demonstrates using std::monostate as a void-like result type for fire-and-forget
-// async operations where the completion matters but no value is produced.
+// Demonstrates using std::monostate as a void-like result type for
+// fire-and-forget async operations where the completion matters but no value is
+// produced.
 TEST(Promise, MonostateVoidLike) {
     Promise<std::monostate> promise;
     Future<std::monostate> future = promise.GetFuture();
@@ -258,42 +307,100 @@ TEST(Promise, ExpectedFailure) {
     EXPECT_EQ(result->error(), "timed out");
 }
 
+TEST(WhenAll, NoFutures) {
+    Future<std::monostate> result = WhenAll();
+    EXPECT_TRUE(result.IsFinished());
+}
 
-TEST(Promise, SimplePromise) {
-    Promise<int> promise;
-    Future<int> future = promise.GetFuture();
+// Verifies WhenAll resolves when all futures are already resolved.
+TEST(WhenAll, AllAlreadyResolved) {
+    Promise<int> p1;
+    Promise<std::string> p2;
 
-    Future<float> future2 = future.Then(
-        [](int const& inner) { return static_cast<float>(inner * 2.0); },
-        [](Continuation<void()> f) { f(); });
+    p1.Resolve(1);
+    p2.Resolve("hello");
 
-    promise.Resolve(43);
+    Future<std::monostate> result = WhenAll(p1.GetFuture(), p2.GetFuture());
+    auto& r = result.WaitForResult(std::chrono::seconds(5));
+    ASSERT_TRUE(r.has_value());
+}
 
-    auto& result = future2.WaitForResult(std::chrono::seconds(5));
-    ASSERT_TRUE(result.has_value());
-    EXPECT_FLOAT_EQ(*result, 86.0f);
+// Verifies WhenAll resolves only after all futures resolve, using futures of
+// mixed value types. The original futures still hold their results afterward.
+TEST(WhenAll, ResolvesAfterAll) {
+    Promise<int> p1;
+    Promise<std::string> p2;
+
+    Future<int> f1 = p1.GetFuture();
+    Future<std::string> f2 = p2.GetFuture();
+
+    Future<std::monostate> result = WhenAll(f1, f2);
+
+    EXPECT_FALSE(result.IsFinished());
+    p1.Resolve(42);
+    EXPECT_FALSE(result.IsFinished());
+    p2.Resolve("done");
+
+    auto& r = result.WaitForResult(std::chrono::seconds(5));
+    ASSERT_TRUE(r.has_value());
+    EXPECT_EQ(*f1.GetResult(), 42);
+    EXPECT_EQ(*f2.GetResult(), "done");
 }
 
-TEST(Promise, ASIOTest) {
-    boost::asio::io_context ioc;
-    auto work = boost::asio::make_work_guard(ioc);
-    std::thread ioc_thread([&]() { ioc.run(); });
+// Verifies that WhenAny resolves with the index of the first future to resolve.
+TEST(WhenAny, FirstResolved) {
+    Promise<int> p0;
+    Promise<int> p1;
+    Promise<int> p2;
 
-    Promise<int> promise;
-    Future<int> future = promise.GetFuture();
+    Future<std::size_t> result =
+        WhenAny(p0.GetFuture(), p1.GetFuture(), p2.GetFuture());
 
-    Future<float> future2 = future.Then(
-        [](int const& inner) { return static_cast<float>(inner * 2.0); },
-        [&ioc](Continuation<void()> f) {
-            boost::asio::post(ioc, [f = std::move(f)]() mutable { f(); });
-        });
+    EXPECT_FALSE(result.IsFinished());
+    p1.Resolve(42);
 
-    promise.Resolve(42);
+    auto& r = result.WaitForResult(std::chrono::seconds(5));
+    ASSERT_TRUE(r.has_value());
+    EXPECT_EQ(*r, 1u);
+}
 
-    auto& result = future2.WaitForResult(std::chrono::seconds(5));
-    ASSERT_TRUE(result.has_value());
-    EXPECT_FLOAT_EQ(*result, 84.0f);
+// Verifies that WhenAny works with futures of mixed value types, and that
+// resolving a later future after the winner does not change the result.
+TEST(WhenAny, MixedTypesFirstWins) {
+    Promise<int> p0;
+    Promise<std::string> p1;
+
+    Future<int> f0 = p0.GetFuture();
+    Future<std::string> f1 = p1.GetFuture();
+
+    Future<std::variant<int, std::string>> result = WhenAny(f0, f1).Then(
+        [f0, f1](size_t const& index) -> std::variant<int, std::string> {
+            if (index == 0) {
+                return f0.GetResult().value();
+            } else {
+                return f1.GetResult().value();
+            }
+        },
+        [](Continuation<void()> f) { f(); });
 
-    work.reset();
-    ioc_thread.join();
+    p1.Resolve("hello");
+    p0.Resolve(99);
+
+    auto& r = result.WaitForResult(std::chrono::seconds(5));
+    ASSERT_TRUE(r.has_value());
+    EXPECT_EQ(std::get<std::string>(*result.GetResult()), "hello");
+}
+
+// Verifies that WhenAny resolves immediately if a future is already resolved.
+TEST(WhenAny, AlreadyResolved) {
+    Promise<int> p0;
+    Promise<int> p1;
+
+    p0.Resolve(42);
+
+    Future<std::size_t> result = WhenAny(p0.GetFuture(), p1.GetFuture());
+
+    auto& r = result.WaitForResult(std::chrono::seconds(5));
+    ASSERT_TRUE(r.has_value());
+    EXPECT_EQ(*r, 0u);
 }