How to observe progress while consuming a lazy sequence?

Question

I'd like to know if the following way of "observing" a sequence while consuming it is correct. I've read the following SO answer(s) but I'm a bit surprised in that I've read several times that the "correct" way to do so was to use laziness (and hence explicitely make your sequence lazy even if it wasn't), yet the word "lazy" isn't even mentioned once there:

How to implement the Observer Design Pattern in a pure functional way?

So basically I have a lazy sequence and I want to keep that lazy sequence 100% pure and I don't want the observer to leak anything into that lazy sequence. In my example I'm simply using (range 100000) but any lazy sequence would do.

Then I'm using some mutability (in Clojure using an atom) doing something like this (the code is runnable, you can copy/paste as is in a REPL):

(let [a (atom (range 100000))]
  (loop [i 0]
    (if (= 0 (mod i 10)) (println "i: " i)) ; the observer
    (if (= 100 i)
      (take 1 @a)
      (do (reset! a (rest @a)) (recur (inc i))))))

The point is not that I'm using a mutable atom here but that the implementation of the lazy sequence doesn't know at all that it is being observed. The observer can obviously be fancier: like actually notifying observers instead of using side-effects to print i (once again: printing i here is just an example).

Is this a "correct" way of observing while consuming a lazy sequence?

If this is not correct, how would you observe while consuming a lazy sequence in Clojure?

Alternatively, how would you do it in Haskell?

Answer 1

If you merely want to intersperse a side effect during the consumption, then yes, the sensible thing to do in Clojure would be to wrap in another lazy-sequence.

(defn lazy-report
  "Wrap s in a lazy sequence that will call f, presumably for side effects, 
   on every nth element of s during consumption."
  [s f n] 
  (let [g (cons (comp first (juxt identity f)) (repeat (dec n) identity))]
    (map #(% %2) (rest (cycle g)) s)))

---

(println "sum:" (reduce + (lazy-report (range 1 1000) #(println "at:" %) 100)))
;=> at: 100
;   at: 200
; ...
;   at: 900
;   sum: 499500

Answer 2

Here's how I would do this in Haskell. What you're asking for is a lazy data structure that occasionally produces some value. This can be modeled as a stream with skipping.

data Stream a
    = Done 
    | Skip    (Stream a)
    | Yield a (Stream a) 

A Stream can either Yield a value and the rest of the stream, Skip and return the rest of the stream, or it is Done which means the stream has been fully consumed.

We can evaluate a Stream that produces side-effects by simply recursing through it, executing the side-effects in Yield, until we reach the end of the Stream.

eval :: Monad m => Stream (m a) -> m ()
eval Done         = return ()
eval (Yield ma s) = ma >> eval s
eval (Skip     s) = eval s

We can build a Stream from a lazy sequence (a list in this case) given a "decision" function that chooses whether to produce a side-effect or skip for each element in the original sequence.

observe :: Monad m => (a -> Maybe (m b)) -> [a] -> m ()
observe f = eval . go
    where go []     = Done
          go (x:xs) = case f x of
              Nothing -> Skip     (go xs)
              Just mb -> Yield mb (go xs)

If we wanted to observe other types of sequences (including lists), we can generalize observe to work for any Foldable instance with this nicely succinct definition:

observe :: (Foldable f, Monad m) => (a -> Maybe (m b)) -> f a -> m ()
observe f = eval . foldr (maybe Skip Yield . f) Done

The final product,

f :: Int -> Maybe (IO ())
f x | x `rem` 10 == 0 = Just (print x)
    | otherwise       = Nothing

main = observe f [0..100]

This works for infinite sequences as well.