yarpl/test/FlowableCreateSubscribe_test.cpp
/*
 * Verbose usage of interfaces with inline class implementations.
 *
 * - Demonstrating subscribe, onSubscribe, request, onNext, cancel.
 */
Verbose usage of interfaces with inline class implementations.
- Demonstrating subscribe, onSubscribe, request, onNext, cancel.
Verbose usage of interfaces with inline class implementations . - Demonstrating subscribe onSubscribe request onNext cancel .
TEST(FlowableCreateSubscribe, SubscribeRequestAndCancel) {
  class InfiniteIntegersSource : public Subscription {
    std::unique_ptr<Subscriber<int>> subscriber_;
    std::atomic_bool isCancelled{false};
   public:
    static void subscribe(std::unique_ptr<Subscriber<int>> subscriber) {
      InfiniteIntegersSource s_(std::move(subscriber));
    }
    void start() {
      subscriber_->onSubscribe(this);
    }
    void cancel() override {
      isCancelled = true;
      subscriber_->onComplete();
    }
    void request(int64_t n) override {
      for (auto i = 1; i < n; i++) {
        if (isCancelled) {
          return;
        }
        subscriber_->onNext(i);
      }
    }
   protected:
    InfiniteIntegersSource(std::unique_ptr<Subscriber<int>> subscriber)
        : subscriber_(std::move(subscriber)) {
      subscriber_->onSubscribe(this);
    }
  };
  class MySubscriber : public Subscriber<int> {
    Subscription* subscription;
   public:
    void onNext(const int& value) override {
      std::cout << "received& " << value << std::endl;
      if (value == 6) {
        subscription->cancel();
      }
    }
    void onError(const std::exception_ptr e) override {}
    void onComplete() override {}
    void onSubscribe(Subscription* s) override {
      std::cout << "onSubscribe in subscriber" << std::endl;
      subscription = s;
      subscription->request((long)10);
    }
  };
  auto a =
      Flowable<int>::create([](std::unique_ptr<Subscriber<int>> subscriber) {
        InfiniteIntegersSource::subscribe(std::move(subscriber));
      });
  auto ts = TestSubscriber<int>::create(std::make_unique<MySubscriber>());
  a->subscribe(ts->unique_subscriber());
  ts->awaitTerminalEvent();
  ts->assertValueCount(6);
}
TEST ( FlowableCreateSubscribe , SubscribeRequestAndCancel ) { class InfiniteIntegersSource : public Subscription { std :: unique_ptr < Subscriber < int > > subscriber_ ; std :: atomic_bool isCancelled { false } ; public : static void subscribe ( std :: unique_ptr < Subscriber < int > > subscriber ) { InfiniteIntegersSource s_ ( std :: move ( subscriber ) ) ; } void start ( ) { subscriber_ -> onSubscribe ( this ) ; } void cancel ( ) override { isCancelled = true ; subscriber_ -> onComplete ( ) ; } void request ( int64_t n ) override { for ( auto i = 1 ; i < n ; i ++ ) { if ( isCancelled ) { return ; } subscriber_ -> onNext ( i ) ; } } protected : InfiniteIntegersSource ( std :: unique_ptr < Subscriber < int > > subscriber ) : subscriber_ ( std :: move ( subscriber ) ) { subscriber_ -> onSubscribe ( this ) ; } } ; class MySubscriber : public Subscriber < int > { Subscription * subscription ; public : void onNext ( const int & value ) override { std :: cout << " " << value << std :: endl ; if ( value == 6 ) { subscription -> cancel ( ) ; } } void onError ( const std :: exception_ptr e ) override { } void onComplete ( ) override { } void onSubscribe ( Subscription * s ) override { std :: cout << " " << std :: endl ; subscription = s ; subscription -> request ( ( long ) 10 ) ; } } ; auto a = Flowable < int > :: create ( [ ] ( std :: unique_ptr < Subscriber < int > > subscriber ) { InfiniteIntegersSource :: subscribe ( std :: move ( subscriber ) ) ; } ) ; auto ts = TestSubscriber < int > :: create ( std :: make_unique < MySubscriber > ( ) ) ; a -> subscribe ( ts -> unique_subscriber ( ) ) ; ts -> awaitTerminalEvent ( ) ; ts -> assertValueCount ( 6 ) ; }
Verbose usage of interfaces with inline class implementations.
Verbose usage of interfaces with inline class implementations .
// Subscription that emits integers forever as long as requested // complete downstream like take(...) would // NOTE: Do not implement real code like this // This is NOT safe at all since request(n) can be called concurrently // This assumes synchronous execution which this unit test does // just to test the most basic wiring of all the types together. // Subscriber that requests 10 items, then cancels after receiving 6 // create Flowable around InfiniteIntegersSource