Remove XResource classes (#4126)

Author: akarnokd

src/main/java/io/reactivex/Completable.java

@@ -60,7 +60,7 @@ public interface CompletableTransformer extends Function<Completable, Completabl
      * @return the source or its wrapper Completable
      * @throws NullPointerException if source is null
      */
-    static Completable wrap(CompletableConsumable source) {
+    public static Completable wrap(CompletableConsumable source) {
         Objects.requireNonNull(source, "source is null");
         if (source instanceof Completable) {
             return (Completable)source;
@@ -175,10 +175,8 @@ public static Completable concat(Publisher<? extends CompletableConsumable> sour
      * when the Completable is subscribed to.
      * @return the created Completable instance
      * @throws NullPointerException if onSubscribe is null
-     * @deprecated
      */
     @SchedulerSupport(SchedulerSupport.NONE)
-    @Deprecated // FIXME temporary
     public static Completable create(CompletableConsumable onSubscribe) {
         Objects.requireNonNull(onSubscribe, "onSubscribe is null");
         if (onSubscribe instanceof Completable) {
@@ -798,7 +796,6 @@ public final Throwable get(long timeout, TimeUnit unit) {
      * @throws NullPointerException if onLift is null
      */
     @SchedulerSupport(SchedulerSupport.NONE)
-    @Deprecated // FIXME temporary
     public final Completable lift(final CompletableOperator onLift) {
         Objects.requireNonNull(onLift, "onLift is null");
         return new CompletableLift(this, onLift);

src/main/java/io/reactivex/disposables/CompositeDisposable.java

@@ -15,13 +15,14 @@
 import java.util.*;
 
 import io.reactivex.exceptions.CompositeException;
+import io.reactivex.internal.disposables.DisposableContainer;
 import io.reactivex.internal.functions.Objects;
 import io.reactivex.internal.util.*;
 
 /**
  * A disposable container that can hold onto multiple other disposables.
  */
-public final class CompositeDisposable implements Disposable {
+public final class CompositeDisposable implements Disposable, DisposableContainer {
 
     OpenHashSet<Disposable> resources;
 
@@ -156,6 +157,18 @@ public void clear() {
         dispose(set);
     }
 
+    public int size() {
+        if (disposed) {
+            return 0;
+        }
+        synchronized (this) {
+            if (disposed) {
+                return 0;
+            }
+            return resources.size();
+        }
+    }
+    
     void dispose(OpenHashSet<Disposable> set) {
         if (set == null) {
             return;

src/main/java/io/reactivex/disposables/Disposables.java

@@ -13,12 +13,13 @@
 
 package io.reactivex.disposables;
 
-import io.reactivex.functions.Consumer;
-import io.reactivex.internal.disposables.EmptyDisposable;
-import io.reactivex.internal.functions.Functions;
 import java.util.concurrent.Future;
+
 import org.reactivestreams.Subscription;
 
+import io.reactivex.internal.disposables.EmptyDisposable;
+import io.reactivex.internal.functions.Functions;
+
 /**
  * Utility class to help create disposables by wrapping
  * other types.
@@ -52,21 +53,4 @@ public static Disposable empty() {
     public static Disposable disposed() {
         return EmptyDisposable.INSTANCE;
     }
-
-    private static final Consumer<Disposable> DISPOSER = new Consumer<Disposable>() {
-        @Override
-        public void accept(Disposable d) {
-            d.dispose();
-        }
-    };
-    
-    /**
-     * Returns a consumer that calls dispose on the received Disposable.
-     * @return the consumer that calls dispose on the received Disposable.
-     * @deprecated that generic resource management will be removed
-     */
-    @Deprecated
-    public static Consumer<Disposable> consumeAndDispose() {
-        return DISPOSER;
-    }
 }

src/main/java/io/reactivex/disposables/FutureDisposable.java

@@ -15,6 +15,9 @@
 import java.util.concurrent.Future;
 
 final class FutureDisposable extends ReferenceDisposable<Future<?>> {
+    /** */
+    private static final long serialVersionUID = 6545242830671168775L;
+    
     private final boolean allowInterrupt;
 
     FutureDisposable(Future<?> run, boolean allowInterrupt) {

src/main/java/io/reactivex/disposables/RefCountDisposable.java

@@ -56,6 +56,9 @@ public boolean isDisposed() {
     }
 
     static final class InnerDisposable extends ReferenceDisposable<RefCountDisposable> {
+        /** */
+        private static final long serialVersionUID = -6066815451193282256L;
+
         InnerDisposable(RefCountDisposable parent) {
             super(parent);
         }

src/main/java/io/reactivex/disposables/ReferenceDisposable.java

@@ -17,6 +17,9 @@
 import java.util.concurrent.atomic.AtomicReference;
 
 abstract class ReferenceDisposable<T> extends AtomicReference<T> implements Disposable {
+    /** */
+    private static final long serialVersionUID = 6537757548749041217L;
+
     ReferenceDisposable(T value) {
         super(Objects.requireNonNull(value, "value is null"));
     }

src/main/java/io/reactivex/disposables/RunnableDisposable.java

@@ -13,6 +13,9 @@
 package io.reactivex.disposables;
 
 final class RunnableDisposable extends ReferenceDisposable<Runnable> {
+    /** */
+    private static final long serialVersionUID = -8219729196779211169L;
+
     RunnableDisposable(Runnable value) {
         super(value);
     }

src/main/java/io/reactivex/disposables/SubscriptionDisposable.java

@@ -15,6 +15,9 @@
 import org.reactivestreams.Subscription;
 
 final class SubscriptionDisposable extends ReferenceDisposable<Subscription> {
+    /** */
+    private static final long serialVersionUID = -707001650852963139L;
+
     SubscriptionDisposable(Subscription value) {
         super(value);
     }

src/main/java/io/reactivex/internal/disposables/ArrayCompositeResource.java renamed to src/main/java/io/reactivex/internal/disposables/ArrayCompositeDisposable.java

@@ -16,30 +16,20 @@
 import java.util.concurrent.atomic.AtomicReferenceArray;
 
 import io.reactivex.disposables.Disposable;
-import io.reactivex.functions.Consumer;
 
 /**
- * A composite resource with a fixed number of slots.
+ * A composite disposable with a fixed number of slots.
  *
  * <p>Note that since the implementation leaks the methods of AtomicReferenceArray, one must be
  * careful to only call setResource, replaceResource and dispose on it. All other methods may lead to undefined behavior
  * and should be used by internal means only.
- * 
- * @param <T> the resource tpye
- * @deprecated Use more type-specific and inlined resource management
  */
-@Deprecated
-public final class ArrayCompositeResource<T> extends AtomicReferenceArray<Object> implements Disposable {
+public final class ArrayCompositeDisposable extends AtomicReferenceArray<Disposable> implements Disposable {
     /** */
     private static final long serialVersionUID = 2746389416410565408L;
 
-    final Consumer<? super T> disposer;
-
-    static final Object DISPOSED = new Object();
-    
-    public ArrayCompositeResource(int capacity, Consumer<? super T> disposer) {
+    public ArrayCompositeDisposable(int capacity) {
         super(capacity);
-        this.disposer = disposer;
     }
 
     /**
@@ -48,17 +38,16 @@ public ArrayCompositeResource(int capacity, Consumer<? super T> disposer) {
      * @param resource
      * @return true if the resource has ben set, false if the composite has been disposed
      */
-    @SuppressWarnings("unchecked")
-    public boolean setResource(int index, T resource) {
+    public boolean setResource(int index, Disposable resource) {
         for (;;) {
-            Object o = get(index);
-            if (o == DISPOSED) {
-                disposer.accept(resource);
+            Disposable o = get(index);
+            if (o == DisposableHelper.DISPOSED) {
+                resource.dispose();;
                 return false;
             }
             if (compareAndSet(index, o, resource)) {
                 if (o != null) {
-                    disposer.accept((T)o);
+                    o.dispose();
                 }
                 return true;
             }
@@ -71,31 +60,29 @@ public boolean setResource(int index, T resource) {
      * @param resource
      * @return the old resource, can be null
      */
-    @SuppressWarnings("unchecked")
-    public T replaceResource(int index, T resource) {
+    public Disposable replaceResource(int index, Disposable resource) {
         for (;;) {
-            Object o = get(index);
-            if (o == DISPOSED) {
-                disposer.accept(resource);
+            Disposable o = get(index);
+            if (o == DisposableHelper.DISPOSED) {
+                resource.dispose();
                 return null;
             }
             if (compareAndSet(index, o, resource)) {
-                return (T)o;
+                return o;
             }
         }
     }
 
     @Override
-    @SuppressWarnings("unchecked")
     public void dispose() {
-        if (get(0) != DISPOSED) {
+        if (get(0) != DisposableHelper.DISPOSED) {
             int s = length();
             for (int i = 0; i < s; i++) {
-                Object o = get(i);
-                if (o != DISPOSED) {
-                    o = getAndSet(i, DISPOSED);
-                    if (o != DISPOSED && o != null) {
-                        disposer.accept((T)o);
+                Disposable o = get(i);
+                if (o != DisposableHelper.DISPOSED) {
+                    o = getAndSet(i, DisposableHelper.DISPOSED);
+                    if (o != DisposableHelper.DISPOSED && o != null) {
+                        o.dispose();
                     }
                 }
             }
@@ -104,6 +91,6 @@ public void dispose() {
 
     @Override
     public boolean isDisposed() {
-        return get(0) == DISPOSED;
+        return get(0) == DisposableHelper.DISPOSED;
     }
 }

src/main/java/io/reactivex/internal/disposables/CompositeResource.java renamed to src/main/java/io/reactivex/internal/disposables/DisposableContainer.java

@@ -13,11 +13,16 @@
 
 package io.reactivex.internal.disposables;
 
-public interface CompositeResource<T> {
+import io.reactivex.disposables.Disposable;
+
+/**
+ * Common interface to add and remove disposables from a container.
+ */
+public interface DisposableContainer {
 
-    boolean add(T resource);
+    boolean add(Disposable d);
 
-    boolean remove(T resource);
+    boolean remove(Disposable d);
 
-    boolean delete(T resource);
+    boolean delete(Disposable d);
 }