Implement group-by

Note that since 02ccc4d, the val-fn
variant of groupBy is unimplemented so for now an exception is thrown if
it's used.

Author: daveray

language-adaptors/rxjava-clojure/README.md

@@ -72,7 +72,7 @@ Blocking operators, which are useful for testing, but should otherwise be avoide
 ## Open Issues
 
 * The missing stuff mentioned below
-* `group-by`
+* `group-by` val-fn variant isn't implemented in RxJava
 * There are some functions for defining customer Observables and Operators (`subscriber`, `operator*`, `observable*`). I don't think these are really enough for serious operator implementation, but I'm hesitant to guess at an abstraction at this point. These will probably change dramatically.
 
 ## What's Missing

language-adaptors/rxjava-clojure/src/main/clojure/rx/lang/clojure/core.clj

@@ -3,6 +3,7 @@
                             distinct do drop drop-while
                             empty every?
                             filter first future
+                            group-by
                             interpose into
                             keep keep-indexed
                             map mapcat map-indexed
@@ -16,7 +17,9 @@
             Observable
             Observer Observable$Operator Observable$OnSubscribe
             Subscriber Subscription]
-           [rx.observables BlockingObservable]
+           [rx.observables
+            BlockingObservable
+            GroupedObservable]
            [rx.subscriptions Subscriptions]
            [rx.util.functions Action0 Action1 Func0 Func1 Func2]))
 
@@ -453,7 +456,33 @@
   [^Observable xs]
   (.takeFirst xs))
 
-; TODO group-by
+(defn ^Observable group-by
+  "Returns an Observable of clojure.lang.MapEntry where the key is the result of
+  (key-fn x) and the val is an Observable of (val-fn x) for each key. If val-fn is
+  omitted, it defaults to identity.
+
+  This returns a clojure.lang.MapEntry rather than rx.observables.GroupedObservable
+  for some vague consistency with clojure.core/group-by and so that clojure.core/key,
+  clojure.core/val and destructuring will work as expected.
+
+  See:
+    clojure.core/group-by
+    rx.Observable/groupBy
+    rx.observables.GroupedObservable
+  "
+  ([key-fn ^Observable xs]
+   (->> (.groupBy xs (iop/fn* key-fn))
+        (map (fn [^GroupedObservable go]
+               (clojure.lang.MapEntry. (.getKey go) go)))))
+  ([key-fn val-fn ^Observable xs]
+   ; TODO reinstate once this is implemented
+   ; see https://github.com/Netflix/RxJava/commit/02ccc4d727a9297f14219549208757c6e0efce2a
+   (throw (UnsupportedOperationException. "groupBy with val-fn is currently unimplemented in RxJava"))
+   (->> (.groupBy xs
+                  (iop/fn* key-fn)
+                  (iop/fn* val-fn))
+        (map (fn [^GroupedObservable go]
+               (clojure.lang.MapEntry. (.getKey go) go))))))
 
 (defn interpose
   [sep xs]

language-adaptors/rxjava-clojure/src/test/clojure/rx/lang/clojure/core_test.clj

@@ -256,6 +256,36 @@
   (is (= []
          (b/into [] (rx/first (rx/empty))))))
 
+(deftest test-group-by
+  (let [xs [{:k :a :v 1} {:k :b :v 2} {:k :a :v 3} {:k :c :v 4}]]
+    (testing "with just a key-fn"
+      (is (= [[:a {:k :a :v 1}]
+              [:b {:k :b :v 2}]
+              [:a {:k :a :v 3}]
+              [:c {:k :c :v 4}]]
+             (->> xs
+                  (rx/seq->o)
+                  (rx/group-by :k)
+                  (rx/mapcat (fn [[k vo :as me]]
+                               (is (instance? clojure.lang.MapEntry me))
+                               (rx/map #(vector k %) vo)))
+                  (b/into [])))))
+
+    ; TODO reinstate once this is implemented
+    ; see https://github.com/Netflix/RxJava/commit/02ccc4d727a9297f14219549208757c6e0efce2a
+    #_(testing "with a val-fn"
+      (is (= [[:a 1]
+              [:b 2]
+              [:a 3]
+              [:c 4]]
+             (->> xs
+                  (rx/seq->o)
+                  (rx/group-by :k :v)
+                  (rx/mapcat (fn [[k vo :as me]]
+                               (is (instance? clojure.lang.MapEntry me))
+                               (rx/map #(vector k %) vo)))
+                  (b/into [])))))
+    ))
 (deftest test-interpose
   (is (= (interpose \, [1 2 3])
          (b/into [] (rx/interpose \, (rx/seq->o [1 2 3]))))))