如何獲取 Flux 的最后一項而不用 reduce() 或 last() 折疊它？這是我的用例：1）我有Flux<T>根據狀態生成的生成器。2）當內部Flux完成時，它會改變影響我在生成器中發出的下一個對象的狀態Flux。示意性地看起來像這樣static class State { int secret = 2; int iteration = 0;}Random rand = new Random(1024);Flux<Integer> stream = Flux.<Flux<Integer>, State>generate(State::new, (state, sink) -> { System.out.println(String.format("Generate: %d", state.secret)); Flux<Integer> inner = Flux.range(1, rand.nextInt(10)); sink.next(inner.doOnComplete(() -> { // How do I get last item of `inner` here ? // For example I'd like to decrement `state.secret` by last value of `inner` })); return state;}).flatMap(Function.identity());UPD：我沒有標記我的答案，因為黑客被證明是不可靠的。有可能.generate()在前一個被完全消耗之前被調用Flux，因此使得值last不正確。
1 回答
幕布斯6054654
TA貢獻1876條經驗 獲得超7個贊
第一個版本不可靠。我又黑了一個：
static <T> Flux<T> expandOnLastItem(Supplier<Flux<T>> seed, Function<T, Flux<T>> generator) {
return Flux.just(new AtomicReference<T>())
.flatMap(last -> Flux.just(seed.get().materialize())
.flatMap(Function.identity())
.expand(v -> {
if (v.hasValue()) {
last.set(v.get());
} else if (v.isOnComplete() && last.get() != null) {
Flux<T> res = generator.apply(last.get());
last.set(null);
return res.materialize();
}
return Flux.empty();
})
.filter(s -> !s.isOnComplete())
.dematerialize());
}
可以用作
static Flux<Integer> getPage(int pageId, int size) {
return Flux.defer(() -> {
if (pageId < 3) {
System.out.println("Returning data for pageId: " + pageId);
return Flux.range(pageId * 100, size);
} else {
System.out.println("Returning empty for pageId: " + pageId);
return Flux.empty();
}
});
}
expandOnLastItem(
() -> getPage(0, 5),
lastId -> {
System.out.println(" Expanding. Last item: " + lastId);
int curPage = lastId / 100;
return getPage(curPage + 1, 5);
})
.reduce(0L, (count, value) -> {
System.out.println("==> " + value);
return count + 1;
})
.block();
所以我通過改變生成器中的狀態變量來破解它。它可以工作，但功能不是很好。如果其他人可以提出替代方案，我將不勝感激。
Random rand = new Random(1024);
Flux.<Flux<String>, State>generate(State::new, (state, sink) -> {
if (state.iteration < 4) {
final int count = rand.nextInt(10) + 1;
System.out.println(String.format("*** Generate %d: start %d (count %d)", state.iteration, state.secret, count));
Flux<Integer> inner = Flux.range(state.secret, count);
final int[] last = {Integer.MIN_VALUE};
sink.next(
inner
.doOnNext(value -> {
last[0] = value;
})
.map(value -> String.format("Iter %d value %d", state.iteration, value))
.doOnComplete(() -> {
System.out.println(String.format("Inner complete (last item was %d)", last[0]));
state.secret = last[0];
state.iteration += 1;
}));
} else {
System.out.println("Generate complete");
sink.complete();
}
return state;
})
.flatMap(Function.identity())
.map(value -> {
System.out.println(String.format("Ext map: %s", value));
return value;
})
.buffer(5)
.flatMapIterable(Function.identity())
.subscribe(value -> System.out.println(String.format(" ---> %s", value)));
System.out.println("Exiting");
添加回答
舉報
0/150
提交
取消