Merge pull request #1005 from suncelesta/scala-observable-update

add toMap from Java Observable

Author: benjchristensen

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

@@ -573,4 +573,28 @@ class RxScalaDemo extends JUnitSuite {
     val o : Observable[Seq[Char]] = List("red".toList, "green".toList, "blue".toList).toObservable.elementAtOrDefault(3, "black".toSeq)
     println(o.toBlockingObservable.single)
   }
+
+  @Test def toMapExample1(): Unit = {
+    val o : Observable[String] = List("alice", "bob", "carol").toObservable
+    val keySelector = (s: String) => s.head
+    val m = o.toMap(keySelector)
+    println(m.toBlockingObservable.single)
+  }
+
+  @Test def toMapExample2(): Unit = {
+    val o : Observable[String] = List("alice", "bob", "carol").toObservable
+    val keySelector = (s: String) => s.head
+    val valueSelector = (s: String) => s.tail
+    val m = o.toMap(keySelector, valueSelector)
+    println(m.toBlockingObservable.single)
+  }
+
+  @Test def toMapExample3(): Unit = {
+    val o : Observable[String] = List("alice", "bob", "carol").toObservable
+    val keySelector = (s: String) => s.head
+    val valueSelector = (s: String) => s.tail
+    val mapFactory = () => Map(('s',"tart"))
+    val m = o.toMap(keySelector, valueSelector, mapFactory)
+    println(m.toBlockingObservable.single)
+  }
 }

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

@@ -20,6 +20,8 @@ import rx.functions.FuncN
 import rx.Observable.OnSubscribeFunc
 import rx.lang.scala.observables.ConnectableObservable
 import scala.concurrent.duration
+import java.util
+import collection.JavaConversions._
 
 
 /**
@@ -2353,7 +2355,7 @@ trait Observable[+T]
    * 
    * @param index
    *            the zero-based index of the item to retrieve
-   * @param defaultValue
+   * @param default
    *            the default item
    * @return an Observable that emits the item at the specified position in the sequence emitted by the source
    *         Observable, or the default item if that index is outside the bounds of the source sequence
@@ -2364,6 +2366,69 @@ trait Observable[+T]
     val thisJava = asJavaObservable.asInstanceOf[rx.Observable[U]]
     toScalaObservable[U](thisJava.elementAtOrDefault(index, default))
   }
+
+  /**
+   * Return an Observable that emits a single Map containing all items emitted by the source Observable,
+   * mapped by the keys returned by a specified {@code keySelector} function.
+   * <p>
+   * <img width="640" src="https://raw.github.com/wiki/Netflix/RxJava/images/rx-operators/toMap.png">
+   * <p>
+   * If more than one source item maps to the same key, the Map will contain the latest of those items.
+   *
+   * @param keySelector
+     * the function that extracts the key from a source item to be used in the Map
+   * @return an Observable that emits a single item: a Map containing the mapped items from the source
+   *         Observable
+   */
+  def toMap[K] (keySelector: T => K): Observable[Map[K, T]]= {
+    val thisJava = asJavaObservable.asInstanceOf[rx.Observable[T]]
+    val o: rx.Observable[util.Map[K, T]] = thisJava.toMap[K](keySelector)
+    toScalaObservable[util.Map[K,T]](o).map(m => m.toMap)
+  }
+
+  /**
+   * Return an Observable that emits a single Map containing values corresponding to items emitted by the
+   * source Observable, mapped by the keys returned by a specified {@code keySelector} function.
+   * <p>
+   * <img width="640" src="https://raw.github.com/wiki/Netflix/RxJava/images/rx-operators/toMap.png">
+   * <p>
+   * If more than one source item maps to the same key, the Map will contain a single entry that
+   * corresponds to the latest of those items.
+   *
+   * @param keySelector
+     *            the function that extracts the key from a source item to be used in the Map
+   * @param valueSelector
+     *            the function that extracts the value from a source item to be used in the Map
+   * @return an Observable that emits a single item: a HashMap containing the mapped items from the source
+   *         Observable
+   */
+  def toMap[K, V] (keySelector: T => K, valueSelector: T => V) : Observable[Map[K, V]] = {
+    val thisJava = asJavaObservable.asInstanceOf[rx.Observable[T]]
+    val o: rx.Observable[util.Map[K, V]] = thisJava.toMap[K, V](keySelector, valueSelector)
+    toScalaObservable[util.Map[K, V]](o).map(m => m.toMap)
+  }
+
+  /**
+   * Return an Observable that emits a single Map, returned by a specified {@code mapFactory} function, that
+   * contains keys and values extracted from the items emitted by the source Observable.
+   * <p>
+   * <img width="640" src="https://raw.github.com/wiki/Netflix/RxJava/images/rx-operators/toMap.png">
+   *
+   * @param keySelector
+     *            the function that extracts the key from a source item to be used in the Map
+   * @param valueSelector
+     *            the function that extracts the value from the source items to be used as value in the Map
+   * @param mapFactory
+     *            the function that returns a Map instance to be used
+   * @return an Observable that emits a single item: a Map that contains the mapped items emitted by the
+   *         source Observable
+   */
+  def toMap[K, V] (keySelector: T => K, valueSelector: T => V, mapFactory: () => Map[K, V]): Observable[Map[K, V]] = {
+    val thisJava = asJavaObservable.asInstanceOf[rx.Observable[T]]
+    val o: rx.Observable[util.Map[K, V]] = thisJava.toMap[K, V](keySelector, valueSelector)
+    toScalaObservable[util.Map[K, V]](o).map(m => mapFactory() ++ m.toMap)
+  }
+
 }
 
 /**