标签: rx-java2

我在 Kotlin 有以下密封类

sealed class Result<T, E> {
    class Success<T, E>(val result: T) : Result<T, E>()
    class Failure<T, E>(val failure: E) : Result<T, E>()
}

create()并从函数返回 Result 实体

并将以下内容与 RxJava2 一起使用，

    return Single.fromCallable { repository.create(details) }
            .map {
                return@map when(result) {
                        is Result.Success -> CreatedState.Success(result.result)
                        is Result.Failure -> CreatedState.Error(result.failure)
                }
            }

据我所知，该功能似乎按预期工作。然而当我对此进行单元测试时，

    val testSubscriber = underTest.create(newDetails).test()

我得到一个并在异常列表中java.lang.IllegalStateException看到一个（当执行一个时），我无法理解它是如何发生的，因为我的何时检查是详尽的。或者我在那里错过了什么？kotlin.NoWhenBranchMatchedExceptionassertComplete()

蒂亚！

当在后台使用而不是在 Activity/Fragment 中使用时，如何在 RxJava2 中处理单个观察者。

我知道如果在 Activity/Fragment 中使用 Single，则可以将其添加到复合处置器中，并在 Activity/Fragment 的 onDestroy 方法中对其进行处置。

但是当从其他地方调用它时如何处理它呢？

另外，如果从 Activity/Fragment 调用 Single，并且我们想在调用 onDestroy 之前处置它，还有哪些其他方法来处置 Single？

我有一个返回 Completable 的函数，Completable.error(RuntimeException("message"))如果另一个函数失败或Completable.complete()失败，则该函数返回。

我试图为此编写一个单元测试，并查看流程正确地转到错误和成功代码，但在我的测试中，我无法使用区分它们

underTest.unregisterFromService().test().assertComplete().assertNoErrors()

有谁知道如何Completable.error()在单元测试中检查该值？

@GET("/city/{cityId}/category/all")
Observable<MyDictionary> getDictionaries(@Path(value = "cityId", encoded = true) String cityId, @HeaderMap Map<String, String> headers);

调用方法检索数据：

service.getDictionaries(cityId, headersMap)
                .subscribeOn(Schedulers.io())
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe(resp ->
                {
                     /*...............*/

                }, throwable ->
                {
                     /*...............*/
                });

我可以轻松得到答复。但我还需要知道响应头。如何在我的代码中访问它们？有小费吗？我只获取正文，如何获取标题？

我正在尝试将Firestore 方法包装到 CompletableEmitter 中，以便能够在成功/完成时返回 Completable.complete() ，在失败时返回 Completable.error() 。（该任务是一个 Firestore 操作，用于更新集合中的文档（该文档中的 1 个字段））

这是firestore method我试图包装成一个返回Task<Void>. 这是未将其包装在 Completable 中的代码：

public static Completable updateDocument(DocumentReference ref, Map<String,Object> fieldsAndValues){
    ref.update(fieldsAndValues)
                .addOnSuccessListener(aVoid -> {
                    Timber.d("SUCCESS");
                }).addOnCompleteListener(task -> {
                    Timber.d("COMPLETE");
                }).addOnFailureListener(e -> {
                    Timber.d("ERROR");
                });
        return Completable.complete();    //just for the sake of returning complete to finish the method...
}

上面的代码按原样执行。

这是我尝试包装该代码：

public static Completable updateDocument(DocumentReference ref, Map<String,Object> fieldsAndValues){
    return Completable.create(emitter -> ref.update(fieldsAndValues)
       .addOnSuccessListener(aVoid -> emitter.onComplete())
       .addOnCompleteListener(task -> emitter.onComplete())
       .addOnFailureListener(emitter::onError)); …

我期望flatMapCompletable在Completable正常完成时会调用给定的Action。但是，它不像我想的那样工作。这是一个简单的例子：

    PublishProcessor<String> processor = PublishProcessor.create();

    processor.flatMapCompletable(s2 -> {
        System.out.println("s2 " + s2);
        return Completable.complete();
    }).subscribe(() -> {
        System.out.println("done"); // it does not come here
    });

这是预期的行为吗？如果是这样，我如何检查“完成任务”是否已完成？Completable.complete().doFinally()？

我有一个返回的API FamilyDetails.我使用改造来获取它们,并建模如下:

Observable<FamilyDetails> getFamilyDetails(@Query int personId);

获取后,这些详细信息将用于更新a Person.我的API FamilyDetails会在有一些信息显示时立即提供,这意味着细节可能不完整,因此complete = false当服务器尚未完成获取所有家庭详细信息时,响应中会有一个标记 - 反过来,我的应用程序应该FamilyDetails如果我们complete = false在第一个响应中收到,则发送另一个请求以获取剩余请求,并且只要细节仍然不完整,就继续这样做.

我"有点"实现了我想要的,但不完全是这个代码:

personApiService.getFamilyDetails(personId)
    // repeat call
    .flatMap(familyDetails -> personApiService.getFamilyDetails(personId))
    // until family details are complete
    .takeUntil(familyDetails -> familyDetails.isComplete())
    // update the person object with the family details
    .flatMap(familyDetails -> Observable.just(updatePerson(familyDetails))
    //subscribe!
    .subscribe(personSubscriber);

我的personSubscriber返回更新了Person.

我有这个实现的问题是,部分更新不通过personSubscriber的onNext(),因为我只得到一个呼叫onNext()与更新Person其完整的对象FamilyDetails.

我想使用RxJava对此进行建模,具有以下要求:

• 如果我收到不完整的详细信息,我会更新Person 这些详细信息所属的对象,并通过它提供onNext().
• 如果我们收到不完整的细节,我们继续查询完整的详细信息的API,并提供最新Person通过的物体 …

我正在使用RXJava2在广播接收器中发送电子邮件，我想知道何时应该退订该事件。代码基本上是：

        getSmsMmsObservable()
            .subscribeOn(Schedulers.io())
            .observeOn(AndroidSchedulers.mainThread())
            .doOnError(throwable -> Timber.e(throwable, "Error sending mail."))
            .map(smsMmsAddress1 -> {
                smsMmsAddress = smsMmsAddress1;
                return doInBackgroundSendEmail();
            })
            .map(stringSingle -> {
                mMsgResponse = stringSingle;
                this.done = true;
                return deleteFile();
            })
            .subscribe(success -> {
                if (success) {
                    Toast.makeText(context, "Message Status: " + mMsgResponse, Toast.LENGTH_LONG).show();
                }

            });

我什么时候退订（接收器中没有onPause或onDestroy），我怎么知道onReceive何时完成？接收者通过清单进行注册。我考虑过做一个可观​​察的复合物，然后在subscribe（）部分中将其处理。

    compositeDisposable.add(
    getSmsMmsObservable()
            .subscribeOn(Schedulers.io())
            .observeOn(AndroidSchedulers.mainThread())
            .doOnError(throwable -> Timber.e(throwable, "Error sending mail."))
            .map(smsMmsAddress1 -> {
                smsMmsAddress = smsMmsAddress1;
                return doInBackgroundSendEmail();
            })
            .map(stringSingle -> {
                mMsgResponse = stringSingle;
                this.done = true;
                return deleteFile();
            })
            .subscribe(success -> …

我正在将一个应用程序从RxJava 1迁移到2。我有这样的代码：

RxJava 1

public Completable update() {
    return client.fetchNotes()
        .map(toNote())
        .toList()
        .doOnNext(save())
        .toCompletable();
}

在上面的代码中，toList()返回Observable<List<Note>>。

现在，我正在尝试将该代码转换为RxJava 2：

RxJava 2

public Completable update() {
    return client.fetchNotes()
        .map(toNote())
        .toList()
        .doOnNext(save()) // <-- compilation error here
        .ignoreElements();
}

在RxJava 2中，toList()返回Single<List<Note>>，因此我无法链接doOnNext(save())到它，因为doOnNext()中不存在Single<T>.class。

在两种情况下如何获得相同的行为？基本上，在save()方法中，我将Notes 存储在数据库中，并且必须来自List<Note>；Database我使用的界面不允许一一存储。这就是为什么我需要使用toList()：

interface Database {
    void store(List<Note> notes);
}

我有存储库类。在这些课程中，我这样简单collection("..").get()。

override fun getTestCollectionItems(): Observable<TestModel> {

    return Observable.create { subscriber ->

        firebaseFirestore.collection(TEST_COLLECTION)
                .get()
                .addOnCompleteListener { task ->

                    if (task.isSuccessful()) {
                        for (document in task.getResult()) {
                            if (document.exists()) {
                                val documentModel = document.toObject(TestModel::class.java)
                                subscriber.onNext(documentModel)
                            }
                        }
                        subscriber.onComplete()
                    } else {
                        subscriber.onError(task.exception!!)
                    }
                }
    }
}

但是我找到了实时Firecloud选项。如果我将侦听器移至存储库，那么它的含义是否正确？

我尝试了下一个：

override fun getRealTimeCollection() : Observable<TestModel> {

    return Observable.create { subscriber ->

        firebaseFirestore.collection(TEST_COLLECTION).document("3lPtYZEEhPdfvZ1wfHIP")
            .addSnapshotListener(EventListener<DocumentSnapshot> { snapshot, e ->
                if (e != null) {
                    Log.w("test", "Listen failed.", e)
                    subscriber.onError(e)
                    return@EventListener
                }

                if (snapshot != …