我对使用Observable.Publish进行多播处理的生命周期有点困惑。如何正确使用connect?令人意外的是,我发现我不需要调用connect来启动多播观察者的订阅。
var multicast = source.Publish();
var field0 = multicast.Select(record => record.field0);
var field1 = multicast.Select(record => record.field1);
// Do I need t*emphasized text*o call here?
var disposable = multicast.connect()
// Does calling
disposable.Dispose();
// unsubscribe field0 and field1?
编辑
当我没有显式调用IConnectableObservable的Connect方法时,我为什么能够成功订阅是我的疑惑。然而,我在IConnectableObservable上调用了Await方法,该方法隐式地调用了Connect方法。
Public Async Function MonitorMeasurements() As Task
Dim cts = New CancellationTokenSource
Try
Using dialog = New TaskDialog(Of Unit)(cts)
Dim measurementPoints =
MeasurementPointObserver(timeout:=TimeSpan.FromSeconds(2)).
TakeUntil(dialog.CancelObserved).Publish()
Dim viewModel = New MeasurementViewModel(measurementPoints)
dialog.Content = New MeasurementControl(viewModel)
dialog.Show()
Await measurementPoints
End Using
Catch ex As TimeoutException
MessageBox.Show(ex.Message)
Catch ex As Exception
MessageBox.Show(ex.Message)
End Try
End Function
请注意,我的TaskDialog公开了一个名为CancelObserved的observable,用于在按下取消按钮时观察。
解决方案
解决方案由@asti发布的链接中提供。以下是来自该链接中RX团队的引用:
请注意,使用await会通过导致订阅来使observable序列变热。本次发布中包括IConnectableObservable的await支持,这将导致连接序列到其源以及订阅它。如果没有Connect调用,await操作将永远无法完成。