Program.fs

open System
open System.Collections.Concurrent
open System.Collections.Generic
open System.Threading
open Microsoft.FSharp.Reflection

type IObservable = obj

type DeferredValueException() =
    inherit Exception()

type ObservableException(message : string, innerException : Exception) =
    inherit Exception(message, innerException)

type IObservableLastValue<'a> =
    abstract LastValue : Choice<Exception, 'a> option

type Monitor(values : ConcurrentDictionary<IObservable, Choice<Exception, obj>>) =
    [<ThreadStatic; DefaultValue>]
    static val mutable private current : Monitor option

    let subscriptions = HashSet<IObservable>()

    static member Current : Monitor =
        match Monitor.current with
        | Some m -> m
        | None -> failwithf "Not inside Observable.Computed"
        
    static member SetCurrent(m : Monitor) : IDisposable =
        let oldCurrent = Monitor.current
        Monitor.current <- Some m
        { new IDisposable with member t.Dispose() = Monitor.current <- oldCurrent }

    member t.GetValue<'a>(obs : IObservable<'a>) : 'a =
        ignore (subscriptions.Add(obs))
        let value =
            match obs with
            | :? IObservableLastValue<'a> as lv -> lv.LastValue
            | _ ->
                match values.TryGetValue(obs) with
                | true, Choice1Of2 ex -> Some (Choice1Of2 ex)
                | true, Choice2Of2 value -> Some (Choice2Of2 (unbox value))
                | false, _ -> None

        match value with
        | Some (Choice1Of2 ex) -> raise (ObservableException(ex.Message, ex))
        | Some (Choice2Of2 value) -> value
        | None -> raise (DeferredValueException())
        
    member t.Subscriptions = subscriptions

type Observable<'a>() =
    let syncRoot = obj()
    let mutable subs : IObserver<'a> list = [ ]
    let mutable lastValue : Choice<Exception, 'a> option = None

    let rec removeFirst value =
        function
        | x :: xs when x = value -> xs
        | _ :: xs -> removeFirst value xs
        | [ ] -> [ ]

    member t.Publish(value) =
        let subs =
            lock syncRoot <| fun () ->
                lastValue <- Some (Choice2Of2 value)
                subs

        for sub in subs do sub.OnNext(value)

    member t.PublishError(ex) =
        let subs =
            lock syncRoot <| fun () ->
                lastValue <- Some (Choice1Of2 ex)
                subs

        for sub in subs do sub.OnError(ex)

    interface IObservable<'a> with
        member t.Subscribe(sub) =
            lock syncRoot <| fun () ->
                subs <- sub :: subs

            { new IDisposable with
                member x.Dispose() =
                    lock syncRoot <| fun () ->
                        subs <- removeFirst sub subs
            }

    interface IObservableLastValue<'a> with
        member t.LastValue = lastValue

type Observer<'a>(next : 'a -> unit, error : Exception -> unit) =
    interface IObserver<'a> with
        member t.OnNext(value) =
            next value
            
        member t.OnError(ex) =
            error ex
            
        member t.OnCompleted() =
            ()

module private Impl =

    let memoize (fn : 'a -> 'b) : 'a -> 'b =
        let dict = Dictionary()
        fun a ->
            let found, b = lock dict <| fun () -> dict.TryGetValue(a)
            if found then
                b
            else
                let b = fn a
                lock dict <| fun () -> dict.[a] <- b
                b

    let subscribeAny : Type -> (IObservable * (obj -> unit) * (Exception -> unit)) -> IDisposable =
        memoize <| fun typ ->
            let finder (iface : Type) =
                iface.IsGenericType && iface.GetGenericTypeDefinition() = typedefof<IObservable<_>>

            let iface =
                match Array.tryFind finder (typ.GetInterfaces()) with
                | Some typ -> typ
                | None -> failwithf "%O does not implement IObservable<T>" typ
                
            let methd = iface.GetMethod("Subscribe")
            let valueType = iface.GetGenericArguments().[0]
            let functionType = FSharpType.MakeFunctionType(valueType, typeof<unit>)
            let observerType = typedefof<Observer<_>>.MakeGenericType([| valueType |])
            let observerCtor = observerType.GetConstructor([| functionType; typeof<Exception -> unit> |])

            fun (observable, next, error) ->
                let next = FSharpValue.MakeFunction(functionType, next >> box)
                let observerObj = observerCtor.Invoke([| next; error |])
                unbox (methd.Invoke(observable, [| observerObj |]))
        
module Observable =

    let computed<'a> (fn : unit -> 'a) : Observable<'a> =
        let values = ConcurrentDictionary()
        let syncRoot = obj()
        let subs = Dictionary<IObservable, IDisposable>()
        let observable = Observable<'a>()

        let rec refresh () =
            let value, obses =
                let m = Monitor(values)
                let value =
                    using (Monitor.SetCurrent(m)) <| fun _ ->
                        try
                            Some (Choice2Of2 (fn ()))
                        with
                        | :? DeferredValueException -> None
                        | :? ObservableException as ex -> Some (Choice1Of2 ex.InnerException)
                        | ex -> Some (Choice1Of2 ex)

                value, m.Subscriptions

            lock syncRoot <| fun () ->
                for pair in Array.ofSeq subs do
                    if not (obses.Contains(pair.Key)) then
                        pair.Value.Dispose()
                        ignore (subs.Remove(pair.Key))

                for obs in obses do
                    if not (subs.ContainsKey(obs)) then
                        let on value =
                            values.[obs] <- value
                            refresh ()

                        subs.[obs] <- Impl.subscribeAny (obs.GetType()) (obs, Choice2Of2 >> on, Choice1Of2 >> on)

            match value with
            | Some (Choice1Of2 ex) -> observable.PublishError(ex)
            | Some (Choice2Of2 value) -> observable.Publish(value)
            | None -> ()

        refresh ()
        observable
        

[<AutoOpen>]
module ObservableExtensions =

    type IObservable<'a> with
        member t.Value : 'a =
            Monitor.Current.GetValue(t)

        member t.Subscribe(next : 'a -> unit, error : Exception -> unit) : IDisposable =
            t.Subscribe(Observer(next, error))

[<EntryPoint>]
let main args =
    let obs1 = Observable()
    use d1 = obs1.Subscribe(printfn "Got obs1: %d", fun ex -> printfn "Error obs1: %s" ex.Message)
    let obs2 = Observable()
    use d2 = obs2.Subscribe (printfn "Got obs2: %d", fun ex -> printfn "Error obs2: %s" ex.Message)
    let c =
        Observable.computed <| fun () ->
            if obs2.Value < 10 then
                failwithf "%d is too low" obs2.Value
            else
                obs1.Value + obs2.Value

    use dc = c.Subscribe (printfn "Got computed: %d", fun ex -> printfn "Error computed: %s" ex.Message)
    obs1.Publish(10)

    let r = Random()
    while true do
        obs2.Publish(r.Next(0, 20))
        Thread.Sleep(1000)

    0