Scala Bindings Refactor

Author: AppliedDuality

language-adaptors/rxjava-scala/src/examples/java/rx/lang/scala/examples/MovieLibUsage.java

@@ -15,25 +15,27 @@
  */
 package rx.lang.scala.examples;
 
-import org.junit.Test;
-
 import rx.Observable;
-import rx.util.functions.Action1;
+import rx.lang.scala.examples.Movie;
+import rx.lang.scala.examples.MovieLib;
 
+import static rx.lang.scala.ImplicitFunctionConversions.toScalaObservable;
 
 public class MovieLibUsage {
-	
-	Action1<Movie> moviePrinter = new Action1<Movie>() {
-		public void call(Movie m) {
-			System.out.println("A movie of length " + m.lengthInSeconds() + "s");
-		}
-	};
-	
-	@Test
-	public void test() {
-		MovieLib lib = new MovieLib(Observable.from(new Movie(3000), new Movie(1000), new Movie(2000)));
-		
-		lib.longMovies().subscribe(moviePrinter);		
-	}
+
+    public static void main(String[] args) {
+
+        Observable<Movie> movies = Observable.from(
+                new Movie(3000),
+                new Movie(1000),
+                new Movie(2000)
+        );
+
+        MovieLib lib = new MovieLib(toScalaObservable(movies));
+
+        lib.longMovies().asJavaObservable().subscribe(m ->
+                System.out.println("A movie of length " + m.lengthInSeconds() + "s")
+        );
+    }
 
 }

language-adaptors/rxjava-scala/src/examples/scala/rx/lang/scala/examples/MovieLib.scala renamed to language-adaptors/rxjava-scala/src/examples/scala/MovieLib.scala

@@ -1,12 +1,12 @@
 /**
  * Copyright 2013 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.
@@ -20,11 +20,11 @@ import rx.lang.scala.Observable
 class Movie(val lengthInSeconds: Int) { }
 
 class MovieLib(val moviesStream: Observable[Movie]) {
-  
+
   val threshold = 1200
-  
+
   def shortMovies: Observable[Movie] = moviesStream.filter(_.lengthInSeconds <= threshold)
-  
+
   def longMovies: Observable[Movie] = moviesStream.filter(_.lengthInSeconds > threshold)
 
 }

language-adaptors/rxjava-scala/src/examples/scala/rx/lang/scala/examples/Olympics.scala renamed to language-adaptors/rxjava-scala/src/examples/scala/Olympics.scala

@@ -20,7 +20,7 @@ import scala.concurrent.duration._
 
 object Olympics {
   case class Medal(val year: Int, val games: String, val discipline: String, val medal: String, val athlete: String, val country: String)
-  
+
   def mountainBikeMedals: Observable[Medal] = Observable(
     Observable(
       Medal(1996, "Atlanta 1996", "cross-country men", "Gold", "Bart BRENTJENS", "Netherlands"),
@@ -31,7 +31,7 @@ object Olympics {
       Medal(1996, "Atlanta 1996", "cross-country women", "Bronze", "Susan DEMATTEI", "United States of America")
     ),
     fourYearsEmpty,
-    Observable( 
+    Observable(
       Medal(2000, "Sydney 2000", "cross-country women", "Gold", "Paola PEZZO", "Italy"),
       Medal(2000, "Sydney 2000", "cross-country women", "Silver", "Barbara BLATTER", "Switzerland"),
       Medal(2000, "Sydney 2000", "cross-country women", "Bronze", "Marga FULLANA", "Spain"),
@@ -40,7 +40,7 @@ object Olympics {
       Medal(2000, "Sydney 2000", "cross-country men", "Bronze", "Christoph SAUSER", "Switzerland")
     ),
     fourYearsEmpty,
-    Observable( 
+    Observable(
       Medal(2004, "Athens 2004", "cross-country men", "Gold", "Julien ABSALON", "France"),
       Medal(2004, "Athens 2004", "cross-country men", "Silver", "Jose Antonio HERMIDA RAMOS", "Spain"),
       Medal(2004, "Athens 2004", "cross-country men", "Bronze", "Bart BRENTJENS", "Netherlands"),
@@ -49,7 +49,7 @@ object Olympics {
       Medal(2004, "Athens 2004", "cross-country women", "Bronze", "Sabine SPITZ", "Germany")
     ),
     fourYearsEmpty,
-    Observable( 
+    Observable(
       Medal(2008, "Beijing 2008", "cross-country women", "Gold", "Sabine SPITZ", "Germany"),
       Medal(2008, "Beijing 2008", "cross-country women", "Silver", "Maja WLOSZCZOWSKA", "Poland"),
       Medal(2008, "Beijing 2008", "cross-country women", "Bronze", "Irina KALENTYEVA", "Russian Federation"),
@@ -67,12 +67,12 @@ object Olympics {
       Medal(2012, "London 2012", "cross-country women", "Bronze", "Georgia GOULD", "United States of America")
     )
   ).concat
-    
+
   // speed it up :D
   val fourYears = 4000.millis
-  
+
   val neverUsedDummyMedal = Medal(3333, "?", "?", "?", "?", "?")
-  
+
   def fourYearsEmpty: Observable[Medal] = {
     // TODO this should return an observable which emits nothing during fourYears and then completes
     // Because of https://github.com/Netflix/RxJava/issues/388, we get non-terminating tests
@@ -82,5 +82,5 @@ object Olympics {
     // But we just return empty, which completes immediately
     Observable()
   }
-  
+
 }

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

@@ -15,8 +15,12 @@
  */
 package rx.lang.scala
 
+import java.lang.Exception
 import java.{ lang => jlang }
+import rx.lang.scala._
 import rx.util.functions._
+import scala.collection.Seq
+import rx.lang.scala.subscriptions.Subscription
 
 /**
  * These function conversions convert between Scala functions and Rx `Func`s and `Action`s.
@@ -28,20 +32,28 @@ object ImplicitFunctionConversions {
   import language.implicitConversions
 
   implicit def schedulerActionToFunc2[T](action: (Scheduler, T) => Subscription) =
-    new Func2[rx.Scheduler, T, Subscription] {
-      def call(s: rx.Scheduler, t: T): Subscription = {
-        action(s, t)
+    new Func2[rx.Scheduler, T, rx.Subscription] {
+      def call(s: rx.Scheduler, t: T): rx.Subscription = {
+        action(s, t).asJavaSubscription
       }
-    }  
-  
-  implicit def scalaSchedulerToJavaScheduler(s: Scheduler): rx.Scheduler = s.asJava
-  
+    }
+
+  implicit def toJavaSubscription(s: Subscription): rx.Subscription = s.asJavaSubscription
+  implicit def toScalaSubscription(s: rx.Subscription): Subscription = Subscription(s)
+
+  implicit def scalaSchedulerToJavaScheduler(s: Scheduler): rx.Scheduler = s.asJavaScheduler
   implicit def javaSchedulerToScalaScheduler(s: rx.Scheduler): Scheduler = Scheduler(s)
+
+  implicit def toJavaObserver[T](s: Observer[T]): rx.Observer[_ >: T] = s.asJavaObserver
+  implicit def toScalaObserver[T](s: rx.Observer[T]): Observer[T] = Observer(s)
+
+  implicit def toJavaObservable[T](s: Observable[T]): rx.Observable[_ <: T] = s.asJavaObservable
+  implicit def toScalaObservable[T](s: rx.Observable[T]): Observable[T] = Observable(s)
 
   implicit def scalaFunction1ToOnSubscribeFunc[T](f: rx.lang.scala.Observer[T] => Subscription) =
     new rx.Observable.OnSubscribeFunc[T] {
       def onSubscribe(obs: rx.Observer[_ >: T]): rx.Subscription = {
-        f(obs)
+        f(Observer(obs))
       }
     }
 
@@ -55,6 +67,10 @@ object ImplicitFunctionConversions {
       def call(): Unit = f()
     }
 
+  implicit def Action1toScalaFunction1ProducingUnit[A](f: Action1[A]): (A=>Unit) = {
+    a => f(a)
+  }
+
   implicit def scalaFunction1ProducingUnitToAction1[A](f: (A => Unit)): Action1[A] =
     new Action1[A] {
       def call(a: A): Unit = f(a)

language-adaptors/rxjava-scala/src/main/scala/rx/lang/scala/Notification.scala renamed to language-adaptors/rxjava-scala/src/main/scala/Notification.scala

@@ -1,22 +1,7 @@
-/**
- * Copyright 2013 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.lang.scala
 
 /**
- * Emitted by Observables returned by [[Observable.materialize]].
+ * Emitted by Observables returned by [[rx.lang.scala.Observable.materialize]].
  */
 sealed trait Notification[+T] {
   def asJava: rx.Notification[_ <: T]
@@ -58,7 +43,7 @@ object Notification {
   }
 
   class OnError[+T](val asJava: rx.Notification[_ <: T]) extends Notification[T] {
-    def error: Throwable = asJava.getThrowable()
+    def error: Throwable = asJava.getThrowable
   }
 
   object OnError {