Added more functions

Author: Thorium

src/FSharp.Control.R3/Observable.fs

@@ -2,4 +2,33 @@ module FSharp.Control.R3.Observable
 
 open R3
 
+/// Maps the given observable with the given function
 let inline map (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))
+
+/// Filters the observable elements of a sequence based on a predicate
+let inline filter (f : 't -> bool) source = ObservableExtensions.Where (source, f)
+
+/// 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)
+
+/// Converts the elements of the sequence to the specified type
+let inline cast<'T, 'CastType> (source) = ObservableExtensions.Cast<'T, 'CastType> (source)
+
+/// Concatenates the second observable sequence to the first observable sequence
+/// upn the successful termination of the first
+let inline concat source = ObservableExtensions.Concat source
+
+/// Returns an observable sequence that only contains distinct elements
+let inline distinct source = ObservableExtensions.Distinct source
+
+/// 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? )
+let inline take (count : int) (source) = ObservableExtensions.Take (source, count)

src/FSharp.Control.R3/TaskObservable.fs

@@ -7,6 +7,7 @@ open FSharp.Control.R3
 
 module Observable =
 
+    /// Maps the given observable with the given function
     let mapAsync (options : ProcessingOptions) (f : CancellationToken -> 't -> Task<'r>) source =
         let selector x ct = ValueTask<'r> (f ct x)
         ObservableExtensions.SelectAwait (
@@ -20,10 +21,28 @@ module Observable =
 
     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)
 
     let iterAsync cancellationToken options (action : CancellationToken -> 't -> Task<unit>) source =
         source
         |> mapAsync options action
         |> length cancellationToken
         :> Task
+
+    /// 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)
+
+    /// 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