JavaRX Pagination - наблюдайте в каждом интерфейсе, а не в конце - Generic Paginator

Question

Я работаю с разбитым на страницы API. Я использовал следующее решение, предоставленное Adam Millerchip , и оно работает хорошо.

import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.Optional;
import java.util.function.Function;

import io.reactivex.Flowable;
import io.reactivex.Single;
import io.reactivex.processors.BehaviorProcessor;

public class Pagination {

    // Fetch all pages and return the items contained in those pages, using the provided page fetcher function
    public static <T> Flowable<T> fetchItems(Function<Integer, Single<Page<T>>> fetchPage) {
        // Processor issues page indices
        BehaviorProcessor<Integer> processor = BehaviorProcessor.createDefault(0);
        // When an index number is issued, fetch the corresponding page
        return processor.concatMap(index -> fetchPage.apply(index).toFlowable())
                        // when returning the page, update the processor to get the next page (or stop)
                        .doOnNext(page -> {
                            if (page.hasNext()) {
                                processor.onNext(page.getNextPageIndex());
                            } else {
                                processor.onComplete();
                            }
                        })
                        .concatMapIterable(Page::getElements);
    }

    public static void main(String[] args) {
        fetchItems(Pagination::examplePageFetcher).subscribe(System.out::println);
    }

    // A function to fetch a page of our paged data
    private static Single<Page<String>> examplePageFetcher(int index) {
        return Single.just(pages.get(index));
    }

    // Create some paged data
    private static ArrayList<Page<String>> pages = new ArrayList<>(3);

    static {
        pages.add(new Page<>(Arrays.asList("one", "two"), Optional.of(1)));
        pages.add(new Page<>(Arrays.asList("three", "four"), Optional.of(2)));
        pages.add(new Page<>(Arrays.asList("five"), Optional.empty()));
    }

    static class Page<T> {
        private List<T> elements;
        private Optional<Integer> nextPageIndex;

        public Page(List<T> elements, Optional<Integer> nextPageIndex) {
            this.elements = elements;
            this.nextPageIndex = nextPageIndex;
        }

        public List<T> getElements() {
            return elements;
        }

        public int getNextPageIndex() {
            return nextPageIndex.get();
        }

        public boolean hasNext() {
            return nextPageIndex.isPresent();
        }
    }
}

Но у меня есть 2 вопроса:

• В этой реализации элементы обрабатываются в конце (подписка (System.out :: println)), когда все страницы загружены. Это может вызвать проблемы с памятью, если собранные данные многочисленны. Я бы предпочел обрабатывать их (сохранение базы данных) сразу же после их загрузки (в .doOnNext (page -> {}). Я смог это сделать, но «грязным способом» (добавить код сохранения базы данных в doOnNext). Как я могу это сделать?

• В моей реализации класса "page" я использую собственный десериализатор Gson. И я не знаю, как обращаться с общими данными. Мне пришлось написать "list.add (( MyGenericClass ) context.deserialize (anArray.getAsJsonObject (), MyGenericClass .class));" где я хотел бы что-то вроде "list.add (( T ) context.deserialize (anArray.getAsJsonObject (), T .class));". Как я могу сделать вещи действительно общими?

  public static JsonDeserializer<Paginator> deserializer = new JsonDeserializer<Paginator>() {
  @Override
  public Paginator deserialize(JsonElement json, Type typeOfT, JsonDeserializationContext context) throws JsonParseException {
      JsonObject jsonObject = json.getAsJsonObject();
      Paginator paginator = new Paginator(null, Optional.of(1));
      if (jsonObject.get("data") != null && !jsonObject.get("data").isJsonNull()) {
          JsonArray array = jsonObject.getAsJsonArray("data");
          List<MyGenericClass> list = new ArrayList<>();
          for (JsonElement anArray : array) {
              list.add((MyGenericClass)context.deserialize(anArray.getAsJsonObject(), MyGenericClass.class));
          }
          paginator.setData(list);
      }
      paginator.setCurrent_page(jsonAsInt(jsonObject, "current_page",-1));
      paginator.setFrom(jsonAsInt(jsonObject,"from",-1));
      paginator.setTo(jsonAsInt(jsonObject,"to",-1));
      paginator.setTotal(jsonAsInt(jsonObject,"total",-1));
      paginator.setLast_page(jsonAsInt(jsonObject, "last_page", -1));
      paginator.setNextPage(); // calculate next page
      return paginator;
  }
  };
  

Answer 1

Как уже упоминалось Адам Миллерчип , вам нужно обрабатывать каждый элемент в одной подписке на выборку. Вот пример:

List<Integer> dataSource = new ArrayList<>(10);

    public void fetch(int bufferSize) {
        Observable.fromIterable(dataSource) //Use special constructor to get stream from the iterable
                .buffer(bufferSize) //Take N items from the stream...
                .flatMapIterable((bunch) -> bunch) //... and then process them separately
                .subscribe(this::processItem); //here you can get every item arriving from the buffer
    }

После опустошения буфера - другая часть данных будет извлечена и передана в буфер. И так до тех пор, пока ваш источник Observable (Observable.fromIterable(dataSource)) не будет излучать onComplete или onError.

Answer 2

Чтобы ответить на ваш первый вопрос:

В этой реализации элементы обрабатываются в конце (подписка (System.out :: println)), когда все страницы загружены. "

Это неверно. Весь смысл реактивного программирования состоит в том, чтобы избежать этого. fetchItems() возвращает Flowable<T>, который фактически не извлекает какие-либо элементы, пока что-то не подпишется на него. Когда вы подписываетесь на что-то, подписчик получает уведомление каждый раз, когда товар готов. Вы должны вызвать subscribe() и передать функцию, которая будет вызываться каждый раз, когда элемент готов. В моем примере я передаю System.out::println, который печатает значения, но вы можете реализовать свой собственный обработчик, который сохраняет в базу данных.

Я бы предпочел обрабатывать их (сохранение базы данных) сразу же после их загрузки (в .doOnNext (page -> {})

Это сбивает с толку разницу между издателем и потребителем. Издатель производит предметы - в моем примере это Flowable<T>, который производит предметы типа T. Потребитель потребляет товары, которые производит издатель. doOnNext() является функцией издателя . Там написано: «Когда вы что-то публикуете, тоже делайте этот побочный эффект». В моем примере побочным эффектом является выдача номера следующей страницы для извлечения. Вы не должны обрабатывать сохранение БД там, вы должны написать свою собственную функцию обратного вызова ( Потребитель ) или Подписчик , чтобы обработать ее, и предоставить это для вызова подписки.