feat: move async functions to specific modules and add where, XML comments and lambda inlining

Author: xperiandri

src/FSharp.Control.R3/AsyncObservable.fs

@@ -7,29 +7,81 @@ 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
-        )
+    /// Applies an accumulator function over an observable sequence, returning the
+    /// result of the aggregation as a single element in the result sequence
+    let aggregate seed (f : 'r -> 't -> 'r) source = async {
+        let! ct = Async.CancellationToken
+        return!
+            ObservableExtensions.AggregateAsync (source, seed, f, ct)
+            |> Async.AwaitTask
+    }
 
-    let length source = async {
+    /// Determines whether all elements of an observable satisfy a predicate
+    let all (f : 't -> bool) source = async {
         let! ct = Async.CancellationToken
         return!
-            ObservableExtensions.CountAsync (source, ct)
+            ObservableExtensions.AllAsync (source, f, ct)
+            |> Async.AwaitTask
+    }
+
+    /// Determines whether an observable sequence contains a specified value
+    /// which satisfies the given predicate
+    let existsAsync source = async {
+        let! ct = Async.CancellationToken
+        return!
+            ObservableExtensions.AnyAsync (source, ct)
+            |> Async.AwaitTask
+    }
+
+    /// Returns the first element of an observable sequence
+    let firstAsync source = async {
+        let! ct = Async.CancellationToken
+        return!
+            ObservableExtensions.FirstAsync (source, ct)
             |> Async.AwaitTask
     }
 
-    let inline iter (action : 't -> unit) source = async {
+    /// <summary>
+    /// Invokes an action for each element in the observable sequence, and propagates all observer
+    /// messages through the result sequence.
+    /// </summary>
+    /// <remarks>
+    /// This method can be used for debugging, logging, etc. of query behavior
+    /// by intercepting the message stream to run arbitrary actions for messages on the pipeline.
+    /// </remarks>
+    let iter (action : 't -> unit) source = async {
         let! ct = Async.CancellationToken
         return!
             ObservableExtensions.ForEachAsync (source, action, ct)
             |> Async.AwaitTask
     }
 
+    /// Returns the last element of an observable sequence till its completion or cancellation
+    let length source = async {
+        let! ct = Async.CancellationToken
+        return!
+            ObservableExtensions.CountAsync (source, ct)
+            |> Async.AwaitTask
+    }
+
+    /// Maps the given observable with the given asynchronous function
+    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
+        )
+
+    /// <summary>
+    /// Invokes an asynchronous action for each element in the observable sequence, and propagates all observer
+    /// messages through the result sequence.
+    /// </summary>
+    /// <remarks>
+    /// This method can be used for debugging, logging, etc. of query behavior
+    /// by intercepting the message stream to run arbitrary actions for messages on the pipeline.
+    /// </remarks>
     let iterAsync options (action : 't -> Async<unit>) source = source |> mapAsync options action |> length |> Async.Ignore

src/FSharp.Control.R3/Observable.fs

@@ -2,19 +2,11 @@ module FSharp.Control.R3.Observable
 
 open R3
 
-/// Applies an accumulator function over an observable sequence, returning the
-/// result of the aggregation as a single element in the result sequence
-let inline aggregateAsync cancellationToken seed (f : 'r -> 't -> 'r) source =
-    ObservableExtensions.AggregateAsync (source, seed, f, cancellationToken)
-
-/// Determines whether all elements of an observable satisfy a predicate
-let inline allAsync cancellationToken (f : 't -> bool) source = ObservableExtensions.AllAsync (source, f, cancellationToken)
-
 /// Hides the identy of an observable sequence
 let inline asObservable source : Observable<'Source> = ObservableExtensions.AsObservable source
 
 /// Binds an observable to generate a subsequent observable.
-let inline bind (f : 'T -> Observable<'TNext>) source = ObservableExtensions.SelectMany (source, f)
+let inline bind ([<InlineIfLambda>] f : 'T -> Observable<'TNext>) source = ObservableExtensions.SelectMany (source, f)
 
 /// Converts the elements of the sequence to the specified type
 let inline cast<'T, 'CastType> (source) = ObservableExtensions.Cast<'T, 'CastType> (source)
@@ -26,28 +18,23 @@ let inline concat source = ObservableExtensions.Concat source
 /// Returns an observable sequence that only contains distinct elements
 let inline distinct source = ObservableExtensions.Distinct source
 
-/// Determines whether an observable sequence contains a specified value
-/// which satisfies the given predicate
-let inline existsAsync source = ObservableExtensions.AnyAsync source
-
-/// Returns the first element of an observable sequence
-let inline firstAsync source = ObservableExtensions.FirstAsync source
-
 /// Filters the observable elements of a sequence based on a predicate
-let inline filter (f : 't -> bool) source = ObservableExtensions.Where (source, f)
+let inline filter ([<InlineIfLambda>] f : 't -> bool) source = ObservableExtensions.Where (source, f)
 
 /// Maps the given observable with the given function
-let inline map (f : 't -> 'r) source = ObservableExtensions.Select (source, f)
+let inline map ([<InlineIfLambda>] f : 't -> 'r) source = ObservableExtensions.Select (source, f)
 
 /// Maps the given observable with the given function and the index of the element
-let inline mapi (f : int -> 't -> 'r) source = ObservableExtensions.Select (source, (fun i x -> f x i))
+let inline mapi ([<InlineIfLambda>] f : int -> 't -> 'r) source = ObservableExtensions.Select (source, (fun i x -> f x i))
 
 /// Bypasses a specified number of elements in an observable sequence and then returns the remaining elements.
 let inline skip (count : int) (source) = ObservableExtensions.Skip (source, count)
 
-/// Takes n elements (from the beginning of an observable sequence? )
+/// Takes n elements (from the beginning of an observable sequence?)
 let inline take (count : int) (source) = ObservableExtensions.Take (source, count)
 
+/// Filters the observable elements of a sequence based on a predicate
+let inline where ([<InlineIfLambda>] f : 't -> bool) source = ObservableExtensions.Where (source, f)
 
 [<CompilationRepresentation(CompilationRepresentationFlags.ModuleSuffix)>]
 module Builders =
@@ -56,59 +43,59 @@ module Builders =
     /// A reactive query builder.
     /// See http://mnajder.blogspot.com/2011/09/when-reactive-framework-meets-f-30.html
     type RxQueryBuilder () =
-        member __.For (s : Observable<_>, body : _ -> Observable<_>) = s.SelectMany (body)
+        member _.For (s : Observable<_>, body : _ -> Observable<_>) = s.SelectMany (body)
         [<CustomOperation("select", AllowIntoPattern = true)>]
-        member __.Select (s : Observable<_>, [<ProjectionParameter>] selector : _ -> _) = s.Select (selector)
+        member _.Select (s : Observable<_>, [<ProjectionParameter>] selector : _ -> _) = s.Select (selector)
         [<CustomOperation("where", MaintainsVariableSpace = true, AllowIntoPattern = true)>]
-        member __.Where (s : Observable<_>, [<ProjectionParameter>] predicate : _ -> bool) = s.Where (predicate)
+        member _.Where (s : Observable<_>, [<ProjectionParameter>] predicate : _ -> bool) = s.Where (predicate)
         [<CustomOperation("takeWhile", MaintainsVariableSpace = true, AllowIntoPattern = true)>]
-        member __.TakeWhile (s : Observable<_>, [<ProjectionParameter>] predicate : _ -> bool) = s.TakeWhile (predicate)
+        member _.TakeWhile (s : Observable<_>, [<ProjectionParameter>] predicate : _ -> bool) = s.TakeWhile (predicate)
         [<CustomOperation("take", MaintainsVariableSpace = true, AllowIntoPattern = true)>]
-        member __.Take (s : Observable<_>, count : int) = s.Take (count)
+        member _.Take (s : Observable<_>, count : int) = s.Take (count)
         [<CustomOperation("skipWhile", MaintainsVariableSpace = true, AllowIntoPattern = true)>]
-        member __.SkipWhile (s : Observable<_>, [<ProjectionParameter>] predicate : _ -> bool) = s.SkipWhile (predicate)
+        member _.SkipWhile (s : Observable<_>, [<ProjectionParameter>] predicate : _ -> bool) = s.SkipWhile (predicate)
         [<CustomOperation("skip", MaintainsVariableSpace = true, AllowIntoPattern = true)>]
-        member __.Skip (s : Observable<_>, count : int) = s.Skip (count)
-        member __.Zero () = Observable.Empty (TimeProvider.System)
-        member __.Yield (value) = Observable.Return (value, TimeProvider.System)
+        member _.Skip (s : Observable<_>, count : int) = s.Skip (count)
+        member _.Zero () = Observable.Empty (TimeProvider.System)
+        member _.Yield (value) = Observable.Return (value, TimeProvider.System)
         [<CustomOperation("count")>]
-        member __.Count (s : Observable<_>) = ObservableExtensions.CountAsync (s)
+        member _.Count (s : Observable<_>) = ObservableExtensions.CountAsync (s)
         [<CustomOperation("all")>]
-        member __.All (s : Observable<_>, [<ProjectionParameter>] predicate : _ -> bool) = s.AllAsync (new Func<_, bool> (predicate))
+        member _.All (s : Observable<_>, [<ProjectionParameter>] predicate : _ -> bool) = s.AllAsync (new Func<_, bool> (predicate))
         [<CustomOperation("contains")>]
-        member __.Contains (s : Observable<_>, key) = s.ContainsAsync (key)
+        member _.Contains (s : Observable<_>, key) = s.ContainsAsync (key)
         [<CustomOperation("distinct", MaintainsVariableSpace = true, AllowIntoPattern = true)>]
-        member __.Distinct (s : Observable<_>) = s.Distinct ()
+        member _.Distinct (s : Observable<_>) = s.Distinct ()
         [<CustomOperation("exactlyOne")>]
-        member __.ExactlyOne (s : Observable<_>) = s.SingleAsync ()
+        member _.ExactlyOne (s : Observable<_>) = s.SingleAsync ()
         [<CustomOperation("exactlyOneOrDefault")>]
-        member __.ExactlyOneOrDefault (s : Observable<_>) = s.SingleOrDefaultAsync ()
+        member _.ExactlyOneOrDefault (s : Observable<_>) = s.SingleOrDefaultAsync ()
         [<CustomOperation("find")>]
-        member __.Find (s : Observable<_>, [<ProjectionParameter>] predicate : _ -> bool) = s.FirstAsync (new Func<_, bool> (predicate))
+        member _.Find (s : Observable<_>, [<ProjectionParameter>] predicate : _ -> bool) = s.FirstAsync (new Func<_, bool> (predicate))
         [<CustomOperation("head")>]
-        member __.Head (s : Observable<_>) = s.FirstAsync ()
+        member _.Head (s : Observable<_>) = s.FirstAsync ()
         [<CustomOperation("headOrDefault")>]
-        member __.HeadOrDefault (s : Observable<_>) = s.FirstOrDefaultAsync ()
+        member _.HeadOrDefault (s : Observable<_>) = s.FirstOrDefaultAsync ()
         [<CustomOperation("last")>]
-        member __.Last (s : Observable<_>) = s.LastAsync ()
+        member _.Last (s : Observable<_>) = s.LastAsync ()
         [<CustomOperation("lastOrDefault")>]
-        member __.LastOrDefault (s : Observable<_>) = s.LastOrDefaultAsync ()
+        member _.LastOrDefault (s : Observable<_>) = s.LastOrDefaultAsync ()
         [<CustomOperation("maxBy")>]
-        member __.MaxBy (s : Observable<'a>, [<ProjectionParameter>] valueSelector : 'a -> 'b) = s.MaxByAsync (new Func<'a, 'b> (valueSelector))
+        member _.MaxBy (s : Observable<'a>, [<ProjectionParameter>] valueSelector : 'a -> 'b) = s.MaxByAsync (new Func<'a, 'b> (valueSelector))
         [<CustomOperation("minBy")>]
-        member __.MinBy (s : Observable<'a>, [<ProjectionParameter>] valueSelector : 'a -> 'b) = s.MinByAsync (new Func<'a, 'b> (valueSelector))
+        member _.MinBy (s : Observable<'a>, [<ProjectionParameter>] valueSelector : 'a -> 'b) = s.MinByAsync (new Func<'a, 'b> (valueSelector))
 
         [<CustomOperation("sumBy")>]
-        member inline __.SumBy (s : Observable<_>, [<ProjectionParameter>] valueSelector : _ -> _) =
+        member inline _.SumBy (s : Observable<_>, [<ProjectionParameter>] valueSelector : _ -> _) =
             s
-                .Select(valueSelector)
-                .AggregateAsync (Unchecked.defaultof<_>, new Func<_, _, _> (fun a b -> a + b))
+            |> _.Select(valueSelector)
+            |> _.AggregateAsync(Unchecked.defaultof<_>, new Func<_, _, _> (fun a b -> a + b))
 
         [<CustomOperation("zip", IsLikeZip = true)>]
-        member __.Zip (s1 : Observable<_>, s2 : Observable<_>, [<ProjectionParameter>] resultSelector : _ -> _) =
+        member _.Zip (s1 : Observable<_>, s2 : Observable<_>, [<ProjectionParameter>] resultSelector : _ -> _) =
             s1.Zip (s2, new Func<_, _, _> (resultSelector))
 
         [<CustomOperation("iter")>]
-        member __.Iter (s : Observable<_>, [<ProjectionParameter>] selector : _ -> _) = s.ForEachAsync (selector)
+        member _.Iter (s : Observable<_>, [<ProjectionParameter>] selector : _ -> _) = s.ForEachAsync (selector)
 
     let rxquery = RxQueryBuilder ()

src/FSharp.Control.R3/TaskObservable.fs

@@ -7,7 +7,36 @@ open FSharp.Control.R3
 
 module Observable =
 
-    /// Maps the given observable with the given function
+    /// Applies an accumulator function over an observable sequence, returning the
+    /// result of the aggregation as a single element in the result sequence
+    let inline aggregate cancellationToken seed ([<InlineIfLambda>] f : 'r -> 't -> 'r) source =
+        ObservableExtensions.AggregateAsync (source, seed, f, cancellationToken)
+
+    /// Determines whether all elements of an observable satisfy a predicate
+    let inline all cancellationToken ([<InlineIfLambda>] f : 't -> bool) source = ObservableExtensions.AllAsync (source, f, cancellationToken)
+
+    /// <summary>
+    /// Invokes an action for each element in the observable sequence, and propagates all observer
+    /// messages through the result sequence.
+    /// </summary>
+    /// <remarks>
+    /// This method can be used for debugging, logging, etc. of query behavior
+    /// by intercepting the message stream to run arbitrary actions for messages on the pipeline.
+    /// </remarks>
+    let inline iter cancellationToken ([<InlineIfLambda>] action : 't -> unit) source =
+        ObservableExtensions.ForEachAsync (source, action, cancellationToken)
+
+    /// Determines whether an observable sequence contains a specified value
+    /// which satisfies the given predicate
+    let inline existsAsync cancellationToken source = ObservableExtensions.AnyAsync (source, cancellationToken)
+
+    /// Returns the first element of an observable sequence
+    let inline firstAsync cancellationToken source = ObservableExtensions.FirstAsync (source, cancellationToken)
+
+    /// Returns the length of the observable sequence till its completion or cancellation
+    let length cancellationToken source = ObservableExtensions.CountAsync (source, cancellationToken)
+
+    /// Maps the given observable with the given asynchronous function
     let mapAsync (options : ProcessingOptions) (f : CancellationToken -> 't -> Task<'r>) source =
         let selector x ct = ValueTask<'r> (f ct x)
         ObservableExtensions.SelectAwait (
@@ -19,13 +48,14 @@ module Observable =
             options.MaxConcurrent
         )
 
-    let length cancellationToken source = ObservableExtensions.CountAsync (source, cancellationToken)
-
-    /// Invokes an action for each element in the observable sequence, and propagates all observer
-    /// messages through the result sequence. This method can be used for debugging, logging, etc. of query
-    /// behavior by intercepting the message stream to run arbitrary actions for messages on the pipeline.
-    let inline iter cancellationToken (action : 't -> unit) source = ObservableExtensions.ForEachAsync (source, action, cancellationToken)
-
+    /// <summary>
+    /// Invokes an asynchronous action for each element in the observable sequence, and propagates all observer
+    /// messages through the result sequence.
+    /// </summary>
+    /// <remarks>
+    /// This method can be used for debugging, logging, etc. of query behavior
+    /// by intercepting the message stream to run arbitrary actions for messages on the pipeline.
+    /// </remarks>
     let iterAsync cancellationToken options (action : CancellationToken -> 't -> Task<unit>) source =
         source
         |> mapAsync options action