Minimalistic Streams

0_PushStream.fs

module PushStreams =
  let flip       f l r = f r l
  let pushIf     f r v = if f v then r v else true
  
  let empty          r = true
  let singleton  v   r = r v
  let collect    c s r = flip c r   |> s
  let map        m s r = m >> r     |> s
  let filter     f s r = pushIf f r |> s
  let rec ofList vs  r =
    match vs with
    | []    -> true
    | h::t  -> r h && ofList t r

  let rec toList     ps =
    let mutable vs = []
    let r v = vs <- v::vs; true
    ps r  |> ignore
    vs    |> List.rev

  let test () =
    let res = 
      List.init 10 ((+) 100) 
      |> ofList
      |> filter   (fun v -> v % 2 = 0)
      |> map      string
      |> collect  (List.ofSeq >> ofList)
      |> toList

    printfn "%A" res

1_Transducers.fs

module Transducers =
  let identity    fold s t  = fold s t
  let filtering f fold s t  = if f t then fold s t else s
  let mapping   m fold s t  = fold s (m t)

  let filter f t fold       = t (filtering f fold)
  let map    m t fold       = t (mapping   m fold)

  let sum   t = t (+)
  let print t = t (fun s v -> printfn "%A" v; s)

  let test () =
    let fold =
      identity
      |> filter (fun v -> v % 2 = 0)
      |> map    ((*) 2)
      |> sum

    let res = 
      List.init 10 ((+) 100)     
      |> List.fold fold 0
    printfn "%A" res

2_PushStream.fs

// Slightly more intricate to avoid mutation in aggregation
module PushStreams =
  type [<Struct>] Result<'S> = R of bool*'S

  type PushStream<'S, 'T> = PS of (('S -> 'T -> Result<'S>) -> 'S -> Result<'S>)

  let keepProcessing s = R (true , s) 
  let stopProcessing s = R (false, s) 

  let empty<'S, 'T> : PushStream<'S, 'T> =
    PS <| fun f z -> 
      keepProcessing z

  let singleton v : PushStream<'S, 'T> =
    PS <| fun f z -> 
      f z v

  let ofList vs : PushStream<'S, 'T> =
    PS <| fun f z ->
      let rec loop vs f s =
        match vs with
        | []    -> keepProcessing s
        | h::t  -> 
          let r = f s h
          let (R (c, s)) = r
          if c then loop t f s else r
      loop vs f z

  let filter (filter : 'T -> bool) (PS ps) : PushStream<'S, 'T> =
    PS <| fun f z ->
      let ff s v = if filter v then f s v else keepProcessing s
      ps ff z

  let map (map : 'T -> 'U) (PS ps) : PushStream<'S, 'U> =
    PS <| fun f z ->
      let ff s v = f s (map v) 
      ps ff z

  let collect (collect : 'T -> PushStream<'S, 'U>) (PS ps) : PushStream<'S, 'U> =
    PS <| fun f z ->
      let ff s v = 
        let (PS cf) = collect v
        cf f s
      ps ff z

  let toList (PS ps) : 'T list =
    let f s v = keepProcessing (v::s)
    let (R (_, v)) = ps f []
    v |> List.rev

  let tryFirst (PS ps) : 'T option =
    let f _ v = stopProcessing (Some v)
    let (R (_, v)) = ps f None
    v

  let test () =
    let res = 
      List.init 10 ((+) 100) 
      |> ofList
      |> filter   (fun v -> v % 2 = 0)
      |> map      string
      |> collect  (List.ofSeq >> ofList)
      |> toList
    printfn "%A" res


  let test2 () =
    let res = 
      List.init 10 ((+) 10) 
      |> ofList
      |> filter   (fun v -> v > 105)
      |> tryFirst

    printfn "%A" res