android - Rxjava 3、retrofit2 和多次调用问题

Question

我在实现 RxJava3/Retrofit2 时遇到了麻烦，我需要完成的是：

总体思路是将云数据库同步到设备上的 SqLite 数据库（房间）。 数据库可能会变大，大约 100,000 个寄存器或更多，因此同步过程可能需要一些时间，我的第一次尝试是在一个请求中完成并获取所有寄存器，然后将它们保存到 SqLite（房间），但这在某些情况下，取决于设备产生了一些内存不足的异常，所以经过一些研究，我发现 RxJava 是答案，还实现了一些 API 调用分页。

首先，我试图做一个概念验证，启动第一个改造调用并显示对活动的响应，但我无法让它工作，我被卡住了！！

这是我得到的错误：java.lang.ClassCastException: io.reactivex.rxjava3.internal.observers.LambdaObserver cannot be cast to io.reactivex.rxjava3.core.Observable

我想做的是，我在这里寻求帮助是；第一次调用，我可以获得注册总数并定义页数，基于此，发送多个改造请求，每次我得到响应（List < Item >），将它们保存到房间。

遵循一些代码

毕业典礼

//Retrofit
implementation 'com.squareup.retrofit2:retrofit:2.8.1'
implementation 'com.squareup.retrofit2:converter-gson:2.8.1'
//Retrofit2 Adapter for RxJava 3
implementation "com.github.akarnokd:rxjava3-retrofit-adapter:3.0.0"
//okhttp3 Logging Interceptor
implementation "com.squareup.okhttp3:logging-interceptor:4.5.0"
//RxJava
implementation "io.reactivex.rxjava3:rxjava:3.0.2"
implementation 'io.reactivex.rxjava3:rxandroid:3.0.

类项目

public class Item {
    private String epc;
    private String barcode;
    private String name;
    private String description;
    private String brand;
    @SerializedName("serial_number")
    private String serialNumber;
    private double cost;
    @SerializedName("fk_item_state")
    private int status;
    @SerializedName("fk_category")
    private int category;
    @SerializedName("fk_location")
    private int location;
    private String parent;
    @SerializedName("responsable")
    private String responsible;
    @SerializedName("purchase_date")
    private String purchaseDate;
    @SerializedName("creation_date")
    private String creationDate;
    @SerializedName("last_update")
    private String lastUpdate;
    @SerializedName("inventory_date")
    private String inventoryDate;
    @SerializedName("last_seen")
    private String lastSeen;
...

类 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;
...

接口 FrekuencyApi

import io.reactivex.rxjava3.core.Observable;
import retrofit2.http.GET;
import retrofit2.http.Query;

public interface FrekuencyApi {

    @GET("item")
    Observable<ItemSyncDetails> getItemsPageDetails(
            @Query("pageSize") Integer pageSize,
            @Query("currentPage") Integer currentPage,
            @Query("sortBy") Integer sortBy
    );
...

类 MainActivity

import io.reactivex.rxjava3.android.schedulers.AndroidSchedulers;
import io.reactivex.rxjava3.core.Observable;
import io.reactivex.rxjava3.disposables.CompositeDisposable;
import io.reactivex.rxjava3.schedulers.Schedulers;
import okhttp3.Interceptor;
import okhttp3.OkHttpClient;
import okhttp3.Request;
import okhttp3.logging.HttpLoggingInterceptor;
import retrofit2.Retrofit;
import retrofit2.converter.gson.GsonConverterFactory;
...

public class MainActivity extends AppCompatActivity {

    private TextView tvResult;
    private Button send;
    private ProgressDialog mProgressDialog;
    private ItemSyncDetails mItemSyncDetails;
    private List<Item> mItemsList;
    private CompositeDisposable disposables = new CompositeDisposable();
    private FrekuencyApi frekuencyApi;
    private Retrofit retrofit;
    private int mNumPages;
    private CompositeDisposable compositeDisposable;
    private HttpLoggingInterceptor logging;

    private static final String TAG = "MainActivity";
    private Object handleResults;


    @Override
    protected void onCreate(Bundle savedInstanceState) {
        super.onCreate(savedInstanceState);
        setContentView(R.layout.activity_main);

        tvResult = findViewById(R.id.tv_result);
        send = findViewById(R.id.btnSend);
        mItemSyncDetails = new ItemSyncDetails();
        mProgressDialog = new ProgressDialog(this);

        mItemsList = new ArrayList<Item>();

        HttpLoggingInterceptor interceptor = new HttpLoggingInterceptor();
        interceptor.setLevel(HttpLoggingInterceptor.Level.BODY);
        OkHttpClient client = new OkHttpClient.Builder()
                .addInterceptor(interceptor)
                .addNetworkInterceptor(new Interceptor() {
                    @Override
                    public okhttp3.Response intercept(Chain chain) throws IOException {
                        Request request = chain.request().newBuilder()
                                .build();
                        return chain.proceed(request);
                    }
                }).build();

        retrofit = new Retrofit.Builder()
                .baseUrl("http://192.168.1.10:82/api/v1.0/")
                .client(client)
                .addCallAdapterFactory(RxJava3CallAdapterFactory.create())
                .addConverterFactory(GsonConverterFactory.create())
                .build();

        send.setOnClickListener(new View.OnClickListener() {
            @Override
            public void onClick(View v) {
                //TODO: EVENTO AL HACER CLICK EN BOTON
                getRecordsCount();
            }
        });
    }

    private void getRecordsCount(){
        FrekuencyApi frekuencyApi = retrofit.create(FrekuencyApi.class);
        Observable<ItemSyncDetails> observable = (Observable<ItemSyncDetails>) frekuencyApi.getRecordsCount(1,1,1)
                .subscribeOn(Schedulers.io())
                .retry(3)
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe(this::HandleResults, this::handleError );
    }

    private void HandleResults(ItemSyncDetails itemSyncDetails) {
        this.mItemSyncDetails = itemSyncDetails;
        int pageSize = 100;
        int numPages = itemSyncDetails.getPageCount()/pageSize;
        if (itemSyncDetails.getRecordCount() < pageSize || itemSyncDetails.getRecordCount()%pageSize != 0){
            numPages++;
        }
        this.mNumPages = numPages;
        tvResult.append("Page size: " + pageSize + "\n");
        tvResult.append("Page number: " + this.mNumPages + "\n");
        tvResult.append("Total items: " + itemSyncDetails.getRecordCount() + "\n");
    }

    private void handleError(Throwable throwable) {
        Log.e("Observer", ""+ throwable.toString());
        Toast.makeText(this, "ERROR DE CONEXION",
                Toast.LENGTH_LONG).show();
    }
...

错误日志

D/AndroidRuntime: Shutting down VM
E/AndroidRuntime: FATAL EXCEPTION: main
    Process: com.frekuency.retrofitapptest, PID: 9120
    java.lang.ClassCastException: io.reactivex.rxjava3.internal.observers.LambdaObserver cannot be cast to io.reactivex.rxjava3.core.Observable
        at com.frekuency.retrofitapptest.views.MainActivity.getRecordsCount(MainActivity.java:110)
        at com.frekuency.retrofitapptest.views.MainActivity.access$000(MainActivity.java:39)
        at com.frekuency.retrofitapptest.views.MainActivity$2.onClick(MainActivity.java:98)
        at android.view.View.performClick(View.java:5273)
        at android.view.View$PerformClick.run(View.java:21315)
        at android.os.Handler.handleCallback(Handler.java:743)
        at android.os.Handler.dispatchMessage(Handler.java:95)
        at android.os.Looper.loop(Looper.java:150)
        at android.app.ActivityThread.main(ActivityThread.java:5659)
        at java.lang.reflect.Method.invoke(Native Method)
        at com.android.internal.os.ZygoteInit$MethodAndArgsCaller.run(ZygoteInit.java:822)
        at com.android.internal.os.ZygoteInit.main(ZygoteInit.java:712)
D/OkHttp: --> GET http://192.168.1.10:82/api/v1.0/item?pageSize=1&currentPage=1&sortBy=1
    --> END GET
I/Process: Sending signal. PID: 9120 SIG: 9
I/System: core_booster, getBoosterConfig = false

json 为 - http://192.168.1.10:82/api/v1.0/item?pageSize=1¤tPage=1&sortBy=1

{
  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
}

在此先感谢，非常感谢您的帮助。

我对 MainActivity 类做了一些更改

public class MainActivity extends AppCompatActivity {
...
    @Override
    protected void onCreate(Bundle savedInstanceState) {
        super.onCreate(savedInstanceState);
        setContentView(R.layout.activity_main);

        tvResult = findViewById(R.id.tv_result);
        send = findViewById(R.id.btnSend);
        mItemSyncDetails = new ItemSyncDetails();
        mProgressDialog = new ProgressDialog(this);

        mItemsList = new ArrayList<Item>();

        HttpLoggingInterceptor interceptor = new HttpLoggingInterceptor();
        interceptor.setLevel(HttpLoggingInterceptor.Level.BODY);
        OkHttpClient client = new OkHttpClient.Builder()
                .addInterceptor(interceptor)
                .addNetworkInterceptor(new Interceptor() {
                    @Override
                    public okhttp3.Response intercept(Chain chain) throws IOException {
                        Request request = chain.request().newBuilder()
                                .build();
                        return chain.proceed(request);
                    }
                }).build();

        retrofit = new Retrofit.Builder()
                .baseUrl("http://192.168.1.10:82/api/v1.0/")
                .client(client)
                .addCallAdapterFactory(RxJava3CallAdapterFactory.create())
                .addConverterFactory(GsonConverterFactory.create())
                .build();

        send.setOnClickListener(new View.OnClickListener() {
            @Override
            public void onClick(View v) {
              showProgressDialog("Items");
              getRecordsCount();
            }
        });
    }

private void getRecordsCount(){
    retrofit.create(FrekuencyApi.class)
    .getRecordsCount(1,1,1)
            .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;
    tvResult.append("Tamaño de pagina: " + pageSize + "\n");
    tvResult.append("Numero de paginas: " + this.mNumPages + "\n");
    tvResult.append("Numero total de registros: " + itemSyncDetails.getRecordCount() + "\n");
}

private void getNumPagesHandlerComplete() {
    getAllRecords(mNumPages);
}

private void getAllRecords(int numPages){
    frekuencyApi = retrofit.create(FrekuencyApi.class);
    //numPages: total of pages are the number of times to send the request to API
    Observable.range(1, numPages)
    .concatMap(i -> frekuencyApi.getItemsPageDetails(100,i,1))
            .subscribeOn(Schedulers.newThread())
            .observeOn(AndroidSchedulers.mainThread())
            .subscribe(this::getAllHandleResults, this::handleError,this::handleComplete);
}

private void getAllHandleResults(ItemSyncDetails itemSyncDetails) {
    //TODO: Should it be the right place to save data to DB?
    //Get the progress
    int tmp = getProgress(itemSyncDetails.getPageCount(),
            itemSyncDetails.getCurrentPage(),itemSyncDetails.getPageSize());
    //Update ProgressDialog Progress
    mProgressDialog.setProgress(tmp);
    tvResult.setText("Progreso: "+ tmp + "%\n");
    if (itemSyncDetails.getCurrentPage() == itemSyncDetails.getPageCount()){
        //Showing on screen when last request is done
        tvResult.append("******************\n");
        tvResult.append(itemSyncDetails.getItemList().toString());
        tvResult.append("******************\n");
    }
}

private void handleComplete() {
    //when the last request finished
    tvResult.append("\nProceso de Sincronizacion terminado con exito!");
    closeProgressDialog();
}

private void handleError(Throwable throwable) {
    Log.e("Observer", ""+ throwable.toString());
    tvResult.append("\n******************\n");
    tvResult.append("ERROR DE CONEXION...\n");
    tvResult.append(throwable.toString());
    Toast.makeText(this, "ERROR DE CONEXION",
            Toast.LENGTH_LONG).show();
    closeProgressDialog();
}
...

随着对 MainActivity 类的更改，基本上有两个方法 getRecordsCount() 和 getAllRecords(int numPages)，每个触发一个 RX Java 调用进程，第一个调用一个 Retrofit 调用，并根据第一个调用的答案，调用第二种方法，它将向 API 发送 (n) 请求，其中 n 是参数 numPages，我使用更新的 progressDialog，当孔进程完成时，progressDialog 是 close()。在第一个 RX java 调用完成的 onComplete 方法上，它调用了第二个 RX Java 方法，此时我需要的是：

如何在一个 Rx java 进程中将其改进为更优雅的代码，而不是两次调用？
我正在考虑在 getAllHandleResults() 方法上对数据库进行房间插入调用以获取列表，这是插入数据库的正确位置吗？
你对这个问题有什么建议吗？这个数据库部分也可以使用 RX Java 完成吗？你有一些网页可以让我获得有关它的信息吗？再次感谢您的帮助。

android - Rxjava 3、retrofit2 和多次调用问题

在此先感谢，非常感谢您的帮助。

0 回答 0

Related

Reference