使用Retrofit和Realm纠正RxJava中的流程 [英] Correct flow in RxJava with Retrofit and Realm
问题描述
我正在结合RxJava和Retrofit来实现网络API，并且我将Realm用作数据库.我几乎可以正常工作，但是我想知道这是否是正确的方法和事件流.因此，这是RetrofitApiManager.
public class RetrofitApiManager {
private static final String BASE_URL = "***";
private final ShopApi shopApi;
public RetrofitApiManager(OkHttpClient okHttpClient) {
// GSON INITIALIZATION
Retrofit retrofit = new Retrofit.Builder()
.client(okHttpClient)
.addCallAdapterFactory(RxJavaCallAdapterFactory.create())
.addConverterFactory(GsonConverterFactory.create(gson))
.baseUrl(BASE_URL)
.build();
shopApi = retrofit.create(ShopApi.class);
}
public Observable<RealmResults<Shop>> getShops() {
return shopApi.getShops()
.subscribeOn(Schedulers.io())
.observeOn(AndroidSchedulers.mainThread())
.doOnNext(response -> {
Realm realm = Realm.getDefaultInstance();
realm.executeTransaction(realm1 ->
realm1.copyToRealmOrUpdate(response.shops));
realm.close();
})
.flatMap(response -> {
Realm realm = Realm.getDefaultInstance();
Observable<RealmResults<Shop>> results = realm.where(Shop.class)
.findAllAsync()
.asObservable()
.filter(RealmResults::isLoaded);
realm.close();
return results;
});
}
}
这是在Fragment内获取RealmResults<Shop>的调用.
realm.where(Shop.class)
.findAllAsync()
.asObservable()
.filter(RealmResults::isLoaded)
.first()
.flatMap(shops ->
shops.isEmpty() ? retrofitApiManager.getShops() : Observable.just(shops))
.subscribe(
shops -> initRecyclerView(),
throwable -> processError(throwable));
这是我的问题:
-
像上面的示例一样，这是一种正确的事件链处理方法还是应该以其他方式进行管理?
-
是否可以在
getShops()方法中使用Realm实例并在那里关闭i，还是将其作为参数传递然后以某种方式进行管理会更好?虽然，这个想法似乎对于线程和总是在正确的时间调用Realm.close()有点问题.
1)我会尽力在后台线程上做尽可能多的事情，现在您正在UI线程上做很多工作.
2)
public Observable<RealmResults<Shop>> getShops() {
return shopApi.getShops()
.subscribeOn(Schedulers.io())
.observeOn(AndroidSchedulers.mainThread())
.doOnNext(response -> {
try(Realm realm = Realm.getDefaultInstance()) {
realm.executeTransaction(realm1 ->
realm1.insertOrUpdate(response.shops));
} // auto-close
})
.flatMap(response -> {
try(Realm realm = Realm.getDefaultInstance()) {
Observable<RealmResults<Shop>> results = realm.where(Shop.class)
.findAllAsync()
.asObservable()
.filter(RealmResults::isLoaded);
} // auto-close
return results;
});
}
所有Realm数据都是延迟加载的，因此仅在Realm实例打开时才可用，因此在检索数据后将其关闭很有可能无法正常工作.就您而言，尽管您是在主线程上进行平面映射，所以很可能那里已经有一个打开的实例.
如果您愿意，可以使用copyFromRealm()来获取非托管数据，这些数据可以跨线程移动并且不再与Realm连接，但是它们也会释放实时更新功能并占用更多内存.
它可能会这样做:
public Observable<RealmResults<Shop>> getShops() {
return shopApi.getShops()
.subscribeOn(Schedulers.io())
.doOnNext(response -> {
try(Realm realm = Realm.getDefaultInstance()) {
realm.executeTransaction(realm1 ->
realm1.copyToRealmOrUpdate(response.shops));
} // auto-close
})
.observeOn(AndroidSchedulers.mainThread())
.flatMap(response -> {
Observable<RealmResults<Shop>> results = realm.where(Shop.class)
.findAllAsync()
.asObservable()
.filter(RealmResults::isLoaded);
return results;
});
或者，您可以将网络请求视为副作用，而仅依赖于Realm在发生更改时通知您(当您将网络与DB访问分开时，这是更好的方法，即IMO，例如存储库模式)
public Observable<RealmResults<Shop>> getShops() {
// Realm will automatically notify this observable whenever data is saved from the network
return realm.where(Shop.class).findAllAsync().asObservable()
.filter(RealmResults::isLoaded)
.doOnNext(results -> {
if (results.size() == 0) {
loadShopsFromNetwork();
}
});
}
private void loadShopsFromNetwork() {
shopApi.getShops()
.subscribeOn(Schedulers.io())
.subscribe(response -> {
try(Realm realm = Realm.getDefaultInstance()) {
realm.executeTransaction(r -> r.insertOrUpdate(response.shops));
} // auto-close
});
}
I'm implementing network API with the combination of RxJava and Retrofit, and I use Realm as my database. I got it pretty much working but I'm wondering if it is the correct approach and flow of events. So, here is the RetrofitApiManager.
public class RetrofitApiManager {
private static final String BASE_URL = "***";
private final ShopApi shopApi;
public RetrofitApiManager(OkHttpClient okHttpClient) {
// GSON INITIALIZATION
Retrofit retrofit = new Retrofit.Builder()
.client(okHttpClient)
.addCallAdapterFactory(RxJavaCallAdapterFactory.create())
.addConverterFactory(GsonConverterFactory.create(gson))
.baseUrl(BASE_URL)
.build();
shopApi = retrofit.create(ShopApi.class);
}
public Observable<RealmResults<Shop>> getShops() {
return shopApi.getShops()
.subscribeOn(Schedulers.io())
.observeOn(AndroidSchedulers.mainThread())
.doOnNext(response -> {
Realm realm = Realm.getDefaultInstance();
realm.executeTransaction(realm1 ->
realm1.copyToRealmOrUpdate(response.shops));
realm.close();
})
.flatMap(response -> {
Realm realm = Realm.getDefaultInstance();
Observable<RealmResults<Shop>> results = realm.where(Shop.class)
.findAllAsync()
.asObservable()
.filter(RealmResults::isLoaded);
realm.close();
return results;
});
}
}
And here is the call to get RealmResults<Shop> inside a Fragment.
realm.where(Shop.class)
.findAllAsync()
.asObservable()
.filter(RealmResults::isLoaded)
.first()
.flatMap(shops ->
shops.isEmpty() ? retrofitApiManager.getShops() : Observable.just(shops))
.subscribe(
shops -> initRecyclerView(),
throwable -> processError(throwable));
Here are my questions:
Is it a correct approach to chain events like in the example above or should I manage them in a different way?
Is it OK to use
Realminstance ingetShops()method and close i there or would it be better to pass it as an argument and then manage it somehow? Although, this idea seems to be a bit problematic with threads and callingRealm.close()always at the right time.
1) I would try to do as much as possible on the background thread, right now you are doing a lot of the work on the UI thread.
2)
public Observable<RealmResults<Shop>> getShops() {
return shopApi.getShops()
.subscribeOn(Schedulers.io())
.observeOn(AndroidSchedulers.mainThread())
.doOnNext(response -> {
try(Realm realm = Realm.getDefaultInstance()) {
realm.executeTransaction(realm1 ->
realm1.insertOrUpdate(response.shops));
} // auto-close
})
.flatMap(response -> {
try(Realm realm = Realm.getDefaultInstance()) {
Observable<RealmResults<Shop>> results = realm.where(Shop.class)
.findAllAsync()
.asObservable()
.filter(RealmResults::isLoaded);
} // auto-close
return results;
});
}
All Realm data is lazy-loaded, so it is only available while the Realm instance is open, so closing it after retrieving it has a high chance of not working. In your case though you are flat-mapping on the main thread, so most likely there is already an open instance there.
If you want you can use copyFromRealm() to get unmanaged data out that can be moved across threads and are not connected to Realm anymore, but they will also loose their live update features and take up more memory.
It would probably do this instead:
public Observable<RealmResults<Shop>> getShops() {
return shopApi.getShops()
.subscribeOn(Schedulers.io())
.doOnNext(response -> {
try(Realm realm = Realm.getDefaultInstance()) {
realm.executeTransaction(realm1 ->
realm1.copyToRealmOrUpdate(response.shops));
} // auto-close
})
.observeOn(AndroidSchedulers.mainThread())
.flatMap(response -> {
Observable<RealmResults<Shop>> results = realm.where(Shop.class)
.findAllAsync()
.asObservable()
.filter(RealmResults::isLoaded);
return results;
});
Alternatively you can treat the network request as a side-effect and just depend on Realm notifying you when there is changes (better approach IMO as you separate network from DB access which is e.g. what the Repository pattern is about)
public Observable<RealmResults<Shop>> getShops() {
// Realm will automatically notify this observable whenever data is saved from the network
return realm.where(Shop.class).findAllAsync().asObservable()
.filter(RealmResults::isLoaded)
.doOnNext(results -> {
if (results.size() == 0) {
loadShopsFromNetwork();
}
});
}
private void loadShopsFromNetwork() {
shopApi.getShops()
.subscribeOn(Schedulers.io())
.subscribe(response -> {
try(Realm realm = Realm.getDefaultInstance()) {
realm.executeTransaction(r -> r.insertOrUpdate(response.shops));
} // auto-close
});
}
这篇关于使用Retrofit和Realm纠正RxJava中的流程的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！
