我正在尝试执行以下操作；使用 Retrofit 将云数据库同步到设备上的本地 SqLite 数据库（房间）。DB 可能会变大，大约有 100,000 个或更多寄存器，因此同步过程可能需要一些时间。所以它发送第一个Retrofit请求来获取寄存器的数量，这样它就可以计算出总页数，然后它会发送多个Retrofit Request，从API中获取所有数据，每次请求后，它将数据保存到房间。
现在，我在组合两个 RxJava 调用或进程时遇到了麻烦，同样在第二个 RxJava 进程上，在 Retrofit 调用之后，有一个对象列表的房间插入，但是在洞进程结束后，我注意到没有插入了 100% 的所有记录，每次我运行该进程时，插入的记录数都会发生变化，大约是 80% - 98%，但从来没有 100%，即使所有的 Retrofit 调用都已发送。
请帮助我：
- 如何在一个 RxJava 调用中完成所有过程，而不是像我现在拥有的 2 个？
- 如何将 100% 的记录插入到 Room？
按照代码：
摇篮
def room_version = "2.2.5"
//RxJava 2
implementation "io.reactivex.rxjava2:rxjava:2.2.19"
implementation 'io.reactivex.rxjava2:rxandroid:2.1.1'
//Retrofit
implementation 'com.squareup.retrofit2:retrofit:2.8.1'
implementation 'com.squareup.retrofit2:converter-gson:2.8.1'
//Retrofit2 Adapter for RxJava 2
implementation "com.squareup.retrofit2:adapter-rxjava2:2.8.1"
//okhttp3 Logging Interceptor
implementation "com.squareup.okhttp3:logging-interceptor:4.5.0"
//Room
implementation "androidx.room:room-runtime:$room_version"
annotationProcessor "androidx.room:room-compiler:$room_version"
//RxJava support for Room
implementation "androidx.room:room-rxjava2:$room_version"
ItemSyncDetails
...
public class ItemSyncDetails {
@SerializedName("CurrentPage")
int currentPage;
@SerializedName("PageCount")
int pageCount;
@SerializedName("PageSize")
int pageSize;
@SerializedName("RecordCount")
int recordCount;
@SerializedName("Results")
List<Item> mItemList;
...
}
物道
注意：我没有使用 Observer/Flowable/Maybe/Single，因为我已经能够使其与 RxJava 一起使用
import io.reactivex.Flowable;
@Dao
public interface ItemDao {
@Insert(onConflict = OnConflictStrategy.REPLACE)
long insert(Item item);
@Insert(onConflict = OnConflictStrategy.REPLACE)
List<Long> insertAll(List<Item> items);
...
数据接口
import io.reactivex.rxjava3.core.Observable;
...
public interface DataApi {
@GET("item")
Observable<ItemSyncDetails> getItemsByPage(
@Query("pageSize") Integer pageSize,
@Query("currentPage") Integer currentPage,
@Query("sortBy") Integer sortBy
);
项目存储库
import io.reactivex.Observable;
...
public class ItemRepository {
...
public ItemRepository(Application application) {
mDataApi = RetrofitClient.getRetrofitInstance("http://192.168.1.100").create(DataApi.class);
RfidDatabase db = RfidDatabase.getAppDatabase(application);
itemDao = db.itemDao();
itemList = itemDao.getAllItems();
inserts = 0;
}
public List<Long> insertAllLocal (List<Item> itemList) {
List<Long> items = itemDao.insertAll(itemList);
inserts += items.size();
Log.i(TAG, "************insertAllLocal - ItemRepository: " + inserts + "*************");
Log.i(TAG, "************insertAllLocal - ItemRepository: " + items);
return items;
}
public Observable<ItemSyncDetails> getRecordsCount(){
return mDataApi.getItemsByPage(1,1,1);
}
public Observable<ItemSyncDetails> getItemsPerPage(int pageSize,int currentPage){
return mDataApi.getItemsByPage(pageSize,currentPage,1);
}
...
SyncConfigFragment
import io.reactivex.Observable;
import io.reactivex.android.schedulers.AndroidSchedulers;
import io.reactivex.disposables.CompositeDisposable;
import io.reactivex.functions.Function;
import io.reactivex.schedulers.Schedule
...
public class SyncConfigFragment extends Fragment {
private ItemViewModel itemViewModel;
private ImageView imageSyncItems;
private ProgressDialog progressDialog;
private TextView tvSyncDescriptionItems;
private DataApi service;
private ItemSyncDetails mItemSyncDetails;
private List<Item> mItemlist;
private CompositeDisposable mCompositeDisposable;
private int mNumPages;
private int syncProgress;
...
@Override
public View onCreateView(LayoutInflater inflater, ViewGroup container, Bundle savedInstanceState) {
View view = inflater.inflate(R.layout.fragment_config_sync,container, false);
progressDialog = new ProgressDialog(getActivity());
sharedPref = getActivity().getSharedPreferences(
getString(R.string.sharepref_filename), Context.MODE_PRIVATE);
mItemlist = new ArrayList<Item>();
mCompositeDisposable = new CompositeDisposable();
itemViewModel = ViewModelProviders.of(this).get(ItemViewModel.class);
tvSyncDescriptionItems = view.findViewById(R.id.tvDescriptionSyncItems);
if(sharedPref.contains("last_sync_item")) {
tvSyncDescriptionItems.setText("Última actualización " + sharedPref.getString("last_sync_item",""));
} else{
tvSyncDescriptionItems.setText("No se ha Sincronizado");
}
imageSyncItems = view.findViewById(R.id.imageViewSyncItems);
imageSyncItems.setOnClickListener(clickListener);
return view;
}
private View.OnClickListener clickListener = new View.OnClickListener() {
public void onClick(View v) {
if (v.equals(imageSyncItems)) {
//If I uncomment the next line it does not work
//mCompositeDisposable.add(
mNumPages = 0;
syncProgress = 0;
showProgressDialog("Items");
getRecordsCount();
//); Closing round bracket for mCompositeDisposable
}
}
};//End View.OnClickListener
private void getRecordsCount(){
itemViewModel.getRecordsCount()
.subscribeOn(Schedulers.io())
.retry(3)
.observeOn(AndroidSchedulers.mainThread())
.subscribe(this::HandleResults, this::handleError,this::getNumPagesHandlerComplete );
}
private void HandleResults(ItemSyncDetails itemSyncDetails) {
this.mItemSyncDetails = itemSyncDetails;
int pageSize = 100;
int numPages = itemSyncDetails.getRecordCount()/pageSize;
if (itemSyncDetails.getRecordCount() < pageSize || itemSyncDetails.getRecordCount()%pageSize != 0){
numPages++;
}
this.mNumPages = numPages;
}
private void getNumPagesHandlerComplete() {
getAllRecords(mNumPages);
}
private void handleError(Throwable throwable) {
tvSyncDescriptionItems.setText("**********Error de conexión...");
closeProgressDialog();
}
private void getAllRecords(int numPages){
//numPages: total of pages are the number of times to send the request to API
Observable.range(1, numPages)
.flatMap(i -> itemViewModel.getItemsPerPage(100,i))
.map(new Function<ItemSyncDetails, Integer>() {
@Override
public Integer apply(ItemSyncDetails itemSyncDetails) throws Throwable {
return itemViewModel.insertAllLocal(itemSyncDetails.getItemList()).size();
}
})
.subscribeOn(Schedulers.newThread())
.observeOn(AndroidSchedulers.mainThread())
.subscribe(this::getAllHandleResults, this::handleError,this::handleComplete);
}
private void getAllHandleResults(Integer i) {
progressDialog.setProgress(getProgress(i));
}
private void handleComplete() {
//last request finished
closeProgressDialog();
}
private int getProgress(int newItems){
syncProgress += newItems;
int progress = 0;
if (syncProgress == mItemSyncDetails.getRecordCount()){
progress = 100;
} else {
progress = (100 * syncProgress)/mItemSyncDetails.getRecordCount();
}
return progress;
}
...
}
http://192.168.1.10:82/api/v1.0/item?pageSize=1¤tPage=1&sortBy=1
注意：页面大小可能会改变，我使用的是每页 100 个项目的固定大小。
{
Results: [
{
epc: "202020202020202030303031",
barcode: "0001",
name: "Televisor Samnsung",
description: "0001",
creation_date: "2020-02-26T10:55:06",
last_update: "2020-02-26T10:55:06",
last_seen: "2020-02-26T10:55:06",
brand: "Samnsung",
serial_number: "0001",
parent: "",
fk_category: 1,
responsable: "",
purchase_date: "2020-02-26T10:55:06",
cost: 0,
fk_location: 1008,
fk_item_state: 1,
inventory_date: "2020-02-26T10:55:06"
}
],
CurrentPage: 1,
PageCount: 65565,
PageSize: 1,
RecordCount: 65565
}