Rx - Observable Extensions for Windows Forms Controls

Question

I'm writing Windows Forms Control extensions like the following to create Observables from the controls' events (using Rx and Rx-WinForms 2.2.2):

''' <summary>
''' Returns an IObservable(Of String) that fires when the TextChanged event fires and the text has actually changed.
''' </summary>
''' <remarks>It's useful to follow this with .Throttle if you wish to throttle the events.</remarks>
<Extension()>
Public Function ObserveTextChanged(textbox As TextBox) As IObservable(Of String)
    Return Observable _
           .FromEventPattern(Of EventArgs)(textbox, "TextChanged", New ControlScheduler(textbox)) _
           .Select(Function(a) DirectCast(a.Sender, TextBox).Text) _
           .DistinctUntilChanged
End Function

''' <summary>
''' Returns an IObservable(Of Boolean) that fires when the CheckedChanged event fires.
''' </summary>
<Extension()>
Public Function ObserveCheckedChanged(radio As RadioButton) As IObservable(Of Boolean)
    Return Observable _
           .FromEventPattern(Of EventArgs)(radio, "CheckedChanged", New ControlScheduler(radio)) _
           .Select(Function(a) DirectCast(a.Sender, RadioButton).Checked)
End Function

My intention is to then be able to write code on my Form as follows:

Dim myObserver = MyTextBox.ObserveTextChanged.Throttle(TimeSpan.FromSeconds(0.3))
myObserver.Subscribe(Function(text) DoSomethingWith(text))

I've got some questions:

1. Am I going about this the right way? That is:
   1. Has anyone done something similar and run into threading / disposing issues?
   2. Do I need the ControlScheduler, or can I just append an .ObserveOn(control) at the end, or do I need both?
   3. Or, do I not need ObserveOn at all, and should use SubscribeOn instead?
2. Has this already been done in the Rx framework for WinForms, and I'm simply overlooking it?

Update:

I've split the event capturing and the value observing to two different extension methods in order to provide more flexibility for the subscriber. Sometimes the subscriber wants the value after it has changed, sometimes the subscriber only wants to know that the value has changed.

''' <summary>
''' Returns an IObservable that fires when the TextChanged event fires.
''' </summary>
''' <param name="textbox"></param>
''' <returns></returns>
''' <remarks>It's useful to follow this with .Throttle if you wish to throttle the events.</remarks>
<Extension()>
Public Function ObserveTextChanged(textbox As TextBox) As IObservable(Of Reactive.EventPattern(Of EventArgs))
    Return Observable.FromEventPattern(Of EventArgs)(textbox, "TextChanged")
End Function

''' <summary>
''' Returns an IObservable(Of String) that fires when the TextChanged event fires and the text has actually changed.
''' </summary>
''' <param name="textbox"></param>
''' <returns></returns>
''' <remarks>It's useful to follow this with .Throttle if you wish to throttle the events.</remarks>
<Extension()>
Public Function ObserveTextChangedResult(textbox As TextBox) As IObservable(Of String)
    '** May need to inject a .ObserveOn(textbox) just before the .Select if we get threading issues.
    Return textbox.ObserveTextChanged.Select(Function(a) textbox.Text).DistinctUntilChanged
End Function

''' <summary>
''' Returns an IObservable that fires when the CheckedChanged event fires.
''' </summary>
''' <param name="radio"></param>
''' <returns></returns>
''' <remarks></remarks>
<Extension()>
Public Function ObserveCheckedChanged(radio As RadioButton) As IObservable(Of Reactive.EventPattern(Of EventArgs))
    Return Observable.FromEventPattern(Of EventArgs)(radio, "CheckedChanged")
End Function

''' <summary>
''' Returns an IObservable(Of Boolean) that fires when the CheckedChanged event fires.
''' </summary>
''' <param name="radio"></param>
''' <returns></returns>
''' <remarks></remarks>
<Extension()>
Public Function ObserveCheckedChangedResult(radio As RadioButton) As IObservable(Of Boolean)
    '** May need to inject a .ObserveOn(radio) just before the .Select if we get threading issues.
    Return radio.ObserveCheckedChanged.Select(Function(a) radio.Checked)
End Function

Answer 1

I would generally write your extensions like this:

<Extension()>
Public Function ObserveTextChanged(textbox As TextBox) As IObservable(Of String)
    Return Observable _
        .FromEventPattern(Of EventArgs)(textbox, "TextChanged") _
        .Select(Function(a) textbox.Text) _
        .DistinctUntilChanged()
End Function

<Extension()>
Public Function ObserveCheckedChanged(radio As RadioButton) As IObservable(Of Boolean)
    Return Observable _
        .FromEventPattern(Of ItemCheckedEventArgs)(radio, "CheckedChanged") _
        .Select(Function(a) radio.Checked)
End Function

I removed the ControlScheduler entirely. Now this only because you want to develop a good habit observing on the UI thread at the last possible moment. So, you would use your code like this:

Dim myQuery = MyTextBox _
    .ObserveTextChanged() _
    .Throttle(TimeSpan.FromSeconds(0.3))

Dim mySubscription = myQuery _
    .ObserveOn(Me) _
    .Subscribe(Function(text) DoSomethingWith(text))

The reasoning behind this is that many of the Rx operators will implicitly change the observable's scheduler.

For example:

Dim otherQuery = _
    From text in MyTextBox.ObserveTextChanged() _
    from choices in Observable.Start(Function () GetSpellingChoices(text)) _
    select choices

That will automatically change the scheduler that the query uses to Scheduler.Default. So you would have to add another ObserveOn(control) to the query if your subscriber is going to update the UI.

So the basic way on thinking about it should be that if your subscriber updates the UI then do an ObserveOn just before the Subscribe.

Also, in your extension methods, I removed the DirectCast(a.Sender, TextBox) code. You don't need it as textbox is already in scope.

Finally you should be doing something with the IDisposable returned from the Subscribe method. You must dispose of the subscription when you close your form otherwise you'll have a lot of event handler wired up preventing garbage collection. I typically create a class-level CompositeDisposable to hold all my subscriptions so that I can dispose all of my subscriptions in one go.

Hopefully this give you a few pointers in the right direction.