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

Question

我正在寻找可以实现特定行为的 Subject 类型(或运算符的某种组合)。

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

我基本上需要的是 ReplaySubject 和 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");
} 

Answer 1

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

A Subject variant that buffers items and allows one or more Observers to exclusively consume one of the items in the buffer asynchronously. If there are no Observers (or they all disposed), the DispatchWorkSubject will keep buffering and later Observers can resume the consumption of the buffer.

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