1. Overview
concat() and merge() are two powerful operators used to combine multiple Observable instances in RxJava.
concat() emits items from each Observable sequentially, waiting for each to complete before moving to the next, while merge() concurrently emits items from all Observable instances as they’re produced.
In this tutorial, we’ll explore scenarios in which concat() and merge() show similar and different behaviors.
2. Synchronous Sources
The concat() and merge() operators behave exactly the same when both sources are synchronous. Let’s simulate this scenario to understand it better.
2.1. Scenario Setup
Let’s start by using the Observable.just() factory method to create three synchronous sources:
Observable<Integer> observable1 = Observable.just(1, 2, 3);
Observable<Integer> observable2 = Observable.just(4, 5, 6);
Observable<Integer> observable3 = Observable.just(7, 8, 9);
Further, let’s create two subscribers, namely, testSubscriberForConcat and testSubscriberForMerge:
TestSubscriber<Integer> testSubscriberForConcat = new TestSubscriber<>();
TestSubscriber<Integer> testSubscriberForMerge = new TestSubscriber<>();
Great! We’ve got everything we need to test the scenario.
2.2. concat() and merge()
First, let’s apply the concat() operator and subscribe the resultant Observable with testSubscriberForConcat:
Observable.concat(observable1, observable2, observable3)
.subscribe(testSubscriberForConcat);
Further, let’s verify that the emissions are in order where items from observable1 appear before observable2, and observable2 before observable3:
testSubscriberForConcat.assertValues(1, 2, 3, 4, 5, 6, 7, 8, 9);
Similarly, we can apply the merge() operator and subscribe the outcome with the testSubscriberForMerge:
Observable.merge(observable1, observable2, observable3).subscribe(testSubscriberForMerge);
Next, let’s verify that the emissions through merge follow the same order as that from the concatenation:
testSubscriberForMerge.assertValues(1, 2, 3, 4, 5, 6, 7, 8, 9);
Lastly, we must note that synchronous Observable instances emit all items immediately and then signal completion. Further, each Observable completes its emission before the next one starts. Consequently, both operators process each Observable sequentially, producing the same output.
As such, whether the sources are synchronous or asynchronous, the general rule is that if we need to maintain the order of emissions by source, we should use concat(). On the other hand, if we want to combine items as they’re emitted from multiple sources, we should use merge().
3. Predictable Asynchronous Sources
In this section, let’s simulate a scenario with asynchronous sources where the order of emissions is predictable.
3.1. Scenario Setup
Let’s create two asynchronous sources, namely, observable1 and observable2:
Observable<Integer> observable1 = Observable.interval(100, TimeUnit.MILLISECONDS)
.map(i -> i.intValue() + 1)
.take(3);
Observable<Integer> observable2 = Observable.interval(30, TimeUnit.MILLISECONDS)
.map(i -> i.intValue() + 4)
.take(7);
We must notice that emissions from observable1 arrive after 100ms, 200ms, and 300ms, respectively. On the other hand, emissions from observable2 arrive at intervals of 30ms.
Now, let’s create the testSubscriberForConcat and testSubscriberforMerge as well:
TestSubscriber<Integer> testSubscriberForConcat = new TestSubscriber<>();
TestSubscriber<Integer> testSubscriberForMerge = new TestSubscriber<>();
Fantastic! We’re ready to test this scenario.
3.2. concat() vs. merge()
First, let’s apply the concat() operator and call subscribe() with testSubscribeForConcat:
Observable.concat(observable1, observable2)
.subscribe(testSubscriberForConcat);
Next, we must call the awaitTerminalEvent() method to ensure that all emissions are received:
testSubscriberForConcat.awaitTerminalEvent();
Now, we can validate that the result contains all items from observable1 followed by all items from observable2:
testSubscriber.assertValues(1, 2, 3, 4, 5, 6, 7, 8, 9, 10);
Further, let’s apply the merge() operator and call subscribe() with testSubscriberForMerge:
Observable.merge(observable1, observable2)
.subscribe(testSubscriberForMerge);
Lastly, let’s wait for the emissions and check the emitted values:
testSubscriberForMerge.awaitTerminalEvent();
testSubscriberForMerge.assertValues(4, 5, 6, 1, 7, 8, 9, 2, 10, 3);
The result contains all items interleaved together in the order of their actual emission from observer1 and observer2.
4. Asynchronous Sources With Race Conditions
In this section, we’ll simulate a scenario with two asynchronous sources in which the order of combined emission is fairly unpredictable.
4.1. Scenario Setup
First, let’s create the two asynchronous sources with exactly the same delay:
Observable<Integer> observable1 = Observable.interval(100, TimeUnit.MILLISECONDS)
.map(i -> i.intValue() + 1)
.take(3);
Observable<Integer> observable2 = Observable.interval(100, TimeUnit.MILLISECONDS)
.map(i -> i.intValue() + 4)
.take(3);
We know that one emission from each source arrives after 100ms, 200ms, and 300ms. However, we can’t predict the exact order because of the race conditions.
Next, let’s create two test subscribers:
TestSubscriber<Integer> testSubscriberForConcat = new TestSubscriber<>();
TestSubscriber<Integer> testSubscriberForMerge = new TestSubscriber<>();
Perfect! We’re good to go now.
4.2. concat() vs. merge()
First, let’s apply the concat() operator, followed by a subscription to testSubscribeForConcat:
Observable.concat(observable1, observable2)
.subscribe(testSubscriberForConcat);
testSubscriberForConcat.awaitTerminalEvent();
Now, let’s verify that the outcome from the concat() operator remains unchanged:
testSubscriberForConcat.assertValues(1, 2, 3, 4, 5, 6);
Further, let’s merge() and subscribe with testSubscriberForMerge:
Observable.merge(observable1, observable2)
.subscribe(testSubscriberForMerge);
testSubscriberForMerge.awaitTerminalEvent();
Next, let’s accumulate all the emissions in a list and verify that it contains all values:
List<Integer> actual = testSubscriberForMerge.getOnNextEvents();
List<Integer> expected = Arrays.asList(1, 2, 3, 4, 5, 6);
assertTrue(actual.containsAll(expected) && expected.containsAll(actual));
Lastly, let’s also log the emissions to see it in action:
21:05:43.252 [main] INFO actual emissions: [4, 1, 2, 5, 3, 6]
We can receive a different order for different runs.
5. Conclusion
In this article, we saw how the concat() and merge() operators in RxJava handle synchronous and asynchronous data sources. Further, we compared scenarios involving predictable and unpredictable patterns of emissions, emphasizing the differences between the two operators.
As always, the code from this article is available over on GitHub.