Using RxJava to quickly obtain massive data

Imagine that when you need some dynamic data, you only need to request the network every time. However, a more efficient approach is to cache the data obtained from the network to disk or memory.

Specifically, the plan is as follows:

1. Occasional networking operations are only used to obtain *** data.

2. Read data as quickly as possible (by fetching previously cached network data).

I will implement this plan by using RxJava.

Basic Mode

Create an Observable<Data> for each data source (network, disk, and memory) and use the concat() and first() operators to construct a simple implementation.

The concat() operator holds multiple Observable objects and concatenates them into a queue in order. The first() operator only takes and emits the first event from the concatenated queue. Therefore, if you use concat().first(), no matter how many data sources there are, only the first event will be retrieved and emitted.

 // Our sources (left as an exercise for the reader)  
 Observable<Data> memory = ...;
 Observable<Data> disk = ...;
 Observable<Data> network = ...; 
 
 // Retrieve the first source with data  
 Observable<Data> source = Observable
  .concat(memory, disk, network)
 .first();

The key to this pattern is that the concat() operator subscribes to all Observable sources only when data is needed. Since the first() operator stops retrieving the queue early, there is no need to access the slower source if there is cached data. In other words, if memory returns the result, there is no need to worry about disk and network being accessed. Conversely, if there is no data in memory or disk, the network request is performed.

Note that the Observable data sources held by concat() are retrieved one by one in order.

Persistent Data

Obviously, the next step is to cache data. If you don't cache the results of network requests to disk, and cache the results of disk accesses to memory, then it's not called caching at all. The next code to write is to persist the network data.

My solution is to have each data source save or cache the data after sending the event.

 Observable<Data> networkWithSave = network.doOnNext( new Action1<Data>() {
 @Override   public   void call(Data data) {
 saveToDisk(data);
 cacheInMemory(data);
 }
 }); 
 
 Observable<Data> diskWithCache = disk.doOnNext( new Action1<Data>() {
 @Override   public   void call(Data data) {
 cacheInMemory(data);
 }
 });

Now, if you use networkWithSave and diskWithCache, the data will be automatically saved after loading.

(Another advantage of this strategy is that networkWithSave and diskWithCache can be used anywhere, not just in our multi-data model.)

Stale data

Unfortunately, right now the code we have that saves the data is overdoing it. It always returns the same data, regardless of whether it's out of date or not. We want to occasionally connect to the server and grab the latest data.

The solution is to use the first() operator to filter, which is to set it to reject worthless data.

 Observable<Data> source = Observable
    .concat(memory, diskWithCache, networkWithSave)
    .first( new Func1<Data, Boolean>() {
 @Override   public Boolean call(Data data) {
 return data.isUpToDate();
 }
 });

Now, we only need to send events that are determined to be the latest data. Therefore, as long as the data of one data source expires, we will continue to retrieve the next data source until the latest data is found.

Comparison of first() and takeFirst() Operators

For this design pattern, either first() or takeFirst() operators can be used.

The difference between the two calling methods is that if the data of all data sources are expired and no valid data is sent as an event, first() will throw a NoSuchElementException (Translator's note: the first() operator always returns false), while the takeFirst() operator will directly call the completion operation without throwing any exception.

Which operator to use depends entirely on whether you need to explicitly handle missing data.

Code Sample

You can check out a sample implementation of all the above code here: https://github.com/dlew/rxjava-multiple-sources-sample.

If you need a real-world example, check out the Gfycat App, which uses this pattern when fetching data. The project doesn’t use all of the features shown above (because it doesn’t need to), but it demonstrates the basic usage of concat().first().