Method getPluginImplementationViaProperty has a Cognitive Complexity of 27 (exceeds 5 allowed). Consider refactoring.

Open

    /* test */ static Object getPluginImplementationViaProperty(Class<?> pluginClass, Properties propsIn) {
        // Make a defensive clone because traversal may fail with ConcurrentModificationException
        // if the properties get changed by something outside RxJava.
        Properties props = (Properties)propsIn.clone();

• Read up
  Read up

Method call has a Cognitive Complexity of 27 (exceeds 5 allowed). Consider refactoring.

Open

    @Override
    public void call(final Subscriber<? super T> child) {

        // when true is a marker to say we are ready to resubscribe to source
        final AtomicBoolean resumeBoundary = new AtomicBoolean(true);

• Read up
  Read up

Method drain has a Cognitive Complexity of 27 (exceeds 5 allowed). Consider refactoring.

Open

    private void drain() {
        if (wip.getAndIncrement() == 0) {
            final Subscriber<? super T> c = child;
            final Queue<Object> q = queue;
            do {

• Read up
  Read up

Method drain has a Cognitive Complexity of 27 (exceeds 5 allowed). Consider refactoring.

Open

    private void drain() {
        if (wip.getAndIncrement() == 0) {
            final Subscriber<? super T> c = child;
            final Queue<Object> q = queue;

• Read up
  Read up

Method emitLoop has 97 lines of code (exceeds 25 allowed). Consider refactoring.

Open

    void emitLoop() {
        final Subscriber<? super T> c = child;

        long toRequest = 0L;
        Producer requestFrom = null;

File OnSubscribeCombineLatest.java has 328 lines of code (exceeds 250 allowed). Consider refactoring.

Open

/**
 * Copyright 2015 Netflix, Inc.
 *
 * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in
 * compliance with the License. You may obtain a copy of the License at

Method amb has 96 lines of code (exceeds 25 allowed). Consider refactoring.

Open

    public static Completable amb(final Iterable<? extends Completable> sources) {
        requireNonNull(sources);

        return create(new OnSubscribe() {
            @Override

OnSubscribeRedo has 31 methods (exceeds 20 allowed). Consider refactoring.

Open

public final class OnSubscribeRedo<T> implements OnSubscribe<T> {
    final Observable<T> source;
    private final Func1<? super Observable<? extends Notification<?>>, ? extends Observable<?>> controlHandlerFunction;
    final boolean stopOnComplete;
    final boolean stopOnError;

OneItemPerf has 31 methods (exceeds 20 allowed). Consider refactoring.

Open

@BenchmarkMode(Mode.Throughput)
@OutputTimeUnit(TimeUnit.SECONDS)
@State(Scope.Thread)
public class OneItemPerf {

Method drain has 94 lines of code (exceeds 25 allowed). Consider refactoring.

Open

        void drain() {
            if (wip.getAndIncrement() != 0) {
                return;
            }

File TestSubscriber.java has 323 lines of code (exceeds 250 allowed). Consider refactoring.

Open

/**
 * Copyright 2014 Netflix, Inc.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.

Method emitLoop has a Cognitive Complexity of 26 (exceeds 5 allowed). Consider refactoring.

Open

    public void emitLoop() {
        for (;;) {
            long localRequested;
            long localProduced;
            Producer localProducer;

• Read up
  Read up

Method tryEnableCancelPolicy has a Cognitive Complexity of 26 (exceeds 5 allowed). Consider refactoring.

Open

    public static boolean tryEnableCancelPolicy(ScheduledExecutorService executor) {
        if (SHOULD_TRY_ENABLE_CANCEL_POLICY) { // NOPMD
            final boolean isInstanceOfScheduledThreadPoolExecutor = executor instanceof ScheduledThreadPoolExecutor;

            Method methodToCall;

• Read up
  Read up

Method dispatch has 92 lines of code (exceeds 25 allowed). Consider refactoring.

Open

        void dispatch() {
            // standard construct of emitter loop (blocking)
            // if there is an emission going on, indicate that more work needs to be done
            // the exact nature of this work needs to be determined from other data structures
            synchronized (this) {

Method call has 92 lines of code (exceeds 25 allowed). Consider refactoring.

Open

    @Override
    public void call(final CompletableSubscriber s) {
        final CompositeSubscription set = new CompositeSubscription();

        s.onSubscribe(set);

AtomicPerf has 30 methods (exceeds 20 allowed). Consider refactoring.

Open

@BenchmarkMode(Mode.Throughput)
@OutputTimeUnit(TimeUnit.SECONDS)
public class AtomicPerf {
    @State(Scope.Thread)
    public static class Times {

File OperatorSwitch.java has 317 lines of code (exceeds 250 allowed). Consider refactoring.

Open

/**
 * Copyright 2014 Netflix, Inc.
 *
 * Licensed under the Apache License, Version 2.0 (the "License"); you may not
 * use this file except in compliance with the License. You may obtain a copy of

Method drain has 87 lines of code (exceeds 25 allowed). Consider refactoring.

Open

        void drain() {
            if (wip.getAndIncrement() != 0) {
                return;
            }
            int missed = 1;

Method init has 84 lines of code (exceeds 25 allowed). Consider refactoring.

Open

    @SuppressWarnings({ "rawtypes", "unchecked", "deprecation"})
    static void init() {
        onError = new Action1<Throwable>() {
            @Override
            public void call(Throwable e) {

Similar blocks of code found in 2 locations. Consider refactoring.

Open

        @SuppressWarnings("unchecked")
        void remove(ReplayProducer<T> rp) {
            for (;;) {
                ReplayProducer<T>[] a = get();
                if (a == TERMINATED || a == EMPTY) {

• Read up
  Read up