AsyncSeq.takeUntil, AsyncSeq.skipUntil

Author: eulerfx

src/AsyncSeq/Async.fs

@@ -12,6 +12,8 @@ open System.Threading.Tasks
 module AsyncOps =
 
   let unit : Async<unit> = async.Return()
+
+  let never : Async<unit> = Async.Sleep -1
 
 
 [<AutoOpen>]
@@ -105,9 +107,12 @@ module AsyncExtensions =
 
         }
 
-      /// An async computation which does nothing.
+      /// An async computation which does nothing and completes immediatly.
       static member inline unit = AsyncOps.unit
 
+      /// An async computation which does nothing and never completes.
+      static member inline never = AsyncOps.never
+
       /// Creates an async computation which maps a function f over the 
       /// value produced by the specified asynchronous computation.
       static member inline map f a = async.Bind(a, f >> async.Return)

src/AsyncSeq/AsyncSeq.fs

@@ -518,6 +518,18 @@ module AsyncSeq =
         else return Nil
     | Nil -> return Nil }
 
+  /// Returns elements from the argument async sequence until the specified signal completes or
+  /// the sequences completes.
+  let rec takeUntil (signal:Async<unit>) (s:AsyncSeq<'a>) : AsyncSeq<'a> =
+    Async.chooseBoth (signal |> Async.map Choice1Of2) (s |> Async.map Choice2Of2)
+    |> Async.map (fun (first,second) ->
+      match first with
+      | Choice1Of2 _ -> Nil
+      | Choice2Of2 Nil -> Nil
+      | Choice2Of2 (Cons(a,tl)) ->        
+        let signal = second |> Async.map (function Choice1Of2 x -> x | _ -> failwith "unexpected state")
+        Cons(a, takeUntil signal tl))
+
   /// Skips elements from an asynchronous sequence while the specified 
   /// predicate holds and then returns the rest of the sequence. The 
   /// predicate is evaluated asynchronously.
@@ -530,10 +542,21 @@ module AsyncSeq =
         else return v
     | Nil -> return Nil }
 
+  /// Skips elements from an async sequence until the specified signal completes.
+  let rec skipUntil (signal:Async<unit>) (s:AsyncSeq<'a>) : AsyncSeq<'a> =
+    Async.chooseBoth (signal |> Async.map Choice1Of2) (s |> Async.map Choice2Of2)
+    |> Async.bind (fun (first,second) ->
+      match first with
+      | Choice1Of2 _ -> second |> Async.map (function Choice2Of2 tl -> tl | _ -> failwith "unexpected state")
+      | Choice2Of2 Nil -> Nil |> async.Return
+      | Choice2Of2 (Cons(_,tl)) -> 
+        let signal = second |> Async.map (function Choice1Of2 x -> x | _ -> failwith "unexpected state")
+        skipUntil signal tl)
+
   /// Returns elements from an asynchronous sequence while the specified 
   /// predicate holds. The predicate is evaluated synchronously.
   let rec takeWhile p (input : AsyncSeq<'T>) = 
-    takeWhileAsync (p >> async.Return) input
+    takeWhileAsync (p >> async.Return) input  
 
   /// Skips elements from an asynchronous sequence while the specified 
   /// predicate holds and then returns the rest of the sequence. The 

tests/AsyncSeq.Tests/AsyncSeqTests.fs

@@ -249,4 +249,35 @@ let ``AsyncSeq.distinctUntilChangedWithAsync``() =
     else false |> async.Return
   let actual = s |> AsyncSeq.distinctUntilChangedWithAsync c
   let expected = [1;2;3;4;5;1] |> AsyncSeq.ofSeq
-  Assert.True(EQ expected actual)
+  Assert.True(EQ expected actual)
+
+
+[<Test>]
+let ``AsyncSeq.takeUntil should complete immediately with completed signal``() =  
+  let s = asyncSeq {
+    do! Async.Sleep 10
+    yield 1
+    yield 2
+  }
+  let actual = AsyncSeq.takeUntil Async.unit s
+  Assert.True(EQ AsyncSeq.empty actual)
+
+
+[<Test>]
+let ``AsyncSeq.takeUntil should take entire sequence with never signal``() =
+  let expected = [1;2;3;4] |> AsyncSeq.ofSeq
+  let actual = expected |> AsyncSeq.takeUntil Async.never
+  Assert.True(EQ expected actual)
+
+
+[<Test>]
+let ``AsyncSeq.skipUntil should not skip with completed signal``() =
+  let expected = [1;2;3;4] |> AsyncSeq.ofSeq
+  let actual = expected |> AsyncSeq.skipUntil Async.unit
+  Assert.True(EQ expected actual)
+
+
+[<Test>]
+let ``AsyncSeq.skipUntil should skip everything with never signal``() =
+  let actual = [1;2;3;4] |> AsyncSeq.ofSeq |> AsyncSeq.skipUntil Async.never
+  Assert.True(EQ AsyncSeq.empty actual)