2.x: Cleaunp - rename fields to upstream and downstream (ReactiveX#6129)

* 2.x: Cleaunp - rename fields to upstream and downstream

* Make links in MaybeEmitter.setDisposable

Author: akarnokd

Diff for: src/jmh/java/io/reactivex/MemoryPerf.java

@@ -35,11 +35,11 @@ static long memoryUse() {
 
     static final class MyRx2Subscriber implements FlowableSubscriber<Object> {
 
-        org.reactivestreams.Subscription s;
+        org.reactivestreams.Subscription upstream;
 
         @Override
         public void onSubscribe(Subscription s) {
-            this.s = s;
+            this.upstream = s;
         }
 
         @Override
@@ -61,11 +61,11 @@ public void onNext(Object t) {
     static final class MyRx2Observer implements io.reactivex.Observer<Object>, io.reactivex.SingleObserver<Object>,
     io.reactivex.MaybeObserver<Object>, io.reactivex.CompletableObserver {
 
-        Disposable s;
+        Disposable upstream;
 
         @Override
-        public void onSubscribe(Disposable s) {
-            this.s = s;
+        public void onSubscribe(Disposable d) {
+            this.upstream = d;
         }
 
         @Override

Diff for: src/main/java/io/reactivex/Completable.java

@@ -1622,11 +1622,11 @@ public final Completable doFinally(Action onFinally) {
      *     // and subsequently this class has to send a Disposable to the downstream.
      *     // Note that relaying the upstream's Disposable directly is not allowed in RxJava
      *     @Override
-     *     public void onSubscribe(Disposable s) {
+     *     public void onSubscribe(Disposable d) {
      *         if (upstream != null) {
-     *             s.cancel();
+     *             d.dispose();
      *         } else {
-     *             upstream = s;
+     *             upstream = d;
      *             downstream.onSubscribe(this);
      *         }
      *     }

Diff for: src/main/java/io/reactivex/CompletableEmitter.java

@@ -59,15 +59,15 @@ public interface CompletableEmitter {
     void onError(@NonNull Throwable t);
 
     /**
-     * Sets a Disposable on this emitter; any previous Disposable
-     * or Cancellation will be disposed/cancelled.
+     * Sets a Disposable on this emitter; any previous {@link Disposable}
+     * or {@link Cancellable} will be disposed/cancelled.
      * @param d the disposable, null is allowed
      */
     void setDisposable(@Nullable Disposable d);
 
     /**
-     * Sets a Cancellable on this emitter; any previous Disposable
-     * or Cancellation will be disposed/cancelled.
+     * Sets a Cancellable on this emitter; any previous {@link Disposable}
+     * or {@link Cancellable} will be disposed/cancelled.
      * @param c the cancellable resource, null is allowed
      */
     void setCancellable(@Nullable Cancellable c);

Diff for: src/main/java/io/reactivex/CompletableSource.java

@@ -24,8 +24,8 @@ public interface CompletableSource {
 
     /**
      * Subscribes the given CompletableObserver to this CompletableSource instance.
-     * @param cs the CompletableObserver, not null
-     * @throws NullPointerException if {@code cs} is null
+     * @param co the CompletableObserver, not null
+     * @throws NullPointerException if {@code co} is null
      */
-    void subscribe(@NonNull CompletableObserver cs);
+    void subscribe(@NonNull CompletableObserver co);
 }

Diff for: src/main/java/io/reactivex/Flowable.java

@@ -10787,7 +10787,7 @@ public final Single<T> lastOrError() {
      *
      *     // In the subscription phase, the upstream sends a Subscription to this class
      *     // and subsequently this class has to send a Subscription to the downstream.
-     *     // Note that relaying the upstream's Subscription directly is not allowed in RxJava
+     *     // Note that relaying the upstream's Subscription instance directly is not allowed in RxJava
      *     &#64;Override
      *     public void onSubscribe(Subscription s) {
      *         if (upstream != null) {

Diff for: src/main/java/io/reactivex/FlowableEmitter.java

@@ -51,15 +51,15 @@
 public interface FlowableEmitter<T> extends Emitter<T> {
 
     /**
-     * Sets a Disposable on this emitter; any previous Disposable
-     * or Cancellation will be disposed/cancelled.
-     * @param s the disposable, null is allowed
+     * Sets a Disposable on this emitter; any previous {@link Disposable}
+     * or {@link Cancellable} will be disposed/cancelled.
+     * @param d the disposable, null is allowed
      */
-    void setDisposable(@Nullable Disposable s);
+    void setDisposable(@Nullable Disposable d);
 
     /**
-     * Sets a Cancellable on this emitter; any previous Disposable
-     * or Cancellation will be disposed/cancelled.
+     * Sets a Cancellable on this emitter; any previous {@link Disposable}
+     * or {@link Cancellable} will be disposed/cancelled.
      * @param c the cancellable resource, null is allowed
      */
     void setCancellable(@Nullable Cancellable c);

Diff for: src/main/java/io/reactivex/Maybe.java

@@ -3237,11 +3237,11 @@ public final Single<Boolean> isEmpty() {
      *     // and subsequently this class has to send a Disposable to the downstream.
      *     // Note that relaying the upstream's Disposable directly is not allowed in RxJava
      *     &#64;Override
-     *     public void onSubscribe(Disposable s) {
+     *     public void onSubscribe(Disposable d) {
      *         if (upstream != null) {
-     *             s.cancel();
+     *             d.dispose();
      *         } else {
-     *             upstream = s;
+     *             upstream = d;
      *             downstream.onSubscribe(this);
      *         }
      *     }

Diff for: src/main/java/io/reactivex/MaybeEmitter.java

@@ -67,15 +67,15 @@ public interface MaybeEmitter<T> {
     void onComplete();
 
     /**
-     * Sets a Disposable on this emitter; any previous Disposable
-     * or Cancellation will be unsubscribed/cancelled.
-     * @param s the disposable, null is allowed
+     * Sets a Disposable on this emitter; any previous {@link Disposable}
+     * or {@link Cancellable} will be disposed/cancelled.
+     * @param d the disposable, null is allowed
      */
-    void setDisposable(@Nullable Disposable s);
+    void setDisposable(@Nullable Disposable d);
 
     /**
-     * Sets a Cancellable on this emitter; any previous Disposable
-     * or Cancellation will be unsubscribed/cancelled.
+     * Sets a Cancellable on this emitter; any previous {@link Disposable}
+     * or {@link Cancellable} will be disposed/cancelled.
      * @param c the cancellable resource, null is allowed
      */
     void setCancellable(@Nullable Cancellable c);

Diff for: src/main/java/io/reactivex/Observable.java

@@ -9409,11 +9409,11 @@ public final Single<T> lastOrError() {
      *     // and subsequently this class has to send a Disposable to the downstream.
      *     // Note that relaying the upstream's Disposable directly is not allowed in RxJava
      *     &#64;Override
-     *     public void onSubscribe(Disposable s) {
+     *     public void onSubscribe(Disposable d) {
      *         if (upstream != null) {
-     *             s.dispose();
+     *             d.dispose();
      *         } else {
-     *             upstream = s;
+     *             upstream = d;
      *             downstream.onSubscribe(this);
      *         }
      *     }

Diff for: src/main/java/io/reactivex/ObservableEmitter.java

@@ -50,15 +50,15 @@
 public interface ObservableEmitter<T> extends Emitter<T> {
 
     /**
-     * Sets a Disposable on this emitter; any previous Disposable
-     * or Cancellation will be unsubscribed/cancelled.
+     * Sets a Disposable on this emitter; any previous {@link Disposable}
+     * or {@link Cancellable} will be disposed/cancelled.
      * @param d the disposable, null is allowed
      */
     void setDisposable(@Nullable Disposable d);
 
     /**
-     * Sets a Cancellable on this emitter; any previous Disposable
-     * or Cancellation will be unsubscribed/cancelled.
+     * Sets a Cancellable on this emitter; any previous {@link Disposable}
+     * or {@link Cancellable} will be disposed/cancelled.
      * @param c the cancellable resource, null is allowed
      */
     void setCancellable(@Nullable Cancellable c);

Diff for: src/main/java/io/reactivex/Single.java

@@ -2708,11 +2708,11 @@ public final T blockingGet() {
      *     // and subsequently this class has to send a Disposable to the downstream.
      *     // Note that relaying the upstream's Disposable directly is not allowed in RxJava
      *     @Override
-     *     public void onSubscribe(Disposable s) {
+     *     public void onSubscribe(Disposable d) {
      *         if (upstream != null) {
-     *             s.cancel();
+     *             d.dispose();
      *         } else {
-     *             upstream = s;
+     *             upstream = d;
      *             downstream.onSubscribe(this);
      *         }
      *     }

Diff for: src/main/java/io/reactivex/SingleEmitter.java

@@ -63,14 +63,14 @@ public interface SingleEmitter<T> {
 
     /**
      * Sets a Disposable on this emitter; any previous Disposable
-     * or Cancellation will be unsubscribed/cancelled.
-     * @param s the disposable, null is allowed
+     * or Cancellable will be disposed/cancelled.
+     * @param d the disposable, null is allowed
      */
-    void setDisposable(@Nullable Disposable s);
+    void setDisposable(@Nullable Disposable d);
 
     /**
-     * Sets a Cancellable on this emitter; any previous Disposable
-     * or Cancellation will be unsubscribed/cancelled.
+     * Sets a Cancellable on this emitter; any previous {@link Disposable}
+     * or {@link Cancellable} will be disposed/cancelled.
      * @param c the cancellable resource, null is allowed
      */
     void setCancellable(@Nullable Cancellable c);

Diff for: src/main/java/io/reactivex/internal/observers/BasicFuseableObserver.java

@@ -28,13 +28,13 @@
 public abstract class BasicFuseableObserver<T, R> implements Observer<T>, QueueDisposable<R> {
 
     /** The downstream subscriber. */
-    protected final Observer<? super R> actual;
+    protected final Observer<? super R> downstream;
 
     /** The upstream subscription. */
-    protected Disposable s;
+    protected Disposable upstream;
 
     /** The upstream's QueueDisposable if not null. */
-    protected QueueDisposable<T> qs;
+    protected QueueDisposable<T> qd;
 
     /** Flag indicating no further onXXX event should be accepted. */
     protected boolean done;
@@ -44,26 +44,26 @@ public abstract class BasicFuseableObserver<T, R> implements Observer<T>, QueueD
 
     /**
      * Construct a BasicFuseableObserver by wrapping the given subscriber.
-     * @param actual the subscriber, not null (not verified)
+     * @param downstream the subscriber, not null (not verified)
      */
-    public BasicFuseableObserver(Observer<? super R> actual) {
-        this.actual = actual;
+    public BasicFuseableObserver(Observer<? super R> downstream) {
+        this.downstream = downstream;
     }
 
     // final: fixed protocol steps to support fuseable and non-fuseable upstream
     @SuppressWarnings("unchecked")
     @Override
-    public final void onSubscribe(Disposable s) {
-        if (DisposableHelper.validate(this.s, s)) {
+    public final void onSubscribe(Disposable d) {
+        if (DisposableHelper.validate(this.upstream, d)) {
 
-            this.s = s;
-            if (s instanceof QueueDisposable) {
-                this.qs = (QueueDisposable<T>)s;
+            this.upstream = d;
+            if (d instanceof QueueDisposable) {
+                this.qd = (QueueDisposable<T>)d;
             }
 
             if (beforeDownstream()) {
 
-                actual.onSubscribe(this);
+                downstream.onSubscribe(this);
 
                 afterDownstream();
             }
@@ -97,7 +97,7 @@ public void onError(Throwable t) {
             return;
         }
         done = true;
-        actual.onError(t);
+        downstream.onError(t);
     }
 
     /**
@@ -106,7 +106,7 @@ public void onError(Throwable t) {
      */
     protected final void fail(Throwable t) {
         Exceptions.throwIfFatal(t);
-        s.dispose();
+        upstream.dispose();
         onError(t);
     }
 
@@ -116,24 +116,24 @@ public void onComplete() {
             return;
         }
         done = true;
-        actual.onComplete();
+        downstream.onComplete();
     }
 
     /**
      * Calls the upstream's QueueDisposable.requestFusion with the mode and
      * saves the established mode in {@link #sourceMode} if that mode doesn't
      * have the {@link QueueDisposable#BOUNDARY} flag set.
      * <p>
-     * If the upstream doesn't support fusion ({@link #qs} is null), the method
+     * If the upstream doesn't support fusion ({@link #qd} is null), the method
      * returns {@link QueueDisposable#NONE}.
      * @param mode the fusion mode requested
      * @return the established fusion mode
      */
     protected final int transitiveBoundaryFusion(int mode) {
-        QueueDisposable<T> qs = this.qs;
-        if (qs != null) {
+        QueueDisposable<T> qd = this.qd;
+        if (qd != null) {
             if ((mode & BOUNDARY) == 0) {
-                int m = qs.requestFusion(mode);
+                int m = qd.requestFusion(mode);
                 if (m != NONE) {
                     sourceMode = m;
                 }
@@ -149,22 +149,22 @@ protected final int transitiveBoundaryFusion(int mode) {
 
     @Override
     public void dispose() {
-        s.dispose();
+        upstream.dispose();
     }
 
     @Override
     public boolean isDisposed() {
-        return s.isDisposed();
+        return upstream.isDisposed();
     }
 
     @Override
     public boolean isEmpty() {
-        return qs.isEmpty();
+        return qd.isEmpty();
     }
 
     @Override
     public void clear() {
-        qs.clear();
+        qd.clear();
     }
 
     // -----------------------------------------------------------

Diff for: src/main/java/io/reactivex/internal/observers/BlockingBaseObserver.java

@@ -24,7 +24,7 @@ public abstract class BlockingBaseObserver<T> extends CountDownLatch
     T value;
     Throwable error;
 
-    Disposable d;
+    Disposable upstream;
 
     volatile boolean cancelled;
 
@@ -34,7 +34,7 @@ public BlockingBaseObserver() {
 
     @Override
     public final void onSubscribe(Disposable d) {
-        this.d = d;
+        this.upstream = d;
         if (cancelled) {
             d.dispose();
         }
@@ -48,7 +48,7 @@ public final void onComplete() {
     @Override
     public final void dispose() {
         cancelled = true;
-        Disposable d = this.d;
+        Disposable d = this.upstream;
         if (d != null) {
             d.dispose();
         }

Diff for: src/main/java/io/reactivex/internal/observers/BlockingFirstObserver.java

@@ -24,7 +24,7 @@ public final class BlockingFirstObserver<T> extends BlockingBaseObserver<T> {
     public void onNext(T t) {
         if (value == null) {
             value = t;
-            d.dispose();
+            upstream.dispose();
             countDown();
         }
     }