Update Observables to match Flowables, make it actually work (#407)

It didn't really compile in the first place...  Copy some Flowable tests over to
use Observable.

Author: alexmalyshev

experimental/yarpl/include/yarpl/Flowables.h

@@ -91,7 +91,7 @@ class Flowables {
   }
 
   template <typename T>
-  static Reference<Flowable<T>> error(const std::exception_ptr ex) {
+  static Reference<Flowable<T>> error(std::exception_ptr ex) {
     auto lambda = [ex](Subscriber<T>& subscriber, int64_t) {
       subscriber.onError(ex);
       return std::make_tuple(static_cast<int64_t>(0), true);

experimental/yarpl/include/yarpl/Observables.h

@@ -10,57 +10,36 @@ namespace observable {
 class Observables {
  public:
   static Reference<Observable<int64_t>> range(int64_t start, int64_t end) {
-    auto lambda = [ start, end, i = start ](
-        Subscriber<int64_t> & subscriber, int64_t requested) mutable {
-      int64_t emitted = 0;
-      bool done = false;
-
-      while (i < end && emitted < requested) {
-        subscriber.onNext(i++);
-        ++emitted;
+    auto lambda = [ start, end ](Reference<Observer<int64_t>> observer) {
+      for (int64_t i = start; i < end; ++i) {
+        observer->onNext(i);
       }
-
-      if (i >= end) {
-        subscriber.onComplete();
-        done = true;
-      }
-
-      return std::make_tuple(requested, done);
+      observer->onComplete();
     };
 
     return Observable<int64_t>::create(std::move(lambda));
   }
 
   template <typename T>
   static Reference<Observable<T>> just(const T& value) {
-    auto lambda = [value](Subscriber<T>& subscriber, int64_t) {
+    auto lambda = [value](Reference<Observer<T>> observer) {
       // # requested should be > 0.  Ignoring the actual parameter.
-      subscriber.onNext(value);
-      subscriber.onComplete();
-      return std::make_tuple(static_cast<int64_t>(1), true);
+      observer->onNext(value);
+      observer->onComplete();
     };
 
     return Observable<T>::create(std::move(lambda));
   }
 
   template <typename T>
-  static Reference<Observable<T>> just(std::initializer_list<T> list) {
-    auto lambda = [ list, it = list.begin() ](
-        Subscriber<T> & subscriber, int64_t requested) mutable {
-      int64_t emitted = 0;
-      bool done = false;
-
-      while (it != list.end() && emitted < requested) {
-        subscriber.onNext(*it++);
-        ++emitted;
-      }
+  static Reference<Observable<T>> justN(std::initializer_list<T> list) {
+    std::vector<T> vec(list);
 
-      if (it == list.end()) {
-        subscriber.onComplete();
-        done = true;
+    auto lambda = [v = std::move(vec)](Reference<Observer<T>> observer) {
+      for (auto const& elem : v) {
+        observer->onNext(elem);
       }
-
-      return std::make_tuple(static_cast<int64_t>(emitted), done);
+      observer->onComplete();
     };
 
     return Observable<T>::create(std::move(lambda));
@@ -70,14 +49,37 @@ class Observables {
       typename T,
       typename OnSubscribe,
       typename = typename std::enable_if<std::is_callable<
-          OnSubscribe(Reference<Subscriber<T>>),
+          OnSubscribe(Reference<Observer<T>>),
           void>::value>::type>
-
   static Reference<Observable<T>> create(OnSubscribe&& function) {
     return Reference<Observable<T>>(new FromPublisherOperator<T, OnSubscribe>(
         std::forward<OnSubscribe>(function)));
   }
 
+  template <typename T>
+  static Reference<Observable<T>> empty() {
+    auto lambda = [](Reference<Observer<T>> observer) {
+      observer->onComplete();
+    };
+    return Observable<T>::create(std::move(lambda));
+  }
+
+  template <typename T>
+  static Reference<Observable<T>> error(std::exception_ptr ex) {
+    auto lambda = [ex](Reference<Observer<T>> observer) {
+      observer->onError(ex);
+    };
+    return Observable<T>::create(std::move(lambda));
+  }
+
+  template <typename T, typename ExceptionType>
+  static Reference<Observable<T>> error(const ExceptionType& ex) {
+    auto lambda = [ex](Reference<Observer<T>> observer) {
+      observer->onError(std::make_exception_ptr(ex));
+    };
+    return Observable<T>::create(std::move(lambda));
+  }
+
  private:
   Observables() = delete;
 };

experimental/yarpl/test/FlowableTest.cpp

@@ -10,6 +10,10 @@ namespace yarpl {
 namespace flowable {
 namespace {
 
+void unreachable() {
+  EXPECT_TRUE(false);
+}
+
 template <typename T>
 class CollectingSubscriber : public Subscriber<T> {
  public:
@@ -56,15 +60,15 @@ class CollectingSubscriber : public Subscriber<T> {
     return error_;
   }
 
-  std::string errorMsg() const {
+  const std::string& errorMsg() const {
     return errorMsg_;
   }
 
  private:
   std::vector<T> values_;
+  std::string errorMsg_;
   bool complete_{false};
   bool error_{false};
-  std::string errorMsg_;
 };
 
 /// Construct a pipeline with a collecting subscriber against the supplied
@@ -194,10 +198,6 @@ TEST(FlowableTest, FlowableEmpty) {
   EXPECT_EQ(collector->error(), false);
 }
 
-void unreachable() {
-  EXPECT_TRUE(false);
-}
-
 TEST(FlowableTest, SubscribersComplete) {
   EXPECT_EQ(0u, Refcounted::objects());
 

experimental/yarpl/test/Observable_test.cpp

@@ -3,18 +3,90 @@
 #include <folly/Baton.h>
 #include <gtest/gtest.h>
 #include <atomic>
-#include "Tuple.h"
+
 #include "yarpl/Observable.h"
+#include "yarpl/Observables.h"
 #include "yarpl/ThreadScheduler.h"
 #include "yarpl/flowable/Subscriber.h"
 #include "yarpl/flowable/Subscribers.h"
 #include "yarpl/observable/Observers.h"
 #include "yarpl/observable/Subscriptions.h"
 
+#include "Tuple.h"
+
 // TODO can we eliminate need to import both of these?
 using namespace yarpl;
 using namespace yarpl::observable;
 
+namespace {
+
+void unreachable() {
+  EXPECT_TRUE(false);
+}
+
+template <typename T>
+class CollectingObserver : public Observer<T> {
+ public:
+  static_assert(
+      std::is_copy_constructible<T>::value,
+      "CollectingSubscriber needs to copy the value in order to collect it");
+
+  void onNext(T next) override {
+    Observer<T>::onNext(next);
+    values_.push_back(std::move(next));
+  }
+
+  void onComplete() override {
+    Observer<T>::onComplete();
+    complete_ = true;
+  }
+
+  void onError(std::exception_ptr ex) override {
+    Observer<T>::onError(ex);
+    error_ = true;
+
+    try {
+      std::rethrow_exception(ex);
+    } catch (const std::exception& e) {
+      errorMsg_ = e.what();
+    }
+  }
+
+  const std::vector<T>& values() const {
+    return values_;
+  }
+
+  bool complete() const {
+    return complete_;
+  }
+
+  bool error() const {
+    return error_;
+  }
+
+  const std::string& errorMsg() const {
+    return errorMsg_;
+  }
+
+ private:
+  std::vector<T> values_;
+  std::string errorMsg_;
+  bool complete_{false};
+  bool error_{false};
+};
+
+/// Construct a pipeline with a collecting observer against the supplied
+/// observable.  Return the items that were sent to the observer.  If some
+/// exception was sent, the exception is thrown.
+template <typename T>
+std::vector<T> run(Reference<Observable<T>> observable) {
+  auto collector = make_ref<CollectingObserver<T>>();
+  observable->subscribe(collector);
+  return collector->values();
+}
+
+} // namespace
+
 TEST(Observable, SingleOnNext) {
   {
     ASSERT_EQ(std::size_t{0}, Refcounted::objects());
@@ -262,3 +334,124 @@ TEST(Observable, toFlowableWithCancel) {
   }
   ASSERT_EQ(std::size_t{0}, Refcounted::objects());
 }
+
+TEST(Observable, Just) {
+  ASSERT_EQ(0u, Refcounted::objects());
+
+  EXPECT_EQ(run(Observables::just(22)), std::vector<int>{22});
+  EXPECT_EQ(
+      run(Observables::justN({12, 34, 56, 98})),
+      std::vector<int>({12, 34, 56, 98}));
+  EXPECT_EQ(
+      run(Observables::justN({"ab", "pq", "yz"})),
+      std::vector<const char*>({"ab", "pq", "yz"}));
+
+  EXPECT_EQ(0u, Refcounted::objects());
+}
+
+TEST(Observable, Range) {
+  ASSERT_EQ(0u, Refcounted::objects());
+
+  auto observable = Observables::range(10, 14);
+  EXPECT_EQ(run(std::move(observable)), std::vector<int64_t>({10, 11, 12, 13}));
+
+  EXPECT_EQ(0u, Refcounted::objects());
+}
+
+TEST(Observable, RangeWithMap) {
+  ASSERT_EQ(0u, Refcounted::objects());
+
+  auto observable = Observables::range(1, 4)
+                      ->map([](int64_t v) { return v * v; })
+                      ->map([](int64_t v) { return v * v; })
+                      ->map([](int64_t v) { return std::to_string(v); });
+  EXPECT_EQ(
+      run(std::move(observable)), std::vector<std::string>({"1", "16", "81"}));
+
+  EXPECT_EQ(0u, Refcounted::objects());
+}
+
+// TODO: Hits ASAN errors.
+TEST(Observable, DISABLED_SimpleTake) {
+  ASSERT_EQ(0u, Refcounted::objects());
+
+  EXPECT_EQ(
+      run(Observables::range(0, 100)->take(3)),
+      std::vector<int64_t>({0, 1, 2}));
+
+  EXPECT_EQ(0u, Refcounted::objects());
+}
+
+TEST(Observable, Error) {
+  auto observable = Observables::error<int>(std::runtime_error("something broke!"));
+  auto collector = make_ref<CollectingObserver<int>>();
+  observable->subscribe(collector);
+
+  EXPECT_EQ(collector->complete(), false);
+  EXPECT_EQ(collector->error(), true);
+  EXPECT_EQ(collector->errorMsg(), "something broke!");
+}
+
+TEST(Observable, ErrorPtr) {
+  auto observable = Observables::error<int>(
+      std::make_exception_ptr(std::runtime_error("something broke!")));
+  auto collector = make_ref<CollectingObserver<int>>();
+  observable->subscribe(collector);
+
+  EXPECT_EQ(collector->complete(), false);
+  EXPECT_EQ(collector->error(), true);
+  EXPECT_EQ(collector->errorMsg(), "something broke!");
+}
+
+TEST(Observable, Empty) {
+  auto observable = Observables::empty<int>();
+  auto collector = make_ref<CollectingObserver<int>>();
+  observable->subscribe(collector);
+
+  EXPECT_EQ(collector->complete(), true);
+  EXPECT_EQ(collector->error(), false);
+}
+
+TEST(Observable, ObserversComplete) {
+  EXPECT_EQ(0u, Refcounted::objects());
+
+  auto observable = Observables::empty<int>();
+  EXPECT_EQ(1u, Refcounted::objects());
+
+  bool completed = false;
+
+  auto observer = Observers::create<int>(
+    [](int) { unreachable(); },
+    [](std::exception_ptr) { unreachable(); },
+    [&] { completed = true; }
+  );
+
+  observable->subscribe(std::move(observer));
+  observable.reset();
+
+  EXPECT_EQ(0u, Refcounted::objects());
+
+  EXPECT_TRUE(completed);
+}
+
+TEST(Observable, ObserversError) {
+  EXPECT_EQ(0u, Refcounted::objects());
+
+  auto observable = Observables::error<int>(std::runtime_error("Whoops"));
+  EXPECT_EQ(1u, Refcounted::objects());
+
+  bool errored = false;
+
+  auto observer = Observers::create<int>(
+    [](int) { unreachable(); },
+    [&](std::exception_ptr) { errored = true; },
+    [] { unreachable(); }
+  );
+
+  observable->subscribe(std::move(observer));
+  observable.reset();
+
+  EXPECT_EQ(0u, Refcounted::objects());
+
+  EXPECT_TRUE(errored);
+}