Merge pull request #84 from sajagi/map2

Added Observable.map2 and Store.map2 functions

src/Sutil/Observable.fs

@@ -24,33 +24,23 @@ module Observable =
             member x.OnCompleted () =
                 if not stopped then stopped <- true; x.Completed ()
 
-    let zip<'A,'B> (a:IObservable<'A>) (b:IObservable<'B>) : IObservable<'A*'B> =
-        { new System.IObservable<'A*'B> with
-            member _.Subscribe( h : IObserver<'A*'B> ) =
-                let mutable valueA = None
-                let mutable valueB = None
+    let map2<'A, 'B, 'Res> (f: 'A -> 'B -> 'Res) (a: IObservable<'A>) (b: IObservable<'B>) : IObservable<'Res> =
+        { new IObservable<'Res> with
+            member _.Subscribe(h: IObserver<'Res>) =
+                let mutable valueA, valueB = None, None
 
                 let notify() =
-                    // unfortunately, there's no Option.iter2
-                    Option.map2 (fun a b -> h.OnNext(a, b)) valueA valueB
-                    |> ignore
+                     if valueA.IsSome && valueB.IsSome then h.OnNext (f valueA.Value valueB.Value)
 
-                let disposeA = a.Subscribe( fun v ->
-                    valueA <- Some v
-                    notify()
-                )
+                let disposeA = a.Subscribe(fun v -> valueA <- Some v; notify())
+                let disposeB = b.Subscribe(fun v -> valueB <- Some v; notify())
 
-                let disposeB = b.Subscribe( fun v ->
-                    valueB <- Some v
-                    notify()
-                )
-
-                Helpers.disposable (fun _ ->
-                    disposeA.Dispose()
-                    disposeB.Dispose()
-                )
+                Helpers.disposable(fun _ -> disposeA.Dispose(); disposeB.Dispose())
         }
 
+    let zip<'A,'B> (a:IObservable<'A>) (b:IObservable<'B>) : IObservable<'A*'B> =
+        map2<'A, 'B, 'A * 'B> (fun a b -> a, b) a b
+
     let distinctUntilChangedCompare<'T> (eq:'T -> 'T -> bool) (source:IObservable<'T>) : IObservable<'T> =
         { new System.IObservable<'T> with
             member _.Subscribe( h : IObserver<'T> ) =

src/Sutil/Store.fs

@@ -90,6 +90,21 @@ module Store =
     /// </example>
     let map<'A, 'B> (callback: 'A -> 'B) (store: IObservable<'A>) = store |> Observable.map callback
 
+    /// <summary>
+    /// Returns an observable that will resolve to the result of said callback applied to two observables
+    /// </summary>
+    /// <example>
+    /// <code>
+    ///     let subscription: IObservable&lt;int&gt; =
+    ///         Store.map2 (fun value1 value2 -> value1 * value2) intStore1 intStore2
+    ///
+    ///     (* after you are done with the subscription *)
+    ///
+    ///     subscription.Dispose()
+    /// </code>
+    /// </example>
+    let map2<'A, 'B, 'Res> (callback: 'A -> 'B -> 'Res) (storeA: IObservable<'A>) (storeB: IObservable<'B>) = Observable.map2<'A, 'B, 'Res> callback storeA storeB
+
     /// <summary>
     /// Applies a predicate function to obtain an observable of the elements that evaluated to true
     /// </summary>

tests/test/ObservableTest.fs

@@ -47,6 +47,28 @@ describe "Sutil.Observable" <| fun () ->
         Expect.areEqual(3,n2) // - a b
     }
 
+    it "map2" <| fun () -> promise {
+        let s1 = Store.make 3
+        let s2 = Store.make 5
+
+        let mutable result = 0
+        let mapped = (s1, s2) ||> Observable.map2 (fun s1 s2 -> s1 * s2)
+        use _ = Store.subscribe (fun x -> result <- x) mapped
+
+        Expect.areEqual(15, result)
+    }
+
+    it "zip" <| fun () -> promise {
+        let s1 = Store.make 3
+        let s2 = Store.make 5
+
+        let mutable result = (0,0)
+        let mapped = (s1, s2) ||> Observable.zip
+        use _ = Store.subscribe (fun x -> result <- x) mapped
+
+        Expect.areEqual((3,5), result)
+    }
+
     //it "exists" <| fun () ->
     //    let s1 = Store.make '-'
     //    let s2 =