使用RXJava 2对Realm进行异步读/写 [英] Asynchronous read / write to Realm using RXJava 2

问题描述

我首次使用 RXJava 2实现异步操作

目标：

获取 json 来自服务器的数据库 Retrofit2 。如果成功，则将数据写入 Realm 并在记录后立即获取数据并发送到适配器 RecyclerView 。

Get json data from the server with library Retrofit2. If successful, then write the data to Realm and immediately after the record get the data back and send to the adapter of RecyclerView.

所以，我用这种方式意识到了这一切：

So, I realized all this in this way:

private void fetchChatsFromNetwork(int count, AccessDataModel accessDataModel) {

    String accessToken = accessDataModel.getAccessToken();

    MyApplication.getRestApi().getChats(count, accessToken, Constants.api_version)
            .subscribeOn(Schedulers.io())
            .observeOn(AndroidSchedulers.mainThread())
            .subscribeWith(new DisposableSubscriber<ChatsModel>() {
                @Override
                public void onNext(ChatsModel chatsModel) {
                    if (chatsRepository.hasData()) {

                        chatsRepository.updateChatsData(chatsModel)
                                .subscribe(new DisposableObserver<ChatsModel>() {
                                    @Override
                                    public void onNext(ChatsModel localChatsModel) {
                                        Log.d(TAG, "DO, onSuccess updated!");
                                        iGetChatsCallback.onGetChatsSuccess(localChatsModel);
                                    }

                                    @Override
                                    public void onError(Throwable e) {
                                        Log.d(TAG, "DO, onError when update!");
                                        iGetChatsCallback.onGetChatsError(e.getMessage());
                                    }

                                    @Override
                                    public void onComplete() {
                                        dispose();
                                        Log.d(TAG, "DO, onComplete!");
                                    }
                                });

                    } else {
                        chatsRepository.insertChatsData(chatsModel)
                                .subscribe(new DisposableObserver<ChatsModel>() {
                                    @Override
                                    public void onNext(ChatsModel localChatsModel) {
                                        iGetChatsCallback.onGetChatsSuccess(localChatsModel);
                                        Log.d(TAG, "DO, onSuccess inserted!");
                                    }

                                    @Override
                                    public void onError(Throwable e) {
                                        iGetChatsCallback.onGetChatsError(e.getMessage());
                                        Log.d(TAG, "DO, onError when inserting!");
                                    }

                                    @Override
                                    public void onComplete() {
                                        dispose();
                                        Log.d(TAG, "DO, onComplete!");
                                    }
                                });
                    }
                }

                @Override
                public void onError(Throwable t) {
                    Log.d(TAG, "onError" + t.getMessage());
                }

                @Override
                public void onComplete() {
                    Log.d(TAG, "onComplete");
                }
            });
}

我在 Realm 在订阅者 MyApplication.getRestApi（）的 onNext（）方法中.GetChats（）

I write the data in Realm in the onNext() method of the subscriber MyApplication.getRestApi().GetChats()

以下是条目代码：

public Observable<ChatsModel> updateChatsData(final ChatsModel chatsModel) {

    return Observable.create(new ObservableOnSubscribe<ChatsModel>() {
        @Override
        public void subscribe(ObservableEmitter<ChatsModel> e) throws Exception {
            if (chatsModel != null) {
                realm.executeTransactionAsync(
                        realm -> realm.copyToRealmOrUpdate(chatsModel),
                        () -> {
                            Log.d(LOG_TAG, "Data success updated!");
                            ChatsModel localChatsModel = getAllChatsData();
                            e.onNext(localChatsModel);
                            e.onComplete();
                        },
                        error -> {
                            Log.d(LOG_TAG, "Update data failed!");
                            e.onError(error);
                        });
            }

        }
    });

}

updateChatsData（）异步写入并在另一个类中声明。

The updateChatsData() writes asynchronously and declared in another class.

正如你所看到的， fetchChatsFromNetwork（）方法写得很麻烦，所以在我看来

As you can see, the fetchChatsFromNetwork() method is written cumbersome or so it seems to me

问题：

我是否正确，如果不是，那怎么办？是正确的吗？

Whether I am doing right or not, if not, how would that be correct?

推荐答案

private void fetchChatsFromNetwork(int count, AccessDataModel accessDataModel) {    
    String accessToken = accessDataModel.getAccessToken();
    Single<ChatsModel> chats = MyApplication.getRestApi().getChats(count, accessToken, Constants.api_version);
    chats.doOnNext((chats) -> {
        chatsRepository.insertOrUpdate(chats);
    }).subscribeOn(Schedulers.io())
    .subscribe();
}

和

public void updateChatsData(final ChatsModel chatsModel) {
    try(Realm realm = Realm.getDefaultInstance()) {
        realm.executeTransaction(r -> {
            r.insertOrUpdate(chatsModel);
        });
    }
}

和

public Flowable<List<ChatsModel>> getAllChatsData(Realm realm) {
    RealmQuery<ChatsModel> query = realm.where(ChatsModel.class);
    if(realm.isAutoRefresh()) {
        return query.findAllAsync().asFlowable().filter(RealmResults::isLoaded);
    } else {
        return Flowable.just(query.findAll());
    }
}

这篇关于使用RXJava 2对Realm进行异步读/写的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！