diff --git a/src/main/java/io/reactivex/rxjava3/disposables/ActionDisposable.java b/src/main/java/io/reactivex/rxjava3/disposables/ActionDisposable.java index dd74d600c6..32ad478656 100644 --- a/src/main/java/io/reactivex/rxjava3/disposables/ActionDisposable.java +++ b/src/main/java/io/reactivex/rxjava3/disposables/ActionDisposable.java @@ -17,7 +17,7 @@ import io.reactivex.rxjava3.internal.util.ExceptionHelper; /** - * A Disposable container that manages an Action instance. + * A Disposable container that manages an {@link Action} instance. */ final class ActionDisposable extends ReferenceDisposable { @@ -35,4 +35,9 @@ protected void onDisposed(@NonNull Action value) { throw ExceptionHelper.wrapOrThrow(ex); } } + + @Override + public String toString() { + return "ActionDisposable(disposed=" + isDisposed() + ", " + get() + ")"; + } } diff --git a/src/main/java/io/reactivex/rxjava3/disposables/AutoCloseableDisposable.java b/src/main/java/io/reactivex/rxjava3/disposables/AutoCloseableDisposable.java new file mode 100644 index 0000000000..8c0f2be2dc --- /dev/null +++ b/src/main/java/io/reactivex/rxjava3/disposables/AutoCloseableDisposable.java @@ -0,0 +1,45 @@ +/** + * Copyright (c) 2016-present, RxJava Contributors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in + * compliance with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software distributed under the License is + * distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See + * the License for the specific language governing permissions and limitations under the License. + */ + +package io.reactivex.rxjava3.disposables; + +import io.reactivex.rxjava3.annotations.NonNull; +import io.reactivex.rxjava3.plugins.RxJavaPlugins; + +/** + * A disposable container that manages an {@link AutoCloseable} instance. + * @since 3.0.0 + */ +final class AutoCloseableDisposable extends ReferenceDisposable { + + private static final long serialVersionUID = -6646144244598696847L; + + AutoCloseableDisposable(AutoCloseable value) { + super(value); + } + + @Override + protected void onDisposed(@NonNull AutoCloseable value) { + try { + value.close(); + } catch (Throwable ex) { + RxJavaPlugins.onError(ex); + } + } + + @Override + public String toString() { + return "AutoCloseableDisposable(disposed=" + isDisposed() + ", " + get() + ")"; + } + +} diff --git a/src/main/java/io/reactivex/rxjava3/disposables/Disposables.java b/src/main/java/io/reactivex/rxjava3/disposables/Disposables.java index 2ff83d64ff..39b1b76b25 100644 --- a/src/main/java/io/reactivex/rxjava3/disposables/Disposables.java +++ b/src/main/java/io/reactivex/rxjava3/disposables/Disposables.java @@ -35,8 +35,8 @@ private Disposables() { } /** - * Construct a Disposable by wrapping a Runnable that is - * executed exactly once when the Disposable is disposed. + * Construct a {@code Disposable} by wrapping a {@link Runnable} that is + * executed exactly once when the {@code Disposable} is disposed. * @param run the Runnable to wrap * @return the new Disposable instance */ @@ -47,8 +47,8 @@ public static Disposable fromRunnable(@NonNull Runnable run) { } /** - * Construct a Disposable by wrapping a Action that is - * executed exactly once when the Disposable is disposed. + * Construct a {@code Disposable} by wrapping a {@link Action} that is + * executed exactly once when the {@code Disposable} is disposed. * @param run the Action to wrap * @return the new Disposable instance */ @@ -59,10 +59,13 @@ public static Disposable fromAction(@NonNull Action run) { } /** - * Construct a Disposable by wrapping a Future that is - * cancelled exactly once when the Disposable is disposed. + * Construct a {@code Disposable} by wrapping a {@link Future} that is + * cancelled exactly once when the {@code Disposable} is disposed. + *

+ * The {@code Future} is cancelled with {@code mayInterruptIfRunning == true}. * @param future the Future to wrap * @return the new Disposable instance + * @see #fromFuture(Future, boolean) */ @NonNull public static Disposable fromFuture(@NonNull Future future) { @@ -71,10 +74,10 @@ public static Disposable fromFuture(@NonNull Future future) { } /** - * Construct a Disposable by wrapping a Future that is - * cancelled exactly once when the Disposable is disposed. + * Construct a {@code Disposable} by wrapping a {@link Future} that is + * cancelled exactly once when the {@code Disposable} is disposed. * @param future the Future to wrap - * @param allowInterrupt if true, the future cancel happens via Future.cancel(true) + * @param allowInterrupt if true, the future cancel happens via {@code Future.cancel(true)} * @return the new Disposable instance */ @NonNull @@ -84,8 +87,8 @@ public static Disposable fromFuture(@NonNull Future future, boolean allowInte } /** - * Construct a Disposable by wrapping a Subscription that is - * cancelled exactly once when the Disposable is disposed. + * Construct a {@code Disposable} by wrapping a {@link Subscription} that is + * cancelled exactly once when the {@code Disposable} is disposed. * @param subscription the Runnable to wrap * @return the new Disposable instance */ @@ -96,8 +99,33 @@ public static Disposable fromSubscription(@NonNull Subscription subscription) { } /** - * Returns a new, non-disposed Disposable instance. - * @return a new, non-disposed Disposable instance + * Construct a {@code Disposable} by wrapping an {@link AutoCloseable} that is + * closed exactly once when the {@code Disposable} is disposed. + * @param autoCloseable the AutoCloseable to wrap + * @return the new Disposable instance + * @since 3.0.0 + */ + @NonNull + public static Disposable fromAutoCloseable(@NonNull AutoCloseable autoCloseable) { + Objects.requireNonNull(autoCloseable, "autoCloseable is null"); + return new AutoCloseableDisposable(autoCloseable); + } + + /** + * Construct an {@link AutoCloseable} by wrapping a {@code Disposable} that is + * disposed when the returned {@code AutoCloseable} is closed. + * @param disposable the Disposable instance + * @return the new AutoCloseable instance + * @since 3.0.0 + */ + @NonNull + public static AutoCloseable toAutoCloseable(@NonNull Disposable disposable) { + return disposable::dispose; + } + + /** + * Returns a new, non-disposed {@code Disposable} instance. + * @return a new, non-disposed {@code Disposable} instance */ @NonNull public static Disposable empty() { @@ -105,8 +133,8 @@ public static Disposable empty() { } /** - * Returns a disposed Disposable instance. - * @return a disposed Disposable instance + * Returns a shared, disposed {@code Disposable} instance. + * @return a shared, disposed {@code Disposable} instance */ @NonNull public static Disposable disposed() { diff --git a/src/main/java/io/reactivex/rxjava3/disposables/FutureDisposable.java b/src/main/java/io/reactivex/rxjava3/disposables/FutureDisposable.java index 0a900f6ed1..9649bcd7d3 100644 --- a/src/main/java/io/reactivex/rxjava3/disposables/FutureDisposable.java +++ b/src/main/java/io/reactivex/rxjava3/disposables/FutureDisposable.java @@ -16,7 +16,7 @@ import java.util.concurrent.atomic.AtomicReference; /** - * A Disposable container that cancels a Future instance. + * A Disposable container that cancels a {@link Future} instance. */ final class FutureDisposable extends AtomicReference> implements Disposable { diff --git a/src/main/java/io/reactivex/rxjava3/disposables/RunnableDisposable.java b/src/main/java/io/reactivex/rxjava3/disposables/RunnableDisposable.java index 43eee3105c..34203e86a9 100644 --- a/src/main/java/io/reactivex/rxjava3/disposables/RunnableDisposable.java +++ b/src/main/java/io/reactivex/rxjava3/disposables/RunnableDisposable.java @@ -15,7 +15,7 @@ import io.reactivex.rxjava3.annotations.NonNull; /** - * A disposable container that manages a Runnable instance. + * A disposable container that manages a {@link Runnable} instance. */ final class RunnableDisposable extends ReferenceDisposable { diff --git a/src/test/java/io/reactivex/rxjava3/disposables/DisposablesTest.java b/src/test/java/io/reactivex/rxjava3/disposables/DisposablesTest.java index d5ea5d0d49..244ca873f0 100644 --- a/src/test/java/io/reactivex/rxjava3/disposables/DisposablesTest.java +++ b/src/test/java/io/reactivex/rxjava3/disposables/DisposablesTest.java @@ -25,6 +25,7 @@ import org.reactivestreams.Subscription; import io.reactivex.rxjava3.core.RxJavaTest; +import io.reactivex.rxjava3.exceptions.TestException; import io.reactivex.rxjava3.functions.Action; import io.reactivex.rxjava3.internal.disposables.DisposableHelper; import io.reactivex.rxjava3.plugins.RxJavaPlugins; @@ -34,11 +35,19 @@ public class DisposablesTest extends RxJavaTest { @Test public void unsubscribeOnlyOnce() { - Runnable dispose = mock(Runnable.class); - Disposable subscription = Disposables.fromRunnable(dispose); - subscription.dispose(); - subscription.dispose(); - verify(dispose, times(1)).run(); + Runnable run = mock(Runnable.class); + + Disposable d = Disposables.fromRunnable(run); + + assertTrue(d.toString(), d.toString().contains("RunnableDisposable(disposed=false, ")); + + d.dispose(); + assertTrue(d.toString(), d.toString().contains("RunnableDisposable(disposed=true, ")); + + d.dispose(); + assertTrue(d.toString(), d.toString().contains("RunnableDisposable(disposed=true, ")); + + verify(run, times(1)).run(); } @Test @@ -61,22 +70,20 @@ public void utilityClass() { } @Test - public void fromAction() { - class AtomicAction extends AtomicBoolean implements Action { + public void fromAction() throws Throwable { + Action action = mock(Action.class); - private static final long serialVersionUID = -1517510584253657229L; + Disposable d = Disposables.fromAction(action); - @Override - public void run() throws Exception { - set(true); - } - } + assertTrue(d.toString(), d.toString().contains("ActionDisposable(disposed=false, ")); - AtomicAction aa = new AtomicAction(); + d.dispose(); + assertTrue(d.toString(), d.toString().contains("ActionDisposable(disposed=true, ")); - Disposables.fromAction(aa).dispose(); + d.dispose(); + assertTrue(d.toString(), d.toString().contains("ActionDisposable(disposed=true, ")); - assertTrue(aa.get()); + verify(action, times(1)).run(); } @Test @@ -174,4 +181,76 @@ public void setOnceTwice() { RxJavaPlugins.reset(); } } + + @Test + public void fromAutoCloseable() { + AtomicInteger counter = new AtomicInteger(); + + AutoCloseable ac = () -> counter.getAndIncrement(); + + Disposable d = Disposables.fromAutoCloseable(ac); + + assertFalse(d.isDisposed()); + assertEquals(0, counter.get()); + assertTrue(d.toString(), d.toString().contains("AutoCloseableDisposable(disposed=false, ")); + + d.dispose(); + + assertTrue(d.isDisposed()); + assertEquals(1, counter.get()); + assertTrue(d.toString(), d.toString().contains("AutoCloseableDisposable(disposed=true, ")); + + d.dispose(); + + assertTrue(d.isDisposed()); + assertEquals(1, counter.get()); + assertTrue(d.toString(), d.toString().contains("AutoCloseableDisposable(disposed=true, ")); + } + + @Test + public void fromAutoCloseableThrows() throws Throwable { + TestHelper.withErrorTracking(errors -> { + AutoCloseable ac = () -> { throw new TestException(); }; + + Disposable d = Disposables.fromAutoCloseable(ac); + + assertFalse(d.isDisposed()); + + assertTrue(errors.isEmpty()); + + d.dispose(); + + assertTrue(d.isDisposed()); + assertEquals(1, errors.size()); + + d.dispose(); + + assertTrue(d.isDisposed()); + assertEquals(1, errors.size()); + + TestHelper.assertUndeliverable(errors, 0, TestException.class); + }); + } + + @Test + public void toAutoCloseable() throws Exception { + AtomicInteger counter = new AtomicInteger(); + + Disposable d = Disposables.fromAction(() -> counter.getAndIncrement()); + + AutoCloseable ac = Disposables.toAutoCloseable(d); + + assertFalse(d.isDisposed()); + assertEquals(0, counter.get()); + + ac.close(); + + assertTrue(d.isDisposed()); + assertEquals(1, counter.get()); + + ac.close(); + + assertTrue(d.isDisposed()); + assertEquals(1, counter.get()); + } }