android - RxJava运算符，用于缓存项提供程序

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

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

我现在需要优化提供程序，以便它：


发出本地物品
必要时上网
发出在线物品


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

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

我能想到的最好的解决方案是一个操作员，该操作员在出现某种情况之前会发出物品，然后自行退订。我找不到一个。

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

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不会发出不检查其条件的项目，这意味着我不会获得最佳List。 hacky解决方案是将错误检查推迟到下一项，但是这种方式即使在没有必要的情况下也会触发网络请求。通过这种解决方案，我使用的是一个仅包含列表的TrustedItemList和一个布尔值，该布尔值告诉Observable是否可以信任非空项目列表（对于memory和network始终为true，如果表示wasDownloaded()）

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;
}

最佳答案

我最终使用了一个自定义的Observable.Operator，它是从OperatorTakeWhile无耻复制的，唯一的变化是在subscriber.onNext(t)方法中的subscriber.onCompleted()之前调用了onNext。这样，发出了最后一项，即在布尔检查中返回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);