Callbags in purescript

callbag.purs

module Main where

import Prelude

import Data.Array.NonEmpty (NonEmptyArray, (..))
import Data.Traversable (class Traversable, sequence)
import Effect (Effect)
import Effect.Console (log)
import Prelude as P

type Callbag a err = A a err -> Effect Unit

data A a err = Subscribe (C a err -> Effect Unit)
data B 
    = Pull 
    | Unsubscribe
data C a err
    = Talkback (B -> Effect Unit)
    | Push a
    | End
    | Error err


fromTraversable :: forall t a. Traversable t => t a -> Callbag a Unit
fromTraversable t = \(Subscribe sink) -> do
    _ <- sink (Talkback (const $ pure unit))
    _ <- sequence $ P.map (sink <<< Push) t
    sink End

map :: forall a b err. (a -> b) -> Callbag a err -> Callbag b err
map f source = \(Subscribe sink) -> let 
    talkback sourceTb Pull = sourceTb Pull
    talkback sourceTb Unsubscribe = sourceTb Unsubscribe
    handler (Talkback tb) = sink (Talkback $ talkback tb)
    handler (Push v) = sink (Push $ f v)
    handler End = sink End
    handler (Error err) = sink (Error err)
    in source (Subscribe handler)

-- incomplete, pulling requires using mutable refs?
forEach :: forall a err. (a -> Effect Unit) -> Callbag a err -> Effect Unit
forEach f cb = cb $ Subscribe handler
    where
        handler (Talkback tb) = pure unit
        handler (Push v) = f v
        handler _ = pure unit