@@ -0,0 +1,3347 @@
+package com.github.davidmoten.rx2.internal.flowable;%0A%0Aimport static org.junit.Assert.assertEquals;%0A%0Aimport java.util.ArrayList;%0Aimport java.util.Arrays;%0Aimport java.util.List;%0Aimport java.util.concurrent.atomic.AtomicLong;%0Aimport java.util.concurrent.atomic.AtomicReference;%0A%0Aimport org.junit.Test;%0Aimport org.reactivestreams.Subscriber;%0Aimport org.reactivestreams.Subscription;%0A%0Aimport io.reactivex.Flowable;%0Aimport io.reactivex.internal.subscriptions.SubscriptionHelper;%0A%0Apublic class FlowablePassThroughTest %7B%0A%0A    @Test%0A    public void test() %7B%0A        new FlowablePassThrough%3CInteger%3E(Flowable.just(1, 2, 3)) //%0A                .test() //%0A                .assertValues(1, 2, 3) //%0A                .assertComplete();%0A    %7D%0A%0A    @Test%0A    public void testSubscriber() %7B%0A        final List%3CInteger%3E list = new ArrayList%3CInteger%3E();%0A        new FlowablePassThrough%3CInteger%3E(Flowable.just(1, 2, 3)) //%0A                .subscribe(new Subscriber%3CInteger%3E() %7B%0A%0A                    private Subscription s;%0A%0A                    @Override%0A                    public void onSubscribe(Subscription s) %7B%0A                        this.s = s;%0A                        s.request(Long.MAX_VALUE);%0A                    %7D%0A%0A                    @Override%0A                    public void onNext(Integer t) %7B%0A                        list.add(t);%0A                    %7D%0A%0A                    @Override%0A                    public void onError(Throwable t) %7B%0A%0A                    %7D%0A%0A                    @Override%0A                    public void onComplete() %7B%0A%0A                    %7D%0A                %7D);%0A        assertEquals(Arrays.asList(1, 2, 3), list);%0A    %7D%0A%0A    static final class FlowablePassThrough%3CT%3E extends Flowable%3CT%3E %7B%0A%0A        private final Flowable%3CT%3E source;%0A%0A        public FlowablePassThrough(Flowable%3CT%3E source) %7B%0A            this.source = source;%0A        %7D%0A%0A        @Override%0A        protected void subscribeActual(Subscriber%3C? super T%3E child) %7B%0A            source.subscribe(new PassThroughSubscriber%3CT%3E(child));%0A        %7D%0A%0A        @SuppressWarnings(%22serial%22)%0A        private static final class PassThroughSubscriber%3CT%3E extends AtomicLong%0A                implements Subscriber%3CT%3E, Subscription %7B%0A%0A            private final Subscriber%3C? super T%3E child;%0A            private AtomicReference%3CSubscription%3E parent = new AtomicReference%3CSubscription%3E();%0A%0A            public PassThroughSubscriber(Subscriber%3C? super T%3E child) %7B%0A                this.child = child;%0A                child.onSubscribe(this);%0A            %7D%0A%0A            @Override%0A            public void onSubscribe(Subscription parent) %7B%0A                SubscriptionHelper.deferredSetOnce(this.parent, this, parent);%0A            %7D%0A%0A            @Override%0A            public void request(long n) %7B%0A                SubscriptionHelper.deferredRequest(parent, this, n);%0A            %7D%0A%0A            @Override%0A            public void cancel() %7B%0A                SubscriptionHelper.cancel(parent);%0A            %7D%0A%0A            @Override%0A            public void onNext(T t) %7B%0A                System.out.println(t);%0A                child.onNext(t);%0A            %7D%0A%0A            @Override%0A            public void onError(Throwable t) %7B%0A                child.onError(t);%0A            %7D%0A%0A            @Override%0A            public void onComplete() %7B%0A                child.onComplete();%0A            %7D%0A%0A        %7D%0A%0A    %7D%0A%0A%7D%0A

add test of pass through