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

Question

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

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

我现在需要改进提供程序，以便它:

1. 发出本地元素
2. 需要时上网
3. 发布在线项目

(现在它停在第一个好的项目列表)。

我正在尝试 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 不会发出不检查其条件的项目，这意味着我不会得到最好的列表。 hacky 解决方案是将错误检查推迟到下一个项目，但这样即使在不需要时也会触发网络请求。有了这个解决方案，我使用了 TrustedItemList它只包含 List 和一个 bool 值，告诉 Observable 他是否可以信任一个非空的项目列表(对于 memory 和 network 始终为真，如果 wasDownloaded() 对于 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 复制而来，唯一的改变是调用 subscriber.onNext(t) 就在 onNext 方法中的 subscriber.onCompleted() 之前。这样，最后一项，即在 bool 检查中返回 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);