在我的 quarkus 服务中,我正在构建一个自定义累加器来使用Multi
. 流可能是无限的,我想知道在积累了足够的数据后如何提前终止并收集结果?
下面是我的原型:
Multi<Data> sortedStream = getStream();
return this.sortedStream.collectItems().in(
LinkedList::new,
new BiConsumer<LinkedList<Coverage>, Data>() {
@Override
public void accept(LinkedList<Coverage> coverages, Data incoming) {
if (coverages.isEmpty()) {
coverages.add(new Coverage(incoming));
return;
}
if (enough(coverages)) {
// Question: How to early terminate and collect coverage downstream?
}
Coverage last = coverages.getLast();
if (worthAdd(last, incoming)) {
coverages.add(new Coverage(incoming));
} else {
return;
}
}
}
)