AMap deltaA implementation

Author: TheAngryByrd

src/FSharp.Data.Adaptive/AdaptiveHashMap/AdaptiveHashMap.fs

@@ -755,7 +755,85 @@ module AdaptiveHashMapImplementation =
                 changes <- HashMap.add i (Set v) changes
 
             HashMapDelta.ofHashMap changes
-     
+
+    /// Reader for deltaA operations.
+    [<Sealed>]
+    type DeltaAReader<'k, 'a, 'b>(input : amap<'k, 'a>, mapping : HashMap<'k,'a> -> HashMap<'k, aval<'b>>) =
+        inherit AbstractReader<HashMapDelta<'k, 'b>>(HashMapDelta.empty)
+
+        let reader = input.GetReader()
+        do reader.Tag <- "input"
+        let cacheLock = obj()
+        let mutable cache: HashMap<'k, aval<'b>> = HashMap.Empty
+        let mutable targets = MultiSetMap.empty<aval<'b>, 'k>
+        let mutable dirty = HashMap.empty<'k, aval<'b>>
+
+        let consumeDirty() =
+            lock cacheLock (fun () ->
+                let d = dirty
+                dirty <- HashMap.empty
+                d
+            )
+
+        override x.InputChangedObject(t, o) =
+            #if FABLE_COMPILER
+            if isNull o.Tag then
+                let o = unbox<aval<'b>> o
+                for i in MultiSetMap.find o targets do
+                    dirty <- HashMap.add i o dirty
+            #else
+            match o with
+            | :? aval<'b> as o ->
+                lock cacheLock (fun () ->
+                    for i in MultiSetMap.find o targets do
+                        dirty <- HashMap.add i o dirty
+                )
+            | _ ->
+                ()
+            #endif
+
+        override x.Compute t =
+            let mutable dirty = consumeDirty()
+            let old = reader.State
+            let ops = reader.GetChanges t |> HashMapDelta.toHashMap
+
+            let setOps, removeOps =
+                ((HashMap.empty, HashMap.empty), ops)
+                ||> HashMap.fold(fun (sets, rems) i op ->
+                    dirty <- HashMap.remove i dirty
+                    cache <- 
+                        match HashMap.tryRemove i cache with
+                        | Some (o, remaingCache) -> 
+                            let rem, rest = MultiSetMap.remove o i targets
+                            targets <- rest
+                            if rem then o.Outputs.Remove x |> ignore
+                            remaingCache
+                        | None -> cache
+                    match op with
+                    | Set v -> 
+                        HashMap.add i v sets, rems
+                    | Remove -> 
+                        sets, HashMap.add i Remove rems
+                )
+            
+            let mutable changes =
+                setOps
+                |> mapping
+                |> HashMap.map(fun i k ->
+                    cache <- HashMap.add i k cache
+                    let v = k.GetValue t
+                    targets <- MultiSetMap.add k i targets
+                    Set v
+                )
+
+
+            for i, d in dirty do
+                let v = d.GetValue t
+                changes <- HashMap.add i (Set v) changes
+
+            HashMap.union removeOps changes
+            |> HashMapDelta
+
     /// Reader for chooseA operations.
     [<Sealed>]
     type ChooseAReader<'k, 'a, 'b>(input : amap<'k, 'a>, mapping : 'k -> 'a -> aval<Option<'b>>) =
@@ -1333,6 +1411,19 @@ module AMap =
         else
             create (fun () -> MapAReader(map, mapping))
 
+    /// Adaptively applies the given mapping to all changes.
+    let deltaA (mapping: HashMap<'K,'T1> -> HashMap<'K,aval<'T2>>) (map: amap<'K, 'T1>) =
+        if map.IsConstant then
+            let map = force map |> mapping
+            if map |> HashMap.forall (fun _ v -> v.IsConstant) then
+                constant (fun () -> map |> HashMap.map (fun _ v -> AVal.force v))
+            else
+                // TODO better impl possible
+                create (fun () -> MapAReader(ofHashMap map, fun _ v -> v))
+        else
+            create (fun () -> DeltaAReader(map, mapping))
+
+
     /// Adaptively chooses all elements returned by mapping.  
     let chooseA (mapping: 'K ->'T1 -> aval<Option<'T2>>) (map: amap<'K, 'T1>) =
         if map.IsConstant then

src/FSharp.Data.Adaptive/AdaptiveHashMap/AdaptiveHashMap.fsi

@@ -102,10 +102,12 @@ module AMap =
     /// Adaptively intersects the two maps.
     val intersectV : amap<'Key, 'Value1> -> amap<'Key, 'Value2> -> amap<'Key, struct('Value1 * 'Value2)>
 
-
     /// Adaptively applies the given mapping function to all elements and returns a new amap containing the results.
     val mapA : mapping: ('K -> 'V -> aval<'T>) -> map: amap<'K, 'V> -> amap<'K, 'T>
 
+    /// Adaptively applies the given mapping to all changes.
+    val deltaA : mapping: (HashMap<'K,'T1> -> HashMap<'K,aval<'T2>>) -> map: amap<'K, 'T1> -> amap<'K, 'T2>
+
     /// Adaptively chooses all elements returned by mapping.  
     val chooseA : mapping: ('K -> 'V -> aval<option<'T>>) -> map: amap<'K, 'V> -> amap<'K, 'T>
 

src/Test/FSharp.Data.Adaptive.Tests/AMap.fs

@@ -638,4 +638,37 @@ let ``[AMap] mapA``() =
         flag.Value <- true
     )
 
+    res |> AMap.force |> should equal (HashMap.ofList ["A", 2; "B", 4; "C", 6])
+
+
+    
+[<Test>]
+let ``[AMap] deltaA``() =
+    let map = cmap ["A", 1; "B", 2; "C", 3]
+    let flag = cval true
+
+    let res =
+        map |> AMap.deltaA (fun d ->
+            d
+            |> HashMap.map(fun _ v -> flag |> AVal.map (function true -> v | false -> -1))
+        )
+
+    res |> AMap.force |> should equal (HashMap.ofList ["A", 1; "B", 2; "C", 3])
+
+    transact (fun () ->
+        flag.Value <- false
+    )
+
+    res |> AMap.force |> should equal (HashMap.ofList ["A", -1; "B", -1; "C", -1])
+
+    transact (fun () ->
+        map.Value <- map.Value |> HashMap.map (fun _ v -> v * 2)
+    )
+
+    res |> AMap.force |> should equal (HashMap.ofList ["A", -1; "B", -1; "C", -1])
+
+    transact (fun () ->
+        flag.Value <- true
+    )
+
     res |> AMap.force |> should equal (HashMap.ofList ["A", 2; "B", 4; "C", 6])