added seque

1 files changed, 36 insertions, 0 deletions

diff --git a/src/boot.clj b/src/boot.clj
index 832c494d..a842d933 100644
--- a/src/boot.clj
+++ b/src/boot.clj
@@ -2611,3 +2611,39 @@ not-every? (comp not every?))
   "Casts to long[]"
   [xs] `(. clojure.lang.Numbers longs ~xs))
 
+(import '(java.util.concurrent BlockingQueue LinkedBlockingQueue))
+
+(defn seque
+  "Creates a queued seq on another (presumably lazy) seq s. The queued
+  seq will produce a concrete seq in the background, and can get up to
+  n items ahead of the consumer. n-or-q can be an integer n buffer
+  size, or an instance of java.util.concurrent BlockingQueue. Note
+  that reading from a seque can block if the reader gets ahead of the
+  producer."
+  ([s] (seque 100 s))
+  ([n-or-q s]
+   (let [#^BlockingQueue q (if (instance? BlockingQueue n-or-q)
+                             n-or-q
+                             (LinkedBlockingQueue. (int n-or-q)))
+         NIL (Object.) ;nil sentinel since LBQ doesn't support nils
+         agt (agent (seq s))
+         fill (fn [s]
+                (try
+                  (loop [[x & xs :as s] s]
+                    (if s
+                      (if (.offer q (if (nil? x) NIL x))
+                        (recur xs)
+                        s)
+                      (.put q q))) ; q itself is eos sentinel
+                  (catch Exception e
+                    (.put q q)
+                    (throw e)))) 
+         drain (fn drain []
+                 (let [x (.take q)]
+                   (if (identical? x q) ;q itself is eos sentinel
+                     @agt  ;will be nil - touch agent just to propagate errors
+                     (do
+                       (send-off agt fill)
+                       (lazy-cons (if (identical? x NIL) nil x) (drain))))))]
+     (send-off agt fill)
+     (drain))))
\ No newline at end of file