2.x: Cleanup code style, commas, spaces, docs (#6255)

Author: akarnokd

src/jmh/java/io/reactivex/parallel/ParallelPerf.java

@@ -28,7 +28,7 @@
 @BenchmarkMode(Mode.Throughput)
 @Warmup(iterations = 5)
 @Measurement(iterations = 5, time = 1, timeUnit = TimeUnit.SECONDS)
-@Fork(value = 1,jvmArgsAppend = { "-XX:MaxInlineLevel=20" })
+@Fork(value = 1, jvmArgsAppend = { "-XX:MaxInlineLevel=20" })
 @OutputTimeUnit(TimeUnit.SECONDS)
 @State(Scope.Thread)
 public class ParallelPerf implements Function<Integer, Integer> {

src/main/java/io/reactivex/internal/operators/flowable/FlowableGroupBy.java

@@ -430,7 +430,7 @@ public boolean isEmpty() {
         }
     }
 
-    static final class EvictionAction<K, V> implements Consumer<GroupedUnicast<K,V>> {
+    static final class EvictionAction<K, V> implements Consumer<GroupedUnicast<K, V>> {
 
         final Queue<GroupedUnicast<K, V>> evictedGroups;
 
@@ -439,7 +439,7 @@ static final class EvictionAction<K, V> implements Consumer<GroupedUnicast<K,V>>
         }
 
         @Override
-        public void accept(GroupedUnicast<K,V> value) {
+        public void accept(GroupedUnicast<K, V> value) {
             evictedGroups.offer(value);
         }
     }

src/main/java/io/reactivex/internal/queue/MpscLinkedQueue.java

@@ -36,7 +36,7 @@ public MpscLinkedQueue() {
         consumerNode = new AtomicReference<LinkedQueueNode<T>>();
         LinkedQueueNode<T> node = new LinkedQueueNode<T>();
         spConsumerNode(node);
-        xchgProducerNode(node);// this ensures correct construction: StoreLoad
+        xchgProducerNode(node); // this ensures correct construction: StoreLoad
     }
 
     /**

src/main/java/io/reactivex/internal/queue/SpscArrayQueue.java

@@ -89,12 +89,12 @@ public E poll() {
         final long index = consumerIndex.get();
         final int offset = calcElementOffset(index);
         // local load of field to avoid repeated loads after volatile reads
-        final E e = lvElement(offset);// LoadLoad
+        final E e = lvElement(offset); // LoadLoad
         if (null == e) {
             return null;
         }
         soConsumerIndex(index + 1); // ordered store -> atomic and ordered for size()
-        soElement(offset, null);// StoreStore
+        soElement(offset, null); // StoreStore
         return e;
     }
 

src/main/java/io/reactivex/internal/queue/SpscLinkedArrayQueue.java

@@ -92,8 +92,8 @@ public boolean offer(final T e) {
     }
 
     private boolean writeToQueue(final AtomicReferenceArray<Object> buffer, final T e, final long index, final int offset) {
-        soElement(buffer, offset, e);// StoreStore
-        soProducerIndex(index + 1);// this ensures atomic write of long on 32bit platforms
+        soElement(buffer, offset, e); // StoreStore
+        soProducerIndex(index + 1); // this ensures atomic write of long on 32bit platforms
         return true;
     }
 
@@ -103,11 +103,11 @@ private void resize(final AtomicReferenceArray<Object> oldBuffer, final long cur
         final AtomicReferenceArray<Object> newBuffer = new AtomicReferenceArray<Object>(capacity);
         producerBuffer = newBuffer;
         producerLookAhead = currIndex + mask - 1;
-        soElement(newBuffer, offset, e);// StoreStore
+        soElement(newBuffer, offset, e); // StoreStore
         soNext(oldBuffer, newBuffer);
         soElement(oldBuffer, offset, HAS_NEXT); // new buffer is visible after element is
                                                                  // inserted
-        soProducerIndex(currIndex + 1);// this ensures correctness on 32bit platforms
+        soProducerIndex(currIndex + 1); // this ensures correctness on 32bit platforms
     }
 
     private void soNext(AtomicReferenceArray<Object> curr, AtomicReferenceArray<Object> next) {
@@ -135,11 +135,11 @@ public T poll() {
         final long index = lpConsumerIndex();
         final int mask = consumerMask;
         final int offset = calcWrappedOffset(index, mask);
-        final Object e = lvElement(buffer, offset);// LoadLoad
+        final Object e = lvElement(buffer, offset); // LoadLoad
         boolean isNextBuffer = e == HAS_NEXT;
         if (null != e && !isNextBuffer) {
-            soElement(buffer, offset, null);// StoreStore
-            soConsumerIndex(index + 1);// this ensures correctness on 32bit platforms
+            soElement(buffer, offset, null); // StoreStore
+            soConsumerIndex(index + 1); // this ensures correctness on 32bit platforms
             return (T) e;
         } else if (isNextBuffer) {
             return newBufferPoll(lvNextBufferAndUnlink(buffer, mask + 1), index, mask);
@@ -152,10 +152,10 @@ public T poll() {
     private T newBufferPoll(AtomicReferenceArray<Object> nextBuffer, final long index, final int mask) {
         consumerBuffer = nextBuffer;
         final int offsetInNew = calcWrappedOffset(index, mask);
-        final T n = (T) lvElement(nextBuffer, offsetInNew);// LoadLoad
+        final T n = (T) lvElement(nextBuffer, offsetInNew); // LoadLoad
         if (null != n) {
-            soElement(nextBuffer, offsetInNew, null);// StoreStore
-            soConsumerIndex(index + 1);// this ensures correctness on 32bit platforms
+            soElement(nextBuffer, offsetInNew, null); // StoreStore
+            soConsumerIndex(index + 1); // this ensures correctness on 32bit platforms
         }
         return n;
     }
@@ -166,7 +166,7 @@ public T peek() {
         final long index = lpConsumerIndex();
         final int mask = consumerMask;
         final int offset = calcWrappedOffset(index, mask);
-        final Object e = lvElement(buffer, offset);// LoadLoad
+        final Object e = lvElement(buffer, offset); // LoadLoad
         if (e == HAS_NEXT) {
             return newBufferPeek(lvNextBufferAndUnlink(buffer, mask + 1), index, mask);
         }
@@ -178,7 +178,7 @@ public T peek() {
     private T newBufferPeek(AtomicReferenceArray<Object> nextBuffer, final long index, final int mask) {
         consumerBuffer = nextBuffer;
         final int offsetInNew = calcWrappedOffset(index, mask);
-        return (T) lvElement(nextBuffer, offsetInNew);// LoadLoad
+        return (T) lvElement(nextBuffer, offsetInNew); // LoadLoad
     }
 
     @Override
@@ -277,13 +277,13 @@ public boolean offer(T first, T second) {
             producerBuffer = newBuffer;
 
             pi = calcWrappedOffset(p, m);
-            soElement(newBuffer, pi + 1, second);// StoreStore
+            soElement(newBuffer, pi + 1, second); // StoreStore
             soElement(newBuffer, pi, first);
             soNext(buffer, newBuffer);
 
             soElement(buffer, pi, HAS_NEXT); // new buffer is visible after element is
 
-            soProducerIndex(p + 2);// this ensures correctness on 32bit platforms
+            soProducerIndex(p + 2); // this ensures correctness on 32bit platforms
         }
 
         return true;

src/main/java/io/reactivex/internal/schedulers/IoScheduler.java

@@ -151,6 +151,7 @@ public IoScheduler() {
     }
 
     /**
+     * Constructs an IoScheduler with the given thread factory and starts the pool of workers.
      * @param threadFactory thread factory to use for creating worker threads. Note that this takes precedence over any
      *                      system properties for configuring new thread creation. Cannot be null.
      */

src/main/java/io/reactivex/internal/schedulers/SingleScheduler.java

@@ -53,6 +53,8 @@ public SingleScheduler() {
     }
 
     /**
+     * Constructs a SingleScheduler with the given ThreadFactory and prepares the
+     * single scheduler thread.
      * @param threadFactory thread factory to use for creating worker threads. Note that this takes precedence over any
      *                      system properties for configuring new thread creation. Cannot be null.
      */

src/main/java/io/reactivex/internal/util/ExceptionHelper.java

@@ -129,7 +129,7 @@ public static String timeoutMessage(long timeout, TimeUnit unit) {
                 + unit.toString().toLowerCase()
                 + " and has been terminated.";
     }
-    
+
     static final class Termination extends Throwable {
 
         private static final long serialVersionUID = -4649703670690200604L;

src/main/java/io/reactivex/processors/UnicastProcessor.java

@@ -255,7 +255,7 @@ public static <T> UnicastProcessor<T> create(int capacityHint, Runnable onCancel
      * @since 2.0
      */
     UnicastProcessor(int capacityHint) {
-        this(capacityHint,null, true);
+        this(capacityHint, null, true);
     }
 
     /**

src/test/java/io/reactivex/exceptions/ExceptionsTest.java

@@ -59,6 +59,7 @@ public void accept(Integer t1) {
     }
 
     /**
+     * Outdated test: Observer should not suppress errors from onCompleted.
      * https://github.com/ReactiveX/RxJava/issues/3885
      */
     @Ignore("v2 components should not throw")
@@ -200,6 +201,7 @@ public void onNext(Integer t) {
     }
 
     /**
+     * Outdated test: throwing from onError handler.
      * https://github.com/ReactiveX/RxJava/issues/969
      */
     @Ignore("v2 components should not throw")
@@ -237,6 +239,7 @@ public void onNext(Object o) {
     }
 
     /**
+     * Outdated test: throwing from onError.
      * https://github.com/ReactiveX/RxJava/issues/2998
      * @throws Exception on arbitrary errors
      */
@@ -276,6 +279,7 @@ public void onNext(GroupedObservable<Integer, Integer> integerIntegerGroupedObse
     }
 
     /**
+     * Outdated test: throwing from onError.
      * https://github.com/ReactiveX/RxJava/issues/2998
      * @throws Exception on arbitrary errors
      */