Implemented partition-all

There isn't really an rxjava impl equivalent to clojure.core/partition,
so it's omitted.

Author: daveray

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

@@ -7,7 +7,7 @@
                             interleave interpose into
                             keep keep-indexed
                             map mapcat map-indexed
-                            merge next nth partition reduce reductions
+                            merge next nth partition-all reduce reductions
                             rest seq some sort sort-by split-with
                             take take-while throw])
   (:require [rx.lang.clojure.interop :as iop]
@@ -618,7 +618,17 @@
   ([^Observable xs index not-found]
    (.elementAtOrDefault xs index not-found)))
 
-; TODO partition. Use window
+(defn ^Observable partition-all
+  "Returns an Observable of Observables of n items each, at offsets step
+  apart. If step is not supplied, defaults to n, i.e. the partitions
+  do not overlap. May include partitions with fewer than n items at the end.
+
+  See:
+    clojure.core/partition-all
+    rx.Observable/window
+  "
+  ([n ^Observable xs] (.window xs (int n)))
+  ([n step ^Observable xs] (.window xs (int n) (int step))))
 
 (defn ^Observable reduce
   ([f ^Observable xs] (.reduce xs (iop/fn* f)))

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

@@ -416,6 +416,33 @@
   (let [in [:q :r :s :t :u]]
     (is (= (rest in) (b/into [] (rx/rest (rx/seq->o in)))))))
 
+(deftest test-partition-all
+  (are [input-size part-size step] (= (->> (range input-size)
+                                           (partition-all part-size step))
+                                      (->> (range input-size)
+                                           (rx/seq->o)
+                                           (rx/partition-all part-size step)
+                                           (rx/map #(rx/into [] %))
+                                           (rx/concat*)
+                                           (b/into [])))
+       0 1 1
+       10 2 2
+       10 3 2
+       15 30 4)
+
+  (are [input-size part-size] (= (->> (range input-size)
+                                      (partition-all part-size))
+                                 (->> (range input-size)
+                                      (rx/seq->o)
+                                      (rx/partition-all part-size)
+                                      (rx/map #(rx/into [] %))
+                                      (rx/concat*)
+                                      (b/into [])))
+       0 1
+       10 2
+       10 3
+       15 30))
+
 (deftest test-reduce
   (is (= (reduce + 0 (range 4))
          (b/first (rx/reduce + 0 (rx/seq->o (range 4)))))))