Flowables::fromGenerator operator

yarpl/include/yarpl/flowable/Flowable.h

@@ -231,7 +231,8 @@ class Flowable : public virtual Refcounted {
  // Don't destroy a locked mutex.
  lock.unlock();
 
- return release();
+ release();
+ return;
  }
 
  // If no more items can be emitted now, wait for a request(n).

yarpl/include/yarpl/flowable/Flowables.h

@@ -116,6 +116,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(generated, false);
+ } catch(...) {
+ subscriber.onError(std::current_exception());
+ return std::make_tuple(generated, true);
+ } 
+ };
+ return Flowable<T>::create(std::move(lambda));
+ }
+
  private:
  Flowables() = delete;
 };

yarpl/test/FlowableTest.cpp

@@ -16,11 +16,7 @@ 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");
-
- CollectingSubscriber(int64_t requestCount = 100)
+ explicit CollectingSubscriber(int64_t requestCount = 100)
  : requestCount_(requestCount) {}
 
  void onSubscribe(Reference<Subscription> subscription) override {
@@ -64,6 +60,10 @@ class CollectingSubscriber : public Subscriber<T> {
  return errorMsg_;
  }
 
+ void cancelSubscription() {
+ Subscriber<T>::subscription()->cancel();
+ }
+
  private:
  std::vector<T> values_;
  std::string errorMsg_;
@@ -210,6 +210,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());