java - RxJava : Find out if BehaviorSubject was a repeated value or not

Question

我正在制作一个 Android 界面，显示从网络获取的一些数据。我想让它显示最新的可用数据，并且永远不会为空(除非还没有获取任何数据)所以我使用 BehaviorSubject 为订阅者(我的 UI)提供最新的可用信息，同时刷新它更新它的背景。

这行得通，但由于我的 UI 中的另一个要求，我现在必须知道发布的结果是否是从网络中获取的。 (换句话说，我需要知道发布的结果是否是 BehaviorSubject 的保存项。)

我怎样才能做到这一点？如果我需要将它拆分成多个 Observables，那很好，只要我能够获得 BehaviorSubject 的缓存行为(获得最后可用的结果)，同时还能判断返回的结果是来自缓存还是不是。我能想到的一种 hacky 方法是检查响应的时间戳是否相对较快，但这真的很草率，我宁愿想出一种用 RxJava 来做的方法。

Answer 1

正如您在问题中提到的，这可以通过多个 Observable 来实现。本质上，您有两个 Observable:“可以观察到新鲜的响应”，以及“可以观察到缓存的响应”。如果某件事可以被“观察到”，你可以将它表示为一个 Observable。我们将第一个命名为 original，将第二个命名为 replayed。

查看此 JSBin (JavaScript，但这些概念可以直接转换为 Java。据我所知，没有用于这些目的的 JavaBin)。

var original = Rx.Observable.interval(1000)
  .map(function (x) { return {value: x, from: 'original'}; })
  .take(4)
  .publish().refCount();

var replayed = original
  .map(function (x) { return {value: x.value, from: 'replayed'}; })
  .replay(null, 1).refCount();

var merged = Rx.Observable.merge(original, replayed)
  .replay(null, 1).refCount()
  .distinctUntilChanged(function (obj) { return obj.value; });

console.log('subscribe 1st');
merged.subscribe(function (x) { 
  console.log('subscriber1: value ' + x.value + ', from: ' + x.from);
});

setTimeout(function () {
  console.log('    subscribe 2nd');
  merged.subscribe(function (x) {
    console.log('    subscriber2: value ' + x.value + ', from: ' + x.from);
  });
}, 2500);

这里的总体思路是:用指示事件来源的字段 from 注释事件。如果它是 original，则它是一个新鲜的响应。如果它是 replayed，则它是一个缓存的响应。 Observable original 只会发出 from: 'original' 而 Observable replayed 只会发出 from: 'replayed'。在 Java 中，我们需要更多的样板文件，因为您需要创建一个类来表示这些带注释的事件。否则，可以在 RxJava 中找到 RxJS 中的相同运算符。

原始的 Observable 是 publish().refCount() 因为我们只希望这个流的一个实例与所有观察者共享。事实上，在 RxJS 和 Rx.NET 中，share() 是 publish().refCount() 的别名。

重放的 Observable 是 replay(1).refCount() 因为它也像原始 Observable 一样被共享，但是 replay(1) 给了我们缓存行为。

merged Observable 包含原始的和重放的，这是您应该向所有订阅者公开的内容。由于 replayed 将在 original 执行时立即发出，因此我们在事件值上使用 distinctUntilChanged 以忽略立即连续。我们 replay(1).refCount() also the merged 的原因是我们希望 original 和 replay 的合并也是一个流的共享实例所有观察员。我们会为此目的使用 publish().refCount()，但我们不能失去 replayed 包含的重放效果，因此它是 replay(1)。 refCount()，而不是 publish().refCount()。