Cleanup code style

Author: akarnokd

README.md

@@ -1695,11 +1695,11 @@ Map the upstream signals onto some reactive type and relay its events to the dow
 
 Availability:
 
-- `Flowables.flatMap{Completable|Single|Maybe}`
-- `Observables.flatMap{Completable|Single|Maybe}`
+<!-- - `Flowables.flatMap{Completable|Single|Maybe}` -->
+<!-- - `Observables.flatMap{Completable|Single|Maybe}` -->
 - `Single`
   - `SingleTransformers.flatMap` (use with `Single.compose()`)
-  - `Singles.flatMapSingle` (use with `Single.as()`)
+  - `Singles.flatMapCompletable` (use with `Single.as()`)
   - `Singles.flatMapMaybe` (use with `Single.as()`)
   - `Singles.flatMapObservable` (use with `Single.as()`)
   - `Singles.flatMapFlowable` (use with `Single.as()`)

gradle/javadocStyleSheet.css

@@ -15,7 +15,6 @@ code { color: #777; line-height: 1.286em; font-family: "Consolas", "Lucida Conso
 a { text-decoration: none; color: #16569A; /* also try #2E85ED, #0033FF, #6C93C6, #1D7BBE, #1D8DD2 */ }
 a:hover { text-decoration: underline; }
 
-
 table[border="1"] { border: 1px solid #ddd; }
 table[border="1"] td, table[border="1"] th { border: 1px solid #ddd; }
 table[cellpadding="3"] td { padding: 0.5em; }
@@ -47,7 +46,6 @@ h2.title { color: #c81f08; }
   text-decoration: underline;
 }
 
-
 /* Navigation bar fonts and colors */
 .NavBarCell1    { background-color:#E0E6DF; } /* Light mauve */
 .NavBarCell1Rev { background-color:#16569A; color:#FFFFFF} /* Dark Blue */

src/jmh/java/hu/akarnokd/rxjava2/FlatMapExPerf.java

@@ -174,7 +174,6 @@ public void asyncAsync(Blackhole bh) {
         c.await(count * range);
     }
 
-
     @Benchmark
     public void syncSyncB(Blackhole bh) {
         PerfConsumer c = new PerfConsumer(bh);

src/main/java/hu/akarnokd/rxjava2/async/AsyncFlowable.java

@@ -1575,7 +1575,6 @@ public static <T> Future<Object> forEachFuture(
         return forEachFuture(source, onNext, Functions.emptyConsumer(), Functions.EMPTY_ACTION, Schedulers.computation());
     }
 
-
     /**
      * Subscribes to the given source and calls the callback for each emitted item, and surfaces the completion
      * or error through a Future.
@@ -1606,7 +1605,6 @@ public static <T> Future<Object> forEachFuture(
         return forEachFuture(source, onNext, onError, Functions.EMPTY_ACTION, Schedulers.computation());
     }
 
-
     /**
      * Subscribes to the given source and calls the callback for each emitted item, and surfaces the completion
      * or error through a Future.
@@ -1639,7 +1637,6 @@ public static <T> Future<Object> forEachFuture(
         return forEachFuture(source, onNext, onError, onComplete, Schedulers.computation());
     }
 
-
     /**
      * Subscribes to the given source and calls the callback for each emitted item, and surfaces the completion
      * or error through a Future, scheduled on the given scheduler.
@@ -1666,7 +1663,6 @@ public static <T> Future<Object> forEachFuture(
         return forEachFuture(source, onNext, Functions.emptyConsumer(), Functions.EMPTY_ACTION, scheduler);
     }
 
-
     /**
      * Subscribes to the given source and calls the callback for each emitted item, and surfaces the completion
      * or error through a Future, scheduled on the given Scheduler.
@@ -1695,7 +1691,6 @@ public static <T> Future<Object> forEachFuture(
         return forEachFuture(source, onNext, onError, Functions.EMPTY_ACTION, scheduler);
     }
 
-
     /**
      * Subscribes to the given source and calls the callback for each emitted item, and surfaces the completion
      * or error through a Future, scheduled on the given Scheduler.

src/main/java/hu/akarnokd/rxjava2/async/AsyncObservable.java

@@ -41,7 +41,6 @@ private AsyncObservable() {
         throw new IllegalStateException("No instances!");
     }
 
-
     /**
      * Invokes the specified function asynchronously and returns an Observable that emits the result.
      * <p>
@@ -1472,7 +1471,6 @@ public static <T> Future<Object> forEachFuture(
         return forEachFuture(source, onNext, Functions.emptyConsumer(), Functions.EMPTY_ACTION, Schedulers.computation());
     }
 
-
     /**
      * Subscribes to the given source and calls the callback for each emitted item, and surfaces the completion
      * or error through a Future.
@@ -1501,7 +1499,6 @@ public static <T> Future<Object> forEachFuture(
         return forEachFuture(source, onNext, onError, Functions.EMPTY_ACTION, Schedulers.computation());
     }
 
-
     /**
      * Subscribes to the given source and calls the callback for each emitted item, and surfaces the completion
      * or error through a Future.
@@ -1532,7 +1529,6 @@ public static <T> Future<Object> forEachFuture(
         return forEachFuture(source, onNext, onError, onComplete, Schedulers.computation());
     }
 
-
     /**
      * Subscribes to the given source and calls the callback for each emitted item, and surfaces the completion
      * or error through a Future, scheduled on the given scheduler.
@@ -1557,7 +1553,6 @@ public static <T> Future<Object> forEachFuture(
         return forEachFuture(source, onNext, Functions.emptyConsumer(), Functions.EMPTY_ACTION, scheduler);
     }
 
-
     /**
      * Subscribes to the given source and calls the callback for each emitted item, and surfaces the completion
      * or error through a Future, scheduled on the given Scheduler.
@@ -1584,7 +1579,6 @@ public static <T> Future<Object> forEachFuture(
         return forEachFuture(source, onNext, onError, Functions.EMPTY_ACTION, scheduler);
     }
 
-
     /**
      * Subscribes to the given source and calls the callback for each emitted item, and surfaces the completion
      * or error through a Future, scheduled on the given Scheduler.
@@ -1641,7 +1635,7 @@ public void run() throws Exception {
             }
         }, new Consumer<Disposable>() {
             @Override
-            public void accept(Disposable s) throws Exception {
+            public void accept(Disposable d) throws Exception {
             }
         });
         d.lazySet(ls);
@@ -1710,8 +1704,8 @@ public void run() {
 
         return new DisposableObservable<T>() {
             @Override
-            protected void subscribeActual(Observer<? super T> s) {
-                subject.subscribe(s);
+            protected void subscribeActual(Observer<? super T> observer) {
+                subject.subscribe(observer);
             }
 
             @Override

src/main/java/hu/akarnokd/rxjava2/async/FlowableFromCallableNull.java

@@ -75,8 +75,8 @@ static final class CallableNullSubscription<T> extends DeferredScalarSubscriptio
 
         private static final long serialVersionUID = -7088349936918117528L;
 
-        CallableNullSubscription(Subscriber<? super T> actual) {
-            super(actual);
+        CallableNullSubscription(Subscriber<? super T> downstream) {
+            super(downstream);
         }
 
     }

src/main/java/hu/akarnokd/rxjava2/async/ObservableFromCallableNull.java

@@ -36,9 +36,9 @@ final class ObservableFromCallableNull<T> extends Observable<T> implements Calla
     }
 
     @Override
-    protected void subscribeActual(Observer<? super T> s) {
-        CallableNullDisposable<T> deferred = new CallableNullDisposable<T>(s);
-        s.onSubscribe(deferred);
+    protected void subscribeActual(Observer<? super T> observer) {
+        CallableNullDisposable<T> deferred = new CallableNullDisposable<T>(observer);
+        observer.onSubscribe(deferred);
 
         if (!deferred.isDisposed()) {
 
@@ -49,14 +49,14 @@ protected void subscribeActual(Observer<? super T> s) {
             } catch (Throwable ex) {
                 Exceptions.throwIfFatal(ex);
                 if (!deferred.isDisposed()) {
-                    s.onError(ex);
+                    observer.onError(ex);
                 }
                 return;
             }
 
             if (!deferred.isDisposed()) {
                 if (v == null) {
-                    s.onComplete();
+                    observer.onComplete();
                 } else {
                     deferred.complete(v);
                 }
@@ -73,8 +73,8 @@ static final class CallableNullDisposable<T> extends DeferredScalarDisposable<T>
 
         private static final long serialVersionUID = -7088349936918117528L;
 
-        CallableNullDisposable(Observer<? super T> actual) {
-            super(actual);
+        CallableNullDisposable(Observer<? super T> downstream) {
+            super(downstream);
         }
 
     }

src/main/java/hu/akarnokd/rxjava2/basetypes/BasicNonoSubscriber.java

@@ -25,20 +25,20 @@
  */
 abstract class BasicNonoSubscriber extends BasicEmptyQueueSubscription implements Subscriber<Object> {
 
-    protected final Subscriber<? super Void> actual;
+    protected final Subscriber<? super Void> downstream;
 
-    protected Subscription s;
+    protected Subscription upstream;
 
-    BasicNonoSubscriber(Subscriber<? super Void> actual) {
-        this.actual = actual;
+    BasicNonoSubscriber(Subscriber<? super Void> downstream) {
+        this.downstream = downstream;
     }
 
     @Override
     public void onSubscribe(Subscription s) {
-        if (SubscriptionHelper.validate(this.s, s)) {
-            this.s = s;
+        if (SubscriptionHelper.validate(this.upstream, s)) {
+            this.upstream = s;
 
-            actual.onSubscribe(this);
+            downstream.onSubscribe(this);
         }
     }
 
@@ -49,7 +49,7 @@ public final void onNext(Object t) {
 
     @Override
     public void cancel() {
-        s.cancel();
+        upstream.cancel();
     }
 
 }

src/main/java/hu/akarnokd/rxjava2/basetypes/BasicRefNonoSubscriber.java

@@ -28,20 +28,20 @@
 abstract class BasicRefNonoSubscriber<R> extends BasicRefQueueSubscription<Void, R> implements Subscriber<Void> {
     private static final long serialVersionUID = -3157015053656142804L;
 
-    protected final Subscriber<? super Void> actual;
+    protected final Subscriber<? super Void> downstream;
 
-    Subscription s;
+    Subscription upstream;
 
-    BasicRefNonoSubscriber(Subscriber<? super Void> actual) {
-        this.actual = actual;
+    BasicRefNonoSubscriber(Subscriber<? super Void> downstream) {
+        this.downstream = downstream;
     }
 
     @Override
     public void onSubscribe(Subscription s) {
-        if (SubscriptionHelper.validate(this.s, s)) {
-            this.s = s;
+        if (SubscriptionHelper.validate(this.upstream, s)) {
+            this.upstream = s;
 
-            actual.onSubscribe(this);
+            downstream.onSubscribe(this);
         }
     }
 
@@ -52,7 +52,7 @@ public final void onNext(Void t) {
 
     @Override
     public void cancel() {
-        s.cancel();
+        upstream.cancel();
     }
 
     @Override

src/main/java/hu/akarnokd/rxjava2/basetypes/BasicRefQueueSubscription.java

@@ -28,7 +28,6 @@
  */
 abstract class BasicRefQueueSubscription<T, R> extends AtomicReference<R> implements QueueSubscription<T> {
 
-
     private static final long serialVersionUID = -6671519529404341862L;
 
     @Override

src/main/java/hu/akarnokd/rxjava2/basetypes/BlockingGetSubscriber.java

@@ -35,7 +35,7 @@ final class BlockingGetSubscriber<T> extends CountDownLatch implements Subscribe
 
     volatile boolean cancelled;
 
-    Subscription s;
+    Subscription upstream;
 
     T value;
     Throwable error;
@@ -46,8 +46,8 @@ final class BlockingGetSubscriber<T> extends CountDownLatch implements Subscribe
 
     @Override
     public void onSubscribe(Subscription s) {
-        if (SubscriptionHelper.validate(this.s, s)) {
-            this.s = s;
+        if (SubscriptionHelper.validate(this.upstream, s)) {
+            this.upstream = s;
             if (cancelled) {
                 s.cancel();
             } else {
@@ -74,7 +74,7 @@ public void onComplete() {
 
     void dispose() {
         cancelled = true;
-        Subscription a = s;
+        Subscription a = upstream;
         if (a != null) {
             a.cancel();
         }

src/main/java/hu/akarnokd/rxjava2/basetypes/FuturePerhapsSubscriber.java

@@ -36,22 +36,22 @@ final class FuturePerhapsSubscriber<T> extends CountDownLatch
     T value;
     Throwable error;
 
-    final AtomicReference<Subscription> s;
+    final AtomicReference<Subscription> upstream;
 
     FuturePerhapsSubscriber() {
         super(1);
-        this.s = new AtomicReference<Subscription>();
+        this.upstream = new AtomicReference<Subscription>();
     }
 
     @Override
     public boolean cancel(boolean mayInterruptIfRunning) {
         for (;;) {
-            Subscription a = s.get();
+            Subscription a = upstream.get();
             if (a == this || a == SubscriptionHelper.CANCELLED) {
                 return false;
             }
 
-            if (s.compareAndSet(a, SubscriptionHelper.CANCELLED)) {
+            if (upstream.compareAndSet(a, SubscriptionHelper.CANCELLED)) {
                 if (a != null) {
                     a.cancel();
                 }
@@ -63,7 +63,7 @@ public boolean cancel(boolean mayInterruptIfRunning) {
 
     @Override
     public boolean isCancelled() {
-        return SubscriptionHelper.isCancelled(s.get());
+        return SubscriptionHelper.isCancelled(upstream.get());
     }
 
     @Override
@@ -108,7 +108,7 @@ public T get(long timeout, TimeUnit unit) throws InterruptedException, Execution
 
     @Override
     public void onSubscribe(Subscription s) {
-        if (SubscriptionHelper.setOnce(this.s, s)) {
+        if (SubscriptionHelper.setOnce(this.upstream, s)) {
             s.request(Long.MAX_VALUE);
         }
     }
@@ -121,13 +121,13 @@ public void onNext(T t) {
     @Override
     public void onError(Throwable t) {
         for (;;) {
-            Subscription a = s.get();
+            Subscription a = upstream.get();
             if (a == this || a == SubscriptionHelper.CANCELLED) {
                 RxJavaPlugins.onError(t);
                 return;
             }
             error = t;
-            if (s.compareAndSet(a, this)) {
+            if (upstream.compareAndSet(a, this)) {
                 countDown();
                 return;
             }
@@ -137,11 +137,11 @@ public void onError(Throwable t) {
     @Override
     public void onComplete() {
         for (;;) {
-            Subscription a = s.get();
+            Subscription a = upstream.get();
             if (a == this || a == SubscriptionHelper.CANCELLED) {
                 return;
             }
-            if (s.compareAndSet(a, this)) {
+            if (upstream.compareAndSet(a, this)) {
                 countDown();
                 return;
             }