Rxjava-订阅前在PublishSubject上发出

Question

考虑一个场景，我们有一个发出字符串的流，我们想将字符串保存在文件中。

我正在使用Publish科目，这很好：

Subject<String> stream = PublishSubject.create();
stream.subscribe(str -> saveFile(str));
mReverseGeocoderStream.onNext("some-string1")
mReverseGeocoderStream.onNext("some-string2")

但是，这不起作用（只有thom-string2被交付）

Subject<String> stream = PublishSubject.create();
mReverseGeocoderStream.onNext("some-string1")
stream.subscribe(str -> saveFile(str));
mReverseGeocoderStream.onNext("some-string2")

有没有办法让第二个场景也能正常工作？

i、 例如，我们是否可以更改PublishSubject以确保它缓冲事件，直到订阅者使用它们？

请注意，BehaviorSubject不是一个选项，因为重新订阅会导致另一个文件保存。它没有“消费事件”的概念。

我找到了Unicastsubject，这几乎是我想要的，除了，当我取消订阅并稍后重新订阅不同的订阅者时，它会在IllegalStateException中失败。

用例：

假设我们有一个android应用程序。它发出网络请求，在网络请求的后面，它需要显示一个对话框。在发出请求时，用户会对应用程序进行背景设置。此时，我们将取消订阅正在侦听信号的观察员，以显示对话框。

网络请求返回并向流发出显示对话框的信号。现在没有人在听。用户将应用前景化。新订阅者将连接到网络请求管理器（ViewModel）。此时，我希望将“未使用”信号传递给订阅者。

注意：我不能使用行为主题。如果我这样做，每次用户背景和前景的应用程序的对话框将显示出来。我希望在显示对话框后使用并完成事件。

Answer 1

做了更多的研究，发现了这个：

https://gist.github.com/xsveda/8c556516079fde97d04b4b7e14a18463

发件人：RxJava中的类队列主题

请注意，它使用中继，但如果不想引入另一个依赖项，则可以轻松地将中继替换为Subject。

我对其他解决方案持开放态度，也许会批评为什么这个解决方案不好。

/**
 * Relay that buffers values when no Observer subscribed and replays them to Observer as requested. Such values are not replayed
 * to any other Observer.
 * <p>
 * This relay holds an unbounded internal buffer.
 * <p>
 * This relay allows only a single Observer at a time to be subscribed to it.
 * <p>
 * If more than one Observer attempts to subscribe to this Relay at the same time, they
 * will receive an IllegalStateException.
 *
 * @param <T> the value type received and emitted by this Relay subclass
 */
public final class CacheRelay<T> extends Relay<T> {

    private final ConcurrentLinkedQueue<T> queue = new ConcurrentLinkedQueue<>();
    private final PublishRelay<T> relay = PublishRelay.create();

    private CacheRelay() {
    }

    public static <T> CacheRelay<T> create() {
        return new CacheRelay<>();
    }

    @Override
    public void accept(T value) {
        if (relay.hasObservers()) {
            relay.accept(value);
        } else {
            queue.add(value);
        }
    }

    @Override
    public boolean hasObservers() {
        return relay.hasObservers();
    }

    @Override
    protected void subscribeActual(Observer<? super T> observer) {
        if (hasObservers()) {
            EmptyDisposable.error(new IllegalStateException("Only a single observer at a time allowed."), observer);
        } else {
            for (T element; (element = queue.poll()) != null; ) {
                observer.onNext(element);
            }
            relay.subscribeActual(observer);
        }
    }
}