Push/Pull Streams

gistfile1.fs

type Stream<'T> = ('T -> unit) -> unit -> bool


let inline ofArray (source : 'T []) : Stream<'T> =
    fun iterf ->
        let i = ref 0
        fun () -> 
            let flag = !i < source.Length
            if not flag then
                false
            else
                iterf source.[!i]
                incr i
                flag
        

let inline map (f : 'T -> 'R) (stream : Stream<'T>) : Stream<'R> =
    fun iterf ->
        stream (fun value -> iterf (f value))


let inline filter (predicate : 'T -> bool) (stream : Stream<'T>) : Stream<'T> =
    fun iterf -> 
        stream (fun value -> if predicate value then iterf value)

let inline run f = while f() do () 
let inline fold (folder : 'State -> 'T -> 'State) (state : 'State) (stream : Stream<'T>) : 'State =
    let accRef = ref state
    let next = stream (fun value -> accRef := folder !accRef value)
    run next 
    !accRef

let inline length (stream : Stream<'T>) : int =
    fold (fun acc _  -> 1 + acc) 0 stream


let inline toSeq (stream : Stream<'T>) : seq<'T> = 
    seq  {
        let current = ref Unchecked.defaultof<'T>
        let next = stream (fun v -> current := v)
        while next () do
            yield !current
    }

let inline zipWith (f : 'T -> 'S -> 'R) (first : Stream<'T>) (second : Stream<'S>) : Stream<'R> =
    let current = ref Unchecked.defaultof<'S>
    let next = second (fun v -> current := v)
    (fun iterf -> 
        first (fun v -> 
                    if not <| next() then
                        failwith "no more elements"
                    iterf (f v !current) ))