Opening/Closing, Timestamped with unapply, BlockingObservable with WithFilter

Author: samuelgruetter

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

@@ -22,6 +22,7 @@ import scala.concurrent.duration._
 import org.junit.{Before, Test, Ignore}
 import org.junit.Assert._
 import rx.lang.scala.concurrency.NewThreadScheduler
+import rx.lang.scala.util.Timestamped
 
 @Ignore // Since this doesn't do automatic testing, don't increase build time unnecessarily
 class RxScalaDemo extends JUnitSuite {
@@ -375,6 +376,13 @@ class RxScalaDemo extends JUnitSuite {
     assertEquals(Seq(10, 9, 8, 7), Observable(10, 7, 8, 9).toSeq.map(_.sortWith(f)).toBlockingObservable.single)
   }
 
+  @Test def timestampExample() {
+    val timestamped = Observable.interval(100 millis).take(3).timestamp.toBlockingObservable
+    for (Timestamped(millis, value) <- timestamped if value > 0) {
+      println(value + " at t = " + millis)
+    }
+  }
+  
   def output(s: String): Unit = println(s)
 
   // blocks until obs has completed

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

@@ -199,7 +199,7 @@ class Observable[+T](val asJava: rx.Observable[_ <: T])
    * @return an Observable that emits timestamped items from the source Observable
    */
   def timestamp: Observable[Timestamped[T]] = {
-    Observable[Timestamped[T]](asJava.timestamp())
+    Observable[rx.util.Timestamped[_ <: T]](asJava.timestamp()).map(Timestamped(_))
   }
 
   /**

language-adaptors/rxjava-scala/src/main/scala/rx/lang/scala/observables/BlockingObservable.scala

@@ -18,19 +18,24 @@ package rx.lang.scala.observables
 import scala.collection.JavaConverters._
 import rx.lang.scala.ImplicitFunctionConversions._
 
+/**
+ * An Observable that provides blocking operators.
+ * 
+ * You can obtain a BlockingObservable from an Observable using [[Observable.toBlockingObservable]]
+ */
 class BlockingObservable[+T](val asJava: rx.observables.BlockingObservable[_ <: T]) 
   extends AnyVal 
 {
 
   /**
    * Invoke a method on each item emitted by the {@link Observable}; block until the Observable
    * completes.
-   * <p>
+   * 
    * NOTE: This will block even if the Observable is asynchronous.
-   * <p>
+   * 
    * This is similar to {@link Observable#subscribe(Observer)}, but it blocks. Because it blocks it does
    * not need the {@link Observer#onCompleted()} or {@link Observer#onError(Throwable)} methods.
-   * <p>
+   * 
    * <img width="640" src="https://github.com/Netflix/RxJava/wiki/images/rx-operators/B.forEach.png">
    *
    * @param onNext
@@ -41,6 +46,10 @@ class BlockingObservable[+T](val asJava: rx.observables.BlockingObservable[_ <:
   def foreach(f: T => Unit): Unit = {
     asJava.forEach(f);
   }
+  
+  def withFilter(p: T => Boolean): WithFilter[T] = {
+    new WithFilter[T](p, asJava)
+  }
 
   // last                 -> use toIterable.last
   // lastOrDefault        -> use toIterable.lastOption
@@ -118,3 +127,23 @@ class BlockingObservable[+T](val asJava: rx.observables.BlockingObservable[_ <:
   }
 
 }
+
+// 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[observables] (p: T => Boolean, asJava: rx.observables.BlockingObservable[_ <: T]) {
+  import rx.lang.scala.ImplicitFunctionConversions._
+  
+  // there's no map and flatMap here, they're only available on Observable
+  
+  def withFilter(q: T => Boolean) = new WithFilter[T]((x: T) => p(x) && q(x), asJava)
+  
+  def foreach(f: T => Unit): Unit = {
+    asJava.forEach((e: T) => {
+      if (p(e)) f(e)
+    })
+  }
+  
+}
+
+

language-adaptors/rxjava-scala/src/main/scala/rx/lang/scala/util/package.scala

@@ -16,33 +16,45 @@
 package rx.lang.scala
 
 package object util {
-  type Closing = rx.util.Closing
-
-  object Closings {
-    def create(): Closing = rx.util.Closings.create()
-  }
-
-  type CompositeException = rx.util.CompositeException
-
-  // TODO not sure if we need this in Scala
-  object Exceptions {
-    def propageate(ex: Throwable) = rx.util.Exceptions.propagate(ex)
-  }
-
-  // rx.util.OnErrorNotImplementedException TODO what's this?
 
+  /**
+   * Tagging interface for objects which can open buffers.
+   * @see [[Observable.buffer]]
+   */
   type Opening = rx.util.Opening
 
-  object Openings {
-    def create(): Opening = rx.util.Openings.create()
-  }
+  /**
+   * Creates an object which can open buffers.
+   * @see [[Observable.buffer]]
+   */
+  def Opening() = rx.util.Openings.create()
+  
+  /**
+   * Tagging interface for objects which can close buffers.
+   * @see [[Observable.buffer]]
+   */
+  type Closing = rx.util.Closing
 
+  /**
+   * Creates an object which can close buffers.
+   * @see [[Observable.buffer]]
+   */
+  def Closing() = rx.util.Closings.create()
+  
   // rx.util.Range not needed because there's a standard Scala Range
 
-  type Timestamped[+T] = rx.util.Timestamped[_ <: T]
+  implicit class Timestamped[+T](val asJava: rx.util.Timestamped[_ <: T]) {}
+  
   object Timestamped {
     def apply[T](timestampMillis: Long, value: T): Timestamped[T] = {
       new rx.util.Timestamped(timestampMillis, value)
     }
+    
+    def unapply[T](v: Timestamped[T]): Option[(Long, T)] = unapply(v.asJava)
+    
+    def unapply[T](v: rx.util.Timestamped[_ <: T]): Option[(Long, T)] = {
+      Some((v.getTimestampMillis, v.getValue))
+    }
   }
+  
 }