@@ -0,0 +1,1476 @@
+package com.github.davidmoten.rx.operators;%0A%0Aimport java.util.ArrayList;%0Aimport java.util.List;%0A%0Aimport org.junit.Test;%0A%0Aimport rx.Observable;%0Aimport rx.Observer;%0Aimport rx.functions.Func0;%0Aimport rx.functions.Func3;%0A%0Aimport com.github.davidmoten.rx.Transformers;%0A%0Apublic class TransformerWithStateTest %7B%0A%0A    @Test%0A    public void testBufferThree() %7B%0A        Func0%3CBuffer%3E initialState = new Func0%3CBuffer%3E() %7B%0A%0A            @Override%0A            public Buffer call() %7B%0A                return new Buffer();%0A            %7D%0A        %7D;%0A%0A        Func3%3CBuffer, Object, Observer%3CObject%3E, Buffer%3E transition = new Func3%3CBuffer, Object, Observer%3CObject%3E, Buffer%3E() %7B%0A%0A            @Override%0A            public Buffer call(Buffer buffer, Object t, Observer%3CObject%3E observer) %7B%0A                List%3CObject%3E list = new ArrayList%3CObject%3E(buffer.list);%0A                list.add(t);%0A                if (list.size() == 3) %7B%0A                    observer.onNext(list);%0A                    return new Buffer();%0A                %7D else %7B%0A                    return new Buffer(list);%0A                %7D%0A            %7D%0A        %7D;%0A        Observable.just(1, 2, 3, 4, 5).compose(Transformers.withState(initialState, transition))%0A                .toList().toBlocking().single();%0A    %7D%0A%0A    private static class Buffer %7B%0A        List%3CObject%3E list;%0A%0A        Buffer(List%3CObject%3E list) %7B%0A            this.list = list;%0A        %7D%0A%0A        Buffer() %7B%0A            this(new ArrayList%3CObject%3E());%0A        %7D%0A%0A    %7D%0A%7D%0A

add test