addCallback API

Summary: This makes it easier to integrate folly::Observer with callback-based code.

Reviewed By: spalamarchuk

Differential Revision: D6888234

fbshipit-source-id: ccf9b57c07ee323ee11496d14f8c90e3331cc7bb

folly/experimental/observer/Observer-inl.h

@@ -72,5 +72,48 @@ const Snapshot<T>& TLObserver<T>::getSnapshotRef() const {
 
   return snapshot;
 }
+
+struct CallbackHandle::Context {
+  Optional<Observer<folly::Unit>> observer;
+  Synchronized<bool> canceled{false};
+};
+
+inline CallbackHandle::CallbackHandle() {}
+
+template <typename T>
+CallbackHandle::CallbackHandle(
+    Observer<T> observer,
+    folly::Function<void(Snapshot<T>)> callback) {
+  context_ = std::make_shared<Context>();
+  context_->observer = makeObserver([observer = std::move(observer),
+                                     callback = std::move(callback),
+                                     context = context_]() mutable {
+    auto rCanceled = context->canceled.rlock();
+    if (*rCanceled) {
+      return folly::unit;
+    }
+    callback(*observer);
+    return folly::unit;
+  });
+}
+
+inline CallbackHandle::~CallbackHandle() {
+  cancel();
+}
+
+inline void CallbackHandle::cancel() {
+  if (!context_) {
+    return;
+  }
+  context_->observer.reset();
+  context_->canceled = true;
+  context_.reset();
+}
+
+template <typename T>
+CallbackHandle Observer<T>::addCallback(
+    folly::Function<void(Snapshot<T>)> callback) const {
+  return CallbackHandle(*this, std::move(callback));
+}
 } // namespace observer
 } // namespace folly

folly/experimental/observer/Observer.h

@@ -118,6 +118,28 @@ class Snapshot {
   const observer_detail::Core* core_;
 };
 
+class CallbackHandle {
+ public:
+  CallbackHandle();
+  template <typename T>
+  CallbackHandle(
+      Observer<T> observer,
+      folly::Function<void(Snapshot<T>)> callback);
+  CallbackHandle(const CallbackHandle&) = delete;
+  CallbackHandle(CallbackHandle&&) = default;
+  CallbackHandle& operator=(const CallbackHandle&) = delete;
+  CallbackHandle& operator=(CallbackHandle&&) = default;
+  ~CallbackHandle();
+
+  // If callback is currently running, waits until it completes.
+  // Callback will never be called after cancel() returns.
+  void cancel();
+
+ private:
+  struct Context;
+  std::shared_ptr<Context> context_;
+};
+
 template <typename T>
 class Observer {
  public:
@@ -137,6 +159,8 @@ class Observer {
     return snapshot.getVersion() < core_->getVersionLastChange();
   }
 
+  CallbackHandle addCallback(folly::Function<void(Snapshot<T>)> callback) const;
+
  private:
   template <typename Observable, typename Traits>
   friend class ObserverCreator;

folly/experimental/observer/test/ObserverTest.cpp

@@ -321,3 +321,35 @@ TEST(Observer, SubscribeCallback) {
   EXPECT_EQ(4, getCallsFinish);
   cobThread.join();
 }
+
+TEST(Observer, SetCallback) {
+  folly::observer::SimpleObservable<int> observable(42);
+  auto observer = observable.getObserver();
+  folly::Baton<> baton;
+  int callbackValue = 0;
+  size_t callbackCallsCount = 0;
+
+  auto callbackHandle =
+      observer.addCallback([&](folly::observer::Snapshot<int> snapshot) {
+        ++callbackCallsCount;
+        callbackValue = *snapshot;
+        baton.post();
+      });
+  baton.wait();
+  baton.reset();
+  EXPECT_EQ(42, callbackValue);
+  EXPECT_EQ(1, callbackCallsCount);
+
+  observable.setValue(43);
+  baton.wait();
+  baton.reset();
+  EXPECT_EQ(43, callbackValue);
+  EXPECT_EQ(2, callbackCallsCount);
+
+  callbackHandle.cancel();
+
+  observable.setValue(44);
+  EXPECT_FALSE(baton.timed_wait(std::chrono::milliseconds{100}));
+  EXPECT_EQ(43, callbackValue);
+  EXPECT_EQ(2, callbackCallsCount);
+}