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

内容导读

互联网集市收集整理的这篇技术教程文章主要介绍了选取元素直到某个字符并将其与RxJava分组，小编现在分享给大家，供广大互联网技能从业者学习和参考。文章包含2582字，纯文字阅读大概需要4分钟。

内容图文

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

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

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

我尝试过使用不同的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.

内容总结

以上是互联网集市为您收集整理的选取元素直到某个字符并将其与RxJava分组全部内容，希望文章能够帮你解决选取元素直到某个字符并将其与RxJava分组所遇到的程序开发问题。 如果觉得互联网集市技术教程内容还不错，欢迎将互联网集市网站推荐给程序员好友。

内容备注

版权声明：本文内容由互联网用户自发贡献，该文观点与技术仅代表作者本人。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如发现本站有涉嫌侵权/违法违规的内容， 请发送邮件至 gblab@vip.qq.com 举报，一经查实，本站将立刻删除。

内容手机端