RxJava 主题的真实世界用例答案

【问题标题】：Real World Use case of RxJava SubjectRxJava 主题的真实世界用例
【发布时间】：2020-02-21 13:19:00
【问题描述】：

有很多很棒的文章解释了 RxJava。但是这些几乎都没有用现实世界的例子来解释这个概念。

所以我基本上理解了RxJava Subject的概念，就像一个管道，它既是observable又是observer。

但我不知道这个 RxJava 主题 在 Android 开发世界中的实际用途是什么。你能详细说明一下吗？

【问题讨论】：

【解决方案1】：

在我的情况下是因为我有一个 Observable 正在等待另一个 Observable 发射的项目，这是异步的，因为它是 Interval。

Scheduler scheduler = RxHelper.scheduler(vertx.getOrCreateContext());


Observable.just(callAnotherObservable)
          .subscribe(item -> System.out.println(item)

public Observable<String> callAnotherObservable(Scheduler scheduler, ){
          Subject subject = ReplaySubject.create(1);
          Observable.interval(100,TimeUnit.MILLISECONDS)
          .map(i->"item to be passed to the other observable")
          .subscribe(subject);
          return subject.observeOn(scheduler).first();//Here we wait for the first emission of the interval Observable.

}

如您所见，我们使用 subject.first() 等待在另一个线程中运行的可观察间隔的第一次发射。

如果您想查看更多“hotObservables”示例https://github.com/politrons/reactive/blob/master/src/test/java/rx/observables/connectable/HotObservable.java

【讨论】：

【解决方案2】：

主题有很多“现实世界”的应用程序，尤其是当您逐渐将代码库从命令式转换为响应式时。它可以充当这两个世界之间的桥梁，您可以在其中使用来自流之外的非反应性代码来影响流。

但是正如您所要求的那样。最近，当用户尝试从活动返回时，我正在实施自定义行为。 RxJava 为我面临的问题提供了非常优雅的解决方案，因此我需要编写一个事件流，对应于用户想要返回。我故意避免使用“按下后退按钮”这句话，因为代码库中有几个地方可以模拟返回的情况，而且总是通过onBackPressed() 方法。

将其转换为单个流需要进行大规模的重构，而这目前不在预算之内。但我不想放弃通过 RxJava 的解决方案，因为它可以使代码更加简洁。使用 BehaviorSubject 给出了答案，因为我只需要在 onBackPressed() 方法中发出一个事件。

【讨论】：

【解决方案3】：

我正在为一个 Android 应用程序构建一个通用的 ReportDownloadManager，我们需要使用一个 Observable，它的 Observer 将在本地下载和存储文件。下载成功或不成功的事件需要由管理器处理，但还需要向使用此 DownloadManager 的活动/服务公开一个 Observable。我相信这是一个很好的用例，它可以使用 Subject 来消费初始 Observable 并为客户端 Observable 生成事件。

import android.app.DownloadManager;
import android.content.Context;
import android.webkit.MimeTypeMap;

import java.io.File;

import io.reactivex.Observable;
import io.reactivex.functions.Consumer;
import io.reactivex.subjects.PublishSubject;

public class ReportDownloadManager {

    private final DownloadManager platformDownloadManager;

    public ReportDownloadManager(Context context) {
        this.platformDownloadManager = (DownloadManager) context.getSystemService(Context.DOWNLOAD_SERVICE);
    }

    public Observable<Object> download(final File file, DownloadAction downloadAction) {
        final PublishSubject<Object> subject = PublishSubject.create();
        downloadAction.execute(file.getName())
                .subscribe(new Consumer<Object>() {
                    @Override
                    public void accept(Object o) throws Exception {
                        platformDownloadManager.addCompletedDownload(file.getName(), "No description", false,
                                MimeTypeMap.getSingleton().getMimeTypeFromExtension("pdf"), file.getAbsolutePath(),
                                file.length(), true);
                        subject.onNext(new Object());
                        subject.onComplete();
                    }
                }, new Consumer<Throwable>() {
                    @Override
                    public void accept(Throwable throwable) throws Exception {
                        subject.onError(throwable);
                    }
                });
        return subject;
    }

    interface DownloadAction {
        Observable<Object> execute(String fileAbsolutePath);
    }

}

【讨论】：

【解决方案4】：

如果您需要我在图表中显示的这种情况

【讨论】：