complete WithFilter

Author: samuelgruetter

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

@@ -1430,6 +1430,10 @@ class Observable[+T](val asJava: rx.Observable[_ <: T])
     new BlockingObservable[T](asJava.toBlockingObservable())
   }
 
+  def withFilter(p: T => Boolean): WithFilter[T] = {
+    new WithFilter[T](p, asJava)
+  }
+  
 }
 
 object Observable {
@@ -1796,18 +1800,25 @@ object Observable {
   }
 }
 
-// Cannot yet have inner class because of this error message:
-// implementation restriction: nested class is not allowed in value class.
-// This restriction is planned to be removed in subsequent releases.  
-class WithFilter[+T] private[scala] (p: T => Boolean, wrapped: rx.Observable[_ <: T]) {
+// Cannot yet have inner class because of this error message: 
+// "implementation restriction: nested class is not allowed in value class.
+// This restriction is planned to be removed in subsequent releases."  
+class WithFilter[+T] private[scala] (p: T => Boolean, asJava: rx.Observable[_ <: T]) {
   import rx.lang.scala.internal.ImplicitFunctionConversions._
 
-  def map[B](f: T => B): Observable[B] = new Observable[B](wrapped.filter(p).map(f))
+  def map[B](f: T => B): Observable[B] = {
+    Observable[B](asJava.filter(p).map[B](f))
+  }
+  
   def flatMap[B](f: T => Observable[B]): Observable[B] = {
-    ??? // TODO
+    Observable[B](asJava.filter(p).flatMap[B]((x: T) => f(x).asJava))
   }
-  def foreach(f: T => Unit): Unit = wrapped.filter(p).toBlockingObservable.forEach(f)
-  def withFilter(p: T => Boolean): Observable[T] = new Observable[T](wrapped.filter(p))
+  
+  def withFilter(q: T => Boolean): Observable[T] = {
+    Observable[T](asJava.filter((x: T) => p(x) && q(x)))
+  }
+  
+  // there is no foreach here, that's only available on BlockingObservable
 }
 
 class UnitTestSuite extends JUnitSuite {

language-adaptors/rxjava-scala/src/main/scala/rx/lang/scala/examples/RxScalaDemo.scala

@@ -119,6 +119,12 @@ class RxScalaDemo extends JUnitSuite {
     for (n <- numbers.toBlockingObservable) println(n+10)
   }
 
+  @Test def testForComprehension() {
+    val observables = Observable(Observable(1, 2, 3), Observable(10, 20, 30))
+    val squares = (for (o <- observables; i <- o if i % 2 == 0) yield i*i)
+    assertEquals(squares.toBlockingObservable.toList, List(4, 100, 400, 900))
+  }
+  
   def output(s: String): Unit = println(s)
 
   // blocks until obs has completed