Rewrite toMultimap

Author: zsxwing

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

@@ -19,6 +19,7 @@ import java.io.IOException
 import java.util.concurrent.CountDownLatch
 import java.util.concurrent.TimeUnit
 
+import scala.collection.mutable
 import scala.concurrent.duration.Duration
 import scala.concurrent.duration.DurationInt
 import scala.concurrent.duration.DurationLong
@@ -881,10 +882,10 @@ class RxScalaDemo extends JUnitSuite {
   }
 
   @Test def toMultimapExample3(): Unit = {
-    val o : Observable[String] = List("alice", "bob", "carol", "allen", "clarke").toObservable
+    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 mapFactory: () => mutable.Map[Char, mutable.Buffer[String]] = () => mutable.Map('d' -> mutable.ListBuffer("oug"))
     val m = o.toMultimap(keySelector, valueSelector, mapFactory)
     println(m.toBlocking.single.mapValues(_.toList))
   }
@@ -893,9 +894,9 @@ 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 = () => Map('d' -> List("oug"))
-    val valueFactor = (k: Char) => List[String]()
-    val m = o.toMultimap(keySelector, valueSelector, mapFactory, valueFactor)
+    val mapFactory: () => mutable.Map[Char, mutable.Buffer[String]] = () => 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

@@ -102,6 +102,7 @@ trait Observable[+T]
   import scala.collection.JavaConverters._
   import scala.collection.Seq
   import scala.concurrent.duration.{Duration, TimeUnit, MILLISECONDS}
+  import scala.collection.mutable
   import rx.functions._
   import rx.lang.scala.observables.BlockingObservable
   import ImplicitFunctionConversions._
@@ -3891,10 +3892,8 @@ trait Observable[+T]
    * @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))
+  def toMultimap[K](keySelector: T => K): Observable[scala.collection.Map[K, Seq[T]]] = {
+    toMultimap(keySelector, k => k)
   }
 
   /**
@@ -3909,10 +3908,8 @@ trait Observable[+T]
    * @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))
+  def toMultimap[K, V](keySelector: T => K, valueSelector: T => V): Observable[scala.collection.Map[K, Seq[V]]] = {
+    toMultimap(keySelector, valueSelector, () => mutable.Map[K, mutable.Buffer[V]]())
   }
 
   /**
@@ -3928,10 +3925,8 @@ trait Observable[+T]
    * @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))
+  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]())
   }
 
   /**
@@ -3948,12 +3943,24 @@ trait Observable[+T]
    * @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)
+  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]]] = {
+    lift {
+      (subscriber: Subscriber[scala.collection.Map[K, Seq[V]]]) => {
+        val map = mapFactory().withDefault(collectionFactory)
+        Subscriber[T](
+          subscriber,
+          (t: T) => {
+            val key = keySelector(t)
+            val value = map(key)
+            value += valueSelector(t)
+            map += key -> value: Unit
+          },
+          e => subscriber.onError(e),
+          () => {
+            subscriber.onNext(map)
+            subscriber.onCompleted()
+          }
+        )
       }
     }
   }