How do I change the Rx Builder implementation to fix the stack overflow exception ?

Question

 

Hi,

I'm trying to come up with an Rx Builder to use Reactive Extension within the F# Computation Expression syntax. How do I fix it so that it doesn't blow the stack? Like the Seq example below. And is there any plans to provide an implementation of the RxBuilder as part of the Reactive Extensions or as part of future versions of the .NET Framework ?

 

open System
open System.Linq
open System.Reactive.Linq

type rxBuilder() = 
 member this.Delay f = Observable.Defer f
 member this.Combine (xs: IObservable<_>, ys : IObservable<_>) = Observable.Concat (xs, ys) 
 member this.Yield x = Observable.Return x
 member this.YieldFrom (xs:IObservable<_>) = xs

let rx = rxBuilder()

let rec f x = seq { yield x ; yield! f (x + 1) }

let rec g x = rx { yield x ; yield! g (x + 1) }


//do f 5 |> Seq.iter (printfn "%A")

do g 5 |> Observable.subscribe (printfn "%A") |> ignore

do System.Console.ReadLine() |> ignore

 

Asked on StackOverflow: http://stackoverflow.com/questions/6162288/how-do-i-change-the-rx-builder-implementation-to-fix-the-stack-overflow-exception

 

Answer 1

What about something like this?

http://rxbuilder.codeplex.com/

(created for the purpose of experimenting with RxBuilder)

 

    type rxBuilder() =    

       member this.Delay (f : unit -> 'a IObservable) = 

                   { new IObservable<_> with

                        member this.Subscribe obv = (f()).Subscribe obv }

       member this.Combine (xs:'a IObservable, ys: 'a IObservable) =

                   { new IObservable<_> with

                        member this.Subscribe obv = xs.Subscribe obv ; 

                                                                    ys.Subscribe obv }

       member this.Yield x = Observable.Return x

       member this.YieldFrom xs = xs

    let rx = rxBuilder()

    let rec f x = rx { yield x 

                            yield! f (x + 1) }

    do f 5 |> Observable.subscribe (fun x -> Console.WriteLine x) |> ignore

    do System.Console.ReadLine() |> ignore

The disposable return value of the xs.subscribe is not wired up. As soon as I try to wire it up, it goes back to blowing the stack.

Any ideas ?

HoloEd

http://fsharpcode.blogspot.com/

Answer 2

I'm interested in this, as well. I have been working on something very similar, though I have a few slightly different implementations for my methods: https://github.com/panesofglass/FSharp.Reactive.

---

Ryan Riley @panesofglass

Answer 3

The basic problem here is that the Concats get nested deeper and deeper.  Typical solutions are either to fold the concats together or to reset the stack by jumping to a new thread.

Answer 4

This has been fixed in Rx v2.0.

---

using (Microsoft.Sql.Cloud.DataProgrammability.Rx) { Signature.Emit("Bart De Smet"); }