Using Observable.Publish with reactive extensions
.net, system.reactive
Solution
`Publish` on a source returns an `IConnectableObservable<T>` which is essentially `IObservable<T>` with a `Connect`method . You can use `Connect` and the `IDisposable` it returns to control the subscription to the source.
Rx is designed to be a fire and forget system. Subscriptions won't be terminated until you explicitly dispose of them, or they complete/error.
i.e., `disp0 = field0.Subscribe(...); disp1 = field1.Subscribe(...)` - the subscriptions won't be terminated until `disp0, disp1` are explicitly disposed - which is independent of the connection to the multicast source.
You can connect and disconnect without disturbing the pipeline below. An easier way to not worry about manually managing the connection is to using `.Publish().RefCount()` which will maintain a connection as long as at least one observer is still subscribed to it. This is known as warming up an observable.
UPDATED FOR AN EDIT IN THE QUESTION
OP was calling `await` on the `IConnectableObservable<T>`.
From Release notes for Rx:
..the use of await makes an observable sequence hot by causing a subscription to take place. Included in this release is await support for IConnectableObservable, which causes connecting the sequence to its source as well as subscribing to it. Without the Connect call, the await operation would never complete.
Example (taken from the same page)
static async void Foo()
{
var xs = Observable.Defer(() =>
{
Console.WriteLine("Operation started!");
return Observable.Interval(TimeSpan.FromSeconds(1)).Take(10);
});
var ys = xs.Publish();
// This doesn't trigger a connection with the source yet.
ys.Subscribe(x => Console.WriteLine("Value = " + x));
// During the asynchronous sleep, nothing will be printed.
await Task.Delay(5000);
// Awaiting causes the connection to be made. Values will be printed now,
// and the code below will return 9 after 10 seconds.
var y = await ys;
Console.WriteLine("Await result = " + y);
}
Problem
I'm a little confused about the lifecycle of using Observable.Publish for multicast handling. How should one use connect correctly? Against intuition I've found I do not need to call connect for the multicast observers to start their subscriptions. ``` 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? ``` EDIT My puzzle was why I was successfully subscribing when I was not calling Connect on the IConnectableObservable explicity. However I was calling Await on the IConnectableObservable which implicitly calls 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 ``` Note my TaskDialog exposes an observable called CancelObserved for when the cancel button is pressed. SOLUTION The solution is posted in a link by @asti. Here is a quote from the RX team in that link Notice use of await makes an observable sequence hot by causing a subscription to take place. Included in this release is await support for IConnectableObservable, which causes connecting the sequence to its source as well as subscribing to it. Without the Connect call, the await operation would never complete