Merge pull request #143 from channaj/sort

Sort functions

Author: dsyme

src/FSharp.Control.AsyncSeq/AsyncSeq.fs

@@ -50,7 +50,7 @@ module internal Utils =
         { new IDisposable with member __.Dispose () = () }
 
     // ----------------------------------------------------------------------------
-    
+
     #if FABLE_COMPILER
     type ExceptionDispatchInfo private (err: exn) =
         member _.SourceException = err
@@ -214,7 +214,7 @@ module AsyncGenerator =
       member x.Disposer =
         g.Disposer
 
-    
+
     static member Bind (g:AsyncGenerator<'a>, cont:unit -> AsyncGenerator<'a>) : AsyncGenerator<'a> =
       #if !FABLE_COMPILER
       match g with
@@ -1062,7 +1062,7 @@ module AsyncSeq =
         mb.Post (Some b) })
       |> Async.map (fun _ -> mb.Post None)
       |> Async.StartChildAsTask
-    
+
     return!
       replicateUntilNoneAsync (Task.chooseTask (err |> Task.taskFault) (async.Delay mb.Receive))
       |> iterAsync id }
@@ -1483,6 +1483,21 @@ module AsyncSeq =
       if (buffer.Count > 0) then
           yield buffer.ToArray() }
 
+  let toSortedSeq fn source =
+    toArrayAsync source |> Async.map fn |> Async.RunSynchronously
+
+  let sort (source:AsyncSeq<'T>) : array<'T> when 'T : comparison =
+    toSortedSeq Array.sort source
+
+  let sortBy (projection:'T -> 'Key) (source:AsyncSeq<'T>) : array<'T> when 'Key : comparison =
+    toSortedSeq (Array.sortBy projection) source
+
+  let sortDescending (source:AsyncSeq<'T>) : array<'T> when 'T : comparison =
+    toSortedSeq Array.sortDescending source
+
+  let sortByDescending (projection:'T -> 'Key) (source:AsyncSeq<'T>) : array<'T> when 'Key : comparison =
+    toSortedSeq (Array.sortByDescending projection) source
+
   #if !FABLE_COMPILER
   let bufferByCountAndTime (bufferSize:int) (timeoutMs:int) (source:AsyncSeq<'T>) : AsyncSeq<'T[]> =
     if (bufferSize < 1) then invalidArg "bufferSize" "must be positive"

src/FSharp.Control.AsyncSeq/AsyncSeq.fsi

@@ -437,6 +437,30 @@ module AsyncSeq =
     /// Flattens an AsyncSeq of asynchronous sequences.
     val concat : AsyncSeq<AsyncSeq<'T>> -> AsyncSeq<'T>
 
+    /// Yields a sequence ordered by keys.
+    /// This function returns a sequence that digests the whole initial sequence as soon as
+    /// that sequence is iterated. As a result this function should not be used with
+    /// large or infinite sequences.
+    val sort : source:AsyncSeq<'T> -> array<'T> when 'T : comparison
+
+    /// Applies a key-generating function to each element of an AsyncSeq and yield an array ordered by keys.
+    /// This function returns an array that digests the whole initial sequence as soon as
+    /// that sequence is iterated. As a result this function should not be used with
+    /// large or infinite sequences.
+    val sortBy : projection:('T -> 'Key) -> source:AsyncSeq<'T> -> array<'T> when 'Key : comparison
+
+    /// Yields an array ordered descending by keys.
+    /// This function returns an array that digests the whole initial sequence as soon as
+    /// that sequence is iterated. As a result this function should not be used with
+    /// large or infinite sequences.
+    val sortDescending : source:AsyncSeq<'T> -> array<'T> when 'T : comparison
+
+    /// Applies a key-generating function to each element of an AsyncSeq and yield an array ordered descending by keys.
+    /// This function returns an array that digests the whole initial sequence as soon as
+    /// that sequence is iterated. As a result this function should not be used with
+    /// large or infinite sequences.
+    val sortByDescending : projection:('T -> 'Key) -> source:AsyncSeq<'T> -> array<'T> when 'Key : comparison
+
     /// Interleaves two async sequences of the same type into a resulting sequence. The provided
     /// sequences are consumed in lock-step.
     val interleave : source1:AsyncSeq<'T> -> source2:AsyncSeq<'T> -> AsyncSeq<'T>
@@ -452,7 +476,7 @@ module AsyncSeq =
     #if !FABLE_COMPILER
     /// Buffer items from the async sequence until a specified buffer size is reached or a specified amount of time is elapsed.
     val bufferByCountAndTime : bufferSize:int -> timeoutMs:int -> source:AsyncSeq<'T> -> AsyncSeq<'T []>
-    
+
     /// Buffers items from the async sequence by the specified time interval.
     /// If no items are received in an intervel and empty array is emitted.
     val bufferByTime : timeMs:int -> source:AsyncSeq<'T> -> AsyncSeq<'T[]>
@@ -490,7 +514,7 @@ module AsyncSeq =
     /// but in parallel, without waiting for a prior mapping operation to complete.
     /// Parallelism is bound by the ThreadPool.
     val mapAsyncParallel : mapping:('T -> Async<'U>) -> s:AsyncSeq<'T> -> AsyncSeq<'U>
-    
+
     /// Applies a key-generating function to each element and returns an async sequence containing unique keys
     /// and async sequences containing elements corresponding to the key.
     ///
@@ -504,7 +528,7 @@ module AsyncSeq =
     /// Note that the resulting async sequence has to be processed in parallel (e.g AsyncSeq.mapAsyncParallel) becaused
     /// completion of sub-sequences depends on completion of other sub-sequences.
     val groupBy<'T, 'Key when 'Key : equality> : projection:('T -> 'Key) -> source:AsyncSeq<'T> -> AsyncSeq<'Key * AsyncSeq<'T>>
-    
+
     #if (NETSTANDARD2_1 || NETCOREAPP3_0)
 
     /// Creates an asynchronous computation that asynchronously yields results from the provided .NET IAsyncEnumerable.

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

@@ -1512,18 +1512,18 @@ let ``AsyncSeq.iterAsyncParallel should propagate exception`` () =
 
 [<Test>]
 let ``AsyncSeq.iterAsyncParallel should cancel and not block forever when run in parallel with another exception-throwing Async`` () =
-    
+
     let handle x = async {
-           do! Async.Sleep 50           
+           do! Async.Sleep 50
     }
 
-    let fakeAsync = async {    
+    let fakeAsync = async {
            do! Async.Sleep 500
            return "fakeAsync"
     }
 
     let makeAsyncSeqBatch () =
-           let rec loop() = asyncSeq {            
+           let rec loop() = asyncSeq {
                let! batch =  fakeAsync |> Async.Catch
                match batch with
                | Choice1Of2 batch ->
@@ -1532,11 +1532,11 @@ let ``AsyncSeq.iterAsyncParallel should cancel and not block forever when run in
                    yield! loop()
                  else
                    yield batch
-                   yield! loop() 
+                   yield! loop()
                | Choice2Of2 err ->
                     printfn "Problem getting batch: %A" err
            }
-       
+
            loop()
 
     let x = makeAsyncSeqBatch () |> AsyncSeq.concatSeq |> AsyncSeq.iterAsyncParallel handle
@@ -1733,6 +1733,36 @@ let ``Async.concat should work``() =
 
       Assert.True(EQ expected actual)
 
+[<Test>]
+let ``AsyncSeq.sort should work for``() =
+  let input = [1; 3; 2; 5; 7; 4; 6] |> AsyncSeq.ofSeq
+  let expected = [|1..7|]
+  let actual = input |> AsyncSeq.sort
+  Assert.AreEqual(expected, actual)
+
+[<Test>]
+let ``AsyncSeq.sortDescending should work``() =
+  let input = [1; 3; 2; Int32.MaxValue; 4; 6; Int32.MinValue; 5; 7; 0] |> AsyncSeq.ofSeq
+  let expected = seq { yield Int32.MaxValue; yield! seq{ 7..-1..0 }; yield Int32.MinValue } |> Array.ofSeq
+  let actual = input |> AsyncSeq.sortDescending
+  Assert.AreEqual(expected, actual)
+
+[<Test>]
+let ``AsyncSeq.sortBy should work``() =
+  let fn x = Math.Abs(x-5)
+  let input = [1; 2; 4; 5; 7] |> AsyncSeq.ofSeq
+  let expected = [|5; 4; 7; 2; 1|]
+  let actual = input |> AsyncSeq.sortBy fn
+  Assert.AreEqual(expected, actual)
+
+[<Test>]
+let ``AsyncSeq.sortByDescending should work``() =
+  let fn x = Math.Abs(x-5)
+  let input = [1; 2; 4; 5; 6; 7;] |> AsyncSeq.ofSeq
+  let expected = [|1; 2; 7; 4; 6; 5;|]
+  let actual = input |> AsyncSeq.sortByDescending fn
+  Assert.AreEqual(expected, actual)
+
 #if (NETSTANDARD2_1 || NETCOREAPP3_0)
 [<Test>]
 let ``AsyncSeq.ofAsyncEnum should roundtrip successfully``() =
@@ -1744,8 +1774,8 @@ let ``AsyncSeq.ofAsyncEnum should roundtrip successfully``() =
 let ``AsyncSeq.toAsyncEnum raises exception``() : unit =
   async {
     let exceptionMessage = "Raised inside AsyncSeq"
-    let exceptionSequence = 
-      asyncSeq { yield failwith exceptionMessage; yield 1 } 
+    let exceptionSequence =
+      asyncSeq { yield failwith exceptionMessage; yield 1 }
       |> AsyncSeq.toAsyncEnum
     let mutable exceptionRaised = false
     try
@@ -1765,9 +1795,9 @@ let ``AsyncSeq.toAsyncEnum raises exception``() : unit =
 let ``AsyncSeq.ofAsyncEnum raises exception``() : unit =
   async {
     let exceptionMessage = "Raised inside AsyncSeq"
-    let exceptionSequence = 
-      asyncSeq { yield failwith exceptionMessage; yield 1 } 
-      |> AsyncSeq.toAsyncEnum 
+    let exceptionSequence =
+      asyncSeq { yield failwith exceptionMessage; yield 1 }
+      |> AsyncSeq.toAsyncEnum
       |> AsyncSeq.ofAsyncEnum
     let mutable exceptionRaised = false
     try