From e202af3ccc608bbafd692428266eedd0de576641 Mon Sep 17 00:00:00 2001 From: Janusz Sobczak Date: Mon, 20 Mar 2023 12:13:44 -0700 Subject: [PATCH] Add value parameter to future listeners Pass the Future result to listeners as a parameter. This makes the AddListener() API a bit more convenient. PiperOrigin-RevId: 518040156 --- internal/platform/future.h | 5 +- internal/platform/future_test.cc | 51 +++++++++---------- .../implementation/listenable_future.h | 4 +- .../windows/listenable_future.h | 3 +- internal/platform/settable_future.h | 32 ++++++++---- 5 files changed, 53 insertions(+), 42 deletions(-) diff --git a/internal/platform/future.h b/internal/platform/future.h index 02e2cd1b..9c7b44c4 100644 --- a/internal/platform/future.h +++ b/internal/platform/future.h @@ -24,6 +24,7 @@ namespace nearby { template class Future final { public: + using FutureCallback = typename SettableFuture::FutureCallback; // Default Future. Does not time out. Future() : impl_(std::make_shared>()) {} @@ -39,8 +40,8 @@ class Future final { virtual ExceptionOr Get(absl::Duration timeout) { return impl_->Get(timeout); } - void AddListener(Runnable runnable, api::Executor* executor) { - impl_->AddListener(std::move(runnable), executor); + void AddListener(FutureCallback callback, api::Executor* executor) { + impl_->AddListener(std::move(callback), executor); } bool IsSet() const { return impl_->IsSet(); } diff --git a/internal/platform/future_test.cc b/internal/platform/future_test.cc index aaacd3e4..821a9dc5 100644 --- a/internal/platform/future_test.cc +++ b/internal/platform/future_test.cc @@ -121,10 +121,9 @@ TEST(FutureTest, CallsListenerOnSet) { { SingleThreadExecutor executor; future.AddListener( - [&]() { - ASSERT_TRUE(future.IsSet()); - ASSERT_TRUE(future.Get().ok()); - ASSERT_EQ(future.Get().GetResult(), kValue); + [&](ExceptionOr result) { + ASSERT_TRUE(result.ok()); + ASSERT_EQ(result.GetResult(), kValue); ++call_count; }, &executor); @@ -144,18 +143,16 @@ TEST(FutureTest, CallsAllListenersOnSet) { { SingleThreadExecutor executor; future.AddListener( - [&]() { - ASSERT_TRUE(future.IsSet()); - ASSERT_TRUE(future.Get().ok()); - ASSERT_EQ(future.Get().GetResult(), kValue); + [&](ExceptionOr result) { + ASSERT_TRUE(result.ok()); + ASSERT_EQ(result.GetResult(), kValue); ++call_count_listener_1; }, &executor); future.AddListener( - [&]() { - ASSERT_TRUE(future.IsSet()); - ASSERT_TRUE(future.Get().ok()); - ASSERT_EQ(future.Get().GetResult(), kValue); + [&](ExceptionOr result) { + ASSERT_TRUE(result.ok()); + ASSERT_EQ(result.GetResult(), kValue); ++call_count_listener_2; }, &executor); @@ -176,10 +173,9 @@ TEST(FutureTest, AddListenerWhenAlreadySetCallsCallback) { { SingleThreadExecutor executor; future.AddListener( - [&]() { - ASSERT_TRUE(future.IsSet()); - ASSERT_TRUE(future.Get().ok()); - ASSERT_EQ(future.Get().GetResult(), kValue); + [&](ExceptionOr result) { + ASSERT_TRUE(result.ok()); + ASSERT_EQ(result.GetResult(), kValue); ++call_count; }, &executor); @@ -196,10 +192,9 @@ TEST(FutureTest, CallsListenerOnSetException) { { SingleThreadExecutor executor; future.AddListener( - [&]() { - ASSERT_TRUE(future.IsSet()); - ASSERT_FALSE(future.Get().ok()); - ASSERT_EQ(future.Get().GetException(), kException); + [&](ExceptionOr result) { + ASSERT_FALSE(result.ok()); + ASSERT_EQ(result.GetException(), kException); ++call_count; }, &executor); @@ -220,10 +215,9 @@ TEST(FutureTest, AddListenerWhenAlreadySetExceptionCallsCallback) { SingleThreadExecutor executor; future.AddListener( - [&]() { - ASSERT_TRUE(future.IsSet()); - ASSERT_FALSE(future.Get().ok()); - ASSERT_EQ(future.Get().GetException(), kException); + [&](ExceptionOr result) { + ASSERT_FALSE(result.ok()); + ASSERT_EQ(result.GetException(), kException); ++call_count; }, &executor); @@ -243,8 +237,13 @@ TEST(FutureTest, TimeoutSetsException) { TEST(FutureTest, TimeoutCallsListeners) { Future future(absl::Milliseconds(10)); CountDownLatch latch(1); - future.AddListener([&]() { latch.CountDown(); }, - &DirectExecutor::GetInstance()); + future.AddListener( + [&](ExceptionOr result) { + ASSERT_FALSE(result.ok()); + ASSERT_EQ(result.exception(), Exception::kTimeout); + latch.CountDown(); + }, + &DirectExecutor::GetInstance()); EXPECT_TRUE(latch.Await().Ok()); diff --git a/internal/platform/implementation/listenable_future.h b/internal/platform/implementation/listenable_future.h index 4dfa96a4..0633c48e 100644 --- a/internal/platform/implementation/listenable_future.h +++ b/internal/platform/implementation/listenable_future.h @@ -20,7 +20,6 @@ #include "internal/platform/exception.h" #include "internal/platform/implementation/executor.h" #include "internal/platform/implementation/future.h" -#include "internal/platform/runnable.h" namespace nearby { namespace api { @@ -31,9 +30,10 @@ namespace api { template class ListenableFuture : public Future { public: + using FutureCallback = absl::AnyInvocable)>; ~ListenableFuture() override = default; - virtual void AddListener(Runnable runnable, Executor* executor) = 0; + virtual void AddListener(FutureCallback callback, Executor* executor) = 0; }; } // namespace api diff --git a/internal/platform/implementation/windows/listenable_future.h b/internal/platform/implementation/windows/listenable_future.h index e12d6539..ede07fd9 100644 --- a/internal/platform/implementation/windows/listenable_future.h +++ b/internal/platform/implementation/windows/listenable_future.h @@ -26,11 +26,12 @@ namespace windows { template class ListenableFuture : public api::ListenableFuture { public: + using FutureCallback = typename api::ListenableFuture::FutureCallback; // TODO(b/184975123): replace with real implementation. ~ListenableFuture() override = default; // TODO(b/184975123): replace with real implementation. - void AddListener(Runnable runnable, api::Executor* executor) {} + void AddListener(FutureCallback callback, api::Executor* executor) {} }; } // namespace windows diff --git a/internal/platform/settable_future.h b/internal/platform/settable_future.h index 12dbf763..406187e3 100644 --- a/internal/platform/settable_future.h +++ b/internal/platform/settable_future.h @@ -20,6 +20,7 @@ #include #include "internal/platform/condition_variable.h" +#include "internal/platform/implementation/listenable_future.h" #include "internal/platform/mutex.h" #include "internal/platform/mutex_lock.h" #include "internal/platform/system_clock.h" @@ -30,6 +31,7 @@ namespace nearby { template class SettableFuture : public api::SettableFuture { public: + using FutureCallback = typename api::ListenableFuture::FutureCallback; SettableFuture() = default; // Creates a SettableFuture that fails with a kTimeout when `timeout` expires. @@ -54,12 +56,15 @@ class SettableFuture : public api::SettableFuture { return false; } - void AddListener(Runnable runnable, api::Executor* executor) override { + void AddListener(FutureCallback callback, api::Executor* executor) override { MutexLock lock(&mutex_); if (done_) { - executor->Execute(std::move(runnable)); + executor->Execute( + [value = GetLocked(), callback = std::move(callback)]() mutable { + callback(std::move(value)); + }); } else { - listeners_.emplace_back(std::make_pair(executor, std::move(runnable))); + listeners_.emplace_back(std::make_pair(executor, std::move(callback))); } } @@ -85,9 +90,7 @@ class SettableFuture : public api::SettableFuture { while (!done_) { completed_.Wait(); } - return exception_.value != Exception::kSuccess - ? ExceptionOr{exception_.value} - : ExceptionOr{value_}; + return GetLocked(); } ExceptionOr Get(absl::Duration timeout) override { @@ -106,9 +109,7 @@ class SettableFuture : public api::SettableFuture { break; } } - return exception_.value != Exception::kSuccess - ? ExceptionOr{exception_.value} - : ExceptionOr{value_}; + return GetLocked(); } private: @@ -124,16 +125,25 @@ class SettableFuture : public api::SettableFuture { return true; } + ExceptionOr GetLocked() { + return exception_.value != Exception::kSuccess + ? ExceptionOr{exception_.value} + : ExceptionOr{value_}; + } + void InvokeAllLocked() { for (auto& item : listeners_) { - item.first->Execute(std::move(item.second)); + item.first->Execute( + [value = GetLocked(), callback = std::move(item.second)]() mutable { + callback(std::move(value)); + }); } listeners_.clear(); } mutable Mutex mutex_; ConditionVariable completed_{&mutex_}; - std::vector> listeners_; + std::vector> listeners_; bool done_{false}; T value_; Exception exception_{Exception::kFailed};