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)))

The sole purpose of ($$) is to apply an enumerator into an iteratee getting a new iterate out of that...

It would only abstract the notion of:

(partial enum-seq 5 [1 2 3 4 5 6]) and the calling of the (it) so you can just use it like:

($$ (enum-seq 5 [1 2 3 4 5 6]) it)

A problem I'm seeing that is a bit annoying though is that ($$ ... ...) is returning an "Iteratee" type instead of a function that returns an Iteratee type (as the first sum-it), and it will cause problems when chaining two ($$) calls together

($$ enum-terminal-input ($$ (enum-seq 5 [1 2 3 4 5 6]) sum-it))

Have you played with duck streams at all yet? I'm trying to think of situations where iteratees and seqs differ in what they provide. At a glance it seems 1 to 1 aside from chunking.

…

On 2011-10-22, at 10:20 AM, Roman Gonzalez ***@***.*** wrote: The sole purpose of ($$) is to _apply_ an enumerator into an iteratee getting a new iterate out of that... It would only abstract the notion of: `(partial enum-seq 5 [1 2 3 4 5 6])` and the calling of the `(it)` so you can just use it like: ``` clojure ($$ (enum-seq 5 [1 2 3 4 5 6]) it) ``` A problem I'm seeing that is a bit annoying though is that ($$ ... ...) is returning an "Iteratee" type instead of a function that returns an Iteratee type (as the first sum-it), and it will cause problems when chaining two ($$) calls together ``` Clojure ($$ enum-terminal-input ($$ (enum-seq 5 [1 2 3 4 5 6]) sum-it)) ``` ## Reply to this email directly or view it on GitHub: https://gist.github.com/1305449

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.

I think the core seq abstraction supports all 3 well. 1: use partition 2: it's just plain old function composition 3: wrap a lazy seq in another that does the transform

…

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

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