Some code clean ups.

Nothing that could change logic or application flow, just minor refactors to be consistent with good practices and clean code.

Author: Tomasz Drodowski

src/main/java/rx/internal/operators/BlockingOperatorNext.java

@@ -79,11 +79,11 @@ public boolean hasNext() {
             }
             // Since an iterator should not be used in different thread,
             // so we do not need any synchronization.
-            if (hasNext == false) {
+            if (!hasNext) {
                 // the iterator has reached the end.
                 return false;
             }
-            if (isNextConsumed == false) {
+            if (!isNextConsumed) {
                 // next has not been used yet.
                 return true;
             }

src/main/java/rx/internal/operators/OnSubscribeRedo.java

@@ -195,7 +195,7 @@ public void call(final Subscriber<? super T> child) {
         final AtomicBoolean resumeBoundary = new AtomicBoolean(true);
 
         // incremented when requests are made, decremented when requests are fulfilled
-        final AtomicLong consumerCapacity = new AtomicLong(0l);
+        final AtomicLong consumerCapacity = new AtomicLong();
 
         final Scheduler.Worker worker = scheduler.createWorker();
         child.add(worker);

src/main/java/rx/internal/operators/OperatorBufferWithSize.java

@@ -158,7 +158,6 @@ public void request(long n) {
                         }
                         if (n == Long.MAX_VALUE) {
                             requestInfinite();
-                            return;
                         } else {
                             if (firstRequest) {
                                 firstRequest = false;

src/main/java/rx/internal/operators/OperatorConcat.java

@@ -229,5 +229,5 @@ public void setProducer(Producer producer) {
             arbiter.setProducer(producer);
         }
 
-    };
+    }
 }

src/main/java/rx/internal/operators/OperatorWindowWithSize.java

@@ -117,7 +117,6 @@ public void onNext(T t) {
                 noWindow = true;
                 if (child.isUnsubscribed()) {
                     unsubscribe();
-                    return;
                 }
             }
         }

src/main/java/rx/internal/operators/OperatorWithLatestFrom.java

@@ -60,7 +60,6 @@ public void onNext(T t) {
                         s.onNext(result);
                     } catch (Throwable e) {
                         Exceptions.throwOrReport(e, this);
-                        return;
                     }
                 }
             }

src/main/java/rx/internal/schedulers/EventLoopsScheduler.java

@@ -160,18 +160,16 @@ public Subscription schedule(Action0 action) {
             if (isUnsubscribed()) {
                 return Subscriptions.unsubscribed();
             }
-            ScheduledAction s = poolWorker.scheduleActual(action, 0, null, serial);
-            
-            return s;
+
+            return poolWorker.scheduleActual(action, 0, null, serial);
         }
         @Override
         public Subscription schedule(Action0 action, long delayTime, TimeUnit unit) {
             if (isUnsubscribed()) {
                 return Subscriptions.unsubscribed();
             }
-            ScheduledAction s = poolWorker.scheduleActual(action, delayTime, unit, timed);
-            
-            return s;
+
+            return poolWorker.scheduleActual(action, delayTime, unit, timed);
         }
     }
 

src/main/java/rx/internal/schedulers/GenericScheduledExecutorService.java

@@ -70,7 +70,6 @@ public void start() {
                     NewThreadWorker.registerExecutor((ScheduledThreadPoolExecutor)exec);
                 }
             }
-            return;
         } else {
             exec.shutdownNow();
         }

src/main/java/rx/internal/util/SubscriptionRandomList.java

@@ -108,7 +108,7 @@ public void clear() {
     }
 
     public void forEach(Action1<T> action) {
-        T[] ss=null;
+        T[] ss = null;
         synchronized (this) {
             if (unsubscribed || subscriptions == null) {
                 return;

src/main/java/rx/observables/AbstractOnSubscribe.java

@@ -597,10 +597,7 @@ protected boolean use() {
          */
         protected void free() {
             int i = inUse.get();
-            if (i <= 0) {
-                return;
-            } else
-            if (inUse.decrementAndGet() == 0) {
+            if (i > 0 && inUse.decrementAndGet() == 0) {
                 parent.onTerminated(state);
             }
         }

src/main/java/rx/observables/AsyncOnSubscribe.java

@@ -29,7 +29,7 @@
 import rx.observers.*;
 import rx.plugins.RxJavaPlugins;
 import rx.subscriptions.CompositeSubscription;
-;
+
 /**
  * A utility class to create {@code OnSubscribe<T>} functions that respond correctly to back
  * pressure requests from subscribers. This is an improvement over

src/main/java/rx/subjects/ReplaySubject.java

@@ -958,7 +958,7 @@ public void evictFinal(NodeList<Object> t1) {
         public boolean test(Object value, long now) {
             return first.test(value, now) || second.test(value, now);
         }
-    };
+    }
 
     /** Maps the values to Timestamped. */
     static final class AddTimestamped implements Func1<Object, Object> {

src/main/java/rx/subscriptions/RefCountSubscription.java

@@ -143,5 +143,5 @@ public void unsubscribe() {
         public boolean isUnsubscribed() {
             return innerDone != 0;
         }
-    };
+    }
 }

src/test/java/rx/CovarianceTest.java

@@ -211,7 +211,7 @@ public Observable<Movie> call(List<List<Movie>> listOfLists) {
 
                 return Observable.from(delta);
             }
-        };
+        }
     };
 
     /*

src/test/java/rx/internal/operators/OnSubscribeCombineLatestTest.java

@@ -251,7 +251,7 @@ public void testCombineLatest3TypesB() {
     }
 
     private Func3<String, String, String, String> getConcat3StringsCombineLatestFunction() {
-        Func3<String, String, String, String> combineLatestFunction = new Func3<String, String, String, String>() {
+        return new Func3<String, String, String, String>() {
 
             @Override
             public String call(String a1, String a2, String a3) {
@@ -268,31 +268,28 @@ public String call(String a1, String a2, String a3) {
             }
 
         };
-        return combineLatestFunction;
     }
 
     private Func2<String, Integer, String> getConcatStringIntegerCombineLatestFunction() {
-        Func2<String, Integer, String> combineLatestFunction = new Func2<String, Integer, String>() {
+        return new Func2<String, Integer, String>() {
 
             @Override
             public String call(String s, Integer i) {
                 return getStringValue(s) + getStringValue(i);
             }
 
         };
-        return combineLatestFunction;
     }
 
     private Func3<String, Integer, int[], String> getConcatStringIntegerIntArrayCombineLatestFunction() {
-        Func3<String, Integer, int[], String> combineLatestFunction = new Func3<String, Integer, int[], String>() {
+        return new Func3<String, Integer, int[], String>() {
 
             @Override
             public String call(String s, Integer i, int[] iArray) {
                 return getStringValue(s) + getStringValue(i) + getStringValue(iArray);
             }
 
         };
-        return combineLatestFunction;
     }
 
     private static String getStringValue(Object o) {

src/test/java/rx/internal/operators/OnSubscribeUsingTest.java

@@ -45,9 +45,9 @@
 public class OnSubscribeUsingTest {
 
     private interface Resource {
-        public String getTextFromWeb();
-
-        public void dispose();
+        String getTextFromWeb();
+        
+        void dispose();
     }
 
     private static class DisposeAction implements Action1<Resource> {

src/test/java/rx/internal/operators/OperatorConcatTest.java

@@ -749,7 +749,7 @@ public void onNext(Integer t) {
                 if (counter.getAndIncrement() % 100 == 0) {
                     System.out.print("testIssue2890NoStackoverflow -> ");
                     System.out.println(counter.get());
-                };
+                }
             }
 
             @Override

src/test/java/rx/internal/operators/OperatorGroupByTest.java

@@ -976,7 +976,7 @@ public String toString() {
 
     Observable<Event> ASYNC_INFINITE_OBSERVABLE_OF_EVENT(final int numGroups, final AtomicInteger subscribeCounter, final AtomicInteger sentEventCounter) {
         return SYNC_INFINITE_OBSERVABLE_OF_EVENT(numGroups, subscribeCounter, sentEventCounter).subscribeOn(Schedulers.newThread());
-    };
+    }
 
     Observable<Event> SYNC_INFINITE_OBSERVABLE_OF_EVENT(final int numGroups, final AtomicInteger subscribeCounter, final AtomicInteger sentEventCounter) {
         return Observable.create(new OnSubscribe<Event>() {
@@ -997,7 +997,7 @@ public void call(final Subscriber<? super Event> op) {
             }
 
         });
-    };
+    }
 
     @Test
     public void testGroupByOnAsynchronousSourceAcceptsMultipleSubscriptions() throws InterruptedException {

src/test/java/rx/internal/operators/OperatorMapTest.java

@@ -91,7 +91,7 @@ public void testMapMany() {
             @Override
             public Observable<String> call(Integer id) {
                 /* simulate making a nested async call which creates another Observable */
-                Observable<Map<String, String>> subObservable = null;
+                Observable<Map<String, String>> subObservable;
                 if (id == 1) {
                     Map<String, String> m1 = getMap("One");
                     Map<String, String> m2 = getMap("Two");

src/test/java/rx/internal/operators/OperatorMergeTest.java

@@ -946,7 +946,7 @@ public Observable<Integer> call(Integer i) {
     }
 
     private Observable<Integer> createInfiniteObservable(final AtomicInteger generated) {
-        Observable<Integer> observable = Observable.from(new Iterable<Integer>() {
+        return Observable.from(new Iterable<Integer>() {
             @Override
             public Iterator<Integer> iterator() {
                 return new Iterator<Integer>() {
@@ -967,7 +967,6 @@ public boolean hasNext() {
                 };
             }
         });
-        return observable;
     }
 
     @Test

src/test/java/rx/internal/operators/OperatorPublishTest.java

@@ -199,7 +199,7 @@ public void call() {
                         sourceUnsubscribed.set(true);
                     }
                 }).share();
-        ;
+
 
         final AtomicBoolean child1Unsubscribed = new AtomicBoolean();
         final AtomicBoolean child2Unsubscribed = new AtomicBoolean();

src/test/java/rx/internal/operators/OperatorSerializeTest.java

@@ -213,7 +213,7 @@ public void run() {
         }
     }
 
-    private static enum TestConcurrencyobserverEvent {
+    private enum TestConcurrencyobserverEvent {
         onCompleted, onError, onNext
     }
 

src/test/java/rx/internal/operators/OperatorTakeUntilPredicateTest.java

@@ -30,7 +30,7 @@
 import rx.functions.Func1;
 import rx.internal.util.UtilityFunctions;
 import rx.observers.TestSubscriber;
-;
+
 
 public class OperatorTakeUntilPredicateTest {
     @Test

src/test/java/rx/internal/operators/OperatorZipTest.java

@@ -594,19 +594,18 @@ public String call(String t1, String t2) {
     }
 
     private Func2<Integer, Integer, Integer> getDivideZipr() {
-        Func2<Integer, Integer, Integer> zipr = new Func2<Integer, Integer, Integer>() {
+        return new Func2<Integer, Integer, Integer>() {
 
             @Override
             public Integer call(Integer i1, Integer i2) {
                 return i1 / i2;
             }
 
         };
-        return zipr;
     }
 
     private Func3<String, String, String, String> getConcat3StringsZipr() {
-        Func3<String, String, String, String> zipr = new Func3<String, String, String, String>() {
+        return new Func3<String, String, String, String>() {
 
             @Override
             public String call(String a1, String a2, String a3) {
@@ -623,31 +622,28 @@ public String call(String a1, String a2, String a3) {
             }
 
         };
-        return zipr;
     }
 
     private Func2<String, Integer, String> getConcatStringIntegerZipr() {
-        Func2<String, Integer, String> zipr = new Func2<String, Integer, String>() {
+        return new Func2<String, Integer, String>() {
 
             @Override
             public String call(String s, Integer i) {
                 return getStringValue(s) + getStringValue(i);
             }
 
         };
-        return zipr;
     }
 
     private Func3<String, Integer, int[], String> getConcatStringIntegerIntArrayZipr() {
-        Func3<String, Integer, int[], String> zipr = new Func3<String, Integer, int[], String>() {
+        return new Func3<String, Integer, int[], String>() {
 
             @Override
             public String call(String s, Integer i, int[] iArray) {
                 return getStringValue(s) + getStringValue(i) + getStringValue(iArray);
             }
 
         };
-        return zipr;
     }
 
     private static String getStringValue(Object o) {
@@ -1147,7 +1143,7 @@ public String call(Integer t1, Integer t2) {
     }
 
     private Observable<Integer> createInfiniteObservable(final AtomicInteger generated) {
-        Observable<Integer> observable = Observable.from(new Iterable<Integer>() {
+        return Observable.from(new Iterable<Integer>() {
             @Override
             public Iterator<Integer> iterator() {
                 return new Iterator<Integer>() {
@@ -1168,7 +1164,6 @@ public boolean hasNext() {
                 };
             }
         });
-        return observable;
     }
 
     Observable<Integer> OBSERVABLE_OF_5_INTEGERS = OBSERVABLE_OF_5_INTEGERS(new AtomicInteger());

src/test/java/rx/internal/util/IndexedRingBufferTest.java

@@ -191,11 +191,8 @@ public Boolean call(String t1) {
             @Override
             public Boolean call(String t1) {
                 list.add(t1);
-                if (i++ == 2) {
-                    return false;
-                } else {
-                    return true;
-                }
+                i++;
+                return i != 3;
             }
 
         }, 0);

src/test/java/rx/observers/SerializedObserverTest.java

@@ -500,7 +500,7 @@ public void run() {
         }
     }
 
-    private static enum TestConcurrencyObserverEvent {
+    private enum TestConcurrencyObserverEvent {
         onCompleted, onError, onNext
     }