xforms 0.10.0: new io namespace, some as a transducing context (and a rf) too

README.md

@@ -12,16 +12,21 @@ More transducers and reducing functions for Clojure(script)!
  * higher-order ones: `by-key`, `into-by-key`, `multiplex`, `transjuxt`, `partition` (2+ args)
  * 1-item ones: `reduce`, `into`, `transjuxt`, `last`, `count`, `avg`, `sd`, `min`, `minimum`, `max`, `maximum`, `str`
  
-*Reducing functions* (in `net.cgrand.xforms.rfs`): `min`, `minimum`, `max`, `maximum`, `str`, `str!`, `avg`, `sd`, `juxt` and `last`.
+*Reducing functions*
 
-Transducing contexts: `transjuxt` (for performing several transductions in a single pass), `into`, `count`.
+ * in `net.cgrand.xforms.rfs`: `min`, `minimum`, `max`, `maximum`, `str`, `str!`, `avg`, `sd`, `last` and `some`.
+ * in `net.cgrand.xforms.io`: `line-out` and `edn-out`.
+
+*Transducing contexts*: `transjuxt` (for performing several transductions in a single pass), `into`, `count` and `some`.
+
+*Reducible views* (in `net.cgrand.xforms.io`): `lines-in` and `edn-in`.
 
 ## Usage
 
 Add this dependency to your project:
 
 ```clj
-[net.cgrand /xforms "0.9.5"]
+[net.cgrand /xforms "0.10.0"]
 ```
 
 ```clj

project.clj

@@ -1,4 +1,4 @@
-(defproject net.cgrand/xforms "0.9.5"
+(defproject net.cgrand/xforms "0.10.0"
   :description "Extra transducers for Clojure"
   :url "https://github.com/cgrand/xforms"
   :license {:name "Eclipse Public License"

src/net/cgrand/xforms.cljc

@@ -5,7 +5,7 @@
              [net.cgrand.macrovich :as macros]
              [net.cgrand.xforms :refer [for kvrf let-complete]])
       :clj (:require [net.cgrand.macrovich :as macros]))
-  (:refer-clojure :exclude [reduce reductions into count for partition str last keys vals min max drop-last take-last])
+  (:refer-clojure :exclude [some reduce reductions into count for partition str last keys vals min max drop-last take-last])
   (:require [#?(:clj clojure.core :cljs cljs.core) :as core]
     [net.cgrand.xforms.rfs :as rf])
   #?(:cljs (:import [goog.structs Queue])))
@@ -367,6 +367,10 @@
                     acc)
                   acc))))))))
   
+  #_(defn zip [xform1 xform2]
+     (fn [rf]
+       (let )))
+  
   (defn take-last [n]
     (fn [rf]
       (let [dq (java.util.ArrayDeque. n)]
@@ -460,7 +464,8 @@
 
 #?(:clj
     (defn window-by-time
-      "Returns a transducer which computes a windowed accumulator over chronologically sorted items.
+      "ALPHA
+   Returns a transducer which computes a windowed accumulator over chronologically sorted items.
    
    timef is a function from one item to its scaled timestamp (as a double). The window length is always 1.0
    so timef must normalize timestamps. For example if timestamps are in seconds (and under the :ts key),
@@ -589,6 +594,11 @@
 
 (def last (reduce rf/last))
 
+(defn some
+  "Process coll through the specified xform and returns the first local true value."
+  [xform coll]
+  (transduce xform rf/some nil coll))
+
 (defn transjuxt
   "Performs several transductions over coll at once. xforms-map can be a map or a sequential collection.
    When xforms-map is a map, returns a map with the same keyset as xforms-map.
@@ -609,4 +619,25 @@
   ([xforms-map coll]
     (transduce (transjuxt xforms-map) rf/last coll)))
 
+#_(defn rollup
+   "Roll-up input data along the provided dimensions (which are functions of one input item),
+   Values of interest are extracted from items using the valfn function and are then summarized
+   by summary-fn (a reducing function over values returned by valfn or summaries).
+   Each level of rollup is a map with two keys: :summary and :details."
+   ([dimensions valfn summary-fn]
+     (let [[dim & dims] (reverse dimensions)]
+       (core/reduce
+         (fn [xform dim]
+           (comp
+             (by-key dim xform)
+             (transjuxt
+               {:detail (into {})
+                :summary (comp vals (map :summary) (reduce summary-fn))})))
+         (comp (by-key dim (map valfn))
+           (transjuxt
+             {:detail (into {})
+              :summary (comp vals (reduce summary-fn))}))
+         dims)))
+   ([dimensions valfn summary-fn coll]
+     (into {} (rollup dimensions valfn summary-fn) coll)))
 )

src/net/cgrand/xforms/io.clj

@@ -0,0 +1,83 @@
+(ns net.cgrand.xforms.io
+  (:require [clojure.java.io :as io]
+    [clojure.edn :as edn]))
+
+(defn keep-opts [m like]
+  (let [ns (namespace like)]
+    (into {}
+      (keep (fn [[k v]]
+              (when (= ns (or (namespace k) ns))
+                [(keyword (name k)) v])))
+      m)))
+
+(defn lines-in
+  "Returns a reducible view over the provided input.
+   Input is read line by line. Coercion of the input is done by io/reader (and opts are passed to it).
+   Input is automatically closed upon completion or error."
+  [in & opts]
+  (let [no-init (Object.)]
+    (reify clojure.lang.IReduce 
+      (reduce [self f] (.reduce self f no-init))
+      (reduce [self f init]
+        (with-open [rdr (apply io/reader in opts)]
+          (let [rdr (cond-> rdr (not (instance? java.io.BufferedReader rdr)) java.io.BufferedReader.)
+                init (if (identical? init no-init)
+                       (or (.readLine rdr) (f))
+                       init)]
+            (loop [state init]
+              (if-some [line (.readLine rdr)]
+                (let [state (f state line)]
+                  (if (reduced? state)
+                    (unreduced state)
+                    (recur state)))
+                state))))))))
+
+(defn lines-out
+  "Reducing function that writes values serialized to its accumulator (a java.io.Writer).
+   Returns the writer."
+  ([w] w)
+  ([^java.io.Writer w line]
+    (doto w
+      (.write (str line))
+      (.newLine))))
+
+(defn edn-in
+  "Returns a reducible view over the provided input.
+   Input is read line by line. Coercion of the input is done by io/reader (and opts are passed to it).
+   Input is automatically closed upon completion or error."
+  [in & {:as opts}]
+  (let [no-init (Object.)]
+    (reify clojure.lang.IReduce 
+     (reduce [self f] (.reduce self f no-init))
+     (reduce [self f init]
+       (with-open [rdr (apply io/reader in (mapcat seq (keep-opts opts ::io/opts)))]
+         (let [rdr (cond-> rdr (not (instance? java.io.PushbackReader rdr)) java.io.PushbackReader.)
+               opts (assoc (keep-opts opts ::edn/opts) :eof no-init)
+               init (if (identical? init no-init)
+                      (let [form (edn/read opts rdr)]
+                        (if (identical? no-init form)
+                          (f)
+                          form))
+                      init)]
+           (loop [state init]
+             (let [form (edn/read opts rdr)]
+               (if (identical? no-init form)
+                 state
+                 (let [state (f state form)]
+                   (if (reduced? state)
+                     (unreduced state)
+                     (recur state))))))))))))
+
+(defn edn-out
+  "Reducing function that writes values serialized as EDN to its accumulator (a java.io.Writer).
+   Returns the writer."
+  ([w] w)
+  ([^java.io.Writer w x]
+    (binding [*out* w
+              *print-length* nil
+              *print-level* nil
+              *print-dup* false
+              *print-meta* false
+              *print-readably* true]
+      (println x)
+      w)))

src/net/cgrand/xforms/rfs.cljc

@@ -1,6 +1,6 @@
 (ns net.cgrand.xforms.rfs
   {:author "Christophe Grand"}
-  (:refer-clojure :exclude [str last min max])
+  (:refer-clojure :exclude [str last min max some])
   #?(:cljs (:require-macros
              [net.cgrand.macrovich :as macros]
              [net.cgrand.xforms.rfs :refer [or-instance?]])
@@ -85,6 +85,12 @@
   ([x] x)
   ([_ x] x))
 
+(defn some
+  "Reducing function that returns the first logical true value."
+  ([] nil)
+  ([x] x)
+  ([_ x] (when x (reduced x))))
+
 (defn str!
   "Like xforms/str but returns a StringBuilder."
   ([] (#?(:clj StringBuilder. :cljs StringBuffer.)))