Organize, Format, Comments and Cleanup

Also included comment => /* package accessible for unit tests */  for private classes made package public to enable unit testing.

Author: benjchristensen

rxjava-core/src/main/java/rx/Observable.java

@@ -1,12 +1,12 @@
 /**
  * Copyright 2013 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
- *
+ * 
  * http://www.apache.org/licenses/LICENSE-2.0
- *
+ * 
  * Unless required by applicable law or agreed to in writing, software
  * distributed under the License is distributed on an "AS IS" BASIS,
  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
@@ -18,12 +18,12 @@
 /**
  * Provides a mechanism for receiving push-based notifications.
  * <p>
- * After an Observer calls an {@link Observable}'s <code>Observable.subscribe</code> method, the {@link Observable}
- * calls the Observer's <code>onNext</code> method to provide notifications. A well-behaved {@link Observable} will
+ * After an Observer calls an {@link Observable}'s <code>Observable.subscribe</code> method, the {@link Observable} calls the Observer's <code>onNext</code> method to provide notifications. A
+ * well-behaved {@link Observable} will
  * call an Observer's <code>onCompleted</code> closure exactly once or the Observer's <code>onError</code> closure exactly once.
  * <p>
  * For more information see the <a href="https://github.com/Netflix/RxJava/wiki/Observable">RxJava Wiki</a>
- *
+ * 
  * @param <T>
  */
 public interface Observer<T> {
@@ -39,7 +39,7 @@ public interface Observer<T> {
      * Notifies the Observer that the {@link Observable} has experienced an error condition.
      * <p>
      * If the {@link Observable} calls this closure, it will not thereafter call <code>onNext</code> or <code>onCompleted</code>.
-     *
+     * 
      * @param e
      */
     public void onError(Throwable e);
@@ -50,7 +50,7 @@ public interface Observer<T> {
      * The {@link Observable} calls this closure 1 or more times, unless it calls <code>onError</code> in which case this closure may never be called.
      * <p>
      * The {@link Observable} will not call this closure again after it calls either <code>onCompleted</code> or <code>onError</code>.
-     *
+     * 
      * @param args
      */
     public void onNext(T args);

rxjava-core/src/main/java/rx/Observer.java

@@ -15,17 +15,17 @@
  */
 package rx;
 
+import java.util.Date;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+
 import rx.subscriptions.CompositeSubscription;
 import rx.subscriptions.MultipleAssignmentSubscription;
 import rx.subscriptions.Subscriptions;
 import rx.util.functions.Action0;
 import rx.util.functions.Action1;
 import rx.util.functions.Func2;
 
-import java.util.Date;
-import java.util.concurrent.TimeUnit;
-import java.util.concurrent.atomic.AtomicBoolean;
-
 /**
  * Represents an object that schedules units of work.
  * <p>

rxjava-core/src/main/java/rx/Scheduler.java

@@ -1,12 +1,12 @@
 /**
  * Copyright 2013 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
- *
+ * 
  * http://www.apache.org/licenses/LICENSE-2.0
- *
+ * 
  * Unless required by applicable law or agreed to in writing, software
  * distributed under the License is distributed on an "AS IS" BASIS,
  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
@@ -15,14 +15,14 @@
  */
 package rx.concurrency;
 
-import rx.Scheduler;
-import rx.Subscription;
-import rx.util.functions.Func2;
-
 import java.util.PriorityQueue;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicInteger;
 
+import rx.Scheduler;
+import rx.Subscription;
+import rx.util.functions.Func2;
+
 /**
  * Schedules work on the current thread but does not execute immediately. Work is put in a queue and executed after the current unit of work is completed.
  */
@@ -35,7 +35,7 @@ public static CurrentThreadScheduler getInstance() {
 
     private static final ThreadLocal<PriorityQueue<TimedAction>> QUEUE = new ThreadLocal<PriorityQueue<TimedAction>>();
 
-    CurrentThreadScheduler() {
+    /* package accessible for unit tests */CurrentThreadScheduler() {
     }
 
     private final AtomicInteger counter = new AtomicInteger(0);

rxjava-core/src/main/java/rx/concurrency/CurrentThreadScheduler.java

@@ -48,23 +48,23 @@ public ExecutorScheduler(ScheduledExecutorService executor) {
     public <T> Subscription schedulePeriodically(final T state, final Func2<? super Scheduler, ? super T, ? extends Subscription> action, long initialDelay, long period, TimeUnit unit) {
         if (executor instanceof ScheduledExecutorService) {
             final CompositeSubscription subscriptions = new CompositeSubscription();
-            
+
             ScheduledFuture<?> f = ((ScheduledExecutorService) executor).scheduleAtFixedRate(new Runnable() {
                 @Override
                 public void run() {
                     Subscription s = action.call(ExecutorScheduler.this, state);
                     subscriptions.add(s);
                 }
             }, initialDelay, period, unit);
-            
+
             subscriptions.add(Subscriptions.create(f));
             return subscriptions;
-            
+
         } else {
             return super.schedulePeriodically(state, action, initialDelay, period, unit);
         }
     }
-    
+
     @Override
     public <T> Subscription schedule(final T state, final Func2<? super Scheduler, ? super T, ? extends Subscription> action, long delayTime, TimeUnit unit) {
         final DiscardableAction<T> discardableAction = new DiscardableAction<T>(state, action);

rxjava-core/src/main/java/rx/concurrency/ExecutorScheduler.java

@@ -1,12 +1,12 @@
 /**
  * Copyright 2013 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
- *
+ * 
  * http://www.apache.org/licenses/LICENSE-2.0
- *
+ * 
  * Unless required by applicable law or agreed to in writing, software
  * distributed under the License is distributed on an "AS IS" BASIS,
  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
@@ -15,35 +15,35 @@
  */
 package rx.concurrency;
 
+import java.util.concurrent.TimeUnit;
+
 import rx.Scheduler;
 import rx.Subscription;
 import rx.util.functions.Func2;
 
-import java.util.concurrent.TimeUnit;
-
 /**
  * Executes work immediately on the current thread.
  */
 public final class ImmediateScheduler extends Scheduler {
-  private static final ImmediateScheduler INSTANCE = new ImmediateScheduler();
+    private static final ImmediateScheduler INSTANCE = new ImmediateScheduler();
 
-  public static ImmediateScheduler getInstance() {
-    return INSTANCE;
-  }
+    public static ImmediateScheduler getInstance() {
+        return INSTANCE;
+    }
 
-  ImmediateScheduler() {
-  }
+    /* package accessible for unit tests */ImmediateScheduler() {
+    }
 
-  @Override
-  public <T> Subscription schedule(T state, Func2<? super Scheduler, ? super T, ? extends Subscription> action) {
-    return action.call(this, state);
-  }
+    @Override
+    public <T> Subscription schedule(T state, Func2<? super Scheduler, ? super T, ? extends Subscription> action) {
+        return action.call(this, state);
+    }
 
-  @Override
-  public <T> Subscription schedule(T state, Func2<? super Scheduler, ? super T, ? extends Subscription> action, long dueTime, TimeUnit unit) {
-    // since we are executing immediately on this thread we must cause this thread to sleep
-    long execTime = now() + unit.toMillis(dueTime);
+    @Override
+    public <T> Subscription schedule(T state, Func2<? super Scheduler, ? super T, ? extends Subscription> action, long dueTime, TimeUnit unit) {
+        // since we are executing immediately on this thread we must cause this thread to sleep
+        long execTime = now() + unit.toMillis(dueTime);
 
-    return schedule(state, new SleepingAction<T>(action, this, execTime));
-  }
+        return schedule(state, new SleepingAction<T>(action, this, execTime));
+    }
 }

rxjava-core/src/main/java/rx/concurrency/ImmediateScheduler.java

@@ -1,12 +1,12 @@
 /**
  * Copyright 2013 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
- *
+ * 
  * http://www.apache.org/licenses/LICENSE-2.0
- *
+ * 
  * Unless required by applicable law or agreed to in writing, software
  * distributed under the License is distributed on an "AS IS" BASIS,
  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
@@ -30,16 +30,14 @@ public SleepingAction(Func2<? super Scheduler, ? super T, ? extends Subscription
         this.execTime = execTime;
     }
 
-
     @Override
     public Subscription call(Scheduler s, T state) {
         if (execTime > scheduler.now()) {
             long delay = execTime - scheduler.now();
-            if (delay> 0) {
+            if (delay > 0) {
                 try {
                     Thread.sleep(delay);
-                }
-                catch (InterruptedException e) {
+                } catch (InterruptedException e) {
                     Thread.currentThread().interrupt();
                     throw new RuntimeException(e);
                 }

rxjava-core/src/main/java/rx/concurrency/SleepingAction.java

@@ -15,18 +15,23 @@
  */
 package rx.observables;
 
+import java.util.Iterator;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.Future;
+import java.util.concurrent.atomic.AtomicReference;
+
 import rx.Observable;
 import rx.Observer;
 import rx.Subscription;
-import rx.operators.*;
+import rx.operators.OperationMostRecent;
+import rx.operators.OperationNext;
+import rx.operators.OperationToFuture;
+import rx.operators.OperationToIterator;
+import rx.operators.SafeObservableSubscription;
+import rx.operators.SafeObserver;
 import rx.util.functions.Action1;
 import rx.util.functions.Func1;
 
-import java.util.Iterator;
-import java.util.concurrent.CountDownLatch;
-import java.util.concurrent.Future;
-import java.util.concurrent.atomic.AtomicReference;
-
 /**
  * An extension of {@link Observable} that provides blocking operators.
  * <p>

rxjava-core/src/main/java/rx/observables/BlockingObservable.java

@@ -1,12 +1,12 @@
 /**
  * Copyright 2013 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
- *
+ * 
  * http://www.apache.org/licenses/LICENSE-2.0
- *
+ * 
  * Unless required by applicable law or agreed to in writing, software
  * distributed under the License is distributed on an "AS IS" BASIS,
  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
@@ -19,7 +19,6 @@
 import rx.Observer;
 import rx.Subscription;
 import rx.operators.OperationRefCount;
-import rx.util.functions.Func1;
 
 /**
  * A ConnectableObservable resembles an ordinary {@link Observable}, except that it does not begin
@@ -32,7 +31,7 @@
  * For more information see
  * <a href="https://github.com/Netflix/RxJava/wiki/Connectable-Observable-Operators">Connectable
  * Observable Operators</a> at the RxJava Wiki
- *
+ * 
  * @param <T>
  */
 
@@ -44,13 +43,14 @@ protected ConnectableObservable(OnSubscribeFunc<T> onSubscribe) {
 
     /**
      * Call a ConnectableObservable's connect() method to instruct it to begin emitting the
-     * items from its underlying {@link Observable} to its {@link Observer}s. 
+     * items from its underlying {@link Observable} to its {@link Observer}s.
      */
     public abstract Subscription connect();
 
     /**
      * Returns an observable sequence that stays connected to the source as long
      * as there is at least one subscription to the observable sequence.
+     * 
      * @return a {@link Observable}
      */
     public Observable<T> refCount() {

rxjava-core/src/main/java/rx/observables/ConnectableObservable.java

@@ -1,12 +1,12 @@
 /**
  * Copyright 2013 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
- *
+ * 
  * http://www.apache.org/licenses/LICENSE-2.0
- *
+ * 
  * Unless required by applicable law or agreed to in writing, software
  * distributed under the License is distributed on an "AS IS" BASIS,
  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
@@ -19,13 +19,14 @@
 import rx.util.functions.Func1;
 
 /**
- * An {@link Observable} that has been grouped by a key whose value can be obtained using
- * {@link #getKey()} <p>
- *
+ * An {@link Observable} that has been grouped by a key whose value can be obtained using {@link #getKey()} <p>
+ * 
  * @see Observable#groupBy(Func1)
- *
- * @param <K> the type of the key
- * @param <T> the type of the elements in the group
+ * 
+ * @param <K>
+ *            the type of the key
+ * @param <T>
+ *            the type of the elements in the group
  */
 public class GroupedObservable<K, T> extends Observable<T> {
     private final K key;
@@ -37,11 +38,10 @@ public GroupedObservable(K key, OnSubscribeFunc<T> onSubscribe) {
 
     /**
      * Returns the key the elements in this observable were grouped by.
-     *
+     * 
      * @return the key the elements in this observable were grouped by
      */
     public K getKey() {
         return key;
     }
 }
-

rxjava-core/src/main/java/rx/observables/GroupedObservable.java

@@ -61,7 +61,7 @@ protected interface ChunkCreator {
      * 
      * @param <T>
      *            The type of objects which this {@link Chunk} can hold.
-     * @param <C> 
+     * @param <C>
      *            The type of object being tracked by the {@link Chunk}
      */
     protected abstract static class Chunk<T, C> {