Haskell Iteratees in clojure (naive impl)

iteratee.clj

(ns iteratee
  (:require [clojure.contrib.types :as adt]))

;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;

(adt/defadt ::stream
  eof
  (chunks xs))

(adt/defadt ::iteratee
  (yield value stream)
  (continue consumer)
  (error e))

(defn sum [xs] (reduce + 0 xs))

(defn $$ [enum it] (enum (it)))

(defn run_ [it]
  (adt/match it
    (error e) e
    (yield result _) result
    (continue consumer)
      (do
        (let [next-it (consumer eof)]
          (do
            (adt/match next-it
              (error e) e
              (yield result _) result
              (continue _) (do
                (println "missbehaving iteratee")
                nil)))))))

(defn enum-seq [chunk-size xs0 it]
  (let [ [a xs] (split-at chunk-size xs0) ]
    (if (seq a)
        (adt/match it
          (error _) it
          (yield _ _) it
          (continue consumer)
              (recur chunk-size xs (consumer (chunks a))))

        it)))


(defn sum-it []
  (letfn [(go [acc stream]
            (adt/match stream
              eof
                (yield acc eof)
              (chunks xs)
                (do
                  (continue (partial go (+ (sum xs) acc))))))]
   (continue (partial go 0))))

(defn -main [& args]
  (run_ ($$ (partial enum-seq 5 [1 2 3 4 5 6]) sum-it)))

Async event handling where you don't want to block the consumer when the source doesn't have new data but isn't finished yet might be the answer to my question.

…

On 2011-10-22, at 11:00 AM, Roman Gonzalez ***@***.*** wrote: I read the clojure.contrib.stream-utils lib and I found a hard time to understand it... don't know exactly how the actors involved work. 1) Chunking is one 2) Stream composition 3) Stream transformation Of course in this impl #3 is not supported, but in practice you are able to do that... One problem I'm seeing however is that you are doing recursion on the consumers, (as you can see I'm using recursion on the sum-it, which if there is to many things to sum, it is likely that is going to throw a StackLimitException). We definitely need to play with the stream-utils (first understand it of course) and see if that library supports all of this features. ## Reply to this email directly or view it on GitHub: https://gist.github.com/1305449