@@ -0,0 +1,3797 @@
+package au.gov.amsa.util.rx;%0A%0Aimport java.util.ArrayList;%0Aimport java.util.List;%0A%0Aimport rx.Observable.Operator;%0Aimport rx.Subscriber;%0Aimport rx.functions.Func1;%0A%0A/**%0A * Buffers items into lists that can overlap but not more than two at a time.%0A * The commencement of the next buffer is signalled by a change in the value of%0A * %7B@code startFunction%7D and the finish of a buffer is signalled by a change in%0A * the value of the %7B@code whileFunction%7D.%0A *%0A * @param %3CT%3E%0A *            generic type%0A */%0Apublic class OperatorDynamicBuffer%3CT%3E implements Operator%3CList%3CT%3E, T%3E %7B%0A%0A    private static final Object NOT_SET = new Object();%0A    private static final Object NULL_SENTINEL = new Object();%0A    private final Func1%3CT, ?%3E startFunction;%0A    private final Func1%3CT, ?%3E whileFunction;%0A%0A    /**%0A     * Constructor.%0A     * %0A     * @param startFunction%0A     *            a change in the value of this function signals the start of a%0A     *            new buffer%0A     * @param whileFunction%0A     *            a change in the value of this function signals the finish of a%0A     *            buffer%0A     */%0A    public OperatorDynamicBuffer(Func1%3CT, ?%3E startFunction, Func1%3CT, ?%3E whileFunction) %7B%0A        this.startFunction = startFunction;%0A        this.whileFunction = whileFunction;%0A    %7D%0A%0A    @Override%0A    public Subscriber%3C? super T%3E call(Subscriber%3C? super List%3CT%3E%3E child) %7B%0A        return new Subscriber%3CT%3E(child) %7B%0A%0A            private List%3CT%3E list;%0A            private List%3CT%3E list2;%0A            private Object lastStartValue = NOT_SET;%0A            private Object lastWhileValue = NOT_SET;%0A%0A            @Override%0A            public void onCompleted() %7B%0A                if (list != null && !isUnsubscribed())%0A                    child.onNext(list);%0A                if (list2 != null && !isUnsubscribed())%0A                    child.onNext(list2);%0A                if (!isUnsubscribed())%0A                    child.onCompleted();%0A            %7D%0A%0A            @Override%0A            public void onError(Throwable e) %7B%0A                child.onError(e);%0A            %7D%0A%0A            @Override%0A            public void onNext(T t) %7B%0A                if (list == null)%0A                    list = new ArrayList%3CT%3E();%0A                Object startValue = replaceNull(startFunction.call(t));%0A                Object whileValue = replaceNull(whileFunction.call(t));%0A                if (lastWhileValue == NOT_SET) %7B%0A                    list.add(t);%0A                    request(1);%0A                %7D else %7B%0A                    if (lastWhileValue.equals(whileValue)) %7B%0A                        list.add(t);%0A                        request(1);%0A                    %7D else %7B%0A                        // emit list 1%0A                        child.onNext(list);%0A                        list = null;%0A                    %7D%0A                    if (lastStartValue != NOT_SET && !lastStartValue.equals(startValue)) %7B%0A                        // startFunction value has changed so we are ready to%0A                        // start another buffer%0A                        if (list2 == null) %7B%0A                            list2 = new ArrayList%3CT%3E();%0A                        %7D else %7B%0A                            throw new RuntimeException(%22unexpected%22);%0A                        %7D%0A                        list2.add(t);%0A                    %7D%0A                    if (list == null && list2 != null) %7B%0A                        // move second list into first position%0A                        list = list2;%0A                        list2 = null;%0A                    %7D%0A                %7D%0A                lastStartValue = startValue;%0A                lastWhileValue = whileValue;%0A            %7D%0A        %7D;%0A    %7D%0A%0A    private static Object replaceNull(Object o) %7B%0A        if (o == null)%0A            return NULL_SENTINEL;%0A        else%0A            return o;%0A    %7D%0A%7D%0A

add OperatorDynamicBuffer