Move RxThreadFactory out from NewThreadScheduler

Author: jbripley

rxjava-core/src/main/java/rx/internal/util/RxThreadFactory.java

@@ -0,0 +1,37 @@
+/**
+ * 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.util;
+
+import java.util.concurrent.ThreadFactory;
+import java.util.concurrent.atomic.AtomicLongFieldUpdater;
+
+public final class RxThreadFactory implements ThreadFactory {
+    final String prefix;
+    volatile long counter;
+    static final AtomicLongFieldUpdater<RxThreadFactory> COUNTER_UPDATER
+            = AtomicLongFieldUpdater.newUpdater(RxThreadFactory.class, "counter");
+
+    public RxThreadFactory(String prefix) {
+        this.prefix = prefix;
+    }
+
+    @Override
+    public Thread newThread(Runnable r) {
+        Thread t = new Thread(r, prefix + COUNTER_UPDATER.incrementAndGet(this));
+        t.setDaemon(true);
+        return t;
+    }
+}

rxjava-core/src/main/java/rx/schedulers/CachedThreadScheduler.java

@@ -18,6 +18,7 @@
 import rx.Scheduler;
 import rx.Subscription;
 import rx.functions.Action0;
+import rx.internal.util.RxThreadFactory;
 import rx.subscriptions.CompositeSubscription;
 import rx.subscriptions.Subscriptions;
 
@@ -27,12 +28,12 @@
 
 /* package */final class CachedThreadScheduler extends Scheduler {
     private static final String WORKER_THREAD_NAME_PREFIX = "RxCachedThreadScheduler-";
-    private static final NewThreadScheduler.RxThreadFactory WORKER_THREAD_FACTORY =
-            new NewThreadScheduler.RxThreadFactory(WORKER_THREAD_NAME_PREFIX);
+    private static final RxThreadFactory WORKER_THREAD_FACTORY =
+            new RxThreadFactory(WORKER_THREAD_NAME_PREFIX);
 
     private static final String EVICTOR_THREAD_NAME_PREFIX = "RxCachedWorkerPoolEvictor-";
-    private static final NewThreadScheduler.RxThreadFactory EVICTOR_THREAD_FACTORY =
-            new NewThreadScheduler.RxThreadFactory(EVICTOR_THREAD_NAME_PREFIX);
+    private static final RxThreadFactory EVICTOR_THREAD_FACTORY =
+            new RxThreadFactory(EVICTOR_THREAD_NAME_PREFIX);
 
     private static final class CachedWorkerPool {
         private final long keepAliveTime;

rxjava-core/src/main/java/rx/schedulers/EventLoopsScheduler.java

@@ -15,17 +15,17 @@
  */
 package rx.schedulers;
 
-import java.util.concurrent.ThreadFactory;
-import java.util.concurrent.TimeUnit;
-
 import rx.Scheduler;
 import rx.Subscription;
 import rx.functions.Action0;
+import rx.internal.util.RxThreadFactory;
 import rx.schedulers.NewThreadScheduler.NewThreadWorker.ScheduledAction;
-import rx.schedulers.NewThreadScheduler.RxThreadFactory;
 import rx.subscriptions.CompositeSubscription;
 import rx.subscriptions.Subscriptions;
 
+import java.util.concurrent.ThreadFactory;
+import java.util.concurrent.TimeUnit;
+
 /* package */class EventLoopsScheduler extends Scheduler {
     /** Manages a fixed number of workers. */
     private static final String THREAD_NAME_PREFIX = "RxComputationThreadPool-";

rxjava-core/src/main/java/rx/schedulers/GenericScheduledExecutorService.java

@@ -15,13 +15,14 @@
  */
 package rx.schedulers;
 
+import rx.Scheduler;
+import rx.internal.util.RxThreadFactory;
+
 import java.util.concurrent.Executor;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
 import java.util.concurrent.ScheduledExecutorService;
 
-import rx.Scheduler;
-
 /**
  * A default {@link ScheduledExecutorService} that can be used for scheduling actions when a {@link Scheduler} implementation doesn't have that ability.
  * <p>
@@ -34,7 +35,7 @@
 /* package */final class GenericScheduledExecutorService {
 
     private static final String THREAD_NAME_PREFIX = "RxScheduledExecutorPool-";
-    private static final NewThreadScheduler.RxThreadFactory THREAD_FACTORY = new NewThreadScheduler.RxThreadFactory(THREAD_NAME_PREFIX);
+    private static final RxThreadFactory THREAD_FACTORY = new RxThreadFactory(THREAD_NAME_PREFIX);
 
     private final static GenericScheduledExecutorService INSTANCE = new GenericScheduledExecutorService();
     private final ScheduledExecutorService executor;

rxjava-core/src/main/java/rx/schedulers/NewThreadScheduler.java

@@ -15,20 +15,16 @@
  */
 package rx.schedulers;
 
-import java.util.concurrent.Executors;
-import java.util.concurrent.Future;
-import java.util.concurrent.ScheduledExecutorService;
-import java.util.concurrent.ThreadFactory;
-import java.util.concurrent.TimeUnit;
-import java.util.concurrent.atomic.AtomicIntegerFieldUpdater;
-import java.util.concurrent.atomic.AtomicLongFieldUpdater;
-
 import rx.Scheduler;
 import rx.Subscription;
 import rx.functions.Action0;
+import rx.internal.util.RxThreadFactory;
 import rx.subscriptions.CompositeSubscription;
 import rx.subscriptions.Subscriptions;
 
+import java.util.concurrent.*;
+import java.util.concurrent.atomic.AtomicIntegerFieldUpdater;
+
 /**
  * Schedules work on a new thread.
  */
@@ -38,24 +34,6 @@ public class NewThreadScheduler extends Scheduler {
     private static final RxThreadFactory THREAD_FACTORY = new RxThreadFactory(THREAD_NAME_PREFIX);
     private static final NewThreadScheduler INSTANCE = new NewThreadScheduler();
 
-    static final class RxThreadFactory implements ThreadFactory {
-        final String prefix;
-        volatile long counter;
-        static final AtomicLongFieldUpdater<RxThreadFactory> COUNTER_UPDATER
-                = AtomicLongFieldUpdater.newUpdater(RxThreadFactory.class, "counter");
-
-        public RxThreadFactory(String prefix) {
-            this.prefix = prefix;
-        }
-        
-        @Override
-        public Thread newThread(Runnable r) {
-            Thread t = new Thread(r, prefix + COUNTER_UPDATER.incrementAndGet(this));
-            t.setDaemon(true);
-            return t;
-        }
-    }
-    
     /* package */static NewThreadScheduler instance() {
         return INSTANCE;
     }

rxjava-core/src/test/java/rx/schedulers/ExecutorSchedulerTest.java

@@ -15,14 +15,16 @@
  */
 package rx.schedulers;
 
-import java.util.concurrent.Executors;
 import rx.Scheduler;
+import rx.internal.util.RxThreadFactory;
+
+import java.util.concurrent.Executors;
 
 public class ExecutorSchedulerTest extends AbstractSchedulerConcurrencyTests {
 
     @Override
     protected Scheduler getScheduler() {
-        return Schedulers.from(Executors.newFixedThreadPool(2, new NewThreadScheduler.RxThreadFactory("TestCustomPool-")));
+        return Schedulers.from(Executors.newFixedThreadPool(2, new RxThreadFactory("TestCustomPool-")));
     }
 
 }