My own sample implementation of a stream library.

sample-stream.hs

module Main (main) where
import Control.Monad
import Control.Monad.Free

main :: IO ()
main = print . runStream $ sourceString |==| reverseString |==| waitString

data Void

data StreamF i o next =
  Notify o next
  | Wait (i -> next)

instance Functor (StreamF i o) where
  fmap f (Notify x y) = Notify x (f y)
  fmap f (Wait c) = Wait (\x -> f (c x))

type Stream i o = Free (StreamF i o)

runStream :: Stream () Void a -> a
runStream (Pure x) = x
runStream (Free (Notify _ next)) = runStream next
runStream (Free (Wait cont))     = runStream (cont ())

notify :: o -> Stream i o ()
notify x = Free $ Notify x (Pure ())

wait :: Stream i o i
wait = Free $ Wait (\x -> Pure x)

sourceString :: Stream () String Int
sourceString = do
  notify "Hogehoge"
  notify "BarBar"
  notify "Hehhe"
  return (-1)

reverseString :: Stream String String Int
reverseString = forever (wait >>= notify . reverse)

waitString :: Stream String Void Int
waitString = do
  x <- wait
  y <- wait
  return (length x + length y)

(|==|) :: Stream i x a -> Stream x o a -> Stream i o a
_                    |==| Pure x               = Pure x
p1                   |==| Free (Notify x next) = Free $ Notify x (p1 |==| next)
Pure x               |==| Free (Wait _)        = Pure x
Free (Notify x next) |==| Free (Wait cont)     = next |==| (cont x)
Free (Wait cont')    |==| Free (Wait cont)     =
                            Free $ Wait (\i -> (cont' i) |==| Free (Wait cont))