2.x: RxJavaPlugins unwrapRunnable

src/main/java/io/reactivex/Scheduler.java

@@ -13,8 +13,6 @@
 
 package io.reactivex;
 
-import java.util.concurrent.TimeUnit;
-
 import io.reactivex.annotations.*;
 import io.reactivex.disposables.Disposable;
 import io.reactivex.exceptions.Exceptions;
@@ -23,6 +21,9 @@
 import io.reactivex.internal.schedulers.*;
 import io.reactivex.internal.util.ExceptionHelper;
 import io.reactivex.plugins.RxJavaPlugins;
+import io.reactivex.schedulers.SchedulerRunnableIntrospection;
+
+import java.util.concurrent.TimeUnit;
 
 /**
  * A {@code Scheduler} is an object that specifies an API for scheduling
@@ -197,7 +198,7 @@ public Disposable schedulePeriodicallyDirect(@NonNull Runnable run, long initial
      * <p>
      * Limit the amount concurrency two at a time without creating a new fix
      * size thread pool:
-     * 
+     *
      * <pre>
      * Scheduler limitScheduler = Schedulers.computation().when(workers -&gt; {
      *  // use merge max concurrent to limit the number of concurrent
@@ -215,20 +216,20 @@ public Disposable schedulePeriodicallyDirect(@NonNull Runnable run, long initial
      * {@link Flowable#zip(org.reactivestreams.Publisher, org.reactivestreams.Publisher, io.reactivex.functions.BiFunction)} where
      * subscribing to the first {@link Flowable} could deadlock the
      * subscription to the second.
-     * 
+     *
      * <pre>
      * Scheduler limitScheduler = Schedulers.computation().when(workers -&gt; {
      *  // use merge max concurrent to limit the number of concurrent
      *  // Flowables two at a time
      *  return Completable.merge(Flowable.merge(workers, 2));
      * });
      * </pre>
-     * 
+     *
      * Slowing down the rate to no more than than 1 a second. This suffers from
      * the same problem as the one above I could find an {@link Flowable}
      * operator that limits the rate without dropping the values (aka leaky
      * bucket algorithm).
-     * 
+     *
      * <pre>
      * Scheduler slowScheduler = Schedulers.computation().when(workers -&gt; {
      *  // use concatenate to make each worker happen one at a time.
@@ -238,7 +239,7 @@ public Disposable schedulePeriodicallyDirect(@NonNull Runnable run, long initial
      *  }));
      * });
      * </pre>
-     * 
+     *
      * <p>History: 2.0.1 - experimental
      * @param <S> a Scheduler and a Subscription
      * @param combine the function that takes a two-level nested Flowable sequence of a Completable and returns
@@ -347,7 +348,7 @@ public long now(@NonNull TimeUnit unit) {
          * Holds state and logic to calculate when the next delayed invocation
          * of this task has to happen (accounting for clock drifts).
          */
-        final class PeriodicTask implements Runnable {
+        final class PeriodicTask implements Runnable, SchedulerRunnableIntrospection {
             @NonNull
             final Runnable decoratedRun;
             @NonNull
@@ -393,11 +394,16 @@ public void run() {
                     sd.replace(schedule(this, delay, TimeUnit.NANOSECONDS));
                 }
             }
+
+            @Override
+            public Runnable getWrappedRunnable() {
+                return this.decoratedRun;
+            }
         }
     }
 
     static class PeriodicDirectTask
-    implements Runnable, Disposable {
+    implements Disposable, Runnable, SchedulerRunnableIntrospection {
         final Runnable run;
         @NonNull
         final Worker worker;
@@ -432,9 +438,14 @@ public void dispose() {
         public boolean isDisposed() {
             return disposed;
         }
+
+        @Override
+        public Runnable getWrappedRunnable() {
+            return run;
+        }
     }
 
-    static final class DisposeTask implements Runnable, Disposable {
+    static final class DisposeTask implements Disposable, Runnable, SchedulerRunnableIntrospection {
         final Runnable decoratedRun;
         final Worker w;
 
@@ -469,5 +480,10 @@ public void dispose() {
         public boolean isDisposed() {
             return w.isDisposed();
         }
+
+        @Override
+        public Runnable getWrappedRunnable() {
+            return this.decoratedRun;
+        }
     }
 }

src/main/java/io/reactivex/schedulers/SchedulerRunnableIntrospection.java

@@ -0,0 +1,35 @@
+/**
+ * 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.schedulers;
+
+import io.reactivex.Scheduler;
+import io.reactivex.annotations.*;
+import io.reactivex.plugins.RxJavaPlugins;
+
+/**
+ * Marker interface to indicate wrapped action inside internal scheduler's task.
+ *
+ * Inside of the {@link RxJavaPlugins#onSchedule(Runnable)}, you can unwrap runnable from internal wrappers like
+ * {@link Scheduler.DisposeTask}.
+ *
+ * @since 2.1.7 - experimental
+ */
+@Experimental
+public interface SchedulerRunnableIntrospection {
+
+    /**
+     * Returns the wrapped action.
+     *
+     * @return the wrapped action, may be null.
+     */
+    @Nullable
+    Runnable getWrappedRunnable();
+}

src/test/java/io/reactivex/schedulers/SchedulerTest.java

@@ -307,4 +307,66 @@ public void customScheduleDirectDisposed() {
 
         assertTrue(d.isDisposed());
     }
+
+    @Test
+    public void unwrapDefaultPeriodicTask() {
+        TestScheduler scheduler = new TestScheduler();
+
+        Runnable runnable = new Runnable() {
+            @Override
+            public void run() {
+            }
+        };
+        SchedulerRunnableIntrospection wrapper = (SchedulerRunnableIntrospection) scheduler.schedulePeriodicallyDirect(runnable, 100, 100, TimeUnit.MILLISECONDS);
+
+        assertEquals(runnable, wrapper.getWrappedRunnable());
+    }
+
+    @Test
+    public void unwrapScheduleDirectTask() {
+        TestScheduler scheduler = new TestScheduler();
+
+        Runnable runnable = new Runnable() {
+            @Override
+            public void run() {
+            }
+        };
+        SchedulerRunnableIntrospection wrapper = (SchedulerRunnableIntrospection) scheduler.scheduleDirect(runnable, 100, TimeUnit.MILLISECONDS);
+        assertEquals(runnable, wrapper.getWrappedRunnable());
+    }
+
+    @Test
+    public void unwrapWorkerPeriodicTask() {
+        final Runnable runnable = new Runnable() {
+            @Override
+            public void run() {
+            }
+        };
+
+        Scheduler scheduler = new Scheduler() {
+            @Override
+            public Worker createWorker() {
+                return new Worker() {
+                    @Override
+                    public Disposable schedule(Runnable run, long delay, TimeUnit unit) {
+                        SchedulerRunnableIntrospection outerWrapper = (SchedulerRunnableIntrospection) run;
+                        SchedulerRunnableIntrospection innerWrapper = (SchedulerRunnableIntrospection) outerWrapper.getWrappedRunnable();
+                        assertEquals(runnable, innerWrapper.getWrappedRunnable());
+                        return (Disposable) innerWrapper;
+                    }
+
+                    @Override
+                    public void dispose() {
+                    }
+
+                    @Override
+                    public boolean isDisposed() {
+                        return false;
+                    }
+                };
+            }
+        };
+
+        scheduler.schedulePeriodicallyDirect(runnable, 100, 100, TimeUnit.MILLISECONDS);
+    }
 }