Add toMultimap to RxScala

Author: zsxwing

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

@@ -865,6 +865,40 @@ class RxScalaDemo extends JUnitSuite {
     println(m.toBlockingObservable.single)
   }
 
+  @Test def toMultimapExample1(): Unit = {
+    val o : Observable[String] = List("alice", "bob", "carol", "allen", "clarke").toObservable
+    val keySelector = (s: String) => s.head
+    val m = o.toMultimap(keySelector)
+    println(m.toBlocking.single.mapValues(_.toList))
+  }
+
+  @Test def toMultimapExample2(): Unit = {
+    val o : Observable[String] = List("alice", "bob", "carol", "allen", "clarke").toObservable
+    val keySelector = (s: String) => s.head
+    val valueSelector = (s: String) => s.tail
+    val m = o.toMultimap(keySelector, valueSelector)
+    println(m.toBlocking.single.mapValues(_.toList))
+  }
+
+  @Test def toMultimapExample3(): Unit = {
+    val o : Observable[String] = List("alice", "bob", "carol", "allen", "clarke").toObservable
+    val keySelector = (s: String) => s.head
+    val valueSelector = (s: String) => s.tail
+    val mapFactory = () => Map('d' -> List("oug"))
+    val m = o.toMultimap(keySelector, valueSelector, mapFactory)
+    println(m.toBlocking.single.mapValues(_.toList))
+  }
+
+  @Test def toMultimapExample4(): Unit = {
+    val o : Observable[String] = List("alice", "bob", "carol", "allen", "clarke").toObservable
+    val keySelector = (s: String) => s.head
+    val valueSelector = (s: String) => s.tail
+    val mapFactory = () => Map('d' -> List("oug"))
+    val valueFactor = (k: Char) => List[String]()
+    val m = o.toMultimap(keySelector, valueSelector, mapFactory, valueFactor)
+    println(m.toBlocking.single)
+  }
+
   @Test def containsExample(): Unit = {
     val o1 = List(1, 2, 3).toObservable.contains(2)
     assertTrue(o1.toBlockingObservable.single)

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

@@ -3880,6 +3880,83 @@ trait Observable[+T]
   def longCount: Observable[Long] = {
     toScalaObservable[java.lang.Long](asJavaObservable.longCount()).map(_.longValue())
   }
+
+  /**
+   * Returns an Observable that emits a single `Map` that contains an `Seq` of items emitted by the
+   * source Observable keyed by a specified keySelector` function.
+   *
+   * <img width="640" height="305" src="https://raw.github.com/wiki/Netflix/RxJava/images/rx-operators/toMultiMap.png">
+   *
+   * @param keySelector the function that extracts the key from the source items to be used as key in the HashMap
+   * @return an Observable that emits a single item: a `Map` that contains an `Seq` of items mapped from
+   *         the source Observable
+   */
+  def toMultimap[K](keySelector: T => K): Observable[Map[K, Seq[T]]] = {
+    val thisJava = asJavaObservable.asInstanceOf[rx.Observable[T]]
+    val o: rx.Observable[java.util.Map[K, java.util.Collection[T]]] = thisJava.toMultimap[K](keySelector)
+    toScalaObservable[java.util.Map[K, java.util.Collection[T]]](o).map(m => m.toMap.mapValues(_.toSeq))
+  }
+
+  /**
+   * Returns an Observable that emits a single `Map` that contains an `Seq` of values extracted by a
+   * specified `valueSelector` function from items emitted by the source Observable, keyed by a
+   * specified `keySelector` function.
+   *
+   * <img width="640" height="305" src="https://raw.github.com/wiki/Netflix/RxJava/images/rx-operators/toMultiMap.png">
+   *
+   * @param keySelector the function that extracts a key from the source items to be used as key in the HashMap
+   * @param valueSelector the function that extracts a value from the source items to be used as value in the HashMap
+   * @return an Observable that emits a single item: a `Map` that contains an `Seq` of items mapped from
+   *         the source Observable
+   */
+  def toMultimap[K, V](keySelector: T => K, valueSelector: T => V): Observable[Map[K, Seq[V]]] = {
+    val thisJava = asJavaObservable.asInstanceOf[rx.Observable[T]]
+    val o: rx.Observable[java.util.Map[K, java.util.Collection[V]]] = thisJava.toMultimap[K, V](keySelector, valueSelector)
+    toScalaObservable[java.util.Map[K, java.util.Collection[V]]](o).map(m => m.toMap.mapValues(_.toSeq))
+  }
+
+  /**
+   * Returns an Observable that emits a single `Map`, returned by a specified mapFactory` function, that
+   * contains an `Seq` of values, extracted by a specified `valueSelector` function from items
+   * emitted by the source Observable and keyed by the `keySelector` function.
+   *
+   * <img width="640" height="305" src="https://raw.github.com/wiki/Netflix/RxJava/images/rx-operators/toMultiMap.png">
+   *
+   * @param keySelector the function that extracts a key from the source items to be used as the key in the Map
+   * @param valueSelector the function that extracts a value from the source items to be used as the value in the Map
+   * @param mapFactory he function that returns a Map instance to be used
+   * @return an Observable that emits a single item: a `Map` that contains a `Seq` items mapped from the source
+   *         Observable
+   */
+  def toMultimap[K, V](keySelector: T => K, valueSelector: T => V, mapFactory: () => Map[K, Seq[V]]): Observable[Map[K, Seq[V]]] = {
+    val thisJava = asJavaObservable.asInstanceOf[rx.Observable[T]]
+    val o: rx.Observable[java.util.Map[K, java.util.Collection[V]]] = thisJava.toMultimap[K, V](keySelector, valueSelector)
+    toScalaObservable[java.util.Map[K, java.util.Collection[V]]](o).map(m => mapFactory() ++ m.toMap.mapValues(_.toSeq))
+  }
+
+  /**
+   * Returns an Observable that emits a single `Map`, returned by a specified `mapFactory` function, that
+   * contains a custom `Seq` of values, extracted by a specified `valueSelector` function from
+   * items emitted by the source Observable, and keyed by the `keySelector` function.
+   *
+   * <img width="640" height="305" src="https://raw.github.com/wiki/Netflix/RxJava/images/rx-operators/toMultiMap.png">
+   *
+   * @param keySelector the function that extracts a key from the source items to be used as the key in the Map
+   * @param valueSelector the function that extracts a value from the source items to be used as the value in the Map
+   * @param mapFactory the function that returns a Map instance to be used
+   * @param collectionFactory the function that returns a Collection instance for a particular key to be used in the Map
+   * @return an Observable that emits a single item: a `Map` that contains the `Seq` of mapped items from
+   *         the source Observable
+   */
+  def toMultimap[K, V](keySelector: T => K, valueSelector: T => V, mapFactory: () => Map[K, Seq[V]], collectionFactory: K => Seq[V]): Observable[Map[K, Seq[V]]] = {
+    val thisJava = asJavaObservable.asInstanceOf[rx.Observable[T]]
+    val o: rx.Observable[java.util.Map[K, java.util.Collection[V]]] = thisJava.toMultimap[K, V](keySelector, valueSelector)
+    toScalaObservable[java.util.Map[K, java.util.Collection[V]]](o).map {
+      m => mapFactory() ++ m.toMap.map {
+        case (k: K, v: java.util.Collection[V]) => (k, collectionFactory(k) ++ v)
+      }
+    }
+  }
 }
 
 /**

language-adaptors/rxjava-scala/src/test/scala/rx/lang/scala/CompletenessTest.scala

@@ -177,6 +177,8 @@ class CompletenessTest extends JUnitSuite {
       "timer(Long, Long, TimeUnit)" -> "timer(Duration, Duration)",
       "timer(Long, Long, TimeUnit, Scheduler)" -> "timer(Duration, Duration, Scheduler)",
       "toList()" -> "toSeq",
+      "toMultimap(Func1[_ >: T, _ <: K], Func1[_ >: T, _ <: V], Func0[_ <: Map[K, Collection[V]]])" -> "toMultimap(T => K, T => V, () => Map[K, Seq[V]])",
+      "toMultimap(Func1[_ >: T, _ <: K], Func1[_ >: T, _ <: V], Func0[_ <: Map[K, Collection[V]]], Func1[_ >: K, _ <: Collection[V]])" -> "toMultimap(T => K, T => V, () => Map[K, Seq[V]], K => Seq[V])",
       "toSortedList()" -> "[Sorting is already done in Scala's collection library, use `.toSeq.map(_.sorted)`]",
       "toSortedList(Func2[_ >: T, _ >: T, Integer])" -> "[Sorting is already done in Scala's collection library, use `.toSeq.map(_.sortWith(f))`]",
       "window(Observable[U])" -> "window(=> Observable[Any])",