Оператор RxJava для провайдера кэшированных товаров

Question

Я пытаюсь реализовать поставщика, который ищет товары в memory, disk, network, в этом порядке. Основная цель этого - избежать сетевых вызовов, если у меня есть правильный локальный кеш. Существует уловка, поскольку мои вызовы в сеть используют фильтры для получения элементов, я мог бы иметь 10 элементов из локального запроса, но все равно нужно перейти в сеть, потому что эти элементы поступают из другого сетевого вызова с различными параметрами запроса.

Прямо сейчас я использую concat с firstOrDefault, проверяя, что список не пустой или пустой. Я внедрил способ проверить, действительно ли я уже вызвал сервер с конкретным запросом, и я использую его для возврата значения null при чтении с диска.

теперь мне нужно уточнить у поставщика, так что:

1. излучает местные пункты
2. пойти онлайн при необходимости
3. излучает онлайн элементы

(Сейчас это останавливается в первом хорошем списке предметов).

Я пытаюсь использовать takeWhile, используя метод, который возвращает true, если данные являются пустыми или пустыми, или если я еще не вызывал сервер для этого запроса. Проблема заключается в том, что takeWhile не выделяет элемент, если проверка для этого элемента ложна, то есть я не получу последний хороший товар (который также является лучшим).

Лучшее решение, о котором я могу думать, это оператор, который испускает элементы до тех пор, пока не появится какое-то условие, а затем отмените подписку на себя. Я не могу его найти.

EDIT: НЕКОТОРЫЙ КОД Решение 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. 
} 

Solution 2) Использованием takeWhile. Проблема с takeWhile является Observable, не будет излучать элемент, который не проверяет его состояние, то есть я не получу лучший список. Хакерное решение - отложить ложную проверку до следующего элемента, но таким образом сетевой запрос будет запущен, даже если это не необходимо. С помощью этого решения я использую TrustedItemList что только содержит перечень и логическое, который говорит Наблюдаемая, если он может доверять непустой список элементов (всегда верно для 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) непосредственно перед 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);