yarpl/test/FlowableCreateSubscribe_test.cpp
/**
 * Assert that memory is not retained when a single Flowable
 * is subscribed to multiple times.
 */
Assert that memory is not retained when a single Flowable
is subscribed to multiple times.
Assert that memory is not retained when a single Flowable is subscribed to multiple times .
TEST(
    FlowableCreateSubscribe,
    TestRetainOnStaticAsyncFlowableWithMultipleSubscribers) {
  static std::atomic<int> subscriptionInstanceCount;
  static std::atomic<int> subscriberInstanceCount;
  class MySubscriber : public Subscriber<Tuple> {
    Subscription* subscription_;
   public:
    MySubscriber() {
      std::cout << "   Subscriber created!!" << std::endl;
      subscriberInstanceCount++;
    }
    ~MySubscriber() {
      std::cout << "   Subscriber destroyed!!" << std::endl;
      subscriberInstanceCount--;
    }
    void onNext(const Tuple& value) override {
      std::cout << "   Subscriber received value& " << value.a << " on thread "
                << std::this_thread::get_id() << std::endl;
    }
    void onError(const std::exception_ptr e) override {}
    void onComplete() override {}
    void onSubscribe(Subscription* s) override {
      subscription_ = std::move(s);
      subscription_->request((u_long)10);
    }
  };
  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)) {
      std::cout << "   Subscription created!!" << std::endl;
      subscriptionInstanceCount++;
    }
    ~Source() {
      std::cout << "   Subscription destroyed!!" << std::endl;
      subscriptionInstanceCount--;
    }
    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();
    }
  };
  {
    static std::atomic_int counter{2};
    auto a = Flowable<Tuple>::create(
        [](std::unique_ptr<Subscriber<Tuple>> subscriber) {
          std::cout << "Flowable subscribed to ... starting new thread ..."
                    << std::endl;
          try {
            std::thread([subscriber_ = std::move(subscriber)]() mutable {
              try {
                std::cout << "*** Running in another thread!!!!!" << std::endl;
                std::this_thread::sleep_for(std::chrono::milliseconds(100));
                Source s(std::move(subscriber_));
                s.start();
                counter--;
                std::cout << "Done thread" << std::endl;
              } catch (std::exception& e) {
                std::cout << e.what() << std::endl;
              }
            }).detach();
          } catch (std::exception& e) {
            std::cout << e.what() << std::endl;
          }
        });
    EXPECT_EQ(0, subscriptionInstanceCount);
    std::cout << "Thread counter: " << counter << std::endl;
    std::cout << "About to subscribe (1) ..." << std::endl;
    { a->subscribe(std::make_unique<MySubscriber>()); }
    while (counter > 1) {
      std::cout << "waiting for completion" << std::endl;
      std::this_thread::sleep_for(std::chrono::milliseconds(50));
    }
    EXPECT_EQ(0, subscriptionInstanceCount);
    std::cout << "About to subscribe (2) ..." << std::endl;
    { a->subscribe(std::make_unique<MySubscriber>()); }
    while (counter > 0) {
      std::cout << "waiting for completion" << std::endl;
      std::this_thread::sleep_for(std::chrono::milliseconds(50));
    }
    EXPECT_EQ(0, subscriptionInstanceCount);
    std::cout << "Finished ... remaining instances == " << Tuple::instanceCount
              << std::endl;
  }
  std::cout << "... after block" << std::endl;
  EXPECT_EQ(0, Tuple::instanceCount);
  EXPECT_EQ(0, subscriptionInstanceCount);
  EXPECT_EQ(0, subscriberInstanceCount);
  std::cout << "-----------------------------" << std::endl;
}
TEST ( FlowableCreateSubscribe , TestRetainOnStaticAsyncFlowableWithMultipleSubscribers ) { static std :: atomic < int > subscriptionInstanceCount ; static std :: atomic < int > subscriberInstanceCount ; class MySubscriber : public Subscriber < Tuple > { Subscription * subscription_ ; public : MySubscriber ( ) { std :: cout << " " << std :: endl ; subscriberInstanceCount ++ ; } ~ MySubscriber ( ) { std :: cout << " " << std :: endl ; subscriberInstanceCount -- ; } void onNext ( const Tuple & value ) override { std :: cout << " " << value . a << " " << std :: this_thread :: get_id ( ) << std :: endl ; } void onError ( const std :: exception_ptr e ) override { } void onComplete ( ) override { } void onSubscribe ( Subscription * s ) override { subscription_ = std :: move ( s ) ; subscription_ -> request ( ( u_long ) 10 ) ; } } ; 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 ) ) { std :: cout << " " << std :: endl ; subscriptionInstanceCount ++ ; } ~ Source ( ) { std :: cout << " " << std :: endl ; subscriptionInstanceCount -- ; } 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 ( ) ; } } ; { static std :: atomic_int counter { 2 } ; auto a = Flowable < Tuple > :: create ( [ ] ( std :: unique_ptr < Subscriber < Tuple > > subscriber ) { std :: cout << " " << std :: endl ; try { std :: thread ( [ subscriber_ = std :: move ( subscriber ) ] ( ) mutable { try { std :: cout << " " << std :: endl ; std :: this_thread :: sleep_for ( std :: chrono :: milliseconds ( 100 ) ) ; Source s ( std :: move ( subscriber_ ) ) ; s . start ( ) ; counter -- ; std :: cout << " " << std :: endl ; } catch ( std :: exception & e ) { std :: cout << e . what ( ) << std :: endl ; } } ) . detach ( ) ; } catch ( std :: exception & e ) { std :: cout << e . what ( ) << std :: endl ; } } ) ; EXPECT_EQ ( 0 , subscriptionInstanceCount ) ; std :: cout << " " << counter << std :: endl ; std :: cout << " " << std :: endl ; { a -> subscribe ( std :: make_unique < MySubscriber > ( ) ) ; } while ( counter > 1 ) { std :: cout << " " << std :: endl ; std :: this_thread :: sleep_for ( std :: chrono :: milliseconds ( 50 ) ) ; } EXPECT_EQ ( 0 , subscriptionInstanceCount ) ; std :: cout << " " << std :: endl ; { a -> subscribe ( std :: make_unique < MySubscriber > ( ) ) ; } while ( counter > 0 ) { std :: cout << " " << std :: endl ; std :: this_thread :: sleep_for ( std :: chrono :: milliseconds ( 50 ) ) ; } EXPECT_EQ ( 0 , subscriptionInstanceCount ) ; std :: cout << " " << Tuple :: instanceCount << std :: endl ; } std :: cout << " " << std :: endl ; EXPECT_EQ ( 0 , Tuple :: instanceCount ) ; EXPECT_EQ ( 0 , subscriptionInstanceCount ) ; EXPECT_EQ ( 0 , subscriberInstanceCount ) ; std :: cout << " " << std :: endl ; }
Assert that memory is not retained when a single Flowable
 is subscribed to multiple times.
Assert that memory is not retained when a single Flowable is subscribed to multiple times .
// ignoring n for tests ... DO NOT DO THIS FOR REAL // create Flowable around Source // force artificial delay (no race condition, just want to // trigger scheduling/async) /* sleep override */ // now do work // assuming pipeline is synchronous, which it is in this test // TODO use condition variable in MySubscriber /* sleep override */ // ensure Subscriber and Subscription are destroyed // TODO use condition variable in MySubscriber /* sleep override */ // ensure Subscriber and Subscription are destroyed