Caching với Realm và RxJava

Trước khi sử dụng Realm, ta có một câu lệnh request API đơn giản:

@GET("weather?units=metric")
Observable<WeatherResponse> getWeather(@Query("q") String city, 
                                       @Query("appid") String apiKey);

Đây là một API call đơn giản sử dụng Retrofit + RxJavaRx. getWeather trả về một WeatherResponse observable, WeatherResponse là một đối tượng được tạo ra từ đoạn JSON trả về.

Đoạn mã request API trên bằng cách subcribe Observable<WeatherResponse> trên IO thread, sau khi request xong, kết quả trả về được chuyển về main thread.

// Request API data on IO Scheduler
service.getWeather("Berlin", getString(R.string.api_key))
    .subscribeOn(Schedulers.io())
    // Read results in Android Main Thread (UI)
    .observeOn(AndroidSchedulers.mainThread())
    .subscribe(this::display, this::processError);

Cách này được nói ngắn gọn như sau: Dữ liệu từ trên server được hiển thị trực tiếp lên ứng dụng thông qua Restful API request (REST Call).

Giờ ta sẽ cho thêm Realm vào giữa đoạn lệnh từ Rest Call đến hiển thị lên giao diện. Response trả về sẽ được lưu vào database trước khi nó được hiển thị. Để làm việc này, trước tiên ta phải tạo một WeatherRealm object để quản lý việc lưu trữ dữ liệu.

Các bước cơ bản như sau:

1. Request API
2. Lấy response trả về lưu vào DB
3. Đọc response từ DB
4. Hiển thị response.

Từ lần thứ hai request API, đầu tiên ta sẽ hiển thị response cũ trước đó đã được lưu vào database và sau đó nhận lại dữ liệu mới nhất từ API.

Các bước cơ bản như sau:

1. Lấy dữ liệu từ DB (Cached data)
2. Hiển thị dữ liệu
3. Request API
4. Lấy response trả về lưu vào DB
5. Lấy dữ liệu từ DB
6. Hiển thị dữ liệu

Các bước trên đã tương đối ổn rồi, giờ ta sẽ thêm phần xử lý luồng

Để tránh việc bị block UI ta sẽ không ghi data vào DB trên main thread, ngoài ra, để tránh việc block network call trên IO thread , ta sẽ xử lý việc này trên Computation thread

Ta sẽ xử lý việc ghi dữ liệu vào DB trên Computation thread để tránh việc bị block UI khi chạy trên Main Thread hoặc làm chậm quá trình request API khi chạy trên IO Thread

Đầu tiên, ta tạo WeatherRealm extend RealmObject để xử lý việc đọc, ghi dữ liệu:

public class WeatherRealm extends RealmObject {
    @PrimaryKey
    private String name;
    private Double temp;

    public String getName() {
        return name;
    }

    public void setName(String name) {
        this.name = name;
    }

    public Double getTemp() {
        return temp;
    }

    public void setTemp(Double temp) {
        this.temp = temp;
    }
}

Lấy dữ liệu từ DB

private Realm realmUI;

protected void onCreate(Bundle savedInstanceState) {
    // ...
    realmUI = Realm.getDefaultInstance();
    // ...
}

private WeatherRealm readFromRealm(String name) {
    return findInRealm(realmUI, name);
}

private WeatherRealm findInRealm(Realm realm, String name) {
    return realm.where(WeatherRealm.class).equalTo("name", name).findFirst();
}

@Override
protected void onDestroy() {
    super.onDestroy();
    realmUI.close();
}

Ta tạo một instance của Realm để xử lý việc đọc dữ liệu trên UI Thread, biến này được gọi là realmUI, đừng quên close nó trong hàm onDestroy() để tránh việc leak bộ nhớ.

Hàm findInRealm chứa câu lệnh query tìm các object trong database theo tên thành phố.

realm.where(WeatherRealm.class).equalTo("name", name).findFirst();

Lưu trữ dữ liệu vào DB

private String writeToRealm(WeatherResponse weatherResponse) {
    Realm realm = Realm.getDefaultInstance();
    realm.executeTransaction(transactionRealm -> {
        WeatherRealm weatherRealm = findInRealm(transactionRealm, weatherResponse.getName());
        if (weatherRealm == null)
            weatherRealm = transactionRealm.createObject(WeatherRealm.class, weatherResponse.getName());
        weatherRealm.setTemp(weatherResponse.getMain().getTemp());
    });
    realm.close();
    return weatherResponse.getName();
}

Để lưu dữ liệu vào DB ta cần một instance khác của Realm trong computation threads của RxJava. Ta sử dụng ExecuteTransaction để thực thi lệnh ghi dữ liệu vào DB, việc sử dụng ExecuteTransaction sẽ tự động commit transaction và nếu không thành công, sẽ tự động close. Ta cũng sẽ thực hiện việc gọi hàm findInRealm trong transtactionRealm bởi vì ta không thể sử dụng RealmObject mà nằm trong một thread khác.

Nếu object không tồn tại, ta sẽ tạo mới với name là khoá chính, nếu object đã tốn tại ta sẽ cập nhật lại dữ liệu thời tiết cho object này.

Khi transaction hoàn thành, mọi thay đổi đã được lưu lại. Đừng quên đóng Realm instance hiện tại.

Cuối cùng ta truyền name(khoá chính) vào câu lệnh tiếp theo trong chuỗi.

Tạo Observable

Giờ đã là lúc để kết hợp lại tất cả mọi thứ với nhau:

String name = "Berlin";
	

	// Request API data on IO Scheduler
	Observable<WeatherRealm> observable =
	        service.getWeather(name, getString(R.string.api_key))
	                // One second delay for demo purposes
	                .delay(1L, java.util.concurrent.TimeUnit.SECONDS)
	                .subscribeOn(Schedulers.io())
	                // Write to Realm on Computation scheduler
	                .observeOn(Schedulers.computation())
	                .map(this::writeToRealm)
	                // Read results in Android Main Thread (UI)
	                .observeOn(AndroidSchedulers.mainThread())
	                .map(this::readFromRealm);
	

	// Read any cached results
	WeatherRealm cachedWeather = readFromRealm(name);
	if (cachedWeather != null)
	    // Merge with the observable from API
	    observable = observable.mergeWith(Observable.just(cachedWeather));
	

	// Subscription happens on Main Thread
	subscription = observable.subscribe(this::display, this::processError);

Đầu tiên, ta tạo một Observable xử lý việc call API, để thấy rõ hơn sự thay đổi, ta sẽ delay 1s.

Tiếp theo ta sẽ lưu data nhận được vào DB trong Computation scheduler

// Write to Realm on Computation scheduler                
.observeOn(Schedulers.computation())                
.map(this::writeToRealm)

Đây là đoạn xử lý việc lưu trữ dữ liệu vào DB với đầu vào là WeatherResponse và đầu ra là name(khoá chính) của WeatherRealm. Nhớ rằng ta sẽ xử lý việc này trên computation scheduler.

Thứ hai, ta sẽ đọc lại dữ liệu đã được cập nhật trước đó bằng khoá chính, lưu ý ta sẽ thực hiện việc này trên main thread vì dữ liệu này sẽ được hiển thị trên UI.

// Read results in Android Main Thread (UI) 
.observeOn(AndroidSchedulers.mainThread()) 
.map(this::readFromRealm);

Kết hợp với dữ liệu đã được cached

// Read any cached results
WeatherRealm cachedWeather = readFromRealm(name);
if (cachedWeather != null) 
    // Merge with the observable from API 
    observable = observable
                 .mergeWith(Observable.just(cachedWeather));

Để làm việc này ta sẽ sử dụng mergeWith. Ta sẽ tạo ra một Observable với cachedWeather từ trong DB. Hàm mergeWith sẽ xử lý việc kết hợp giữa data lấy từ DB và data lấy từ API.

Chạy thử

Sau khi chạy ta sẽ thấy hàm onNext được gọi ra 2 lần, một lần là từ dữ liệu cache và một lần là dữ liệu từ API trả về. Tất nhiên lần gọi đầu tiên gần như ngay lập tức vì dữ liệu đã được lưu sẵn dưới DB, lần gọi thứ 2 sẽ chậm hơn sau 1 giây.

09-25 19:11:59.022 ... MainActivity: City: Berlin, Temp: 18.22
09-25 19:12:00.275 ... MainActivity: City: Berlin, Temp: 18.22

Nếu người dùng có kết nối chậm, thì họ ngay lập tức sẽ được nhìn thấy dữ liệu đã được cache trên giao diện thay vì một thanh loading... đang chạy.

Nếu người dùng không có mạng, họ vẫn có thể thấy được dữ liệu đã được cache và sau thông báo lỗi của Retrofit.

Bài viết gốc https://medium.com/@Miqubel/caching-with-realm-and-rxjava-80f48c5f5e37