RxJava: how to compose multiple Observables with d

Question

I'm learning RxJava and, as my first experiment, trying to rewrite the code in the first run() method in this code (cited on Netflix's blog as a problem RxJava can help solve) to improve its asynchronicity using RxJava, i.e. so it doesn't wait for the result of the first Future (f1.get()) before proceeding on to the rest of the code.

f3 depends on f1. I see how to handle this, flatMap seems to do the trick:

Observable<String> f3Observable = Observable.from(executor.submit(new CallToRemoteServiceA()))
    .flatMap(new Func1<String, Observable<String>>() {
        @Override
        public Observable<String> call(String s) {
            return Observable.from(executor.submit(new CallToRemoteServiceC(s)));
        }
    });

Next, f4 and f5 depend on f2. I have this:

final Observable<Integer> f4And5Observable = Observable.from(executor.submit(new CallToRemoteServiceB()))
    .flatMap(new Func1<Integer, Observable<Integer>>() {
        @Override
        public Observable<Integer> call(Integer i) {
            Observable<Integer> f4Observable = Observable.from(executor.submit(new CallToRemoteServiceD(i)));
            Observable<Integer> f5Observable = Observable.from(executor.submit(new CallToRemoteServiceE(i)));
            return Observable.merge(f4Observable, f5Observable);
        }
    });

Which starts to get weird (mergeing them probably isn't what I want...) but allows me to do this at the end, not quite what I want:

f3Observable.subscribe(new Action1<String>() {
    @Override
    public void call(String s) {
        System.out.println("Observed from f3: " + s);
        f4And5Observable.subscribe(new Action1<Integer>() {
            @Override
            public void call(Integer i) {
                System.out.println("Observed from f4 and f5: " + i);
            }
        });
    }
});

That gives me:

Observed from f3: responseB_responseA
Observed from f4 and f5: 140
Observed from f4 and f5: 5100

which is all the numbers, but unfortunately I get the results in separate invocations, so I can't quite replace the final println in the original code:

System.out.println(f3.get() + " => " + (f4.get() * f5.get()));

I don't understand how to get access to both those return values on the same line. I think there's probably some functional programming fu I'm missing here. How can I do this? Thanks.

Answer 1

It looks like all you really need is a bit more encouragement and perspective on how RX is used. I'd suggest you read more into the documentation as well as marble diagrams (I know they're not always useful). I also suggest looking into the lift() function and operators.

• The entire point of an observable is to concatenate data flow and data manipulation into a single object
• The point of calls to map, flatMap and filter are to manipulate the data in your data flow
• The point of merges are to combine data flows

• The point of operators are to allow you to disrupt a steady stream of observables and define your own operations on a data flow. For example, I coded a moving average operator. That sums up n doubles in an Observable of doubles to return a stream of moving averages. The code literally looked like this

  Observable movingAverage = Observable.from(mDoublesArray).lift(new MovingAverageOperator(frameSize))

You'll be a relieved that a lot of the filtering methods that you take for granted all have lift() under the hood.

With that said; all it takes to merge multiple dependencies is:

• changing all incoming data to a standard data type using map or flatMap
• merging standard data-types to a stream
• using custom operators if one object needs to wait on another, or if you need to order data in the stream. Caution: this approach will slow the stream down
• using to list or subscribe to collect all of that data

Answer 2

I think what you are looking for is switchmap. We ran into a similar issue where we have a session service that handles getting a new session from an api, and we need that session before we can get more data. We can add to the session observable that returns the sessionToken for use in our data call.

getSession returns an observable;

public getSession(): Observable<any>{
  if (this.sessionToken)
    return Observable.of(this.sessionToken);
  else if(this.sessionObservable)
    return this.sessionObservable;
  else {
    // simulate http call 
    this.sessionObservable = Observable.of(this.sessonTokenResponse)
    .map(res => {
      this.sessionObservable = null;
      return res.headers["X-Session-Token"];
    })
    .delay(500)
    .share();
    return this.sessionObservable;
  }
}

and getData takes that observable and appends to it.

public getData() {
  if (this.dataObservable)
    return this.dataObservable;
  else {
    this.dataObservable = this.sessionService.getSession()
      .switchMap((sessionToken:string, index:number) =>{
        //simulate data http call that needed sessionToken
          return Observable.of(this.dataResponse)
          .map(res => {
            this.dataObservable = null;
            return res.body;
          })
          .delay(1200)
        })
        .map ( data => {
          return data;
        })
        .catch(err => {
          console.log("err in data service", err);
         // return err;
        })
        .share();
    return this.dataObservable;
  }
}

You will still need a flatmap to combine the not dependent observables.

Plunkr: http://plnkr.co/edit/hiA1jP?p=info

Where I got the idea to use switch map: http://blog.thoughtram.io/angular/2016/01/06/taking-advantage-of-observables-in-angular2.html

Answer 3

Edit: someone converted the following text, which I had added as an edit on the question, into an answer, which I appreciate, and understand may be the proper SO thing to do, however I do not consider this an answer because it's clearly not the right way to do it. I would not ever use this code nor would I advise anyone to copy it. Other/better solutions and comments welcome!

---

I was able to solve this with the following. I didn't realize you could flatMap an observable more than once, I assumed results could only be consumed once. So I just flatMap f2Observable twice (sorry, I renamed some stuff in the code since my original post), then zip on all the Observables, then subscribe to that. That Map in the zip to aggregate the values is undesirable because of the type juggling. Other/better solutions and comments welcome! The full code is viewable in a gist. Thank you.

Future<Integer> f2 = executor.submit(new CallToRemoteServiceB());
Observable<Integer> f2Observable = Observable.from(f2);
Observable<Integer> f4Observable = f2Observable
    .flatMap(new Func1<Integer, Observable<Integer>>() {
        @Override
        public Observable<Integer> call(Integer integer) {
            System.out.println("Observed from f2: " + integer);
            Future<Integer> f4 = executor.submit(new CallToRemoteServiceD(integer));
            return Observable.from(f4);
        }       
    });     

Observable<Integer> f5Observable = f2Observable
    .flatMap(new Func1<Integer, Observable<Integer>>() {
        @Override
        public Observable<Integer> call(Integer integer) {
            System.out.println("Observed from f2: " + integer);
            Future<Integer> f5 = executor.submit(new CallToRemoteServiceE(integer));
            return Observable.from(f5);
        }       
    });     

Observable.zip(f3Observable, f4Observable, f5Observable, new Func3<String, Integer, Integer, Map<String, String>>() {
    @Override
    public Map<String, String> call(String s, Integer integer, Integer integer2) {
        Map<String, String> map = new HashMap<String, String>();
        map.put("f3", s);
        map.put("f4", String.valueOf(integer));
        map.put("f5", String.valueOf(integer2));
        return map;
    }       
}).subscribe(new Action1<Map<String, String>>() {
    @Override
    public void call(Map<String, String> map) {
        System.out.println(map.get("f3") + " => " + (Integer.valueOf(map.get("f4")) * Integer.valueOf(map.get("f5"))));
    }       
});     

And this yields me the desired output:

responseB_responseA => 714000