yarpl/test/FlowableCreateSubscribe_test.cpp
/**
 * Assert that all items passed through the Flowable get destroyed
 */
Assert that all items passed through the Flowable get destroyed
Assert that all items passed through the Flowable get destroyed
TEST(FlowableCreateSubscribe, ItemsCollectedSynchronously) {
  class Source : public Subscription {
    std::unique_ptr<Subscriber<Tuple>> subscriber_;
    std::atomic_bool isCancelled{false};
   public:
    Source(std::unique_ptr<Subscriber<Tuple>> subscriber)
        : subscriber_(std::move(subscriber)) {}
    void start() {
      subscriber_->onSubscribe(this);
    }
    void cancel() override {
      isCancelled = true;
    }
    void request(int64_t n) override {
      subscriber_->onNext(Tuple{1, 2});
      subscriber_->onNext(Tuple{2, 3});
      subscriber_->onNext(Tuple{3, 4});
      subscriber_->onComplete();
    }
  };
  auto a = Flowable<Tuple>::create(
      [](std::unique_ptr<Subscriber<Tuple>> subscriber) {
        Source s(std::move(subscriber));
        s.start();
      });
  auto ts = TestSubscriber<Tuple>::create();
  a->subscribe(ts->unique_subscriber());
  ts->awaitTerminalEvent();
  ts.reset();
  std::cout << "Remaining tuples == " << Tuple::instanceCount << std::endl;
  EXPECT_EQ(0, Tuple::instanceCount);
  std::cout << "-----------------------------" << std::endl;
}
TEST ( FlowableCreateSubscribe , ItemsCollectedSynchronously ) { class Source : public Subscription { std :: unique_ptr < Subscriber < Tuple > > subscriber_ ; std :: atomic_bool isCancelled { false } ; public : Source ( std :: unique_ptr < Subscriber < Tuple > > subscriber ) : subscriber_ ( std :: move ( subscriber ) ) { } void start ( ) { subscriber_ -> onSubscribe ( this ) ; } void cancel ( ) override { isCancelled = true ; } void request ( int64_t n ) override { subscriber_ -> onNext ( Tuple { 1 , 2 } ) ; subscriber_ -> onNext ( Tuple { 2 , 3 } ) ; subscriber_ -> onNext ( Tuple { 3 , 4 } ) ; subscriber_ -> onComplete ( ) ; } } ; auto a = Flowable < Tuple > :: create ( [ ] ( std :: unique_ptr < Subscriber < Tuple > > subscriber ) { Source s ( std :: move ( subscriber ) ) ; s . start ( ) ; } ) ; auto ts = TestSubscriber < Tuple > :: create ( ) ; a -> subscribe ( ts -> unique_subscriber ( ) ) ; ts -> awaitTerminalEvent ( ) ; ts . reset ( ) ; std :: cout << " " << Tuple :: instanceCount << std :: endl ; EXPECT_EQ ( 0 , Tuple :: instanceCount ) ; std :: cout << " " << std :: endl ; }
Assert that all items passed through the Flowable get destroyed
Assert that all items passed through the Flowable get destroyed
// ignoring n for tests ... DO NOT DO THIS FOR REAL // create Flowable around Source // reset the TestSubscriber since it holds the onNexted values