无法将 API 的响应添加到数据库

Question

我有一个来自 API 的单一 GET 请求。使用Retrofit 和RxJava，我得到了回复。现在我想将我的回复添加到Room 数据库。但我不想使用AsyncTask 我想使用RxJava。我在Dao insertAll中准备了一个方法。如何使用RxJava 将我的列表异步添加到数据库中？ 我的获取请求：

@GET("contacts")
fun getContactModel(): Single<List<Contact>>

我的道insertAll方法：

@Insert(onConflict = OnConflictStrategy.REPLACE)
fun insertAll(contact: List<Contact>?) : Completable

我使用 RxJava 的请求：

val disposable = CompositeDisposable()

disposable.add(contactsRepository.modelSingle()
        .subscribeOn(Schedulers.io())
        .observeOn(AndroidSchedulers.mainThread())
        .subscribeWith(object : DisposableSingleObserver<List<Contact>>() {
            override fun onSuccess(t: List<Contact>) {
                // t - My List from Api
            }

            override fun onError(e: Throwable) {
            }
        })
)

Answer 1

如果contactsRepository.modelSingle() 返回一个Single Observable，那么你可以这样做

disposable.add(contactsRepository.modelSingle()
                   .doOnSuccess { data -> saveToDb(data) } // this will be called on Schedulers.io
                   .subscribeOn(Schedulers.io())
                   .observeOn(AndroidSchedulers.mainThread())
                   .subscribeWith(object : DisposableSingleObserver<List<Contact>>() {
                       override fun onSuccess(t: List<Contact>) {
                           // t - My List from Api
                       }

                       override fun onError(e: Throwable) {
                       }
                   })
        )

如果是另一个 Observable，那么你可以使用 flatMap

disposable.add(contactsRepository.modelSingle()
                   .flatMap { data ->
                       saveToDb(data)
                       Observable.just(data)
                   }
                   .subscribeOn(Schedulers.io())
                   .observeOn(AndroidSchedulers.mainThread())
                   .subscribe()
        )

Answer 2

在 java 中，这是我首先将 Retrofit 与 RxJava 一起使用，并能够从 API 检索 Observable 响应，如下所示

在改造接口服务中使用以下代码

    @GET("contacts")
    Observable<Contacts> getContactsFromAPI

在@Dao Room Class 中实现

 @Insert(onConflict = OnConflictStrategy.REPLACE)
 Completable insertContact(List<Contact> contact);

要从 API 中检索可观察的数据项，请使用以下代码

    RetrofitService getContactService = RetroInstance.getService();
    Observable<Contact> apiData = 
     getContactService.getContactsFromAPI();

     compositeDisposable.add(apiData
            .subscribeOn(Schedulers.io())
            .observeOn(AndroidSchedulers.mainThread())
               .subscribeWith(new DisposableObserver<Contact>() {
                @Override
                public void onNext(List<Contact> contact) {
                    newContact(contact)
                    Log.d(LOG_TAG, "The contacts are : "+ contact);
                }

                @Override
                public void onError(Throwable e) {
                    Log.d(LOG_TAG, "Error: "+ e.getMessage());
                }

                @Override
                public void onComplete() {                
                }
            }));

要将来自 api 的添加检索添加到 Room 数据库，然后创建一个调用的方法，该方法用于将数据插入到 Room 数据库中，并且在编写用于从 api 检索数据的代码上调用此方法

 public void newContact(List<Contacts> contacts) {

 disposable.add(Completable.fromAction(new Action() {
        @Override
        public void run() throws Exception {
        rowIdInserted = 
    contactAppDatabase.userDao().insertContact(contacts);
        }
    }).subscribeOn(Schedulers.io())
        .observeOn(AndroidSchedulers.mainThread())
        .subscribeWith(new DisposableCompletableObserver() {
            @Override
            public void onComplete() {
                Log.d(TAG, "inserted successful");
            }

            @Override
            public void onError(Throwable e) {
            Log.d(LOG_TAG, "tHE new error on create the new user is: " + 
        e.getMessage());
            }
        }));
     }

我认为这有帮助