Rxjava Android how to use the Zip operator
Solution 1:
Zip operator strictly pairs emitted items from observables. It waits for both (or more) items to arrive then merges them. So yes this would be suitable for your needs.
I would use Func2 to chain the result from the first two observables.
Notice this approach would be simpler if you use Retrofit since its api interface may return an observable. Otherwise you would need to create your own observable.
// assuming each observable returns response in the form of String
Observable<String> movOb = Observable.create(...);
// if you use Retrofit
Observable<String> picOb = RetrofitApiManager.getService().uploadPic(...),
Observable.zip(movOb, picOb, new Func2<String, String, MyResult>() {
@Override
public MyResult call(String movieUploadResponse, String picUploadResponse) {
// analyze both responses, upload them to another server
// and return this method with a MyResult type
return myResult;
}
}
)
// continue chaining this observable with subscriber
// or use it for something else
Solution 2:
A small example:
val observableOne = Observable.just("Hello", "World")
val observableTwo = Observable.just("Bye", "Friends")
val zipper = BiFunction<String, String, String> { first, second -> "$first - $second" }
Observable.zip(observableOne, observableTwo, zipper)
.subscribe { println(it) }
This will print:
Hello - Bye
World - Friends
In BiFunction<String, String, String> the first String the type of the first observable, the second String is the type of the second observable, the third String represents the type of the return of your zipper function.
I made small example with that calls two real endpoints using zip in this blog post
Solution 3:
Here I have an example that I did using Zip in asynchronous way, just in case you´re curious
/**
* Since every observable into the zip is created to subscribeOn a diferent thread, it´s means all of them will run in parallel.
* By default Rx is not async, only if you explicitly use subscribeOn.
*/
@Test
public void testAsyncZip() {
scheduler = Schedulers.newThread();
scheduler1 = Schedulers.newThread();
scheduler2 = Schedulers.newThread();
long start = System.currentTimeMillis();
Observable.zip(obAsyncString(), obAsyncString1(), obAsyncString2(), (s, s2, s3) -> s.concat(s2)
.concat(s3))
.subscribe(result -> showResult("Async in:", start, result));
}
/**
* In this example the the three observables will be emitted sequentially and the three items will be passed to the pipeline
*/
@Test
public void testZip() {
long start = System.currentTimeMillis();
Observable.zip(obString(), obString1(), obString2(), (s, s2, s3) -> s.concat(s2)
.concat(s3))
.subscribe(result -> showResult("Sync in:", start, result));
}
public void showResult(String transactionType, long start, String result) {
System.out.println(result + " " +
transactionType + String.valueOf(System.currentTimeMillis() - start));
}
public Observable<String> obString() {
return Observable.just("")
.doOnNext(val -> {
System.out.println("Thread " + Thread.currentThread()
.getName());
})
.map(val -> "Hello");
}
public Observable<String> obString1() {
return Observable.just("")
.doOnNext(val -> {
System.out.println("Thread " + Thread.currentThread()
.getName());
})
.map(val -> " World");
}
public Observable<String> obString2() {
return Observable.just("")
.doOnNext(val -> {
System.out.println("Thread " + Thread.currentThread()
.getName());
})
.map(val -> "!");
}
public Observable<String> obAsyncString() {
return Observable.just("")
.observeOn(scheduler)
.doOnNext(val -> {
System.out.println("Thread " + Thread.currentThread()
.getName());
})
.map(val -> "Hello");
}
public Observable<String> obAsyncString1() {
return Observable.just("")
.observeOn(scheduler1)
.doOnNext(val -> {
System.out.println("Thread " + Thread.currentThread()
.getName());
})
.map(val -> " World");
}
public Observable<String> obAsyncString2() {
return Observable.just("")
.observeOn(scheduler2)
.doOnNext(val -> {
System.out.println("Thread " + Thread.currentThread()
.getName());
})
.map(val -> "!");
}
You can see more examples here https://github.com/politrons/reactive
Solution 4:
zip operator allow you to compose a result from results of two different observable.
You 'll have to give am lambda that will create a result from datas emitted by each observable.
Observable<MovieResponse> movies = ...
Observable<PictureResponse> picture = ...
Observable<Response> response = movies.zipWith(picture, (movie, pic) -> {
return new Response("description", movie.getName(), pic.getUrl());
});
Solution 5:
i have been searching for a simple answer on how to use the Zip operator, and what to do with the Observables i create to pass them to it, i was wondering if i should call subscribe() for every observable or not, non of these answers were simple to find, i had to figure it out by my self, so here is a simple example for using Zip operator on 2 Observables :
@Test
public void zipOperator() throws Exception {
List<Integer> indexes = Arrays.asList(0, 1, 2, 3, 4);
List<String> letters = Arrays.asList("a", "b", "c", "d", "e");
Observable<Integer> indexesObservable = Observable.fromIterable(indexes);
Observable<String> lettersObservable = Observable.fromIterable(letters);
Observable.zip(indexesObservable, lettersObservable, mergeEmittedItems())
.subscribe(printMergedItems());
}
@NonNull
private BiFunction<Integer, String, String> mergeEmittedItems() {
return new BiFunction<Integer, String, String>() {
@Override
public String apply(Integer index, String letter) throws Exception {
return "[" + index + "] " + letter;
}
};
}
@NonNull
private Consumer<String> printMergedItems() {
return new Consumer<String>() {
@Override
public void accept(String s) throws Exception {
System.out.println(s);
}
};
}
the printed result is :
[0] a
[1] b
[2] c
[3] d
[4] e
the final answers to the questions that where in my head were as follows
the Observables passed to the zip() method just need to be created only, they do not need to have any subscribers to them, only creating them is enough ... if you want any observable to run on a scheduler, you can specify this for that Observable ... i also tried the zip() operator on Observables where they should wait for there result, and the Consumable of the zip() was triggered only when both results where ready (which is the expected behavior)