在订阅时以相反的顺序重播缓存的项目

【问题标题】：Replay cached items in reverse order on subscribe在订阅时以相反的顺序重播缓存的项目
【发布时间】：2018-10-19 21:41:09
【问题描述】：

我有一个 ConnectableObservable，它在订阅后将按照原始顺序（从旧到最新）重播最后 x 个项目以及之后的任何后续事件。

我使用这个 Observable 作为事件记录器的后备存储，但是在订阅时，我实际上希望以相反的顺序（从最新到最旧）推送/onNext'ed 重播项目，以便我可以显示最相关的项目优先。

这是否可以使用标准 RX 操作符，还是我必须创建一个自定义操作符？

【问题讨论】：

您可以只使用map() 运算符：map(buffer => buffer.reverse())

标签： rxjs rx-java system.reactive monix

【解决方案1】：

replay() 无法做到这一点，因为您只需要获取未终止源上的缓存项。但是，ReplaySubject 让您可以窥视它并获取可以反转的项目数组，然后与来自同一主题的其余项目连接，但跳过刚刚检索到的快照项目：

ReplaySubject<ItemType> subject = ReplaySubject.create();

source.subscribe(subject);

Observable<ItemType> result = Observable.defer(() -> {
    ItemType[] current = subject.getValues(new ItemType[0]);

    return Observable.range(0, current.length)
        .map(index -> current[current.length - 1 - index])
        .concatWith(subject.skip(current.length));
});

【讨论】：

我想过这样的解决方案，但这些不会受到可观察的第一部分（回放缓冲区）和 concatWith （后半部分）之间的竞争条件的影响?