Merge pull request #1559 from mattrjacobs/scheduler-plugin

More consistent hooks for scheduler plugins.  This is a rework of #1514.

Author: mattrjacobs

rxjava-core/src/main/java/rx/internal/schedulers/NewThreadWorker.java

@@ -18,6 +18,8 @@
 import rx.Scheduler;
 import rx.Subscription;
 import rx.functions.Action0;
+import rx.plugins.RxJavaPlugins;
+import rx.plugins.RxJavaSchedulersHook;
 import rx.subscriptions.Subscriptions;
 
 import java.util.concurrent.*;
@@ -27,11 +29,13 @@
  */
 public class NewThreadWorker extends Scheduler.Worker implements Subscription {
     private final ScheduledExecutorService executor;
+    private final RxJavaSchedulersHook schedulersHook;
     volatile boolean isUnsubscribed;
 
     /* package */
     public NewThreadWorker(ThreadFactory threadFactory) {
         executor = Executors.newScheduledThreadPool(1, threadFactory);
+        schedulersHook = RxJavaPlugins.getInstance().getSchedulersHook();
     }
 
     @Override
@@ -55,7 +59,8 @@ public Subscription schedule(final Action0 action, long delayTime, TimeUnit unit
      * @return
      */
     public ScheduledAction scheduleActual(final Action0 action, long delayTime, TimeUnit unit) {
-        ScheduledAction run = new ScheduledAction(action);
+        Action0 decoratedAction = schedulersHook.onSchedule(action);
+        ScheduledAction run = new ScheduledAction(decoratedAction);
         Future<?> f;
         if (delayTime <= 0) {
             f = executor.submit(run);

rxjava-core/src/main/java/rx/plugins/RxJavaDefaultSchedulers.java

@@ -23,6 +23,7 @@
  * See {@link RxJavaPlugins} or the RxJava GitHub Wiki for information on configuring plugins:
  * <a href="https://github.com/Netflix/RxJava/wiki/Plugins">https://github.com/Netflix/RxJava/wiki/Plugins</a>.
  */
+@Deprecated
 public abstract class RxJavaDefaultSchedulers {
 
     /**

rxjava-core/src/main/java/rx/plugins/RxJavaDefaultSchedulersDefault.java

@@ -22,6 +22,7 @@
  * 
  * @ExcludeFromJavadoc
  */
+@Deprecated
 public class RxJavaDefaultSchedulersDefault extends RxJavaDefaultSchedulers {
 
     private static RxJavaDefaultSchedulersDefault INSTANCE = new RxJavaDefaultSchedulersDefault();

rxjava-core/src/main/java/rx/plugins/RxJavaPlugins.java

@@ -34,7 +34,9 @@ public class RxJavaPlugins {
 
     private final AtomicReference<RxJavaErrorHandler> errorHandler = new AtomicReference<RxJavaErrorHandler>();
     private final AtomicReference<RxJavaObservableExecutionHook> observableExecutionHook = new AtomicReference<RxJavaObservableExecutionHook>();
+    //deprecated
     private final AtomicReference<RxJavaDefaultSchedulers> schedulerOverrides = new AtomicReference<RxJavaDefaultSchedulers>();
+    private final AtomicReference<RxJavaSchedulersHook> schedulersHook = new AtomicReference<RxJavaSchedulersHook>();
 
     /**
      * Retrieves the single {@code RxJavaPlugins} instance.
@@ -52,6 +54,7 @@ public static RxJavaPlugins getInstance() {
     /* package accessible for unit tests */void reset() {
         INSTANCE.errorHandler.set(null);
         INSTANCE.observableExecutionHook.set(null);
+        INSTANCE.schedulersHook.set(null);
     }
 
     /**
@@ -207,4 +210,46 @@ public void registerDefaultSchedulers(RxJavaDefaultSchedulers impl) {
             throw new IllegalStateException("Another strategy was already registered: " + schedulerOverrides.get());
         }
     }
+
+    /**
+     * Retrieves the instance of {@link RxJavaSchedulersHook} to use based on order of precedence as defined
+     * in the {@link RxJavaPlugins} class header.
+     * <p>
+     * Override the default by calling {@link #registerSchedulersHook(RxJavaSchedulersHook)} or by setting
+     * the property {@code rxjava.plugin.RxJavaSchedulersHook.implementation} with the full classname to
+     * load.
+     *
+     * @return the {@link RxJavaSchedulersHook} implementation in use
+     */
+    public RxJavaSchedulersHook getSchedulersHook() {
+        if (schedulersHook.get() == null) {
+            // check for an implementation from System.getProperty first
+            Object impl = getPluginImplementationViaProperty(RxJavaSchedulersHook.class);
+            if (impl == null) {
+                // nothing set via properties so initialize with default
+                schedulersHook.compareAndSet(null, RxJavaSchedulersHook.getDefaultInstance());
+                // we don't return from here but call get() again in case of thread-race so the winner will always get returned
+            } else {
+                // we received an implementation from the system property so use it
+                schedulersHook.compareAndSet(null, (RxJavaSchedulersHook) impl);
+            }
+        }
+        return schedulersHook.get();
+    }
+
+    /**
+     * Registers an {@link RxJavaSchedulersHook} implementation as a global override of any injected or
+     * default implementations.
+     *
+     * @param impl
+     *            {@link RxJavaSchedulersHook} implementation
+     * @throws IllegalStateException
+     *             if called more than once or after the default was initialized (if usage occurs before trying
+     *             to register)
+     */
+    public void registerSchedulersHook(RxJavaSchedulersHook impl) {
+        if (!schedulersHook.compareAndSet(null, impl)) {
+            throw new IllegalStateException("Another strategy was already registered: " + schedulersHook.get());
+        }
+    }
 }

rxjava-core/src/main/java/rx/plugins/RxJavaSchedulersHook.java

@@ -0,0 +1,85 @@
+/**
+ * Copyright 2014 Netflix, Inc.
+ *
+ * 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 rx.plugins;
+
+import rx.Scheduler;
+import rx.functions.Action0;
+
+/**
+ * This plugin class provides 2 ways to customize {@link Scheduler} functionality
+ * 1.  You may redefine entire schedulers, if you so choose.  To do so, override
+ * the 3 methods that return Scheduler (io(), computation(), newThread()).
+ * 2.  You may wrap/decorate an {@link Action0}, before it is handed off to a Scheduler.  The system-
+ * supplied Schedulers (Schedulers.ioScheduler, Schedulers.computationScheduler,
+ * Scheduler.newThreadScheduler) all use this hook, so it's a convenient way to
+ * modify Scheduler functionality without redefining Schedulers wholesale.
+ *
+ * Also, when redefining Schedulers, you are free to use/not use the onSchedule decoration hook.
+ * <p>
+ * See {@link RxJavaPlugins} or the RxJava GitHub Wiki for information on configuring plugins:
+ * <a href="https://github.com/Netflix/RxJava/wiki/Plugins">https://github.com/Netflix/RxJava/wiki/Plugins</a>.
+ */
+public class RxJavaSchedulersHook {
+
+    protected RxJavaSchedulersHook() {
+
+    }
+
+    private final static RxJavaSchedulersHook DEFAULT_INSTANCE = new RxJavaSchedulersHook();
+
+    /**
+     * Scheduler to return from {@link rx.schedulers.Schedulers#computation()} or null if default should be
+     * used.
+     *
+     * This instance should be or behave like a stateless singleton;
+     */
+    public Scheduler getComputationScheduler() {
+        return null;
+    }
+
+    /**
+     * Scheduler to return from {@link rx.schedulers.Schedulers#io()} or null if default should be used.
+     *
+     * This instance should be or behave like a stateless singleton;
+     */
+    public Scheduler getIOScheduler() {
+        return null;
+    }
+
+    /**
+     * Scheduler to return from {@link rx.schedulers.Schedulers#newThread()} or null if default should be used.
+     *
+     * This instance should be or behave like a stateless singleton;
+     */
+    public Scheduler getNewThreadScheduler() {
+        return null;
+    }
+
+    /**
+     * Invoked before the Action is handed over to the scheduler.  Can be used for wrapping/decorating/logging.
+     * The default is just a passthrough.
+     * @param action action to schedule
+     * @return wrapped action to schedule
+     */
+    public Action0 onSchedule(Action0 action) {
+        return action;
+    }
+
+    public static RxJavaSchedulersHook getDefaultInstance() {
+        return DEFAULT_INSTANCE;
+    }
+}