Removed unnecessary upstream.cancel() call for casually finished upst…

…ream sequences. (#6992)

* no upstream.cancel() in FlowablePublishMulticast when the sequence is finished normally via onComplete/onError from upstream;
minor code cleanup - unnecessary Disposable implementation to avoid method name clash

* cleanup in FlowablePublishFunctionTest: refactored anonymous classes to lambdas

* reduced visibility for dispose() and isDisposed() in the inner MulticastProcessor

Co-authored-by: derar <derar@appsflyer.com>

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

@@ -22,7 +22,6 @@
 import org.reactivestreams.*;
 
 import io.reactivex.rxjava3.core.*;
-import io.reactivex.rxjava3.disposables.Disposable;
 import io.reactivex.rxjava3.exceptions.*;
 import io.reactivex.rxjava3.functions.Function;
 import io.reactivex.rxjava3.internal.fuseable.*;
@@ -124,7 +123,7 @@ public void cancel() {
         }
     }
 
-    static final class MulticastProcessor<T> extends Flowable<T> implements FlowableSubscriber<T>, Disposable {
+    static final class MulticastProcessor<T> extends Flowable<T> implements FlowableSubscriber<T> {
 
         @SuppressWarnings("rawtypes")
         static final MulticastSubscription[] EMPTY = new MulticastSubscription[0];
@@ -192,19 +191,19 @@ public void onSubscribe(Subscription s) {
             }
         }
 
-        @Override
-        public void dispose() {
-            SubscriptionHelper.cancel(upstream);
-            if (wip.getAndIncrement() == 0) {
-                SimpleQueue<T> q = queue;
-                if (q != null) {
-                    q.clear();
+        void dispose() {
+            if (!done) {
+                SubscriptionHelper.cancel(upstream);
+                if (wip.getAndIncrement() == 0) {
+                    SimpleQueue<T> q = queue;
+                    if (q != null) {
+                        q.clear();
+                    }
                 }
             }
         }
 
-        @Override
-        public boolean isDisposed() {
+        boolean isDisposed() {
             return upstream.get() == SubscriptionHelper.CANCELLED;
         }