Change the return type, and add tests

Author: zsxwing

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

@@ -870,22 +870,22 @@ class RxScalaDemo extends JUnitSuite {
     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))
+    println(m.toBlocking.single)
   }
 
   @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))
+    println(m.toBlocking.single)
   }
 
   @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: () => mutable.Map[Char, mutable.Buffer[String]] = () => mutable.Map('d' -> mutable.ListBuffer("oug"))
+    val mapFactory = () => mutable.Map('d' -> mutable.Buffer("oug"))
     val m = o.toMultimap(keySelector, valueSelector, mapFactory)
     println(m.toBlocking.single.mapValues(_.toList))
   }
@@ -894,7 +894,7 @@ class RxScalaDemo extends JUnitSuite {
     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: () => mutable.Map[Char, mutable.Buffer[String]] = () => mutable.Map('d' -> mutable.ListBuffer("oug"))
+    val mapFactory = () => mutable.Map('d' -> mutable.ListBuffer("oug"))
     val valueFactory = (k: Char) => mutable.ListBuffer[String]()
     val m = o.toMultimap(keySelector, valueSelector, mapFactory, valueFactory)
     println(m.toBlocking.single)

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

@@ -3913,47 +3913,52 @@ trait Observable[+T]
   }
 
   /**
-   * 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.
+   * Returns an Observable that emits a single `mutable.Map[K, mutable.Buffer[V]]`, returned by a specified `mapFactory` function, that
+   * contains 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
+   * @param mapFactory he function that returns a `mutable.Map[K, mutable.Buffer[V]]` instance to be used
+   * @return an Observable that emits a single item: a `mutable.Map[K, mutable.Buffer[V]]` that contains items mapped
+   *         from the source Observable
    */
-  def toMultimap[K, V](keySelector: T => K, valueSelector: T => V, mapFactory: () => mutable.Map[K, mutable.Buffer[V]]): Observable[scala.collection.Map[K, Seq[V]]] = {
-    toMultimap(keySelector, valueSelector, mapFactory, k => mutable.ListBuffer[V]())
+  def toMultimap[K, V, M <: mutable.Map[K, mutable.Buffer[V]]](keySelector: T => K, valueSelector: T => V, mapFactory: () => M): Observable[M] = {
+    toMultimap[K, V, mutable.Buffer[V], M](keySelector, valueSelector, mapFactory, k => mutable.Buffer[V]())
   }
 
   /**
-   * 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.
+   * Returns an Observable that emits a single `mutable.Map[K, B]`, returned by a specified `mapFactory` function, that
+   * contains 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
+   * @param collectionFactory the function that returns a `mutable.Buffer[V]` instance for a particular key to be used in the Map
+   * @return an Observable that emits a single item: a `mutable.Map[K, B]` that contains mapped items from the source Observable
    */
-  def toMultimap[K, V](keySelector: T => K, valueSelector: T => V, mapFactory: () => mutable.Map[K, mutable.Buffer[V]], collectionFactory: K => mutable.Buffer[V]): Observable[scala.collection.Map[K, Seq[V]]] = {
+  def toMultimap[K, V, B <: mutable.Buffer[V], M <: mutable.Map[K, B]](keySelector: T => K, valueSelector: T => V, mapFactory: () => M, collectionFactory: K => B): Observable[M] = {
+    // It's complicated to convert `mutable.Map[K, mutable.Buffer[V]]` to `java.util.Map[K, java.util.Collection[V]]`,
+    // so RxScala implements `toMultimap` directly.
+    // Choosing `mutable.Buffer/Map` is because `append/update` is necessary to implement an efficient `toMultimap`.
     lift {
-      (subscriber: Subscriber[scala.collection.Map[K, Seq[V]]]) => {
-        val map = mapFactory().withDefault(collectionFactory)
+      (subscriber: Subscriber[M]) => {
+        val map = mapFactory()
         Subscriber[T](
           subscriber,
           (t: T) => {
             val key = keySelector(t)
-            val value = map(key)
-            value += valueSelector(t)
-            map += key -> value: Unit
+            val values = map.get(key) match {
+              case Some(v) => v
+              case None => collectionFactory(key)
+            }
+            values += valueSelector(t)
+            map += key -> values: Unit
           },
           e => subscriber.onError(e),
           () => {

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

@@ -177,8 +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])",
+      "toMultimap(Func1[_ >: T, _ <: K], Func1[_ >: T, _ <: V], Func0[_ <: Map[K, Collection[V]]])" -> "toMultimap(T => K, T => V, () => M)",
+      "toMultimap(Func1[_ >: T, _ <: K], Func1[_ >: T, _ <: V], Func0[_ <: Map[K, Collection[V]]], Func1[_ >: K, _ <: Collection[V]])" -> "toMultimap(T => K, T => V, () => M, K => B)",
       "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])",

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

@@ -15,6 +15,7 @@
  */
 package rx.lang.scala
 
+import scala.collection.mutable
 import scala.collection.mutable.ListBuffer
 import scala.concurrent.{Future, Await}
 import scala.concurrent.duration.Duration
@@ -228,4 +229,42 @@ class ObservableTests extends JUnitSuite {
     val o = Observable.empty.orElse(-1)
     assertEquals(List(-1), o.toBlocking.toList)
   }
+
+  @Test
+  def testToMultimap() {
+    val o = Observable.items("a", "b", "cc", "dd").toMultimap(_.length)
+    val expected = Map(1 -> List("a", "b"), 2 -> List("cc", "dd"))
+    assertEquals(expected, o.toBlocking.single)
+  }
+
+  @Test
+  def testToMultimapWithValueSelector() {
+    val o = Observable.items("a", "b", "cc", "dd").toMultimap(_.length, s => s + s)
+    val expected = Map(1 -> List("aa", "bb"), 2 -> List("cccc", "dddd"))
+    assertEquals(expected, o.toBlocking.single)
+  }
+
+  @Test
+  def testToMultimapWithMapFactory() {
+    val m = mutable.Map[Int, mutable.Buffer[String]]()
+    val o = Observable.items("a", "b", "cc", "dd").toMultimap(_.length, s => s, () => m)
+    val expected = Map(1 -> List("a", "b"), 2 -> List("cc", "dd"))
+    val r = o.toBlocking.single
+    assertTrue(m eq r) // check same instance
+    assertEquals(expected, r)
+  }
+
+  @Test
+  def testToMultimapWithCollectionFactory() {
+    val m = mutable.Map[Int, mutable.Buffer[String]]()
+    val ls = List(mutable.Buffer[String](), mutable.Buffer[String]())
+    val o = Observable.items("a", "b", "cc", "dd").toMultimap(_.length, s => s, () => m, (i: Int) => ls(i - 1))
+    val expected = Map(1 -> List("a", "b"), 2 -> List("cc", "dd"))
+    val r = o.toBlocking.single
+    assertTrue(m eq r) // check same instance
+    assertTrue(ls(0) eq r(1)) // check same instance
+    assertTrue(ls(1) eq r(2)) // check same instance
+    assertEquals(expected, r)
+  }
+
 }