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

/// 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

// 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

// 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

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

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

dbrattli

type OnNext<'a> =  'a -> unit

type Observable<'a> =
    Observable of (OnNext<'a> -> unit) with

    static member Single (value: 'a) =
        let subscribe (onNext: OnNext<'a>) =
            onNext value
        Observable subscribe

dbrattli

type IAsyncDisposable =
    abstract member DisposeAsync: unit -> Async<unit>

type IAsyncEnumerator<'a> =
    inherit IDisposable
    abstract member Current : 'a with get
    abstract member MoveNextAsync : unit -> Async<bool>
    
type IAsyncEnumerable<'a> =
    abstract member GetAsyncEnumerator : unit -> IAsyncEnumerator<'a>

dbrattli

type IAsyncDisposable =
    abstract member DisposeAsync: unit -> Async<unit>

type IAsyncObserver<'a> =
    abstract member OnNextAsync: 'a -> Async<unit>
    abstract member OnErrorAsync: exn -> Async<unit>
    abstract member OnCompletedAsync: unit -> Async<unit>

type IAsyncObservable<'a> =
    abstract member SubscribeAsync: IAsyncObserver<'a> -> Async<IAsyncDisposable>

dbrattli

type Cont<'a> =  'a -> unit

type Continuation<'a> =
    Continuation of (Cont<'a> -> unit) with

    static member Return (value: 'a) =
        let run (cont: Cont<'a>) =
            cont value
        Continuation run