1.x: remove ObjectPool, code style cleanups (#4846)

Author: akarnokd

src/main/java/rx/BackpressureOverflow.java

@@ -26,6 +26,10 @@
 @Beta
 public final class BackpressureOverflow {
 
+    private BackpressureOverflow() {
+        throw new IllegalStateException("No instances!");
+    }
+
     /**
      * Signal a MissingBackressureException due to lack of requests.
      */
@@ -65,10 +69,10 @@ public interface Strategy {
     /**
      * Drop oldest items from the buffer making room for newer ones.
      */
-    static class DropOldest implements BackpressureOverflow.Strategy {
+    static final class DropOldest implements BackpressureOverflow.Strategy {
         static final DropOldest INSTANCE = new DropOldest();
 
-        private DropOldest() {}
+        private DropOldest() { }
 
         @Override
         public boolean mayAttemptDrop() {
@@ -80,10 +84,10 @@ public boolean mayAttemptDrop() {
      * Drop most recent items, but not {@code onError} nor unsubscribe from source
      * (as {code OperatorOnBackpressureDrop}).
      */
-    static class DropLatest implements BackpressureOverflow.Strategy {
+    static final class DropLatest implements BackpressureOverflow.Strategy {
         static final DropLatest INSTANCE = new DropLatest();
 
-        private DropLatest() {}
+        private DropLatest() { }
 
         @Override
         public boolean mayAttemptDrop() {
@@ -94,11 +98,11 @@ public boolean mayAttemptDrop() {
     /**
      * {@code onError} a MissingBackpressureException and unsubscribe from source.
      */
-    static class Error implements BackpressureOverflow.Strategy {
+    static final class Error implements BackpressureOverflow.Strategy {
 
         static final Error INSTANCE = new Error();
 
-        private Error() {}
+        private Error() { }
 
         @Override
         public boolean mayAttemptDrop() throws MissingBackpressureException {

src/main/java/rx/Observable.java

@@ -3470,7 +3470,7 @@ public static Observable<Integer> range(int start, int count) {
         if (start > Integer.MAX_VALUE - count + 1) {
             throw new IllegalArgumentException("start + count can not exceed Integer.MAX_VALUE");
         }
-        if(count == 1) {
+        if (count == 1) {
             return Observable.just(start);
         }
         return Observable.create(new OnSubscribeRange(start, start + (count - 1)));
@@ -11742,7 +11742,7 @@ public final Observable<List<T>> toSortedList(Func2<? super T, ? super T, Intege
      * @return an Observable that emits the items emitted by the source Observable in sorted order
      */
     @Experimental
-    public final Observable<T> sorted(){
+    public final Observable<T> sorted() {
         return toSortedList().flatMapIterable(UtilityFunctions.<List<T>>identity());
     }
 

src/main/java/rx/Scheduler.java

@@ -212,9 +212,9 @@ public long now() {
      *
      * <pre>
      * Scheduler limitScheduler = Schedulers.computation().when(workers -> {
-     * 	// use merge max concurrent to limit the number of concurrent
-     * 	// callbacks two at a time
-     * 	return Completable.merge(Observable.merge(workers), 2);
+     *     // use merge max concurrent to limit the number of concurrent
+     *     // callbacks two at a time
+     *     return Completable.merge(Observable.merge(workers), 2);
      * });
      * </pre>
      * <p>
@@ -230,9 +230,9 @@ public long now() {
      *
      * <pre>
      * Scheduler limitScheduler = Schedulers.computation().when(workers -> {
-     * 	// use merge max concurrent to limit the number of concurrent
-     * 	// Observables two at a time
-     * 	return Completable.merge(Observable.merge(workers, 2));
+     *     // use merge max concurrent to limit the number of concurrent
+     *     // Observables two at a time
+     *     return Completable.merge(Observable.merge(workers, 2));
      * });
      * </pre>
      *
@@ -243,10 +243,10 @@ public long now() {
      *
      * <pre>
      * Scheduler slowScheduler = Schedulers.computation().when(workers -> {
-     * 	// use concatenate to make each worker happen one at a time.
-     * 	return Completable.concat(workers.map(actions -> {
-     * 		// delay the starting of the next worker by 1 second.
-     * 		return Completable.merge(actions.delaySubscription(1, TimeUnit.SECONDS));
+     *     // use concatenate to make each worker happen one at a time.
+     *     return Completable.concat(workers.map(actions -> {
+     *         // delay the starting of the next worker by 1 second.
+     *         return Completable.merge(actions.delaySubscription(1, TimeUnit.SECONDS));
      *    }));
      * });
      * </pre>

src/main/java/rx/exceptions/CompositeException.java

@@ -141,7 +141,7 @@ public synchronized Throwable getCause() { // NOPMD
 
                 List<Throwable> listOfCauses = getListOfCauses(e);
                 // check if any of them have been seen before
-                for(Throwable child : listOfCauses) {
+                for (Throwable child : listOfCauses) {
                     if (seenCauses.contains(child)) {
                         // already seen this outer Throwable so skip
                         e = new RuntimeException("Duplicate found in causal chain so cropping to prevent loop ...");
@@ -288,7 +288,7 @@ private List<Throwable> getListOfCauses(Throwable ex) {
         if (root == null || root == ex) {
             return list;
         } else {
-            while(true) {
+            while (true) {
                 list.add(root);
                 Throwable cause = root.getCause();
                 if (cause == null || cause == root) {
@@ -311,7 +311,7 @@ private Throwable getRootCause(Throwable e) {
         if (root == null || root == e) {
             return e;
         } else {
-            while(true) {
+            while (true) {
                 Throwable cause = root.getCause();
                 if (cause == null || cause == root) {
                     return root;

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

@@ -102,7 +102,7 @@ public void call() {
                     }
                 }
                 if (win) {
-                    while(true) {
+                    while (true) {
                         Object o;
                         while ((o = state.buffer.poll()) != null) {
                             NotificationLite.accept(state.get(), o);

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

@@ -29,7 +29,7 @@
  * Given multiple {@link Observable}s, propagates the one that first emits an item.
  * @param <T> the value type
  */
-public final class OnSubscribeAmb<T> implements OnSubscribe<T>{
+public final class OnSubscribeAmb<T> implements OnSubscribe<T> {
     //give default access instead of private as a micro-optimization
     //for access from anonymous classes below
     final Iterable<? extends Observable<? extends T>> sources;
@@ -336,12 +336,12 @@ private boolean isSelected() {
     }
 
     @SuppressWarnings("serial")
-	static final class Selection<T> extends AtomicReference<AmbSubscriber<T>> {
+    static final class Selection<T> extends AtomicReference<AmbSubscriber<T>> {
         final Collection<AmbSubscriber<T>> ambSubscribers = new ConcurrentLinkedQueue<AmbSubscriber<T>>();
 
         public void unsubscribeLosers() {
             AmbSubscriber<T> winner = get();
-            if(winner != null) {
+            if (winner != null) {
                 unsubscribeOthers(winner);
             }
         }
@@ -437,7 +437,7 @@ public void request(long n) {
     }
 
     static <T> void unsubscribeAmbSubscribers(Collection<AmbSubscriber<T>> ambSubscribers) {
-        if(!ambSubscribers.isEmpty()) {
+        if (!ambSubscribers.isEmpty()) {
             for (AmbSubscriber<T> other : ambSubscribers) {
                 other.unsubscribe();
             }

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

@@ -33,7 +33,7 @@
 public final class OnSubscribeAutoConnect<T> extends AtomicInteger implements OnSubscribe<T> {
     // AtomicInteger aspect of `this` represents the number of clients
 
-	final ConnectableObservable<? extends T> source;
+    final ConnectableObservable<? extends T> source;
     final int numberOfSubscribers;
     final Action1<? super Subscription> connection;
 

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

@@ -78,12 +78,12 @@ public Observable<?> call(Observable<? extends Notification<?>> ts) {
 
                 @Override
                 public Notification<?> call(Notification<?> terminalNotification) {
-                    if(count == 0) {
+                    if (count == 0) {
                         return terminalNotification;
                     }
 
                     num++;
-                    if(num <= count) {
+                    if (num <= count) {
                         return Notification.createOnNext(num);
                     } else {
                         return terminalNotification;
@@ -153,7 +153,7 @@ public static <T> Observable<T> repeat(Observable<T> source, final long count) {
     }
 
     public static <T> Observable<T> repeat(Observable<T> source, final long count, Scheduler scheduler) {
-        if(count == 0) {
+        if (count == 0) {
             return Observable.empty();
         }
         if (count < 0) {

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

@@ -155,7 +155,7 @@ public void onStart() {
 
         @Override
         public void onNext(T t) {
-            if (done){
+            if (done) {
                 return;
             }
             try {

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

@@ -43,7 +43,7 @@
  * @param <V>
  *            the value type of the groups
  */
-public final class OperatorGroupBy<T, K, V> implements Operator<GroupedObservable<K, V>, T>{
+public final class OperatorGroupBy<T, K, V> implements Operator<GroupedObservable<K, V>, T> {
     final Func1<? super T, ? extends K> keySelector;
     final Func1<? super T, ? extends V> valueSelector;
     final int bufferSize;

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

@@ -112,7 +112,7 @@ public void call() {
                 guardedSubscription = gs.get();
 
                 // register any subscribers that are waiting with this new subject
-                for(final Subscriber<? super R> s : waitingForConnect) {
+                for (final Subscriber<? super R> s : waitingForConnect) {
                     subject.unsafeSubscribe(new Subscriber<R>(s) {
                         @Override
                         public void onNext(R t) {

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

@@ -99,7 +99,7 @@ public void onNext(T t) {
                     requested.decrementAndGet();
                 } else {
                     // item dropped
-                    if(onDrop != null) {
+                    if (onDrop != null) {
                         try {
                             onDrop.call(t);
                         } catch (Throwable e) {

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

@@ -21,7 +21,8 @@
 import rx.*;
 import rx.exceptions.*;
 import rx.functions.*;
-import rx.internal.util.*;
+import rx.internal.util.RxRingBuffer;
+import rx.internal.util.atomic.SpscAtomicArrayQueue;
 import rx.internal.util.unsafe.*;
 import rx.observables.ConnectableObservable;
 import rx.subscriptions.Subscriptions;
@@ -244,7 +245,7 @@ static final class PublishSubscriber<T> extends Subscriber<T> implements Subscri
         public PublishSubscriber(AtomicReference<PublishSubscriber<T>> current) {
             this.queue = UnsafeAccess.isUnsafeAvailable()
                     ? new SpscArrayQueue<Object>(RxRingBuffer.SIZE)
-                    : new SynchronizedQueue<Object>(RxRingBuffer.SIZE);
+                    : new SpscAtomicArrayQueue<Object>(RxRingBuffer.SIZE);
 
             this.producers = new AtomicReference<InnerProducer[]>(EMPTY);
             this.current = current;

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

@@ -113,7 +113,7 @@ public void call() {
                     clearProducer();
                 }
             }));
-            child.setProducer(new Producer(){
+            child.setProducer(new Producer() {
 
                 @Override
                 public void request(long n) {

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

@@ -97,7 +97,7 @@ public void setProducer(final Producer producer) {
 
                     @Override
                     public void request(long n) {
-                        if (n >0 && !completed) {
+                        if (n > 0 && !completed) {
                             // because requests may happen concurrently use a CAS loop to
                             // ensure we only request as much as needed, no more no less
                             while (true) {

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

@@ -24,7 +24,7 @@
 import rx.subscriptions.*;
 
 public final class CachedThreadScheduler extends Scheduler implements SchedulerLifecycle {
-    private static final long KEEP_ALIVE_TIME = 60;
+    private static final long KEEP_ALIVE_TIME;
     private static final TimeUnit KEEP_ALIVE_UNIT = TimeUnit.SECONDS;
 
     static final ThreadWorker SHUTDOWN_THREADWORKER;
@@ -41,6 +41,8 @@ public final class CachedThreadScheduler extends Scheduler implements SchedulerL
 
         NONE = new CachedWorkerPool(null, 0, null);
         NONE.shutdown();
+
+        KEEP_ALIVE_TIME = Integer.getInteger("rx.io-scheduler.keepalive", 60);
     }
 
     static final class CachedWorkerPool {

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

@@ -63,9 +63,9 @@
  *
  * <pre>
  * Scheduler limitScheduler = Schedulers.computation().when(workers -> {
- * 	// use merge max concurrent to limit the number of concurrent
- * 	// callbacks two at a time
- * 	return Completable.merge(Observable.merge(workers), 2);
+ *     // use merge max concurrent to limit the number of concurrent
+ *     // callbacks two at a time
+ *     return Completable.merge(Observable.merge(workers), 2);
  * });
  * </pre>
  * <p>
@@ -81,9 +81,9 @@
  *
  * <pre>
  * Scheduler limitScheduler = Schedulers.computation().when(workers -> {
- * 	// use merge max concurrent to limit the number of concurrent
- * 	// Observables two at a time
- * 	return Completable.merge(Observable.merge(workers, 2));
+ *     // use merge max concurrent to limit the number of concurrent
+ *     // Observables two at a time
+ *     return Completable.merge(Observable.merge(workers, 2));
  * });
  * </pre>
  *
@@ -94,11 +94,11 @@
  *
  * <pre>
  * Scheduler slowScheduler = Schedulers.computation().when(workers -> {
- * 	// use concatenate to make each worker happen one at a time.
- * 	return Completable.concat(workers.map(actions -> {
- * 		// delay the starting of the next worker by 1 second.
- * 		return Completable.merge(actions.delaySubscription(1, TimeUnit.SECONDS));
- * 	}));
+ *     // use concatenate to make each worker happen one at a time.
+ *     return Completable.concat(workers.map(actions -> {
+ *         // delay the starting of the next worker by 1 second.
+ *         return Completable.merge(actions.delaySubscription(1, TimeUnit.SECONDS));
+ *     }));
  * });
  * </pre>
  */

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

@@ -52,15 +52,6 @@ public final class IndexedRingBuffer<E> implements Subscription {
     /* package for unit testing */final AtomicInteger index = new AtomicInteger();
     /* package for unit testing */final AtomicInteger removedIndex = new AtomicInteger();
 
-    private static final ObjectPool<IndexedRingBuffer<?>> POOL = new ObjectPool<IndexedRingBuffer<?>>() {
-
-        @Override
-        protected IndexedRingBuffer<?> createObject() {
-            return new IndexedRingBuffer<Object>();
-        }
-
-    };
-
     /* package for unit testing */static final int SIZE;
 
     // default size of ring buffer
@@ -255,9 +246,8 @@ protected IndexedRingBuffer<?> createObject() {
         SIZE = defaultSize;
     }
 
-    @SuppressWarnings("unchecked")
     public static <T> IndexedRingBuffer<T> getInstance() {
-        return (IndexedRingBuffer<T>) POOL.borrowObject();
+        return new IndexedRingBuffer<T>();
     }
 
     /**
@@ -283,7 +273,6 @@ public void releaseToPool() {
 
         index.set(0);
         removedIndex.set(0);
-        POOL.returnObject(this);
     }
 
     @Override