feat: split to Observable, Async.Observable and Task.Observable modules (#2)

Author: xperiandri

src/FSharp.Control.R3/AsyncObservable.fs

@@ -0,0 +1,35 @@
+module FSharp.Control.R3.Async
+
+open R3
+open System.Threading
+open System.Threading.Tasks
+open FSharp.Control.R3
+
+module Observable =
+
+    let mapAsync (options : ProcessingOptions) (f : 't -> Async<'r>) source =
+        let selector x ct = ValueTask<'r> (Async.StartImmediateAsTask (f x, ct))
+        ObservableExtensions.SelectAwait (
+            source,
+            selector,
+            options.AwaitOperation,
+            options.ConfigureAwait,
+            options.CancelOnCompleted,
+            options.MaxConcurrent
+        )
+
+    let length source = async {
+        let! ct = Async.CancellationToken
+        return!
+            ObservableExtensions.CountAsync (source, ct)
+            |> Async.AwaitTask
+    }
+
+    let inline iter (action : 't -> unit) source = async {
+        let! ct = Async.CancellationToken
+        return!
+            ObservableExtensions.ForEachAsync (source, action, ct)
+            |> Async.AwaitTask
+    }
+
+    let iterAsync options (action : 't -> Async<unit>) source = source |> mapAsync options action |> length |> Async.Ignore

src/FSharp.Control.R3/FSharp.Control.R3.fsproj

@@ -16,7 +16,10 @@
 
   <ItemGroup>
     <Compile Include="AssemblyInfo.fs" />
+    <Compile Include="ProcessingOptions.fs" />
     <Compile Include="Observable.fs" />
+    <Compile Include="AsyncObservable.fs" />
+    <Compile Include="TaskObservable.fs" />
   </ItemGroup>
 
   <ItemGroup>

src/FSharp.Control.R3/Observable.fs

@@ -1,28 +1,5 @@
-module FSharp.Control.Observable
+module FSharp.Control.R3.Observable
 
 open R3
-open System.Threading
-open System.Threading.Tasks
-
-let asyncMap (f : 't -> Async<'r>) source =
-    let selector x ct = ValueTask<'r> (Async.StartImmediateAsTask (f x, ct))
-    ObservableExtensions.SelectAwait (source, selector)
-
-let mapAsync (f : CancellationToken -> 't -> Task<'r>) source =
-    let selector x ct = ValueTask<'r> (f ct x)
-    ObservableExtensions.SelectAwait (source, selector)
 
 let inline map (f : 't -> 'r) source = ObservableExtensions.Select (source, f)
-
-let length source = async {
-    let! ct = Async.CancellationToken
-    return!
-        ObservableExtensions.CountAsync (source, ct)
-        |> Async.AwaitTask
-}
-
-let inline iter (action : 't -> unit) source = ObservableExtensions.ForEachAsync (source, action)
-
-let asyncIter (action : 't -> Async<unit>) source = source |> asyncMap action |> length |> Async.Ignore
-
-let iterAsync (action : CancellationToken -> 't -> Task<unit>) source = source |> mapAsync action |> length |> Async.Ignore

src/FSharp.Control.R3/ProcessingOptions.fs

@@ -0,0 +1,60 @@
+namespace FSharp.Control.R3
+
+open R3
+
+type AwaitOperationConfiguration =
+    /// <summary>All values are queued, and the next value waits for the completion of the asynchronous method.</summary>
+    | Sequential
+    /// <summary>Drop new value when async operation is running.</summary>
+    | Drop
+    /// <summary>If the previous asynchronous method is running, it is cancelled and the next asynchronous method is executed.</summary>
+    | Switch
+    /// <summary>All values are sent immediately to the asynchronous method.</summary>
+    | Parallel of
+        /// If set to -1, there is no limit.
+        MaxConcurrent : int
+    /// <summary>All values are sent immediately to the asynchronous method, but the results are queued and passed to the next operator in order.</summary>
+    | SequentialParallel of
+        /// If set to -1, there is no limit.
+        MaxConcurrent : int
+    /// <summary>Send the first value and the last value while the asynchronous method is running.</summary>
+    | ThrottleFirstLast
+
+type ProcessingOptions = {
+    AwaitOperationConfiguration : AwaitOperationConfiguration
+    ConfigureAwait : bool
+    CancelOnCompleted : bool
+} with
+
+    static let ``default`` = {
+        AwaitOperationConfiguration = AwaitOperationConfiguration.Sequential
+        ConfigureAwait = true
+        CancelOnCompleted = false
+    }
+
+    static let ``parallel`` = {
+        AwaitOperationConfiguration = AwaitOperationConfiguration.Parallel -1
+        ConfigureAwait = true
+        CancelOnCompleted = false
+    }
+
+    static member Default = ``default``
+    static member Parallel = ``parallel``
+
+    member this.MaxConcurrent =
+        match this.AwaitOperationConfiguration with
+        | AwaitOperationConfiguration.Sequential -> -1
+        | AwaitOperationConfiguration.Drop -> -1
+        | AwaitOperationConfiguration.Switch -> -1
+        | AwaitOperationConfiguration.Parallel maxConcurrent -> maxConcurrent
+        | AwaitOperationConfiguration.SequentialParallel maxConcurrent -> maxConcurrent
+        | AwaitOperationConfiguration.ThrottleFirstLast -> -1
+
+    member this.AwaitOperation =
+        match this.AwaitOperationConfiguration with
+        | AwaitOperationConfiguration.Sequential -> AwaitOperation.Sequential
+        | AwaitOperationConfiguration.Drop -> AwaitOperation.Drop
+        | AwaitOperationConfiguration.Switch -> AwaitOperation.Switch
+        | AwaitOperationConfiguration.Parallel _ -> AwaitOperation.Parallel
+        | AwaitOperationConfiguration.SequentialParallel _ -> AwaitOperation.SequentialParallel
+        | AwaitOperationConfiguration.ThrottleFirstLast -> AwaitOperation.ThrottleFirstLast

src/FSharp.Control.R3/TaskObservable.fs

@@ -0,0 +1,29 @@
+module FSharp.Control.R3.Task
+
+open R3
+open System.Threading
+open System.Threading.Tasks
+open FSharp.Control.R3
+
+module Observable =
+
+    let mapAsync (options : ProcessingOptions) (f : CancellationToken -> 't -> Task<'r>) source =
+        let selector x ct = ValueTask<'r> (f ct x)
+        ObservableExtensions.SelectAwait (
+            source,
+            selector,
+            options.AwaitOperation,
+            options.ConfigureAwait,
+            options.CancelOnCompleted,
+            options.MaxConcurrent
+        )
+
+    let length cancellationToken source = ObservableExtensions.CountAsync (source, cancellationToken)
+
+    let inline iter cancellationToken (action : 't -> unit) source = ObservableExtensions.ForEachAsync (source, action, cancellationToken)
+
+    let iterAsync cancellationToken options (action : CancellationToken -> 't -> Task<unit>) source =
+        source
+        |> mapAsync options action
+        |> length cancellationToken
+        :> Task

tests/FSharp.Control.R3.Tests/Tests.fs

@@ -1,15 +1,16 @@
 namespace FSharp.Control.R3.Tests
 
 open System
-open FSharp.Control
+open System.Threading.Tasks
+open FSharp.Control.R3.Async
 open Microsoft.VisualStudio.TestTools.UnitTesting
 open Swensen.Unquote
 
 [<TestClass>]
 type ObservableTests () =
 
     [<TestMethod>]
-    member _.``Test length`` () =
+    member _.``Test length`` () : Task =
 
         async {
             use r3Bus = new R3.Subject<int> ()
@@ -27,4 +28,5 @@ type ObservableTests () =
             Assert.AreEqual<int> (0, res)
 
         }
-        |> Async.RunSynchronously
+        |> Async.StartImmediateAsTask
+        :> Task