Search code examples
c++rxcpp

How to create a ReplaySubject with RxCpp?


In my c++ project, I need to create Subjects having an initial value, that may be updated. On each subscription/update, subscribers may trigger then data processing... In a previous Angular (RxJS) project, this kind of behavior was handled with ReplaySubject(1).

I'm not able to reproduce this using c++ rxcpp lib.

I've looked up for documentation, snippets, tutorials, but without success.

Expected pseudocode (typescript):


private dataSub: ReplaySubject<Data> = new ReplaySubject<Data>(1);

private init = false;

// public Observable, immediatly share last published value
get currentData$(): Observable<Data> {

    if (!this.init) {
      return this.initData().pipe(switchMap(
        () => this.dataSub.asObservable()
      ));
    }
    return this.dataSub.asObservable();
  }


Solution

  • I think you are looking for rxcpp::subjects::replay. Please try this:

        auto coordination = rxcpp::observe_on_new_thread();
        rxcpp::subjects::replay<int, decltype(coordination)> test(1, coordination);
    
        // to emit data
        test.get_observer().on_next(1);
        test.get_observer().on_next(2);
        test.get_observer().on_next(3);
    
        // to subscribe
        test.get_observable().subscribe([](auto && v)
            printf("%d\n", v); // this will print '3'
        });