TySS-Dev/nearby
1
0
Fork 0
mirror of https://github.com/kidfromjupiter/nearby.git synced 2026-09-14 22:56:12 -04:00
Code Issues Packages Projects Releases Wiki Activity

Add value parameter to future listeners

Browse Source 
Pass the Future result to listeners as a parameter.
This makes the AddListener() API a bit more convenient.

PiperOrigin-RevId: 518040156
This commit is contained in:
Janusz Sobczak
2023-03-20 12:15:47 -07:00
committed by Copybara-Service
parent ced4369202
commit e202af3ccc
5 changed files with 53 additions and 42 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/platform/future.h
+3 -2
View File
@@ -24,6 +24,7 @@ namespace nearby {
template <typename T>
class Future final {
public:
using FutureCallback = typename SettableFuture<T>::FutureCallback;
// Default Future. Does not time out.
Future() : impl_(std::make_shared<SettableFuture<T>>()) {}
@@ -39,8 +40,8 @@ class Future final {
virtual ExceptionOr<T> 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(); }
internal/platform/future_test.cc
+25 -26
View File
@@ -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<int> 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<int> 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<int> 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<int> 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<int> 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<int> result) {
ASSERT_FALSE(result.ok());
ASSERT_EQ(result.GetException(), kException);
++call_count;
},
&executor);
@@ -243,8 +237,13 @@ TEST(FutureTest, TimeoutSetsException) {
TEST(FutureTest, TimeoutCallsListeners) {
Future<int> future(absl::Milliseconds(10));
CountDownLatch latch(1);
future.AddListener([&]() { latch.CountDown(); },
&DirectExecutor::GetInstance());
future.AddListener(
[&](ExceptionOr<int> result) {
ASSERT_FALSE(result.ok());
ASSERT_EQ(result.exception(), Exception::kTimeout);
latch.CountDown();
},
&DirectExecutor::GetInstance());
EXPECT_TRUE(latch.Await().Ok());
internal/platform/implementation/listenable_future.h
+2 -2
View File
@@ -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 <typename T>
class ListenableFuture : public Future<T> {
public:
using FutureCallback = absl::AnyInvocable<void(ExceptionOr<T>)>;
~ListenableFuture() override = default;
virtual void AddListener(Runnable runnable, Executor* executor) = 0;
virtual void AddListener(FutureCallback callback, Executor* executor) = 0;
};
} // namespace api
internal/platform/implementation/windows/listenable_future.h
+2 -1
View File
@@ -26,11 +26,12 @@ namespace windows {
template <typename T>
class ListenableFuture : public api::ListenableFuture<T> {
public:
using FutureCallback = typename api::ListenableFuture<T>::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
internal/platform/settable_future.h
+21 -11
View File
@@ -20,6 +20,7 @@
#include <vector>
#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 <typename T>
class SettableFuture : public api::SettableFuture<T> {
public:
using FutureCallback = typename api::ListenableFuture<T>::FutureCallback;
SettableFuture() = default;
// Creates a SettableFuture that fails with a kTimeout when `timeout` expires.
@@ -54,12 +56,15 @@ class SettableFuture : public api::SettableFuture<T> {
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<T> {
while (!done_) {
completed_.Wait();
}
return exception_.value != Exception::kSuccess
? ExceptionOr<T>{exception_.value}
: ExceptionOr<T>{value_};
return GetLocked();
}
ExceptionOr<T> Get(absl::Duration timeout) override {
@@ -106,9 +109,7 @@ class SettableFuture : public api::SettableFuture<T> {
break;
}
}
return exception_.value != Exception::kSuccess
? ExceptionOr<T>{exception_.value}
: ExceptionOr<T>{value_};
return GetLocked();
}
private:
@@ -124,16 +125,25 @@ class SettableFuture : public api::SettableFuture<T> {
return true;
}
ExceptionOr<T> GetLocked() {
return exception_.value != Exception::kSuccess
? ExceptionOr<T>{exception_.value}
: ExceptionOr<T>{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<std::pair<api::Executor*, Runnable>> listeners_;
std::vector<std::pair<api::Executor*, FutureCallback>> listeners_;
bool done_{false};
T value_;
Exception exception_{Exception::kFailed};