最佳答案
您可以使用缓冲区运算符.
PublishSubject<Token<Integer>> s = PublishSubject.create();
Observable<Token<Integer>> markers = s.filter(x->x.isMarker());
s.buffer(markers).subscribe(
v->{
Optional<Integer> reduce = v.stream()
.filter(t->!t.isMarker())
.map(t->(Valuetoken<Integer>)t)
.map(Valuetoken::get)
.reduce((a,b)->a+b);
reduce.ifPresent(System.out::println);
}
);
s.onNext(value(12));
s.onNext(value(13));
s.onNext(marker()); // will emit 25
s.onNext(value(10));
s.onNext(value(7));
s.onNext(marker()); // will emit 17
s.onNext(value(10));
s.onNext(value(7)); // Not emitting yet
public abstract class Token<T> {
private static final MarkerToken MARKER = new MarkerToken<>();
public boolean isMarker() {
return false;
}
public static <T> MarkerToken<T> marker() {
return MARKER;
}
public static <T> Valuetoken<T> value(T o) {
return new Valuetoken<>(o);
}
public static class Valuetoken<T> extends Token<T> {
T value;
public Valuetoken(T value) {
this.value = value;
}
public T get() {
return value;
}
}
public static class MarkerToken<T> extends Token<T> {
public boolean isMarker() {
return true;
}
}
}
更新(使用扫描)
前一种方法也将在流关闭时发出,使用此解决方案,您只能发出完整的缓冲区.
消息类用作累加器,它将累加令牌,直到累加结束标记为止.
发生这种情况时,下一条消息将从头开始.
public static class Message<T> {
List<Token<T>> tokens = new ArrayList<>();
public Message<T> append(Token<T> t) {
Message<T> mx = new Message<T>();
if(!isComplete()) {
mx.tokens.addAll(tokens);
}
mx.tokens.add(t);
return mx;
}
public boolean isComplete() {
int n = tokens.size();
return n>0 && tokens.get(n-1).isMarker();
}
public Optional<List<Token<T>>> fullMessage(){
return isComplete() ? Optional.of(tokens):Optional.empty();
}
}
扫描源,您为发出的每个令牌发出消息,然后过滤掉不完整的消息,仅发出标记为已完成的消息.
s.scan(new Message<Integer>(),(a,b) -> a.append(b))
.filter(Message::isComplete)
.map(Message::fullMessage)
.map(Optional::get).subscribe(v -> {
System.out.println(v);
});
s.onNext(value(12));
s.onNext(value(13));
s.onNext(marker());// [V(12),V(13),MARKER]
s.onNext(value(10));
s.onNext(value(7));
s.onNext(marker()); // [V(10),V(7),MARKER]
s.onNext(value(10));
s.onNext(value(127));
s.onComplete(); // Not emitting incomplete messages on the closing of the subject.
原文地址:https://www.jb51.cc/java/532951.html
版权声明:本文内容由互联网用户自发贡献,该文观点与技术仅代表作者本人。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如发现本站有涉嫌侵权/违法违规的内容, 请发送邮件至 dio@foxmail.com 举报,一经查实,本站将立刻删除。