RxJava：缓冲项目，直到当前项目满足某些条件为止

Question

这是我要弄清楚的一个片段：

class RaceCondition {

    Subject<Integer, Integer> subject = PublishSubject.create();

    public void entryPoint(Integer data) {
        subject.onNext(data);
    }

    public void client() {
        subject /*some operations*/
                .buffer(getClosingSelector())
                .subscribe(/*handle results*/);
    }

    private Observable<Integer> getClosingSelector() {
        return subject /* some filtering */;
    }
}

有一个Subject接受来自外部的事件。有一个客户订阅了该主题，该主题可以处理事件并buffer对其进行处理。这里的主要思想是，应根据使用流中的项目计算出的某些条件每次都发出缓冲的项目。

为此，缓冲区边界本身会监听主体。

一个重要的期望行为：每当边界发射该项目时，也应将其包括在以下的发射中buffer。当前配置不是这种情况，因为该项目（至少是我认为的）在到达之前从关闭选择器中发出buffer，因此它不包括在当前发出中，而是留在等待下一个发出。

有没有一种方法可以使关闭选择器本质上等待项目被首先缓冲？如果不是，是否还有另一种方法可以基于下一个传入项目来缓冲和释放项目？

Answer 1

如果我理解正确，则需要缓冲，直到某些谓词允许基于项目为止。您可以使用一组复杂的运算符来执行此操作，但编写自定义运算符可能更容易：

public final class BufferUntil<T> 
implements Operator<List<T>, T>{

    final Func1<T, Boolean> boundaryPredicate;

    public BufferUntil(Func1<T, Boolean> boundaryPredicate) {
        this.boundaryPredicate = boundaryPredicate;
    }

    @Override
    public Subscriber<? super T> call(
            Subscriber<? super List<T>> child) {
        BufferWhileSubscriber parent = 
                new BufferWhileSubscriber(child);
        child.add(parent);
        return parent;
    }

    final class BufferWhileSubscriber extends Subscriber<T> {
        final Subscriber<? super List<T>> actual;

        List<T> buffer = new ArrayList<>();

        /**
         * @param actual
         */
        public BufferWhileSubscriber(
                Subscriber<? super List<T>> actual) {
            this.actual = actual;
        }

        @Override
        public void onNext(T t) {
            buffer.add(t);
            if (boundaryPredicate.call(t)) {
                actual.onNext(buffer);
                buffer = new ArrayList<>();
            }
        }

        @Override
        public void onError(Throwable e) {
            buffer = null;
            actual.onError(e);
        }

        @Override
        public void onCompleted() {
            List<T> b = buffer;
            buffer = null;
            if (!b.isEmpty()) {
                actual.onNext(b);
            }
            actual.onCompleted();
        }
    }
}