选取元素直到某个字符并将其与RxJava分组

前端之家收集整理的这篇文章主要介绍了选取元素直到某个字符并将其与RxJava分组 前端之家小编觉得挺不错的,现在分享给大家,也给大家做个参考。

我对问题有一个简单的设置,但解决方案似乎更复杂.

设置:我有一个热的观察者,它起源于扫描仪,扫描仪将在编码完成时发出每个数字作为不同的元素,并发出R.

问题:从这开始,我想要一个热的可观察对象,它将每个完整代码作为1个元素发出.

RxMarbles example

我尝试过使用不同的flatMap,takeUntil和groupByoperators,但是还没有找到解决方案.

最佳答案
您可以使用缓冲区运算符.

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.f2er.com/java/532951.html

猜你在找的Java相关文章