combineLatest

Author: eulerfx

src/FSharp.Control.AsyncSeq/AsyncSeq.fs

@@ -887,7 +887,7 @@ module AsyncSeq =
 
   let zapp (fs:AsyncSeq<'T -> 'U>) (s:AsyncSeq<'T>) : AsyncSeq<'U> =
       zipWith (|>) s fs
-
+    
   let takeWhileAsync p (source : AsyncSeq<'T>) : AsyncSeq<_> = asyncSeq {
       use ie = source.GetEnumerator() 
       let! move = ie.MoveNext()
@@ -1095,9 +1095,7 @@ module AsyncSeq =
       yield! loop None timeoutMs
     }
 
-  let mergeChoice (source1:AsyncSeq<'T1>) (source2:AsyncSeq<'T2>) : AsyncSeq<Choice<'T1,'T2>> = asyncSeq {
-      use ie1 = source1.GetEnumerator() 
-      use ie2 = source2.GetEnumerator() 
+  let private mergeChoiceEnum (ie1:IAsyncEnumerator<'T1>) (ie2:IAsyncEnumerator<'T2>) : AsyncSeq<Choice<'T1,'T2>> = asyncSeq {
       let! move1T = Async.StartChildAsTask (ie1.MoveNext())
       let! move2T = Async.StartChildAsTask (ie2.MoveNext())
       let! move = Async.chooseTasks move1T move2T
@@ -1134,6 +1132,10 @@ module AsyncSeq =
               b1 := move1n 
       | _ -> failwith "unreachable" }
 
+  let mergeChoice (source1:AsyncSeq<'T1>) (source2:AsyncSeq<'T2>) : AsyncSeq<Choice<'T1,'T2>> = asyncSeq {
+      use ie1 = source1.GetEnumerator() 
+      use ie2 = source2.GetEnumerator()
+      yield! mergeChoiceEnum ie1 ie2 }
 
   let merge (source1:AsyncSeq<'T>) (source2:AsyncSeq<'T>) : AsyncSeq<'T> = 
     mergeChoice source1 source2 |> map (function Choice1Of2 x -> x | Choice2Of2 x -> x)
@@ -1174,6 +1176,37 @@ module AsyncSeq =
                   fin := fin.Value - 1
       }
 
+  let combineLatestAsync (f:'a -> 'b -> Async<'c>) (source1:AsyncSeq<'a>) (source2:AsyncSeq<'b>) : AsyncSeq<'c> =
+    asyncSeq {
+      use en1 = source1.GetEnumerator()
+      use en2 = source2.GetEnumerator()
+      let! a = Async.StartChild (en1.MoveNext())
+      let! b = Async.StartChild (en2.MoveNext())
+      let! a = a
+      let! b = b      
+      match a,b with
+      | Some a, Some b ->
+        let! c = f a b
+        yield c        
+        let merged = mergeChoiceEnum en1 en2
+        use mergedEnum = merged.GetEnumerator()
+        let rec loop (prevA:'a, prevB:'b) = asyncSeq {
+          let! next = mergedEnum.MoveNext ()
+          match next with
+          | None -> ()
+          | Some (Choice1Of2 nextA) ->
+            let! c = f nextA prevB
+            yield c
+            yield! loop (nextA,prevB)                                  
+          | Some (Choice2Of2 nextB) ->
+            let! c = f prevA nextB
+            yield c
+            yield! loop (prevA,nextB) }
+        yield! loop (a,b)
+      | _ -> () }
+    
+  let combineLatest (f:'a -> 'b -> 'c) (source1:AsyncSeq<'a>) (source2:AsyncSeq<'b>) : AsyncSeq<'c> =
+    combineLatestAsync (fun a b -> f a b |> async.Return) source1 source2
 
   let distinctUntilChangedWithAsync (f:'T -> 'T -> Async<bool>) (source:AsyncSeq<'T>) : AsyncSeq<'T> = asyncSeq {
       use ie = source.GetEnumerator() 
@@ -1242,7 +1275,6 @@ module AsyncSeq =
          return Choice1Of2 (asyncSeq { for v in res do yield v })
        }
 
-
   module AsyncSeqSrcImpl =
 
     let private createNode () =

src/FSharp.Control.AsyncSeq/AsyncSeq.fsi

@@ -294,6 +294,16 @@ module AsyncSeq =
     /// Feeds an async sequence of values into an async sequence of functions.
     val zapp : functions:AsyncSeq<('T -> 'U)> -> source:AsyncSeq<'T> -> AsyncSeq<'U>
 
+    /// Merges two async sequences using the specified combine function. The resulting async sequence produces an element when either
+    /// input sequence produces an element, passing the new element from the emitting sequence and the previously emitted element from the other sequence.
+    /// If either of the input sequences is empty, the resulting sequence is empty.
+    val combineLatestAsync : combine:('T -> 'U -> Async<'V>) -> source1:AsyncSeq<'T> -> source2:AsyncSeq<'U> -> AsyncSeq<'V>
+
+    /// Merges two async sequences using the specified combine function. The resulting async sequence produces an element when either
+    /// input sequence produces an element, passing the new element from the emitting sequence and the previously emitted element from the other sequence.
+    /// If either of the input sequences is empty, the resulting sequence is empty.
+    val combineLatest : combine:('T -> 'U -> 'V) -> source1:AsyncSeq<'T> -> source2:AsyncSeq<'U> -> AsyncSeq<'V>
+
     /// Traverses an async sequence an applies to specified function such that if None is returned the traversal short-circuits
     /// and None is returned as the result. Otherwise, the entire sequence is traversed and the result returned as Some.
     val traverseOptionAsync : mapping:('T -> Async<'U option>) -> source:AsyncSeq<'T> -> Async<AsyncSeq<'U> option>

tests/FSharp.Control.AsyncSeq.Tests/AsyncSeqTests.fs

@@ -35,18 +35,23 @@ let EQ (a:AsyncSeq<'a>) (b:AsyncSeq<'a>) =
 
 type Assert with  
   /// Determines equality of two async sequences by convering them to lists, ignoring side-effects.
+  static member AreEqual (expected:AsyncSeq<'a>, actual:AsyncSeq<'a>, message:string) =
+    Assert.AreEqual (expected, actual, 2000, exnEq=(fun _ _ -> true), message=message)
+  /// Determines equality of two async sequences by convering them to lists, ignoring side-effects.
   static member AreEqual (expected:AsyncSeq<'a>, actual:AsyncSeq<'a>) =
-    Assert.AreEqual (expected, actual, 1000, exnEq=(fun _ _ -> true))
+    Assert.AreEqual (expected, actual, 2000, exnEq=(fun _ _ -> true), message=null)
   /// Determines equality of two async sequences by convering them to lists, ignoring side-effects.
   static member AreEqual (expected:AsyncSeq<'a>, actual:AsyncSeq<'a>, timeout) =
-    Assert.AreEqual (expected, actual, timeout=timeout, exnEq=(fun _ _ -> true))
+    Assert.AreEqual (expected, actual, timeout=timeout, exnEq=(fun _ _ -> true), message=null)
   /// Determines equality of two async sequences by convering them to lists, ignoring side-effects.
-  static member AreEqual (expected:AsyncSeq<'a>, actual:AsyncSeq<'a>, timeout, exnEq:exn -> exn -> bool) =
+  static member AreEqual (expected:AsyncSeq<'a>, actual:AsyncSeq<'a>, timeout, exnEq:exn -> exn -> bool, message:string) =
     let exp = expected |> AsyncSeq.toListAsync |> Async.Catch
     let exp = Async.RunSynchronously (exp, timeout)
     let act = actual |> AsyncSeq.toListAsync |> Async.Catch 
     let act = Async.RunSynchronously(act, timeout)
-    let message = sprintf "expected=%A actual=%A" exp act
+    let message = 
+      if message = null then sprintf "expected=%A actual=%A" exp act
+      else sprintf "message=%s expected=%A actual=%A" message exp act
     match exp,act with
     | Choice1Of2 exp, Choice1Of2 act ->
       Assert.True((exp = act), message)
@@ -1219,4 +1224,19 @@ let ``AsyncSeq.groupBy should propagate exception and terminate all groups``() =
     asyncSeq { raise (exn("test")) } 
     |> AsyncSeq.groupBy (fun i -> i % 3) 
     |> AsyncSeq.mapAsyncParallel (snd >> AsyncSeq.toListAsync)
-  Assert.AreEqual(expected, actual)
+  Assert.AreEqual(expected, actual)
+
+[<Test>]
+let ``AsyncSeq.combineLatest should behave like merge after initial``() =  
+  for n in 0..20 do
+    for m in 0..10 do
+      let ls1 = List.init n id
+      let ls2 = List.init m id 
+      // expect each element to increase combined sum by 1
+      // expected count is sum of source counts minus 1 for first result
+      let expectedCount = 
+        if n = 0 || m = 0 then 0
+        else max (n + m - 1) 0
+      let expected = List.init expectedCount id |> AsyncSeq.ofSeq
+      let actual = AsyncSeq.combineLatest (+) (AsyncSeq.ofSeq ls1) (AsyncSeq.ofSeq ls2)
+      Assert.AreEqual(expected, actual, (sprintf "n=%i m=%i" n m))