Fixed type errors in Schedulers.

Added scheduler factory.

Author: AppliedDuality

language-adaptors/rxjava-scala/src/main/scala/rx/lang/scala/ImplicitFunctionConversions.scala

@@ -17,12 +17,12 @@ package rx.lang.scala
 
 import java.lang.Exception
 import java.{ lang => jlang }
-import rx.lang.scala._
+//import rx.lang.scala._
 import rx.util.functions._
 import scala.collection.Seq
 import java.{lang => jlang}
 import scala.language.implicitConversions
-import rx.lang.scala.concurrency.Scheduler
+//import rx.lang.scala.Scheduler
 
 /**
  * These function conversions convert between Scala functions and Rx `Func`s and `Action`s.
@@ -33,10 +33,10 @@ import rx.lang.scala.concurrency.Scheduler
 object ImplicitFunctionConversions {
   import language.implicitConversions
 
-  implicit def schedulerActionToFunc2[T](action: (Scheduler, T) => Subscription) =
+  implicit def schedulerActionToFunc2[T](action: (Scheduler, T) => Subscription): Func2[rx.Scheduler, T, rx.Subscription] with Object {def call(s: rx.Scheduler, t: T): rx.Subscription} =
     new Func2[rx.Scheduler, T, rx.Subscription] {
       def call(s: rx.Scheduler, t: T): rx.Subscription = {
-        action(s, t).asJavaSubscription
+        action(rx.lang.scala.Scheduler(s), t).asJavaSubscription
       }
     }
 

language-adaptors/rxjava-scala/src/main/scala/rx/lang/scala/Scheduler.scala

@@ -23,7 +23,7 @@ import rx.util.functions.{Action0, Action1, Func2}
  * Represents an object that schedules units of work.
  */
 trait Scheduler {
-  def asJavaScheduler: rx.Scheduler
+  val asJavaScheduler: rx.Scheduler
 
   /**
    * Schedules a cancelable action to be executed.
@@ -210,3 +210,12 @@ trait Scheduler {
 
 }
 
+object Scheduler {
+  private [scala] def apply(scheduler: rx.Scheduler): Scheduler = {
+    return new Scheduler() {
+       val asJavaScheduler = scheduler
+    }
+  }
+}
+
+

language-adaptors/rxjava-scala/src/main/scala/rx/lang/scala/Subscription.scala

@@ -41,7 +41,7 @@ object Subscription {
 
   import java.util.concurrent.atomic.AtomicBoolean
   import rx.lang.scala.subscriptions._
-  import rx.lang.scala.Subscription
+
 
   /**
    * Creates an [[rx.lang.scala.Subscription]] from an [[rx.Subscription]].
@@ -52,7 +52,7 @@ object Subscription {
       case x: rx.subscriptions.CompositeSubscription => new CompositeSubscription(x)
       case x: rx.subscriptions.MultipleAssignmentSubscription => new MultipleAssignmentSubscription(x)
       case x: rx.subscriptions.SerialSubscription => new SerialSubscription(x)
-      case x: rx.Subscription => Subscription { x.unsubscribe() } // add isUnsubscribed
+      case x: rx.Subscription => Subscription { x.unsubscribe() }
     }
   }
 

language-adaptors/rxjava-scala/src/main/scala/rx/lang/scala/concurrency/CurrentThreadScheduler.scala

@@ -23,9 +23,9 @@ object CurrentThreadScheduler {
    * Returns a [[rx.lang.scala.Scheduler]] that queues work on the current thread to be executed after the current work completes.
    */
   def apply(): CurrentThreadScheduler =  {
-    new ImmediateScheduler(rx.concurrency.Schedulers.currentThread())
+    new CurrentThreadScheduler(rx.concurrency.Schedulers.currentThread())
   }
 }
 
-class CurrentThreadScheduler private[scala] (val asJavaScheduler: rx.concurrency.CurrentThreadScheduler)
+class CurrentThreadScheduler private[scala] (val asJavaScheduler: rx.Scheduler)
   extends Scheduler {}

language-adaptors/rxjava-scala/src/main/scala/rx/lang/scala/concurrency/ExecutorScheduler.scala

@@ -31,7 +31,7 @@ object ExecutorScheduler {
 }
 
 
-class ExecutorScheduler private[scala] (val asJavaScheduler: rx.concurrency.ExecutorScheduler)
+class ExecutorScheduler private[scala] (val asJavaScheduler: rx.Scheduler)
   extends Scheduler {}
 
 

language-adaptors/rxjava-scala/src/main/scala/rx/lang/scala/concurrency/ImmediateScheduler.scala

@@ -27,7 +27,7 @@ object ImmediateScheduler {
   }
 }
 
-class ImmediateScheduler private[scala] (val asJavaScheduler: rx.concurrency.ImmediateScheduler)
+class ImmediateScheduler private[scala] (val asJavaScheduler: rx.Scheduler)
   extends Scheduler {}
 
 

language-adaptors/rxjava-scala/src/main/scala/rx/lang/scala/concurrency/NewThreadScheduler.scala

@@ -26,5 +26,6 @@ object NewThreadScheduler {
     new NewThreadScheduler(rx.concurrency.Schedulers.newThread())
   }
 }
-class NewThreadScheduler private[scala] (val asJavaScheduler: rx.concurrency.NewThreadScheduler)
+
+class NewThreadScheduler private[scala] (val asJavaScheduler: rx.Scheduler)
   extends Scheduler {}

language-adaptors/rxjava-scala/src/main/scala/rx/lang/scala/concurrency/ScheduledExecutorServiceScheduler.scala

@@ -23,7 +23,7 @@ object ScheduledExecutorServiceScheduler {
   /**
    * Returns a [[rx.lang.scala.Scheduler]] that queues work on an `java.util.concurrent.ScheduledExecutorService`.
    */
-  def apply(executor: ScheduledExecutorService): ExecutorScheduler =  {
+  def apply(executor: ScheduledExecutorService): ScheduledExecutorServiceScheduler =  {
     new ScheduledExecutorServiceScheduler(rx.concurrency.Schedulers.executor(executor))
   }
 }

language-adaptors/rxjava-scala/src/main/scala/rx/lang/scala/concurrency/ThreadPoolForComputationScheduler.scala

@@ -28,7 +28,7 @@ object ThreadPoolForComputationScheduler {
    *
    * Do not perform IO-bound work on this scheduler. Use [[rx.lang.scala.concurrency.Schedulers.threadPoolForIO]] instead.
    */
-  def apply(): ExecutorScheduler =  {
+  def apply(): ThreadPoolForComputationScheduler =  {
     new ThreadPoolForComputationScheduler(rx.concurrency.Schedulers.threadPoolForComputation())
   }
 }

language-adaptors/rxjava-scala/src/main/scala/rx/lang/scala/concurrency/ThreadPoolForIOScheduler.scala

@@ -28,7 +28,7 @@ object ThreadPoolForIOScheduler {
    *
    * Do not perform computational work on this scheduler. Use [[rx.lang.scala.concurrency.Schedulers.threadPoolForComputation]] instead.
    */
-  def apply(): ExecutorScheduler =  {
+  def apply(): ThreadPoolForIOScheduler =  {
     new ThreadPoolForIOScheduler(rx.concurrency.Schedulers.threadPoolForIO())
   }
 }