Provide factories for creating the default scheduler instances. (#3856)

Author: JakeWharton

src/main/java/rx/schedulers/CachedThreadScheduler.java renamed to src/main/java/rx/internal/schedulers/CachedThreadScheduler.java

@@ -13,18 +13,17 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-package rx.schedulers;
+package rx.internal.schedulers;
 
 import java.util.concurrent.*;
 import java.util.concurrent.atomic.*;
 
 import rx.*;
 import rx.functions.Action0;
-import rx.internal.schedulers.*;
 import rx.internal.util.RxThreadFactory;
 import rx.subscriptions.*;
 
-/* package */final class CachedThreadScheduler extends Scheduler implements SchedulerLifecycle {
+public final class CachedThreadScheduler extends Scheduler implements SchedulerLifecycle {
     private static final String WORKER_THREAD_NAME_PREFIX = "RxCachedThreadScheduler-";
     static final RxThreadFactory WORKER_THREAD_FACTORY =
             new RxThreadFactory(WORKER_THREAD_NAME_PREFIX);
@@ -234,4 +233,4 @@ public void setExpirationTime(long expirationTime) {
             this.expirationTime = expirationTime;
         }
     }
-}
+}

src/main/java/rx/internal/schedulers/NewThreadScheduler.java

@@ -0,0 +1,36 @@
+/**
+ * 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.internal.schedulers;
+
+import rx.Scheduler;
+import rx.internal.util.RxThreadFactory;
+
+/**
+ * Schedules work on a new thread.
+ */
+public final class NewThreadScheduler extends Scheduler {
+
+    private static final String THREAD_NAME_PREFIX = "RxNewThreadScheduler-";
+    private static final RxThreadFactory THREAD_FACTORY = new RxThreadFactory(THREAD_NAME_PREFIX);
+
+    public NewThreadScheduler() {
+    }
+
+    @Override
+    public Worker createWorker() {
+        return new NewThreadWorker(THREAD_FACTORY);
+    }
+}

src/main/java/rx/plugins/RxJavaSchedulersHook.java

@@ -17,7 +17,12 @@
 package rx.plugins;
 
 import rx.Scheduler;
+import rx.annotations.Experimental;
 import rx.functions.Action0;
+import rx.internal.schedulers.CachedThreadScheduler;
+import rx.internal.schedulers.EventLoopsScheduler;
+import rx.internal.schedulers.NewThreadScheduler;
+import rx.schedulers.Schedulers;
 
 /**
  * This plugin class provides 2 ways to customize {@link Scheduler} functionality
@@ -35,6 +40,30 @@
  */
 public class RxJavaSchedulersHook {
 
+    /**
+     * Create an instance of the default {@link Scheduler} used for {@link Schedulers#computation()}.
+     */
+    @Experimental
+    public static Scheduler createComputationScheduler() {
+        return new EventLoopsScheduler();
+    }
+
+    /**
+     * Create an instance of the default {@link Scheduler} used for {@link Schedulers#io()}.
+     */
+    @Experimental
+    public static Scheduler createIoScheduler() {
+        return new CachedThreadScheduler();
+    }
+
+    /**
+     * Create an instance of the default {@link Scheduler} used for {@link Schedulers#newThread()}.
+     */
+    @Experimental
+    public static Scheduler createNewThreadScheduler() {
+        return new NewThreadScheduler();
+    }
+
     protected RxJavaSchedulersHook() {
 
     }

src/main/java/rx/schedulers/NewThreadScheduler.java

@@ -16,28 +16,18 @@
 package rx.schedulers;
 
 import rx.Scheduler;
-import rx.internal.schedulers.NewThreadWorker;
-import rx.internal.util.RxThreadFactory;
 
 /**
- * Schedules work on a new thread.
+ * @deprecated This type was never publicly instantiable. Use {@link Schedulers#newThread()}.
  */
+@Deprecated
 public final class NewThreadScheduler extends Scheduler {
-
-    private static final String THREAD_NAME_PREFIX = "RxNewThreadScheduler-";
-    private static final RxThreadFactory THREAD_FACTORY = new RxThreadFactory(THREAD_NAME_PREFIX);
-    private static final NewThreadScheduler INSTANCE = new NewThreadScheduler();
-
-    /* package */static NewThreadScheduler instance() {
-        return INSTANCE;
-    }
-
     private NewThreadScheduler() {
-
+        throw new AssertionError();
     }
 
     @Override
     public Worker createWorker() {
-        return new NewThreadWorker(THREAD_FACTORY);
+        return null;
     }
 }

src/main/java/rx/schedulers/Schedulers.java

@@ -19,6 +19,7 @@
 import rx.internal.schedulers.*;
 import rx.internal.util.RxRingBuffer;
 import rx.plugins.RxJavaPlugins;
+import rx.plugins.RxJavaSchedulersHook;
 
 import java.util.concurrent.Executor;
 
@@ -34,25 +35,27 @@ public final class Schedulers {
     private static final Schedulers INSTANCE = new Schedulers();
 
     private Schedulers() {
-        Scheduler c = RxJavaPlugins.getInstance().getSchedulersHook().getComputationScheduler();
+        RxJavaSchedulersHook hook = RxJavaPlugins.getInstance().getSchedulersHook();
+
+        Scheduler c = hook.getComputationScheduler();
         if (c != null) {
             computationScheduler = c;
         } else {
-            computationScheduler = new EventLoopsScheduler();
+            computationScheduler = RxJavaSchedulersHook.createComputationScheduler();
         }
 
-        Scheduler io = RxJavaPlugins.getInstance().getSchedulersHook().getIOScheduler();
+        Scheduler io = hook.getIOScheduler();
         if (io != null) {
             ioScheduler = io;
         } else {
-            ioScheduler = new CachedThreadScheduler();
+            ioScheduler = RxJavaSchedulersHook.createIoScheduler();
         }
 
-        Scheduler nt = RxJavaPlugins.getInstance().getSchedulersHook().getNewThreadScheduler();
+        Scheduler nt = hook.getNewThreadScheduler();
         if (nt != null) {
             newThreadScheduler = nt;
         } else {
-            newThreadScheduler = NewThreadScheduler.instance();
+            newThreadScheduler = RxJavaSchedulersHook.createNewThreadScheduler();
         }
     }