JavaRX分页-在每个时间间隔而不是最后观察-通用分页器 [英] JavaRX Pagination - observe in each interration rather than at the end - Generic Paginator
问题描述
我正在使用分页API. 我使用了 Adam Millerchip 提供的以下解决方案，并且效果很好.
I'm working with a paginated API. I have used the following solution provided by Adam Millerchip and it works well.
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();
}
}
}
但是我有两个问题:
-
在此实现中，在加载所有页面时在结尾处处理元素(订阅(System.out :: println)).如果收集的数据很多，这可能会导致内存问题.我宁愿在加载它们时(在.doOnNext(page-> {})中立即处理它们(数据库保存).我已经能够做到这一点，但是以一种肮脏的方式"(在数据库中添加数据库保存代码) doOnNext).我该怎么做?
In this implementation elements are processed at the end (subscribe(System.out::println)) when all pages are loaded. This may cause memory problems if gathered data are numerous. I would prefer to process them (data base save) immediately when they are loaded (in the .doOnNext(page -> { }). I have been able to do it but in a "dirty way" (add database save code in the doOnNext). How can I do this ?
在"page"类的实现中，我使用了自定义的Gson解串器.而且我不知道如何处理通用数据.我不得不写"list.add(( MyGenericClass )context.deserialize(anArray.getAsJsonObject()， MyGenericClass .class))"；"我想要类似"list.add(( T )context.deserialize(anArray.getAsJsonObject()， T .class));"的地方.我如何才能保持真正的通用性?
in my implementation of the "page" class I use a custom Gson deserializer. And I don't know how to deal with Generic data. I have had to write "list.add((MyGenericClass)context.deserialize(anArray.getAsJsonObject(), MyGenericClass.class));" where I would want something like "list.add((T)context.deserialize(anArray.getAsJsonObject(), T.class));". How can I keep things realy generic ?
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;
}
};
推荐答案
如 Adam Millerchip 所述，您需要在单个访存订阅中处理每个项目.这是一个示例:
As Adam Millerchip mentioned, you need to process every item in single fetch subscription. Here is an example:
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.
After the buffer emptied - another part of the data will be fetched and passed to the buffer. And so on till your source Observable (Observable.fromIterable(dataSource)) will emit onComplete or onError.
这篇关于JavaRX分页-在每个时间间隔而不是最后观察-通用分页器的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！
