用于缓存项目提供者的 RxJava 运算符

Question

我正在尝试实现一个提供程序，它按此顺序在memory、disk、network 中查找项目。如果我有正确的本地缓存，这样做的主要目的是避免网络调用。有一个问题，因为我对网络的调用使用过滤器来获取项目，我可以从本地查询中获得 10 个项目，但仍然需要转到网络，因为这些项目来自不同的网络调用，具有不同的查询参数。

现在我使用concat 和firstOrDefault，检查列表不为空或为空。我已经实现了一种方法来检查我是否已经使用特定查询调用了服务器，并在从磁盘读取时使用它返回 null。

我现在需要改进提供者，以便它：

1. 发射本地项目
2. 上网如果需要
3. 发射在线物品

（现在它停在第一个好的项目列表上）。

我正在尝试使用takeWhile，如果数据为空或空或者我还没有为该查询调用服务器，则使用返回 true 的方法。问题是如果对该项目的检查为假，takeWhile 不会发出该项目，这意味着我不会得到最后一个好的项目（这也是最好的）。

我能想到的最佳解决方案是一个操作符，它发出项目直到出现特定条件，然后自行取消订阅。我找不到。

编辑：一些代码 解决方案1）使用firstOrDefault：如果!DiskService.wasDownloaded()，则不会发出本地项目，因为DiskService用!DiskService.wasDownloaded()返回一个空List<Item>

public Observable<List<Item>> items() {
    List<Observable> obs = new ArrayList<>();

    Observable<List<Item>> memoryObs = Observable.defer(this::getMemoryItems);

    Observable<List<Item>> diskObs = Observable.defer(this::getDiskItems);

    Observable<List<Item>> networkObs = Observable.defer(this::getNetworkItems);

    Observable<List<Item>> concat = Observable.concat(memoryObs, diskObs, networkObs;


    return concat.firstOrDefault(new ArrayList<>(), this::canAccept);
}

private boolean canAccept(List<Item> data) {
    return data != null && data.size() > 0;
}

//Method in DiskService
public boolean wasDownloaded(){
    return true if the query was performed on the server, false otherwise.
} 

解决方案 2) 使用 takeWhile。 takeWhile 的问题是 Observable 不会发出不检查其条件的项目，这意味着我不会得到最好的列表。 hacky 的解决方案是将错误检查推迟到下一个项目，但这样即使在不需要时也会触发网络请求。有了这个解决方案，我使用了一个 TrustedItemList ，它只包含 List 和一个布尔值，告诉 Observable 他是否可以信任一个非空的项目列表（对于 memory 和 network 总是正确的，如果 @987654337 是正确的@为disk)

public Observable<List<Item>> items() {
    List<Observable> obs = new ArrayList<>();

    Observable<TrustedItemList> memoryObs = Observable.defer(this::getMemoryItems);

    Observable<TrustedItemList> diskObs = Observable.defer(this::getDiskItems);

    Observable<TrustedItemList> networkObs = Observable.defer(this::getNetworkItems);

    Observable<TrustedItemList> concat = Observable.concat(memoryObs, diskObs, networkObs;


    return concat.takeWhile(this::shouldContinueSearching)
                 .filter(trustedItemList -> trustedItemList.items != null && !trustedItemList.items.isEmpty())
                 .map(trustedItemList -> trustedItemList.items);
}

private boolean shouldContinueSearching(TrustedPoiList data) {
     return data == null || data.items == null || data.items.isEmpty() || !data.canTrustIfNotEmpty; 
}

Answer 1

我最终使用了自定义的Observable.Operator，无耻地从OperatorTakeWhile 复制而来，唯一的变化是在onNext 方法中在subscriber.onCompleted() 之前调用subscriber.onNext(t)。这样，最后一项，即在布尔检查中返回 false 的项被发出。

public final class OperatorTakeWhileWithLast<T> implements Observable.Operator<T, T> {

    private final Func2<? super T, ? super Integer, Boolean> predicate;

    public OperatorTakeWhileWithLast(final Func1<? super T, Boolean> underlying) {
        this((input, index) -> {
            return underlying.call(input);
        });
    }

    public OperatorTakeWhileWithLast(Func2<? super T, ? super Integer, Boolean> predicate) {
        this.predicate = predicate;
    }

    @Override
    public Subscriber<? super T> call(final Subscriber<? super T> subscriber) {
        Subscriber<T> s = new Subscriber<T>(subscriber, false) {
            private int counter = 0;
            private boolean done = false;

            @Override
            public void onNext(T t) {
                boolean isSelected;
                try {
                    isSelected = predicate.call(t, counter++);
                } catch (Throwable e) {
                    done = true;
                    Exceptions.throwIfFatal(e);
                    subscriber.onError(OnErrorThrowable.addValueAsLastCause(e, t));
                    unsubscribe();
                    return;
                }
                if (isSelected) {
                    subscriber.onNext(t);
                } else {
                    done = true;
                    subscriber.onNext(t); //Just added this line
                    subscriber.onCompleted();
                    unsubscribe();
                }
            }

            @Override
            public void onCompleted() {
                if (!done) {
                    subscriber.onCompleted();
                }
            }

            @Override
            public void onError(Throwable e) {
                if (!done) {
                    subscriber.onError(e);
                }
            }
        };
        subscriber.add(s);
        return s;
    }
}

我的 items() 方法（解决方案 2）现在以：

return concat.lift(new OperatorTakeWhileWithLast<TrustedItemList>(this::shouldContinueSearching))  
             .filter(trustedItemList -> trustedItemList.items != null && !trustedItemList.items.isEmpty()) 
             .map(trustedItemList -> trustedItemList.items);