rx-java2 - 是否有一个 RxJava 主题可以缓存值并在发出后忘记它们？

Question

我正在寻找一种Subject可以实现某种行为的类型（或某种运算符组合）。

• ASubject已创建
• ASubject被onNext()多次调用，所有这些值都被缓存在Subject
• AConsumer订阅了那个Subject
• Consumer接收所有已缓存的值
• Consumer取消订阅..通过Subject调用dispose()
• Subject用 2个onNext()新值调用A
• AConsumer订阅了那个Subject
• 应该只接收 2 个新Consumer值，因为旧值已经发出。

我需要的基本上是 aReplaySubject和 a之间的特殊情况BehaviourSubject。

ReplaySubject重播所有事件..BehaviourSubject重播最后一个事件。

我想要一个Subject只重播尚未消费的事件。换句话说 ..Subject发出时没有订阅者的事件。

为了完成，这是一个澄清行为的测试用例

@Test
public void test() {
    Subject<String> subject = MyDesiredSubject.create();

    subject.onNext("1");
    subject.onNext("2");

    TestObserver<String> testObserver = subject.test();
    testObserver.assertValues("1", "2");
    testObserver.dispose();

    subject.onNext("3");
    subject.onNext("4");
    testObserver = subject.test();
    testObserver.assertValues("3", "4");
} 

标签： rx-java2

Answer 1

您可以DispatchWorkSubject在https://github.com/akarnokd/RxJava2Extensions中使用。

一种主题变体，它缓冲项目并允许一个或多个观察者异步独占使用缓冲区中的一个项目。如果没有观察者（或它们都被释放），DispatchWorkSubject 将继续缓冲，稍后观察者可以恢复对缓冲区的消耗。

但是，如果您一次只想支持最多1 个订阅者，那么UnicastWorkSubject（也在 Rx2 扩展中）。