Flowables::fromGenerator operator

experimental/yarpl/include/yarpl/Flowable.h

@@ -104,6 +104,10 @@ class Flowable : public virtual Refcounted {
  }
 
  void cancel() override {
+ // if this is the first terminating signal to receive, we need to
+ // make sure we break the reference cycle between subscription and
+ // subscriber
+
  requested_.exchange(CANCELED, std::memory_order_relaxed);
  process();
  }
@@ -161,7 +165,8 @@ class Flowable : public virtual Refcounted {
  // Don't destroy a locked mutex.
  lock.unlock();
 
- return release();
+ releaseInstance();
+ return;
  }
 
  // If no more items can be emitted now, wait for a request(n).
@@ -188,6 +193,12 @@ class Flowable : public virtual Refcounted {
  }
  }
 
+ void releaseInstance() {
+ flowable_.reset();
+ subscriber_.reset();
+ release();
+ }
+
  // The number of items that can be sent downstream. Each request(n)
  // adds n; each onNext consumes 1. If this is MAX, flow-control is
  // disabled: items sent downstream don't consume any longer. A MIN

experimental/yarpl/include/yarpl/Flowables.h

@@ -108,6 +108,25 @@ class Flowables {
  return Flowable<T>::create(std::move(lambda));
  }
 
+ template <typename T, typename TGenerator>
+ static Reference<Flowable<T>> fromGenerator(TGenerator generator) {
+ auto lambda = [generator = std::move(generator)]
+ (Subscriber<T>& subscriber, int64_t requested) {
+ int64_t generated = 0;
+ try {
+ while (generated < requested) {
+ subscriber.onNext(generator());
+ ++generated;
+ }
+ return std::make_tuple(static_cast<int64_t>(generated), false);
+ } catch(const std::exception&) {
+ subscriber.onError(std::current_exception());
+ return std::make_tuple(static_cast<int64_t>(generated), true);
+ }
+ };
+ return Flowable<T>::create(std::move(lambda));
+ }
+
  private:
  Flowables() = delete;
 };

experimental/yarpl/include/yarpl/flowable/FlowableOperator.h

@@ -51,6 +51,8 @@ class FlowableOperator : public Flowable<D> {
  subscriber_->onSubscribe(Reference<::yarpl::flowable::Subscription>(this));
  }
 
+ void onNext(U) override {}
+
  void onComplete() override {
  subscriber_->onComplete();
  upstream_.reset();

experimental/yarpl/include/yarpl/flowable/Subscriber.h

@@ -27,7 +27,7 @@ class Subscriber : public virtual Refcounted {
  subscription_.reset();
  }
 
- virtual void onNext(T) {}
+ virtual void onNext(T) = 0;
 
  protected:
  Subscription* subscription() {

experimental/yarpl/include/yarpl/observable/ObservableOperator.h

@@ -40,7 +40,7 @@ class ObservableOperator : public Observable<D> {
  Subscription(
  Reference<Observable<D>> flowable,
  Reference<Observer<D>> subscriber)
- : flowable_(std::move(flowable)), subscriber_(std::move(subscriber)) {}
+ : observable_(std::move(flowable)), subscriber_(std::move(subscriber)) {}
 
  ~Subscription() {
  subscriber_.reset();
@@ -68,7 +68,7 @@ class ObservableOperator : public Observable<D> {
 
  protected:
  /// The Observable has the lambda, and other creation parameters.
- Reference<Observable<D>> flowable_;
+ Reference<Observable<D>> observable_;
 
  /// This subscription controls the life-cycle of the subscriber. The
  /// subscriber is retained as long as calls on it can be made. (Note:
@@ -117,7 +117,7 @@ class MapOperator : public ObservableOperator<U, D> {
  void onNext(U value) override {
  auto* subscriber =
  ObservableOperator<U, D>::Subscription::subscriber_.get();
- auto* flowable = ObservableOperator<U, D>::Subscription::flowable_.get();
+ auto* flowable = ObservableOperator<U, D>::Subscription::observable_.get();
  auto* map = static_cast<MapOperator*>(flowable);
  subscriber->onNext(map->function_(std::move(value)));
  }

experimental/yarpl/test/FlowableTest.cpp

@@ -17,17 +17,14 @@ void unreachable() {
 template <typename T>
 class CollectingSubscriber : public Subscriber<T> {
  public:
- static_assert(
- std::is_copy_constructible<T>::value,
- "CollectingSubscriber needs to copy the value in order to collect it");
+ explicit CollectingSubscriber(int64_t requested = 100) : requested_(requested) {}
 
  void onSubscribe(Reference<Subscription> subscription) override {
  Subscriber<T>::onSubscribe(subscription);
- subscription->request(100);
+ subscription->request(requested_);
  }
 
  void onNext(T next) override {
- Subscriber<T>::onNext(next);
  values_.push_back(std::move(next));
  }
 
@@ -64,11 +61,16 @@ class CollectingSubscriber : public Subscriber<T> {
  return errorMsg_;
  }
 
+ void cancelSubscription() {
+ Subscriber<T>::subscription()->cancel();
+ }
+
  private:
  std::vector<T> values_;
  std::string errorMsg_;
  bool complete_{false};
  bool error_{false};
+ int64_t requested_;
 };
 
 /// Construct a pipeline with a collecting subscriber against the supplied
@@ -198,6 +200,51 @@ TEST(FlowableTest, FlowableEmpty) {
  EXPECT_EQ(collector->error(), false);
 }
 
+TEST(FlowableTest, FlowableFromGenerator) {
+ EXPECT_EQ(0u, Refcounted::objects());
+
+ auto flowable = Flowables::fromGenerator<std::unique_ptr<int>>(
+ [] {return std::unique_ptr<int>();}
+ );
+ EXPECT_EQ(1u, Refcounted::objects());
+
+ auto collector = make_ref<CollectingSubscriber<std::unique_ptr<int>>>(10);
+ flowable->subscribe(collector);
+
+ EXPECT_EQ(collector->complete(), false);
+ EXPECT_EQ(collector->error(), false);
+ EXPECT_EQ(std::size_t{10}, collector->values().size());
+
+ collector->cancelSubscription();
+
+ flowable.reset();
+ collector.reset();
+ EXPECT_EQ(0u, Refcounted::objects());
+}
+
+TEST(FlowableTest, FlowableFromGeneratorException) {
+ EXPECT_EQ(0u, Refcounted::objects());
+
+ int count = 5;
+ auto flowable = Flowables::fromGenerator<std::unique_ptr<int>>(
+ [&] {
+ while (count--) { return std::unique_ptr<int>(); }
+ throw std::runtime_error("error from generator");
+ });
+ EXPECT_EQ(1u, Refcounted::objects());
+
+ auto collector = make_ref<CollectingSubscriber<std::unique_ptr<int>>>(10);
+ flowable->subscribe(collector);
+
+ EXPECT_EQ(collector->complete(), false);
+ EXPECT_EQ(collector->error(), true);
+ EXPECT_EQ(std::size_t{5}, collector->values().size());
+
+ collector.reset();
+ flowable.reset();
+ EXPECT_EQ(0u, Refcounted::objects());
+}
+
 TEST(FlowableTest, SubscribersComplete) {
  EXPECT_EQ(0u, Refcounted::objects());
 

experimental/yarpl/test/Observable_test.cpp

@@ -455,3 +455,21 @@ TEST(Observable, ObserversError) {
 
  EXPECT_TRUE(errored);
 }
+
+TEST(Observable, CancelReleasesObjects) {
+ EXPECT_EQ(0u, Refcounted::objects());
+
+ auto lambda = [](Reference<Observer<int>> observer) {
+ // we will not send anything
+ };
+ auto observable = Observable<int>::create(std::move(lambda));
+
+ EXPECT_EQ(1u, Refcounted::objects());
+
+ auto collector = make_ref<CollectingObserver<int>>();
+ observable->subscribe(collector);
+
+ observable.reset();
+ collector.reset();
+ EXPECT_EQ(0u, Refcounted::objects());
+}

experimental/yarpl/test/ReferenceTest.cpp

@@ -15,7 +15,9 @@ using yarpl::flowable::Subscriber;
 namespace {
 
 template <class T>
-class MySubscriber : public Subscriber<T> {};
+class MySubscriber : public Subscriber<T> {
+ void onNext(T) override {}
+};
 
 }
 
@@ -36,7 +38,7 @@ TEST(ReferenceTest, Upcast) {
 }
 
 TEST(RefcountedTest, CopyAssign) {
- using Sub = Subscriber<int>;
+ using Sub = MySubscriber<int>;
  Reference<Sub> a(new Sub());
  Reference<Sub> b(a);
  EXPECT_EQ(2u, a->count());
@@ -48,7 +50,7 @@ TEST(RefcountedTest, CopyAssign) {
 }
 
 TEST(RefcountedTest, MoveAssign) {
- using Sub = Subscriber<int>;
+ using Sub = MySubscriber<int>;
  Reference<Sub> a(new Sub());
  Reference<Sub> b(a);
  EXPECT_EQ(2u, a->count());
@@ -59,7 +61,7 @@ TEST(RefcountedTest, MoveAssign) {
 }
 
 TEST(RefcountedTest, CopyAssignTemplate) {
- using Sub = Subscriber<int>;
+ using Sub = MySubscriber<int>;
  Reference<Sub> a(new Sub());
  Reference<Sub> b(a);
  EXPECT_EQ(2u, a->count());
@@ -72,7 +74,7 @@ TEST(RefcountedTest, CopyAssignTemplate) {
 }
 
 TEST(RefcountedTest, MoveAssignTemplate) {
- using Sub = Subscriber<int>;
+ using Sub = MySubscriber<int>;
  Reference<Sub> a(new Sub());
  Reference<Sub> b(a);
  EXPECT_EQ(2u, a->count());