}
        }).toObservable().blockingSingle();

        assertEquals(6, value);
    }

    @SuppressWarnings("unused")
    @Test
    public void reduceWithObjectsObservable() {

            

Reported by PMD.

                          }
        }).toObservable();

        assertNotNull(reduceResult2);
    }

    /**
     * Reduce consumes and produces T so can't do covariance.
     *

            

Reported by PMD.

                          }
        }).toObservable();

        assertNotNull(reduceResult2);
    }

    @Test
    public void reduceInts() {
        Observable<Integer> o = Observable.just(1, 2, 3);

            

Reported by PMD.

                          }
        }).blockingGet();

        assertEquals(6, value);
    }

    @SuppressWarnings("unused")
    @Test
    public void reduceWithObjects() {

            

Reported by PMD.

                          }
        });

        assertNotNull(reduceResult2);
    }

    /**
     * Reduce consumes and produces T so can't do covariance.
     *

            

Reported by PMD.

                          }
        });

        assertNotNull(reduceResult2);
    }

    /**
     * Reduce consumes and produces T so can't do covariance.
     *

            

Reported by PMD.

                   * https://github.com/ReactiveX/RxJava/issues/360#issuecomment-24203016
     */
    @Test
    public void reduceCovariance() {
        // must type it to <Movie>
        Observable<Movie> horrorMovies = Observable.<Movie> just(new HorrorMovie());
        libraryFunctionActingOnMovieObservables(horrorMovies);
    }


            

Reported by PMD.

              
package io.reactivex.rxjava3.observable;

import static org.junit.Assert.*;

import org.junit.Test;

import io.reactivex.rxjava3.core.*;
import io.reactivex.rxjava3.functions.BiFunction;

            

Reported by PMD.

              
import org.junit.Test;

import io.reactivex.rxjava3.core.*;
import io.reactivex.rxjava3.functions.BiFunction;
import io.reactivex.rxjava3.observable.ObservableCovarianceTest.*;

public class ObservableReduceTests extends RxJavaTest {


            

Reported by PMD.

              
import io.reactivex.rxjava3.core.*;
import io.reactivex.rxjava3.functions.BiFunction;
import io.reactivex.rxjava3.observable.ObservableCovarianceTest.*;

public class ObservableReduceTests extends RxJavaTest {

    @Test
    public void reduceIntsObservable() {

            

Reported by PMD.

                          .test()
            .assertResult();

        assertEquals(1, atomicInteger.get());
    }

    @Test
    public void fromRunnableTwice() {
        final AtomicInteger atomicInteger = new AtomicInteger();

            

Reported by PMD.

                  }

    @Test
    public void fromRunnableTwice() {
        final AtomicInteger atomicInteger = new AtomicInteger();

        Runnable run = new Runnable() {
            @Override
            public void run() {

            

Reported by PMD.

                          }
        };

        Completable.fromRunnable(run)
            .test()
            .assertResult();

        assertEquals(1, atomicInteger.get());


            

Reported by PMD.

                          }
        };

        Completable.fromRunnable(run)
            .test()
            .assertResult();

        assertEquals(1, atomicInteger.get());


            

Reported by PMD.

                          .test()
            .assertResult();

        assertEquals(1, atomicInteger.get());

        Completable.fromRunnable(run)
            .test()
            .assertResult();


            

Reported by PMD.

              
        assertEquals(1, atomicInteger.get());

        Completable.fromRunnable(run)
            .test()
            .assertResult();

        assertEquals(2, atomicInteger.get());
    }

            

Reported by PMD.

              
        assertEquals(1, atomicInteger.get());

        Completable.fromRunnable(run)
            .test()
            .assertResult();

        assertEquals(2, atomicInteger.get());
    }

            

Reported by PMD.

                          .test()
            .assertResult();

        assertEquals(2, atomicInteger.get());
    }

    @Test
    public void fromRunnableInvokesLazy() {
        final AtomicInteger atomicInteger = new AtomicInteger();

            

Reported by PMD.

                  }

    @Test
    public void fromRunnableInvokesLazy() {
        final AtomicInteger atomicInteger = new AtomicInteger();

        Completable completable = Completable.fromRunnable(new Runnable() {
            @Override
            public void run() {

            

Reported by PMD.

                          }
        });

        assertEquals(0, atomicInteger.get());

        completable
            .test()
            .assertResult();


            

Reported by PMD.

                          .test()
            .assertResult();

        assertEquals(1, atomicInteger.get());
    }

    @Test
    public void fromActionTwice() {
        final AtomicInteger atomicInteger = new AtomicInteger();

            

Reported by PMD.

                  }

    @Test
    public void fromActionTwice() {
        final AtomicInteger atomicInteger = new AtomicInteger();

        Action run = new Action() {
            @Override
            public void run() throws Exception {

            

Reported by PMD.

                          }
        };

        Completable.fromAction(run)
            .test()
            .assertResult();

        assertEquals(1, atomicInteger.get());


            

Reported by PMD.

                          }
        };

        Completable.fromAction(run)
            .test()
            .assertResult();

        assertEquals(1, atomicInteger.get());


            

Reported by PMD.

                          .test()
            .assertResult();

        assertEquals(1, atomicInteger.get());

        Completable.fromAction(run)
            .test()
            .assertResult();


            

Reported by PMD.

              
        assertEquals(1, atomicInteger.get());

        Completable.fromAction(run)
            .test()
            .assertResult();

        assertEquals(2, atomicInteger.get());
    }

            

Reported by PMD.

              
        assertEquals(1, atomicInteger.get());

        Completable.fromAction(run)
            .test()
            .assertResult();

        assertEquals(2, atomicInteger.get());
    }

            

Reported by PMD.

                          .test()
            .assertResult();

        assertEquals(2, atomicInteger.get());
    }

    @Test
    public void fromActionInvokesLazy() {
        final AtomicInteger atomicInteger = new AtomicInteger();

            

Reported by PMD.

                  }

    @Test
    public void fromActionInvokesLazy() {
        final AtomicInteger atomicInteger = new AtomicInteger();

        Completable completable = Completable.fromAction(new Action() {
            @Override
            public void run() throws Exception {

            

Reported by PMD.

                          }
        });

        assertEquals(0, atomicInteger.get());

        completable
            .test()
            .assertResult();


            

Reported by PMD.

                          }
        }).toFlowable().blockingSingle();

        assertEquals(6, value);
    }

    @SuppressWarnings("unused")
    @Test
    public void reduceWithObjectsFlowable() {

            

Reported by PMD.

                          }
        }).toFlowable();

        assertNotNull(reduceResult2);
    }

    /**
     * Reduce consumes and produces T so can't do covariance.
     *

            

Reported by PMD.

                          }
        }).toFlowable();

        assertNotNull(reduceResult2);
    }

    @Test
    public void reduceInts() {
        Flowable<Integer> f = Flowable.just(1, 2, 3);

            

Reported by PMD.

                          }
        }).toFlowable().blockingSingle();

        assertEquals(6, value);
    }

    @SuppressWarnings("unused")
    @Test
    public void reduceWithObjects() {

            

Reported by PMD.

                          }
        });

        assertNotNull(reduceResult2);
    }

    /**
     * Reduce consumes and produces T so can't do covariance.
     *

            

Reported by PMD.

                          }
        });

        assertNotNull(reduceResult2);
    }

    /**
     * Reduce consumes and produces T so can't do covariance.
     *

            

Reported by PMD.

                   * https://github.com/ReactiveX/RxJava/issues/360#issuecomment-24203016
     */
    @Test
    public void reduceCovariance() {
        // must type it to <Movie>
        Flowable<Movie> horrorMovies = Flowable.<Movie> just(new HorrorMovie());
        libraryFunctionActingOnMovieObservables(horrorMovies);
    }


            

Reported by PMD.

              
package io.reactivex.rxjava3.flowable;

import static org.junit.Assert.*;

import org.junit.Test;

import io.reactivex.rxjava3.core.*;
import io.reactivex.rxjava3.flowable.FlowableCovarianceTest.*;

            

Reported by PMD.

              
import org.junit.Test;

import io.reactivex.rxjava3.core.*;
import io.reactivex.rxjava3.flowable.FlowableCovarianceTest.*;
import io.reactivex.rxjava3.functions.BiFunction;

public class FlowableReduceTests extends RxJavaTest {


            

Reported by PMD.

              import org.junit.Test;

import io.reactivex.rxjava3.core.*;
import io.reactivex.rxjava3.flowable.FlowableCovarianceTest.*;
import io.reactivex.rxjava3.functions.BiFunction;

public class FlowableReduceTests extends RxJavaTest {

    @Test

            

Reported by PMD.

                  }

    @Override
    public void subscribe(Subscriber<? super R>[] subscribers) {
        subscribers = RxJavaPlugins.onSubscribe(this, subscribers);

        if (!validate(subscribers)) {
            return;
        }

            

Reported by PMD.

               */
public final class ParallelMap<T, R> extends ParallelFlowable<R> {

    final ParallelFlowable<T> source;

    final Function<? super T, ? extends R> mapper;

    public ParallelMap(ParallelFlowable<T> source, Function<? super T, ? extends R> mapper) {
        this.source = source;

            

Reported by PMD.

              
    final ParallelFlowable<T> source;

    final Function<? super T, ? extends R> mapper;

    public ParallelMap(ParallelFlowable<T> source, Function<? super T, ? extends R> mapper) {
        this.source = source;
        this.mapper = mapper;
    }

            

Reported by PMD.

              
    static final class ParallelMapSubscriber<T, R> implements FlowableSubscriber<T>, Subscription {

        final Subscriber<? super R> downstream;

        final Function<? super T, ? extends R> mapper;

        Subscription upstream;


            

Reported by PMD.

              
        final Subscriber<? super R> downstream;

        final Function<? super T, ? extends R> mapper;

        Subscription upstream;

        boolean done;


            

Reported by PMD.

              
        final Function<? super T, ? extends R> mapper;

        Subscription upstream;

        boolean done;

        ParallelMapSubscriber(Subscriber<? super R> actual, Function<? super T, ? extends R> mapper) {
            this.downstream = actual;

            

Reported by PMD.

              
        Subscription upstream;

        boolean done;

        ParallelMapSubscriber(Subscriber<? super R> actual, Function<? super T, ? extends R> mapper) {
            this.downstream = actual;
            this.mapper = mapper;
        }

            

Reported by PMD.

              
            try {
                v = Objects.requireNonNull(mapper.apply(t), "The mapper returned a null value");
            } catch (Throwable ex) {
                Exceptions.throwIfFatal(ex);
                cancel();
                onError(ex);
                return;
            }

            

Reported by PMD.

                  }
    static final class ParallelMapConditionalSubscriber<T, R> implements ConditionalSubscriber<T>, Subscription {

        final ConditionalSubscriber<? super R> downstream;

        final Function<? super T, ? extends R> mapper;

        Subscription upstream;


            

Reported by PMD.

              
        final ConditionalSubscriber<? super R> downstream;

        final Function<? super T, ? extends R> mapper;

        Subscription upstream;

        boolean done;


            

Reported by PMD.

                                  @Override
                    public void accept(Object pv) {
                        synchronized (pv) {
                            System.out.println(pv);
                        }
                    }
                });

        System.out.println("**** finished");

            

Reported by PMD.

                                  }
                });

        System.out.println("**** finished");

        Thread.sleep(200); // make sure the event streams receive their interrupt
    }

    /**

            

Reported by PMD.

                      Observable<Object> result = Observable.zip(observables, new Function<Object[], Object>() {
            @Override
            public Object apply(Object[] args) {
                System.out.println("received: " + args);
                Assert.assertEquals("No argument should have been passed", 0, args.length);
                return invoked;
            }
        });


            

Reported by PMD.

                  Consumer<Result> action = new Consumer<Result>() {
        @Override
        public void accept(Result t1) {
            System.out.println("Result: " + t1);
        }
    };

    Consumer<ExtendedResult> extendedAction = new Consumer<ExtendedResult>() {
        @Override

            

Reported by PMD.

                  Consumer<ExtendedResult> extendedAction = new Consumer<ExtendedResult>() {
        @Override
        public void accept(ExtendedResult t1) {
            System.out.println("Result: " + t1);
        }
    };

    @Test
    public void zipWithDelayError() {

            

Reported by PMD.

              public class ObservableZipTests extends RxJavaTest {

    @Test
    public void zipObservableOfObservables() throws Exception {
        ObservableEventStream.getEventStream("HTTP-ClusterB", 20)
                .groupBy(new Function<Event, String>() {
                    @Override
                    public String apply(Event e) {
                        return e.instanceId;

            

Reported by PMD.

                      Observable<HorrorMovie> horrors = Observable.just(new HorrorMovie());
        Observable<CoolRating> ratings = Observable.just(new CoolRating());

        Observable.<Movie, CoolRating, Result> zip(horrors, ratings, combine).blockingForEach(action);
        Observable.<Movie, CoolRating, Result> zip(horrors, ratings, combine).blockingForEach(action);
        Observable.<Media, Rating, ExtendedResult> zip(horrors, ratings, combine).blockingForEach(extendedAction);
        Observable.<Media, Rating, Result> zip(horrors, ratings, combine).blockingForEach(action);
        Observable.<Media, Rating, ExtendedResult> zip(horrors, ratings, combine).blockingForEach(action);


            

Reported by PMD.

                      Observable<HorrorMovie> horrors = Observable.just(new HorrorMovie());
        Observable<CoolRating> ratings = Observable.just(new CoolRating());

        Observable.<Movie, CoolRating, Result> zip(horrors, ratings, combine).blockingForEach(action);
        Observable.<Movie, CoolRating, Result> zip(horrors, ratings, combine).blockingForEach(action);
        Observable.<Media, Rating, ExtendedResult> zip(horrors, ratings, combine).blockingForEach(extendedAction);
        Observable.<Media, Rating, Result> zip(horrors, ratings, combine).blockingForEach(action);
        Observable.<Media, Rating, ExtendedResult> zip(horrors, ratings, combine).blockingForEach(action);


            

Reported by PMD.

                      Observable<CoolRating> ratings = Observable.just(new CoolRating());

        Observable.<Movie, CoolRating, Result> zip(horrors, ratings, combine).blockingForEach(action);
        Observable.<Movie, CoolRating, Result> zip(horrors, ratings, combine).blockingForEach(action);
        Observable.<Media, Rating, ExtendedResult> zip(horrors, ratings, combine).blockingForEach(extendedAction);
        Observable.<Media, Rating, Result> zip(horrors, ratings, combine).blockingForEach(action);
        Observable.<Media, Rating, ExtendedResult> zip(horrors, ratings, combine).blockingForEach(action);

        Observable.<Movie, CoolRating, Result> zip(horrors, ratings, combine);

            

Reported by PMD.

                      Observable<CoolRating> ratings = Observable.just(new CoolRating());

        Observable.<Movie, CoolRating, Result> zip(horrors, ratings, combine).blockingForEach(action);
        Observable.<Movie, CoolRating, Result> zip(horrors, ratings, combine).blockingForEach(action);
        Observable.<Media, Rating, ExtendedResult> zip(horrors, ratings, combine).blockingForEach(extendedAction);
        Observable.<Media, Rating, Result> zip(horrors, ratings, combine).blockingForEach(action);
        Observable.<Media, Rating, ExtendedResult> zip(horrors, ratings, combine).blockingForEach(action);

        Observable.<Movie, CoolRating, Result> zip(horrors, ratings, combine);

            

Reported by PMD.

              public class ObservableSubscribeOnTest extends RxJavaTest {

    @Test
    public void issue813() throws InterruptedException {
        // https://github.com/ReactiveX/RxJava/issues/813
        final CountDownLatch scheduled = new CountDownLatch(1);
        final CountDownLatch latch = new CountDownLatch(1);
        final CountDownLatch doneLatch = new CountDownLatch(1);


            

Reported by PMD.

                                  }

                    observer.onComplete();
                } catch (Throwable e) {
                    observer.onError(e);
                } finally {
                    doneLatch.countDown();
                }
            }

            

Reported by PMD.

                  }

    @Test
    public void onError() {
        TestObserverEx<String> to = new TestObserverEx<>();
        Observable.unsafeCreate(new ObservableSource<String>() {

            @Override
            public void subscribe(Observer<? super String> observer) {

            

Reported by PMD.

                  }

    public static class SlowScheduler extends Scheduler {
        final Scheduler actual;
        final long delay;
        final TimeUnit unit;

        public SlowScheduler() {
            this(Schedulers.computation(), 2, TimeUnit.SECONDS);

            

Reported by PMD.

              
    public static class SlowScheduler extends Scheduler {
        final Scheduler actual;
        final long delay;
        final TimeUnit unit;

        public SlowScheduler() {
            this(Schedulers.computation(), 2, TimeUnit.SECONDS);
        }

            

Reported by PMD.

                  public static class SlowScheduler extends Scheduler {
        final Scheduler actual;
        final long delay;
        final TimeUnit unit;

        public SlowScheduler() {
            this(Schedulers.computation(), 2, TimeUnit.SECONDS);
        }


            

Reported by PMD.

              
        private final class SlowInner extends Worker {

            private final Scheduler.Worker actualInner;

            private SlowInner(Worker actual) {
                this.actualInner = actual;
            }


            

Reported by PMD.

                          @Override
            public Disposable schedule(@NonNull final Runnable action, final long delayTime, @NonNull final TimeUnit delayUnit) {
                TimeUnit common = delayUnit.compareTo(unit) < 0 ? delayUnit : unit;
                long t = common.convert(delayTime, delayUnit) + common.convert(delay, unit);
                return actualInner.schedule(action, t, common);
            }

        }


            

Reported by PMD.

                          @Override
            public Disposable schedule(@NonNull final Runnable action, final long delayTime, @NonNull final TimeUnit delayUnit) {
                TimeUnit common = delayUnit.compareTo(unit) < 0 ? delayUnit : unit;
                long t = common.convert(delayTime, delayUnit) + common.convert(delay, unit);
                return actualInner.schedule(action, t, common);
            }

        }


            

Reported by PMD.

                      to.dispose();
        Thread.sleep(200); // give time for the loop to continue
        to.assertValues(1, 2, 3, 4, 5, 6, 7, 8, 9, 10);
        assertEquals(10, count.get());
    }

    @Test
    public void cancelBeforeActualSubscribe() {
        TestScheduler test = new TestScheduler();

            

Reported by PMD.

               * @param <U> the value type in the queue
 * @param <V> the value type the child subscriber accepts
 */
public abstract class QueueDrainSubscriber<T, U, V> extends QueueDrainSubscriberPad4 implements FlowableSubscriber<T>, QueueDrain<U, V> {

    protected final Subscriber<? super V> downstream;

    protected final SimplePlainQueue<U> queue;


            

Reported by PMD.

               */
public abstract class QueueDrainSubscriber<T, U, V> extends QueueDrainSubscriberPad4 implements FlowableSubscriber<T>, QueueDrain<U, V> {

    protected final Subscriber<? super V> downstream;

    protected final SimplePlainQueue<U> queue;

    protected volatile boolean cancelled;


            

Reported by PMD.

              
    protected final Subscriber<? super V> downstream;

    protected final SimplePlainQueue<U> queue;

    protected volatile boolean cancelled;

    protected volatile boolean done;
    protected Throwable error;

            

Reported by PMD.

              
    protected final SimplePlainQueue<U> queue;

    protected volatile boolean cancelled;

    protected volatile boolean done;
    protected Throwable error;

    public QueueDrainSubscriber(Subscriber<? super V> actual, SimplePlainQueue<U> queue) {

            

Reported by PMD.

              
    protected final SimplePlainQueue<U> queue;

    protected volatile boolean cancelled;

    protected volatile boolean done;
    protected Throwable error;

    public QueueDrainSubscriber(Subscriber<? super V> actual, SimplePlainQueue<U> queue) {

            

Reported by PMD.

              
    protected volatile boolean cancelled;

    protected volatile boolean done;
    protected Throwable error;

    public QueueDrainSubscriber(Subscriber<? super V> actual, SimplePlainQueue<U> queue) {
        this.downstream = actual;
        this.queue = queue;

            

Reported by PMD.

              
    protected volatile boolean cancelled;

    protected volatile boolean done;
    protected Throwable error;

    public QueueDrainSubscriber(Subscriber<? super V> actual, SimplePlainQueue<U> queue) {
        this.downstream = actual;
        this.queue = queue;

            

Reported by PMD.

                  protected volatile boolean cancelled;

    protected volatile boolean done;
    protected Throwable error;

    public QueueDrainSubscriber(Subscriber<? super V> actual, SimplePlainQueue<U> queue) {
        this.downstream = actual;
        this.queue = queue;
    }

            

Reported by PMD.

                  protected volatile boolean cancelled;

    protected volatile boolean done;
    protected Throwable error;

    public QueueDrainSubscriber(Subscriber<? super V> actual, SimplePlainQueue<U> queue) {
        this.downstream = actual;
        this.queue = queue;
    }

            

Reported by PMD.

              
        if (fastEnter()) {
            long r = requested.get();
            if (r != 0L) {
                if (accept(s, value)) {
                    if (r != Long.MAX_VALUE) {
                        produced(1);
                    }
                }

            

Reported by PMD.

                      }

        @Override
        public void onError(Throwable t) {
            if (done) {
                RxJavaPlugins.onError(t);
                return;
            }
            done = true;

            

Reported by PMD.

              import io.reactivex.rxjava3.plugins.RxJavaPlugins;

public final class ObservableDoOnEach<T> extends AbstractObservableWithUpstream<T, T> {
    final Consumer<? super T> onNext;
    final Consumer<? super Throwable> onError;
    final Action onComplete;
    final Action onAfterTerminate;

    public ObservableDoOnEach(ObservableSource<T> source, Consumer<? super T> onNext,

            

Reported by PMD.

              
public final class ObservableDoOnEach<T> extends AbstractObservableWithUpstream<T, T> {
    final Consumer<? super T> onNext;
    final Consumer<? super Throwable> onError;
    final Action onComplete;
    final Action onAfterTerminate;

    public ObservableDoOnEach(ObservableSource<T> source, Consumer<? super T> onNext,
                              Consumer<? super Throwable> onError,

            

Reported by PMD.

              public final class ObservableDoOnEach<T> extends AbstractObservableWithUpstream<T, T> {
    final Consumer<? super T> onNext;
    final Consumer<? super Throwable> onError;
    final Action onComplete;
    final Action onAfterTerminate;

    public ObservableDoOnEach(ObservableSource<T> source, Consumer<? super T> onNext,
                              Consumer<? super Throwable> onError,
                              Action onComplete,

            

Reported by PMD.

                  final Consumer<? super T> onNext;
    final Consumer<? super Throwable> onError;
    final Action onComplete;
    final Action onAfterTerminate;

    public ObservableDoOnEach(ObservableSource<T> source, Consumer<? super T> onNext,
                              Consumer<? super Throwable> onError,
                              Action onComplete,
                              Action onAfterTerminate) {

            

Reported by PMD.

                  }

    static final class DoOnEachObserver<T> implements Observer<T>, Disposable {
        final Observer<? super T> downstream;
        final Consumer<? super T> onNext;
        final Consumer<? super Throwable> onError;
        final Action onComplete;
        final Action onAfterTerminate;


            

Reported by PMD.

              
    static final class DoOnEachObserver<T> implements Observer<T>, Disposable {
        final Observer<? super T> downstream;
        final Consumer<? super T> onNext;
        final Consumer<? super Throwable> onError;
        final Action onComplete;
        final Action onAfterTerminate;

        Disposable upstream;

            

Reported by PMD.

              
    static final class DoOnEachObserver<T> implements Observer<T>, Disposable {
        final Observer<? super T> downstream;
        final Consumer<? super T> onNext;
        final Consumer<? super Throwable> onError;
        final Action onComplete;
        final Action onAfterTerminate;

        Disposable upstream;

            

Reported by PMD.

                  static final class DoOnEachObserver<T> implements Observer<T>, Disposable {
        final Observer<? super T> downstream;
        final Consumer<? super T> onNext;
        final Consumer<? super Throwable> onError;
        final Action onComplete;
        final Action onAfterTerminate;

        Disposable upstream;


            

Reported by PMD.

                  static final class DoOnEachObserver<T> implements Observer<T>, Disposable {
        final Observer<? super T> downstream;
        final Consumer<? super T> onNext;
        final Consumer<? super Throwable> onError;
        final Action onComplete;
        final Action onAfterTerminate;

        Disposable upstream;


            

Reported by PMD.

              
            @Override
            public boolean test(String t1) {
                return t1.equals("two");
            }
        });

        Observer<String> observer = TestHelper.mockObserver();


            

Reported by PMD.

              
            @Override
            public boolean test(String t1) {
                return t1.equals("two");
            }
        });

        Observer<String> observer = TestHelper.mockObserver();


            

Reported by PMD.

              
        observable.subscribe(observer);

        verify(observer, Mockito.never()).onNext("one");
        verify(observer, times(1)).onNext("two");
        verify(observer, Mockito.never()).onNext("three");
        verify(observer, Mockito.never()).onError(any(Throwable.class));
        verify(observer, times(1)).onComplete();
    }

            

Reported by PMD.

                      observable.subscribe(observer);

        verify(observer, Mockito.never()).onNext("one");
        verify(observer, times(1)).onNext("two");
        verify(observer, Mockito.never()).onNext("three");
        verify(observer, Mockito.never()).onError(any(Throwable.class));
        verify(observer, times(1)).onComplete();
    }


            

Reported by PMD.

              
        verify(observer, Mockito.never()).onNext("one");
        verify(observer, times(1)).onNext("two");
        verify(observer, Mockito.never()).onNext("three");
        verify(observer, Mockito.never()).onError(any(Throwable.class));
        verify(observer, times(1)).onComplete();
    }

    @Test

            

Reported by PMD.

                      verify(observer, Mockito.never()).onNext("one");
        verify(observer, times(1)).onNext("two");
        verify(observer, Mockito.never()).onNext("three");
        verify(observer, Mockito.never()).onError(any(Throwable.class));
        verify(observer, times(1)).onComplete();
    }

    @Test
    public void dispose() {

            

Reported by PMD.

                      verify(observer, times(1)).onNext("two");
        verify(observer, Mockito.never()).onNext("three");
        verify(observer, Mockito.never()).onError(any(Throwable.class));
        verify(observer, times(1)).onComplete();
    }

    @Test
    public void dispose() {
        TestHelper.checkDisposed(Observable.range(1, 5).filter(Functions.alwaysTrue()));

            

Reported by PMD.

                  }

    @Test
    public void dispose() {
        TestHelper.checkDisposed(Observable.range(1, 5).filter(Functions.alwaysTrue()));
    }

    @Test
    public void doubleOnSubscribe() {

            

Reported by PMD.

              
    @Test
    public void dispose() {
        TestHelper.checkDisposed(Observable.range(1, 5).filter(Functions.alwaysTrue()));
    }

    @Test
    public void doubleOnSubscribe() {
        TestHelper.checkDoubleOnSubscribeObservable(new Function<Observable<Object>, ObservableSource<Object>>() {

            

Reported by PMD.

                  }

    @Test
    public void doubleOnSubscribe() {
        TestHelper.checkDoubleOnSubscribeObservable(new Function<Observable<Object>, ObservableSource<Object>>() {
            @Override
            public ObservableSource<Object> apply(Observable<Object> o) throws Exception {
                return o.filter(Functions.alwaysTrue());
            }

            

Reported by PMD.