3.x: Remove unnecessary cancel/dispose calls from terminating using (#7121)

Author: akarnokd

src/main/java/io/reactivex/rxjava3/core/FlowableSubscriber.java

@@ -18,8 +18,10 @@
 import io.reactivex.rxjava3.annotations.NonNull;
 
 /**
- * Represents a Reactive-Streams inspired {@link Subscriber} that is RxJava 2 only
- * and weakens rules §1.3 and §3.9 of the specification for gaining performance.
+ * Represents a Reactive-Streams inspired {@link Subscriber} that is RxJava 3 only
+ * and weakens the Reactive Streams rules <a href='https://github.com/reactive-streams/reactive-streams-jvm#1.3'>§1.3</a>
+ * and <a href='https://github.com/reactive-streams/reactive-streams-jvm#3.9'>§3.9</a> of the specification
+ * for gaining performance.
  *
  * <p>History: 2.0.7 - experimental; 2.1 - beta
  * @param <T> the value type

src/main/java/io/reactivex/rxjava3/internal/operators/flowable/FlowableUsing.java

@@ -117,15 +117,13 @@ public void onError(Throwable t) {
                     }
                 }
 
-                upstream.cancel();
                 if (innerError != null) {
                     downstream.onError(new CompositeException(t, innerError));
                 } else {
                     downstream.onError(t);
                 }
             } else {
                 downstream.onError(t);
-                upstream.cancel();
                 disposeResource();
             }
         }
@@ -143,11 +141,9 @@ public void onComplete() {
                     }
                 }
 
-                upstream.cancel();
                 downstream.onComplete();
             } else {
                 downstream.onComplete();
-                upstream.cancel();
                 disposeResource();
             }
         }

src/main/java/io/reactivex/rxjava3/internal/operators/observable/ObservableUsing.java

@@ -115,11 +115,9 @@ public void onError(Throwable t) {
                     }
                 }
 
-                upstream.dispose();
                 downstream.onError(t);
             } else {
                 downstream.onError(t);
-                upstream.dispose();
                 disposeResource();
             }
         }
@@ -137,11 +135,9 @@ public void onComplete() {
                     }
                 }
 
-                upstream.dispose();
                 downstream.onComplete();
             } else {
                 downstream.onComplete();
-                upstream.dispose();
                 disposeResource();
             }
         }