dbrattli

data IOAction a = Return a
                | Put String (IOAction a)
                | Get (String -> IOAction a)

get   = Get Return
put s = Put s (Return ())

seqio :: IOAction a -> (a -> IOAction b) -> IOAction b
seqio (Return a) f = f a
seqio (Put s io) f = Put s (seqio io f)

dbrattli

type Promise<'a> () =
    let mutable result : Result<'a, exn> option = None
    let thenCbs = List<Callback<'a>> ()
    let catchCbs = List<Callback<exn>> ()

    member this.Resolve (value: 'a) =
        result <- Some <| Ok value
        for cb in thenCbs do
            cb value

dbrattli

// Setters and getters
type Getter<'a> = unit -> 'a
type Setter<'a> = 'a -> unit

// An Enumerator is a getter of values. An Observer is a setter of values
type Enumerator<'a> = unit -> 'a
type Observer<'a> = 'a -> unit

// An Enumerable is a getter of Enumerators. An Observable is a setter of Observers
type Enumerable<'a> = unit -> Enumerator<'a>

dbrattli

/// Convert async observable to async sequence, non-blocking.
/// Producer will be awaited until item is consumed by the async
/// enumerator.
let toAsyncSeq (source: IAsyncObservable<'a>) : AsyncSeq<'a> =
    let ping = new AutoResetEvent false
    let pong = new AutoResetEvent false
    let mutable latest : Notification<'a> = OnCompleted

    let _obv n =
        async {

dbrattli

let single<'a> (value: 'a) : IObservable<'a> =
    let noop = fun () -> ()
    let subscribe (obs: IObserver<'a>) : IDisposable =
        obs.OnNext value
        obs.OnCompleted ()
        { new IDisposable with member __.Dispose () = noop () }

    { new IObservable<'a> with member __.Subscribe obs' = subscribe obs' }

dbrattli

// An Async Enumerator is a getter of Async values. An Async Observer is an Async setter of values
type AsyncEnumerator<'a> = unit -> Async<'a>
type AsyncObserver<'a> = 'a -> Async<unit>

// An Async sequence is a getter of Async Enumerators. An Async Observable is an Async setter of Async Observers
type AsyncSequence<'a> = unit -> AsyncEnumerator<'a>
type AsyncObservable<'a> = AsyncObserver<'a> -> Async<unit>

type AsyncSequence'<'a> = unit -> (unit -> Async<'a>)
type AsyncObservable'<'a> = ('a -> Async<unit>) -> Async<unit>

dbrattli

open System.Threading

// A (synchronous) function taking a long time before returning a value
let blockingFunc () =
    Thread.Sleep 5000
    42

dbrattli

// A (synchronous) function returning a value
let func () =
    42
    
let value = func ()

dbrattli

// A (synchronous) value
let value = 42

dbrattli

type IDisposable =
    abstract member Dispose: unit -> unit

type IObserver<'a> =
    abstract member OnNext: 'a -> unit
    abstract member OnError: exn -> unit
    abstract member OnCompleted: unit -> unit

type IObservable<'a> =
    abstract member Subscribe: IObserver<'a> -> IDisposable