reactiveDictionary.fs

open System.Collections.Generic
open System


type Notifications<'key, 'item> =
    | Add of addKey : 'key * addItem : 'item
    | Update of updateKey : 'key *  updateItem : 'item
    | Remove of removeKey : 'key
    | RemoveAll

type Operations<'key, 'item> =
| Notification of notification:Notifications<'key, 'item>
| TryGet of tryKey : 'key * reply : AsyncReplyChannel<'item option>
| Subscribe of subscribe: IObserver<Notifications<'key, 'item>>
| Unsubscribe of unsubscribe: IObserver<Notifications<'key, 'item>>

type IReadOnlyReactiveDictionary<'key, 'item> =
    abstract member TryGet : 'key -> Async<'item option>
    inherit IObservable<Notifications<'key, 'item>>

type IReactiveDictionary<'key, 'item> =
    abstract member Add : 'key -> 'item -> unit
    abstract member Update : 'key -> 'item -> unit
    abstract member Remove : 'key -> unit
    abstract member RemoveAll : unit -> unit

let reactiveDictionary<'a,'b when 'a : equality>() =
    let actor = MailboxProcessor.Start (fun inbox ->
        let subscriptions = ResizeArray ()
        let dictionary = Dictionary ()
        let rec loop () = async {
            let! msg = inbox.Receive ()
            try
                match msg with
                | Subscribe s -> subscriptions.Add s
                | Unsubscribe s -> subscriptions.Remove s |> ignore
                | TryGet (key, reply) ->
                    match dictionary.TryGetValue key with
                    | true, v -> Some v
                    | _ -> None
                    |> reply.Reply
                | Notification notification ->
                    match notification with
                    | Add (key, value)
                    | Update (key, value) -> dictionary.[key]<-value
                    | Remove key -> dictionary.Remove key |> ignore
                    | RemoveAll -> dictionary.Clear ()

                    for s in subscriptions do s.OnNext notification
            with e -> eprintf "%A" e
            return! loop () }
        loop ())

    let reader =
        { new IReadOnlyReactiveDictionary<'a,'b>  with
            member __.TryGet key = actor.PostAndAsyncReply (fun x-> TryGet (key, x))
            member __.Subscribe s =
                Subscribe s |> actor.Post
                { new IDisposable with
                    member __.Dispose () = Unsubscribe s |> actor.Post } }
    let writer =
        { new IReactiveDictionary<'a,'b>  with
            member __.Add key value = Add (key, value) |> Notification |> actor.Post
            member __.Update  key value = Update (key, value) |> Notification |> actor.Post
            member __.Remove key = Remove key |> Notification |> actor.Post
            member __.RemoveAll () = RemoveAll |> Notification |> actor.Post
        }
    reader, writer

let r,w = reactiveDictionary<int,int>()

r.Subscribe null
w.Add 2 2