缓存项提供程序的RxJava运算符

Question

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

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

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

1. 发出本地项目
2. 如果需要就上网
3. 发出在线项目

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

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

我能想到的最好的解决方案是一个操作员发出项目，直到某个条件出现，然后自己退订。我找不到一个。

编辑：一些代码解决方案1）使用firstOrDefault：如果！则不会发出本地项目！DiskService.was下载（），因为DiskService返回空列表

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不会发出不检查其状态的项，这意味着我不会得到最佳列表。黑客的解决方案是将错误检查推迟到下一项，但这样即使在不必要时也会触发网络请求。在这个解决方案中，我使用了一个只包含列表的TrustedItemList和一个布尔值，它告诉被观察者是否可以信任一个非空的项目列表（对于内存和网络总是true，对于磁盘总是true，如果下载了则为true（））

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

匿名用户

我最终使用了一个自定义的可观察的。运营商，无耻地从运营商Takewhile复制而来，只需更改呼叫用户。onNext（t）就在订阅服务器之前。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);