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

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

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

标签：rx-java2,rx-java,kotlin,rx-kotlin,java
来源： https://codeday.me/bug/20191210/2104473.html