Windows窗体控件的可观察扩展

Question

我正在编写Windows窗体控件扩展，像下面这样从控件的事件中创建observable(使用Rx和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

我的意图是能够在我的表单上编写代码，如下所示:

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

我有一些问题:

1. 我这样做对吗?那就是:
   1. 是否有人做过类似的事情并遇到线程/处置问题?
   2. 我是否需要ControlScheduler，或者我可以在末尾附加一个。observeon (control)，或者我需要两者吗?
   或者，我根本不需要ObserveOn，而应该使用SubscribeOn ?

这已经在WinForms的Rx框架中完成了吗，我只是忽略了它?

更新:

为了给订阅者提供更大的灵活性，我将事件捕获和值观察拆分为两个不同的扩展方法。有时订阅者想要的值已经改变，有时订阅者只想知道值已经改变。

''' <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

我一般会这样写你的扩展:

<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

我完全删除了ControlScheduler。现在，这只是因为您希望在最后可能的时刻养成在UI线程上观察的好习惯。所以，你可以像这样使用你的代码:

Dim myQuery = MyTextBox _
    .ObserveTextChanged() _
    .Throttle(TimeSpan.FromSeconds(0.3))
Dim mySubscription = myQuery _
    .ObserveOn(Me) _
    .Subscribe(Function(text) DoSomethingWith(text))

这样做的原因是许多Rx操作符将隐式地改变可观察对象的调度程序。

例如:

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

将自动将查询使用的调度器更改为Scheduler.Default。因此，如果您的订阅者要更新UI，则必须在查询中添加另一个ObserveOn(control)。

因此，考虑它的基本方法应该是，如果您的订阅者更新UI，那么在Subscribe之前执行ObserveOn。

另外，在您的扩展方法中，我删除了DirectCast(a.Sender, TextBox)代码。你不需要它，因为textbox已经在作用域中了。

最后，您应该使用从Subscribe方法返回的IDisposable做一些事情。您必须在关闭表单时处置订阅，否则将会有许多事件处理程序连接起来防止垃圾收集。我通常创建一个类级别的CompositeDisposable来保存所有的订阅，这样我就可以一次处理所有的订阅。