{"id":924,"date":"2009-11-23T13:35:39","date_gmt":"2009-11-23T12:35:39","guid":{"rendered":"http:\/\/www.navision-blog.de\/2009\/11\/23\/mapping-the-reactive-framework-rx-operators-for-f\/"},"modified":"2009-12-21T16:54:37","modified_gmt":"2009-12-21T15:54:37","slug":"mapping-the-reactive-framework-rx-operators-for-f","status":"publish","type":"post","link":"http:\/\/www.navision-blog.de\/blog\/2009\/11\/23\/mapping-the-reactive-framework-rx-operators-for-f\/","title":{"rendered":"Mapping the Reactive Framework (Rx) operators for F#"},"content":{"rendered":"<p>The \u201c<a href=\"http:\/\/msdn.microsoft.com\/en-us\/devlabs\/ee794896.aspx\">Reactive Extensions for .NET (Rx)<\/a>\u201d comes with lot\u2019s of operators for using IObservable&lt;T&gt;. This code mimics the signature of the default F# sequence combinators and allows to use observables like sequences. It is a similar approach like Matthews Podwysocki\u2019s blog post about <a href=\"http:\/\/weblogs.asp.net\/podwysocki\/archive\/2009\/02\/23\/adding-parallel-extensions-to-f.aspx\">mapping the IParallelEnumerable<\/a>.<\/p>\n<p>I will update this post from time to time to include more of the operators.<\/p>\n<ul>\n<li>Update: 25.11.2009 \u2013 new operators mapped <\/li>\n<li>Update: 21.11.2009 \u2013 Updated to new Rx release <\/li>\n<\/ul>\n<div style=\"font-family: courier new; background: white; color: black; font-size: 10pt\">\n<p style=\"margin: 0px\"><span style=\"color: blue\">module<\/span> RxExtensions.Observable<\/p>\n<p style=\"margin: 0px\">&#160;<\/p>\n<p style=\"margin: 0px\"><span style=\"color: blue\">open<\/span> System.Linq<\/p>\n<p style=\"margin: 0px\"><span style=\"color: blue\">open<\/span> System<\/p>\n<p style=\"margin: 0px\"><span style=\"color: blue\">open<\/span> System.Threading<\/p>\n<p style=\"margin: 0px\"><span style=\"color: blue\">open<\/span> System.Windows.Threading<\/p>\n<p style=\"margin: 0px\">&#160;<\/p>\n<p style=\"margin: 0px\"><span style=\"color: blue\">type<\/span> &#8216;a observable = IObservable&lt;&#8216;a&gt;<\/p>\n<p style=\"margin: 0px\"><span style=\"color: blue\">type<\/span> &#8216;a observer = IObserver&lt;&#8216;a&gt;<\/p>\n<p style=\"margin: 0px\">&#160;<\/p>\n<p style=\"margin: 0px\"><span style=\"color: green\">\/\/\/ converts a lambda in a System.Action<\/span><\/p>\n<p style=\"margin: 0px\"><span style=\"color: blue\">let<\/span> asAction f = <span style=\"color: blue\">new<\/span> System.Action(f)<\/p>\n<p style=\"margin: 0px\">&#160;<\/p>\n<p style=\"margin: 0px\"><span style=\"color: green\">\/\/\/ System.Action whichs does nothing<\/span><\/p>\n<p style=\"margin: 0px\"><span style=\"color: blue\">let<\/span> doNothing = asAction (<span style=\"color: blue\">fun<\/span> () <span style=\"color: blue\">-&gt;<\/span> ())<\/p>\n<p style=\"margin: 0px\">&#160;<\/p>\n<p style=\"margin: 0px\"><span style=\"color: green\">\/\/\/ Creates an observer<\/span><\/p>\n<p style=\"margin: 0px\"><span style=\"color: blue\">let<\/span> createObserver next error completed =<\/p>\n<p style=\"margin: 0px\">&#160; {<span style=\"color: blue\">new<\/span> System.IObserver&lt;_&gt; <span style=\"color: blue\">with<\/span><\/p>\n<p style=\"margin: 0px\">&#160;&#160;&#160;&#160;&#160; <span style=\"color: blue\">member<\/span> this.OnCompleted() = completed()<\/p>\n<p style=\"margin: 0px\">&#160;&#160;&#160;&#160;&#160; <span style=\"color: blue\">member<\/span> this.OnError(e) = error e<\/p>\n<p style=\"margin: 0px\">&#160;&#160;&#160;&#160;&#160; <span style=\"color: blue\">member<\/span> this.OnNext(args) = next args}<\/p>\n<p style=\"margin: 0px\">&#160;<\/p>\n<p style=\"margin: 0px\"><span style=\"color: green\">\/\/\/ Creates a new observable <\/span><\/p>\n<p style=\"margin: 0px\"><span style=\"color: blue\">let<\/span> create f =<\/p>\n<p style=\"margin: 0px\">&#160; Observable.Create&lt;_&gt;(<span style=\"color: blue\">fun<\/span> x <span style=\"color: blue\">-&gt;<\/span><\/p>\n<p style=\"margin: 0px\">&#160;&#160;&#160; f x<\/p>\n<p style=\"margin: 0px\">&#160;&#160;&#160; doNothing)&#160;&#160; <\/p>\n<p style=\"margin: 0px\">&#160;<\/p>\n<p style=\"margin: 0px\"><span style=\"color: green\">\/\/\/ Creates a observable from a async<\/span><\/p>\n<p style=\"margin: 0px\"><span style=\"color: blue\">let<\/span> ofAsync async =<\/p>\n<p style=\"margin: 0px\">&#160; create<\/p>\n<p style=\"margin: 0px\">&#160;&#160;&#160; (<span style=\"color: blue\">fun<\/span> obs <span style=\"color: blue\">-&gt;<\/span><\/p>\n<p style=\"margin: 0px\">&#160;&#160;&#160;&#160;&#160;&#160; Async.StartWithContinuations<\/p>\n<p style=\"margin: 0px\">&#160;&#160;&#160;&#160;&#160;&#160;&#160;&#160; (async,obs.OnNext,obs.OnError,obs.OnError))<\/p>\n<p style=\"margin: 0px\">&#160;<\/p>\n<p style=\"margin: 0px\"><span style=\"color: green\">\/\/\/ Gets a dispatcher Schdeuler for the current dispatcher<\/span><\/p>\n<p style=\"margin: 0px\"><span style=\"color: blue\">let<\/span> getDispatcherScheduler _ = <\/p>\n<p style=\"margin: 0px\">&#160; <span style=\"color: blue\">new<\/span> DispatcherScheduler(Dispatcher.CurrentDispatcher)<\/p>\n<p style=\"margin: 0px\">&#160;<\/p>\n<p style=\"margin: 0px\"><span style=\"color: green\">\/\/\/ Generates an observable from an IEvent<\/span><\/p>\n<p style=\"margin: 0px\"><span style=\"color: blue\">let<\/span> fromEvent (event:IEvent&lt;_,_&gt;) = create (<span style=\"color: blue\">fun<\/span> x <span style=\"color: blue\">-&gt;<\/span> event.Add x.OnNext)<\/p>\n<p style=\"margin: 0px\">&#160;<\/p>\n<p style=\"margin: 0px\"><span style=\"color: green\">\/\/\/ Generates an empty observable <\/span><\/p>\n<p style=\"margin: 0px\"><span style=\"color: blue\">let<\/span> empty&lt;&#8216;a&gt; = Observable.Empty&lt;&#8216;a&gt;() <\/p>\n<p style=\"margin: 0px\">&#160;<\/p>\n<p style=\"margin: 0px\"><span style=\"color: green\">\/\/\/ Takes the head of the elements<\/span><\/p>\n<p style=\"margin: 0px\"><span style=\"color: blue\">let<\/span> head = Observable.First <\/p>\n<p style=\"margin: 0px\">&#160;<\/p>\n<p style=\"margin: 0px\"><span style=\"color: green\">\/\/\/ Merges the two observables<\/span><\/p>\n<p style=\"margin: 0px\"><span style=\"color: blue\">let<\/span> mergeWith obs1 obs2 = Observable.Merge(obs2, obs1)<\/p>\n<p style=\"margin: 0px\">&#160;<\/p>\n<p style=\"margin: 0px\"><span style=\"color: green\">\/\/\/ Merges all observables<\/span><\/p>\n<p style=\"margin: 0px\"><span style=\"color: blue\">let<\/span> mergeAll (observables:IObservable&lt;IObservable&lt;&#8216;a&gt;&gt;) = <\/p>\n<p style=\"margin: 0px\">&#160; Observable.Merge observables <\/p>\n<p style=\"margin: 0px\">&#160;<\/p>\n<p style=\"margin: 0px\"><span style=\"color: green\">\/\/\/ Merges all observables<\/span><\/p>\n<p style=\"margin: 0px\"><span style=\"color: blue\">let<\/span> merge (observables:(IObservable&lt;&#8216;a&gt;) seq) = <\/p>\n<p style=\"margin: 0px\">&#160; Observable.Merge observables <\/p>\n<p style=\"margin: 0px\">&#160;<\/p>\n<p style=\"margin: 0px\"><span style=\"color: green\">\/\/\/ Creates a range as an observable<\/span><\/p>\n<p style=\"margin: 0px\"><span style=\"color: blue\">let<\/span> range start count = Observable.Range(start, count)<\/p>\n<p style=\"margin: 0px\">&#160;<\/p>\n<p style=\"margin: 0px\"><span style=\"color: green\">\/\/\/ Converts a seq in an observable<\/span><\/p>\n<p style=\"margin: 0px\"><span style=\"color: blue\">let<\/span> toObservable (seq: &#8216;a seq) = Observable.ToObservable seq<\/p>\n<p style=\"margin: 0px\">&#160;<\/p>\n<p style=\"margin: 0px\"><span style=\"color: green\">\/\/\/ Converts a observable in a seq<\/span><\/p>\n<p style=\"margin: 0px\"><span style=\"color: blue\">let<\/span> toEnumerable = Observable.ToEnumerable<\/p>\n<p style=\"margin: 0px\">&#160;<\/p>\n<p style=\"margin: 0px\"><span style=\"color: green\">\/\/\/ Subscribes to the Observable with all 3 callbacks<\/span><\/p>\n<p style=\"margin: 0px\"><span style=\"color: blue\">let<\/span> subscribeComplete next error completed (observable: &#8216;a observable) = <\/p>\n<p style=\"margin: 0px\">&#160;&#160; observable.Subscribe(<\/p>\n<p style=\"margin: 0px\">&#160;&#160;&#160;&#160; (<span style=\"color: blue\">fun<\/span> x <span style=\"color: blue\">-&gt;<\/span> next x), <\/p>\n<p style=\"margin: 0px\">&#160;&#160;&#160;&#160; (<span style=\"color: blue\">fun<\/span> e <span style=\"color: blue\">-&gt;<\/span> error e), <\/p>\n<p style=\"margin: 0px\">&#160;&#160;&#160;&#160; (<span style=\"color: blue\">fun<\/span> () <span style=\"color: blue\">-&gt;<\/span> completed()))<\/p>\n<p style=\"margin: 0px\">&#160;<\/p>\n<p style=\"margin: 0px\"><span style=\"color: green\">\/\/\/ Subscribes to the Observable with a<\/span><\/p>\n<p style=\"margin: 0px\"><span style=\"color: green\">\/\/\/ next and an error-function<\/span><\/p>\n<p style=\"margin: 0px\"><span style=\"color: blue\">let<\/span> subscribeWithError next error observable = <\/p>\n<p style=\"margin: 0px\">&#160; subscribeComplete next error (<span style=\"color: blue\">fun<\/span> () <span style=\"color: blue\">-&gt;<\/span> ()) observable<\/p>\n<p style=\"margin: 0px\">&#160;<\/p>\n<p style=\"margin: 0px\"><span style=\"color: green\">\/\/\/ Subscribes to the Observable with just a next-function<\/span><\/p>\n<p style=\"margin: 0px\"><span style=\"color: blue\">let<\/span> subscribe next observable = <\/p>\n<p style=\"margin: 0px\">&#160; subscribeWithError next ignore observable<\/p>\n<p style=\"margin: 0px\">&#160;<\/p>\n<p style=\"margin: 0px\"><span style=\"color: green\">\/\/\/ throttles the observable for the given interval<\/span><\/p>\n<p style=\"margin: 0px\"><span style=\"color: blue\">let<\/span> throttle interval observable = <\/p>\n<p style=\"margin: 0px\">&#160; Observable.Throttle(observable,interval)&#160;&#160;&#160; <\/p>\n<p style=\"margin: 0px\">&#160;<\/p>\n<p style=\"margin: 0px\"><span style=\"color: green\">\/\/\/ throttles the observable scheduled on the current dispatcher<\/span><\/p>\n<p style=\"margin: 0px\"><span style=\"color: blue\">let<\/span> throttleOnCurrentDispatcher interval observable = <\/p>\n<p style=\"margin: 0px\">&#160; Observable.Throttle(<\/p>\n<p style=\"margin: 0px\">&#160;&#160;&#160;&#160; observable,getDispatcherScheduler(),interval) <\/p>\n<p style=\"margin: 0px\">&#160;<\/p>\n<p style=\"margin: 0px\"><span style=\"color: green\">\/\/\/ samples the observable at the given interval<\/span><\/p>\n<p style=\"margin: 0px\"><span style=\"color: blue\">let<\/span> sample interval observable = <\/p>\n<p style=\"margin: 0px\">&#160; Observable.Sample(observable,interval)&#160;&#160;&#160; <\/p>\n<p style=\"margin: 0px\">&#160;<\/p>\n<p style=\"margin: 0px\"><span style=\"color: green\">\/\/\/ samples the observable at the given interval <\/span><\/p>\n<p style=\"margin: 0px\"><span style=\"color: green\">\/\/\/ scheduled on the current dispatcher<\/span><\/p>\n<p style=\"margin: 0px\"><span style=\"color: blue\">let<\/span> sampleOnCurrentDispatcher interval observable = <\/p>\n<p style=\"margin: 0px\">&#160; Observable.Sample(<\/p>\n<p style=\"margin: 0px\">&#160;&#160;&#160; observable,getDispatcherScheduler(),interval) <\/p>\n<p style=\"margin: 0px\">&#160;<\/p>\n<p style=\"margin: 0px\"><span style=\"color: green\">\/\/\/ returns the observable sequence that reacts first.<\/span><\/p>\n<p style=\"margin: 0px\"><span style=\"color: blue\">let<\/span> takeFirstOf2Reactions obs1 obs2 = <\/p>\n<p style=\"margin: 0px\">&#160; Observable.Amb(obs1,obs2)<\/p>\n<p style=\"margin: 0px\">&#160;<\/p>\n<p style=\"margin: 0px\"><span style=\"color: green\">\/\/\/ returns the observable sequence that reacts first.<\/span><\/p>\n<p style=\"margin: 0px\"><span style=\"color: blue\">let<\/span> amb (obs: IObservable&lt;&#8216;a&gt; seq) = <\/p>\n<p style=\"margin: 0px\">&#160; Observable.Amb obs&#160;&#160; <\/p>\n<p style=\"margin: 0px\">&#160;<\/p>\n<p style=\"margin: 0px\"><span style=\"color: green\">\/\/\/ returns the observable sequence that reacts first.<\/span><\/p>\n<p style=\"margin: 0px\"><span style=\"color: blue\">let<\/span> takeFirstReaction (obs: IObservable&lt;&#8216;a&gt; seq) = <\/p>\n<p style=\"margin: 0px\">&#160; Observable.Amb obs&#160;&#160; <\/p>\n<p style=\"margin: 0px\">&#160;<\/p>\n<p style=\"margin: 0px\"><span style=\"color: green\">\/\/\/ Matches when both observable sequences <\/span><\/p>\n<p style=\"margin: 0px\"><span style=\"color: green\">\/\/\/ have an available value. <\/span><\/p>\n<p style=\"margin: 0px\"><span style=\"color: blue\">let<\/span> both obs1 obs2 = Observable.And(obs1,obs2)<\/p>\n<p style=\"margin: 0px\">&#160;<\/p>\n<p style=\"margin: 0px\"><span style=\"color: green\">\/\/\/ Merges two observable sequences<\/span><\/p>\n<p style=\"margin: 0px\"><span style=\"color: green\">\/\/\/ into one observable sequence.<\/span><\/p>\n<p style=\"margin: 0px\"><span style=\"color: blue\">let<\/span> zip obs1 obs2 =&#160;&#160;&#160; <\/p>\n<p style=\"margin: 0px\">&#160;&#160; Observable.Zip(obs1, obs2, Func&lt;_,_,_&gt;(<span style=\"color: blue\">fun<\/span> a b <span style=\"color: blue\">-&gt;<\/span> a, b))<\/p>\n<p style=\"margin: 0px\">&#160;<\/p>\n<p style=\"margin: 0px\"><span style=\"color: green\">\/\/\/ Merges two observable sequences into one observable sequence <\/span><\/p>\n<p style=\"margin: 0px\"><span style=\"color: green\">\/\/\/ whenever one of the observable sequences has a new value.<\/span><\/p>\n<p style=\"margin: 0px\"><span style=\"color: green\">\/\/\/&#160;&#160;&#160; ==&gt; More results than zip<\/span><\/p>\n<p style=\"margin: 0px\"><span style=\"color: blue\">let<\/span> combineLatest obs1 obs2 =&#160;&#160;&#160; <\/p>\n<p style=\"margin: 0px\">&#160;&#160; Observable.CombineLatest(<\/p>\n<p style=\"margin: 0px\">&#160;&#160;&#160;&#160; obs1, obs2, Func&lt;_,_,_&gt;(<span style=\"color: blue\">fun<\/span> a b <span style=\"color: blue\">-&gt;<\/span> a, b))&#160;&#160;&#160;&#160; <\/p>\n<p style=\"margin: 0px\">&#160;<\/p>\n<p style=\"margin: 0px\"><span style=\"color: green\">\/\/\/ Concats the two observables to one observable<\/span><\/p>\n<p style=\"margin: 0px\"><span style=\"color: blue\">let<\/span> concat observable = <\/p>\n<p style=\"margin: 0px\">&#160; Observable.SelectMany(<\/p>\n<p style=\"margin: 0px\">&#160;&#160;&#160; observable,<\/p>\n<p style=\"margin: 0px\">&#160;&#160;&#160; Func&lt;_,_&gt;(<span style=\"color: blue\">fun<\/span> (x:IObservable&lt;&#8216;a&gt;) <span style=\"color: blue\">-&gt;<\/span> x)) <\/p>\n<p style=\"margin: 0px\">&#160;<\/p>\n<p style=\"margin: 0px\"><span style=\"color: green\">\/\/\/ maps the given observable with the given function<\/span><\/p>\n<p style=\"margin: 0px\"><span style=\"color: blue\">let<\/span> map f observable = <\/p>\n<p style=\"margin: 0px\">&#160; Observable.Select(observable,Func&lt;_,_&gt;(f))&#160;&#160; <\/p>\n<p style=\"margin: 0px\">&#160;<\/p>\n<p style=\"margin: 0px\"><span style=\"color: green\">\/\/\/ maps the given observable with the given function<\/span><\/p>\n<p style=\"margin: 0px\"><span style=\"color: blue\">let<\/span> mapi f observable = <\/p>\n<p style=\"margin: 0px\">&#160; Observable.Select(observable,Func&lt;_,_,_&gt;(<span style=\"color: blue\">fun<\/span> x i <span style=\"color: blue\">-&gt;<\/span>f i x))&#160;&#160; <\/p>\n<p style=\"margin: 0px\">&#160;<\/p>\n<p style=\"margin: 0px\"><span style=\"color: green\">\/\/\/ Filters all elements where the given predicate is satified<\/span><\/p>\n<p style=\"margin: 0px\"><span style=\"color: blue\">let<\/span> filter f observable = <\/p>\n<p style=\"margin: 0px\">&#160; Observable.Where(observable, Func&lt;_,_&gt;(f)) <\/p>\n<p style=\"margin: 0px\">&#160;<\/p>\n<p style=\"margin: 0px\"><span style=\"color: green\">\/\/\/ Splits the observable into two observables<\/span><\/p>\n<p style=\"margin: 0px\"><span style=\"color: green\">\/\/\/ Containing the elements for which the predicate returns<\/span><\/p>\n<p style=\"margin: 0px\"><span style=\"color: green\">\/\/\/ true and false respectively<\/span><\/p>\n<p style=\"margin: 0px\"><span style=\"color: blue\">let<\/span> partition predicate observable =<\/p>\n<p style=\"margin: 0px\">&#160; filter predicate observable,<\/p>\n<p style=\"margin: 0px\">&#160; filter (predicate &gt;&gt; not) observable<\/p>\n<p style=\"margin: 0px\">&#160;<\/p>\n<p style=\"margin: 0px\"><span style=\"color: green\">\/\/\/ Skips n elements<\/span><\/p>\n<p style=\"margin: 0px\"><span style=\"color: blue\">let<\/span> skip n observable = Observable.Skip(observable, n)<\/p>\n<p style=\"margin: 0px\">&#160;<\/p>\n<p style=\"margin: 0px\"><span style=\"color: green\">\/\/\/ Skips elements while the predicate is satisfied<\/span><\/p>\n<p style=\"margin: 0px\"><span style=\"color: blue\">let<\/span> skipWhile f observable = <\/p>\n<p style=\"margin: 0px\">&#160; Observable.SkipWhile(observable, Func&lt;_,_&gt;(f)) <\/p>\n<p style=\"margin: 0px\">&#160;<\/p>\n<p style=\"margin: 0px\"><span style=\"color: green\">\/\/\/ Runs all observable sequences in parallel <\/span><\/p>\n<p style=\"margin: 0px\"><span style=\"color: green\">\/\/\/ and combines their first values. <\/span><\/p>\n<p style=\"margin: 0px\"><span style=\"color: blue\">let<\/span> forkJoin (observables: (&#8216;a observable) seq) =<\/p>\n<p style=\"margin: 0px\">&#160; Observable.ForkJoin observables<\/p>\n<p style=\"margin: 0px\">&#160;<\/p>\n<p style=\"margin: 0px\"><span style=\"color: green\">\/\/\/ Counts the elements<\/span><\/p>\n<p style=\"margin: 0px\"><span style=\"color: blue\">let<\/span> length = Observable.Count<\/p>\n<p style=\"margin: 0px\">&#160;<\/p>\n<p style=\"margin: 0px\"><span style=\"color: green\">\/\/\/ Takes n elements<\/span><\/p>\n<p style=\"margin: 0px\"><span style=\"color: blue\">let<\/span> take n observable =<\/p>\n<p style=\"margin: 0px\">&#160; Observable.Take(observable, n)&#160;&#160; <\/p>\n<p style=\"margin: 0px\">&#160;<\/p>\n<p style=\"margin: 0px\"><span style=\"color: green\">\/\/\/ Determines whether the given observable is empty&#160; <\/span><\/p>\n<p style=\"margin: 0px\"><span style=\"color: blue\">let<\/span> isEmpty observable = Observable.IsEmpty observable<\/p>\n<p style=\"margin: 0px\">&#160;<\/p>\n<p style=\"margin: 0px\"><span style=\"color: green\">\/\/\/ Determines whether the given observable is not empty&#160; <\/span><\/p>\n<p style=\"margin: 0px\"><span style=\"color: blue\">let<\/span> isNotEmpty observable = not (Observable.IsEmpty observable)<\/p>\n<p style=\"margin: 0px\">&#160;<\/p>\n<p style=\"margin: 0px\"><span style=\"color: green\">\/\/\/ Determines whether an observable sequence <\/span><\/p>\n<p style=\"margin: 0px\"><span style=\"color: green\">\/\/\/ contains a specified value<\/span><\/p>\n<p style=\"margin: 0px\"><span style=\"color: green\">\/\/\/ which satisfies the given predicate<\/span><\/p>\n<p style=\"margin: 0px\"><span style=\"color: blue\">let<\/span> exists predicate observable =<\/p>\n<p style=\"margin: 0px\">&#160; observable<\/p>\n<p style=\"margin: 0px\">&#160;&#160;&#160; |&gt; skipWhile (predicate &gt;&gt; not)<\/p>\n<p style=\"margin: 0px\">&#160;&#160;&#160; |&gt; isNotEmpty<\/p>\n<p style=\"margin: 0px\">&#160;<\/p>\n<p style=\"margin: 0px\"><span style=\"color: green\">\/\/\/ Continues an observable sequence that is terminated <\/span><\/p>\n<p style=\"margin: 0px\"><span style=\"color: green\">\/\/\/ by an exception with the next observable sequence. <\/span><\/p>\n<p style=\"margin: 0px\"><span style=\"color: blue\">let<\/span> catch (newObservable:IObservable&lt;&#8216;a&gt;) failingObservable = <\/p>\n<p style=\"margin: 0px\">&#160; Observable.Catch(failingObservable,newObservable)&#160; <\/p>\n<p style=\"margin: 0px\">&#160;<\/p>\n<p style=\"margin: 0px\"><span style=\"color: green\">\/\/\/ Takes elements while the predicate is satisfied<\/span><\/p>\n<p style=\"margin: 0px\"><span style=\"color: blue\">let<\/span> takeWhile f observable = <\/p>\n<p style=\"margin: 0px\">&#160; Observable.TakeWhile(observable, Func&lt;_,_&gt;(f)) <\/p>\n<p style=\"margin: 0px\">&#160;<\/p>\n<p style=\"margin: 0px\"><span style=\"color: green\">\/\/\/ Iterates through the observable <\/span><\/p>\n<p style=\"margin: 0px\"><span style=\"color: green\">\/\/\/ and performs the given side-effect<\/span><\/p>\n<p style=\"margin: 0px\"><span style=\"color: blue\">let<\/span> perform f observable =<\/p>\n<p style=\"margin: 0px\">&#160; Observable.Do(observable,<span style=\"color: blue\">fun<\/span> x <span style=\"color: blue\">-&gt;<\/span> f x)<\/p>\n<p style=\"margin: 0px\">&#160;<\/p>\n<p style=\"margin: 0px\"><span style=\"color: green\">\/\/\/ Invokes finallyAction after source observable <\/span><\/p>\n<p style=\"margin: 0px\"><span style=\"color: green\">\/\/\/ sequence terminates normally or by an exception. <\/span><\/p>\n<p style=\"margin: 0px\"><span style=\"color: blue\">let<\/span> performFinally f observable =<\/p>\n<p style=\"margin: 0px\">&#160; Observable.Finally(observable,<span style=\"color: blue\">fun<\/span> _ <span style=\"color: blue\">-&gt;<\/span> f())<\/p>\n<p style=\"margin: 0px\">&#160;<\/p>\n<p style=\"margin: 0px\"><span style=\"color: green\">\/\/\/ Folds the observable<\/span><\/p>\n<p style=\"margin: 0px\"><span style=\"color: blue\">let<\/span> fold f seed observable = <\/p>\n<p style=\"margin: 0px\">&#160; Observable.Aggregate(observable, seed, Func&lt;_,_,_&gt;(f))&#160;&#160; <\/p>\n<p style=\"margin: 0px\">&#160;<\/p>\n<p style=\"margin: 0px\"><span style=\"color: green\">\/\/\/ Retruns an observable from a async pattern&#160; <\/span><\/p>\n<p style=\"margin: 0px\"><span style=\"color: blue\">let<\/span> fromAsync beginF endF =<\/p>\n<p style=\"margin: 0px\">&#160;&#160; Observable.FromAsyncPattern&lt;_&gt;(<\/p>\n<p style=\"margin: 0px\">&#160;&#160;&#160;&#160; Func&lt;_,_,_&gt;(<span style=\"color: blue\">fun<\/span> x y <span style=\"color: blue\">-&gt;<\/span> beginF(x,y)),<\/p>\n<p style=\"margin: 0px\">&#160;&#160;&#160;&#160;&#160;&#160; (<span style=\"color: blue\">fun<\/span> x <span style=\"color: blue\">-&gt;<\/span> endF x)).Invoke()<\/p>\n<p style=\"margin: 0px\">&#160;<\/p>\n<p style=\"margin: 0px\"><span style=\"color: green\">\/\/\/ Runs all observable sequences in parallel <\/span><\/p>\n<p style=\"margin: 0px\"><span style=\"color: green\">\/\/\/ and combines their first values. <\/span><\/p>\n<p style=\"margin: 0px\"><span style=\"color: blue\">let<\/span> subscribeAll next observables =<\/p>\n<p style=\"margin: 0px\">&#160; observables |&gt; Seq.map (subscribe next) |&gt; Seq.toList&#160;&#160;&#160;&#160;&#160; <\/p>\n<p style=\"margin: 0px\">&#160;<\/p>\n<p style=\"margin: 0px\"><span style=\"color: blue\">type<\/span> IObservable&lt;&#8216;a&gt; <span style=\"color: blue\">with<\/span><\/p>\n<p style=\"margin: 0px\">&#160; <span style=\"color: green\">\/\/\/ Subscribes to the Observable with just a next-function<\/span><\/p>\n<p style=\"margin: 0px\">&#160; <span style=\"color: blue\">member<\/span> this.Subscribe(next) = <\/p>\n<p style=\"margin: 0px\">&#160;&#160;&#160; subscribe next this<\/p>\n<p style=\"margin: 0px\">&#160;<\/p>\n<p style=\"margin: 0px\">&#160; <span style=\"color: green\">\/\/\/ Subscribes to the Observable with a next <\/span><\/p>\n<p style=\"margin: 0px\">&#160; <span style=\"color: green\">\/\/\/ and an error-function<\/span><\/p>\n<p style=\"margin: 0px\">&#160; <span style=\"color: blue\">member<\/span> this.Subscribe(next,error) = <\/p>\n<p style=\"margin: 0px\">&#160;&#160;&#160; subscribeWithError next error this<\/p>\n<p style=\"margin: 0px\">&#160;<\/p>\n<p style=\"margin: 0px\">&#160; <span style=\"color: green\">\/\/\/ Subscribes to the Observable with all 3 callbacks<\/span><\/p>\n<p style=\"margin: 0px\">&#160; <span style=\"color: blue\">member<\/span> this.Subscribe(next,error,completed) = <\/p>\n<p style=\"margin: 0px\">&#160;&#160;&#160; subscribeComplete next error completed this<\/p>\n<p style=\"margin: 0px\">&#160;<\/p>\n<p style=\"margin: 0px\"><span style=\"color: blue\">open<\/span> System.Net<\/p>\n<p style=\"margin: 0px\">&#160;<\/p>\n<p style=\"margin: 0px\"><span style=\"color: blue\">type<\/span> WebRequest <span style=\"color: blue\">with<\/span><\/p>\n<p style=\"margin: 0px\">&#160; <span style=\"color: blue\">member<\/span> this.GetRequestStreamAsync() =<\/p>\n<p style=\"margin: 0px\">&#160;&#160;&#160; fromAsync <\/p>\n<p style=\"margin: 0px\">&#160;&#160;&#160;&#160; this.BeginGetRequestStream <\/p>\n<p style=\"margin: 0px\">&#160;&#160;&#160;&#160; this.EndGetRequestStream<\/p>\n<p style=\"margin: 0px\">&#160;<\/p>\n<p style=\"margin: 0px\">&#160; <span style=\"color: blue\">member<\/span> this.GetResponseAsync() =<\/p>\n<p style=\"margin: 0px\">&#160;&#160;&#160; fromAsync <\/p>\n<p style=\"margin: 0px\">&#160;&#160;&#160;&#160;&#160; this.BeginGetResponse <\/p>\n<p style=\"margin: 0px\">&#160;&#160;&#160;&#160;&#160; this.EndGetResponse <\/p>\n<p style=\"margin: 0px\">&#160;<\/p>\n<p style=\"margin: 0px\">&#160; <span style=\"color: blue\">member<\/span> this.GetResponseStreamAsync() =<\/p>\n<p style=\"margin: 0px\">&#160;&#160;&#160; fromAsync <\/p>\n<p style=\"margin: 0px\">&#160;&#160;&#160;&#160;&#160; this.BeginGetRequestStream <\/p>\n<p style=\"margin: 0px\">&#160;&#160;&#160;&#160;&#160; this.EndGetRequestStream<\/p>\n<p style=\"margin: 0px\">&#160;<\/p>\n<p style=\"margin: 0px\"><span style=\"color: blue\">type<\/span> Async&lt;&#8216;a&gt; <span style=\"color: blue\">with<\/span><\/p>\n<p style=\"margin: 0px\">&#160; <span style=\"color: blue\">member<\/span> this.ToObservable() = ofAsync this<\/p>\n<\/p><\/div>\n","protected":false},"excerpt":{"rendered":"<p>The \u201cReactive Extensions for .NET (Rx)\u201d comes with lot\u2019s of operators for using IObservable&lt;T&gt;. This code mimics the signature of the default F# sequence combinators and allows to use observables like sequences. It is a similar approach like Matthews Podwysocki\u2019s blog post about mapping the IParallelEnumerable. I will update this post from time to time [&hellip;]<\/p>\n","protected":false},"author":1,"featured_media":0,"comment_status":"open","ping_status":"open","sticky":false,"template":"","format":"standard","meta":[],"categories":[448],"tags":[534,580,665,579],"_links":{"self":[{"href":"http:\/\/www.navision-blog.de\/blog\/wp-json\/wp\/v2\/posts\/924"}],"collection":[{"href":"http:\/\/www.navision-blog.de\/blog\/wp-json\/wp\/v2\/posts"}],"about":[{"href":"http:\/\/www.navision-blog.de\/blog\/wp-json\/wp\/v2\/types\/post"}],"author":[{"embeddable":true,"href":"http:\/\/www.navision-blog.de\/blog\/wp-json\/wp\/v2\/users\/1"}],"replies":[{"embeddable":true,"href":"http:\/\/www.navision-blog.de\/blog\/wp-json\/wp\/v2\/comments?post=924"}],"version-history":[{"count":7,"href":"http:\/\/www.navision-blog.de\/blog\/wp-json\/wp\/v2\/posts\/924\/revisions"}],"predecessor-version":[{"id":939,"href":"http:\/\/www.navision-blog.de\/blog\/wp-json\/wp\/v2\/posts\/924\/revisions\/939"}],"wp:attachment":[{"href":"http:\/\/www.navision-blog.de\/blog\/wp-json\/wp\/v2\/media?parent=924"}],"wp:term":[{"taxonomy":"category","embeddable":true,"href":"http:\/\/www.navision-blog.de\/blog\/wp-json\/wp\/v2\/categories?post=924"},{"taxonomy":"post_tag","embeddable":true,"href":"http:\/\/www.navision-blog.de\/blog\/wp-json\/wp\/v2\/tags?post=924"}],"curies":[{"name":"wp","href":"https:\/\/api.w.org\/{rel}","templated":true}]}}