Using Observable.Publish with reactive extensions

Question

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

Answer 1

Publish on a source returns an IConnectableObservable<T> which is essentially IObservable<T> with a Connectmethod . 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);
}

Answer 2

Publish allows you to share a subscription. This obviously is most useful for making a Cold observable sequence Hot. i.e. Taking a sequence that causes some subscription side effect (a connection to a network perhaps) to occur and ensuring that the side effect is performed once and the results of the sequence shared among the consumers.

In practice, you call publish on your cold sequence, subscribe your consumers and then Connect the published sequence after the subscriptions to mitigate any race conditions.

So basically, what you have done above.

It is largely pointless for already Hot sequences e.g. Subjects, FromEventPattern or something that is already published and connected.

Disposing of the value from the Connect() method will 'disconnect' the sequence preventing the consumers from getting any more values. You can also dispose of the consumer subscriptions if any of these want to detach earlier.

Having said all of this, you appear to be doing the correct thing. What is the problem you are seeing? I am assuming that you are connecting to an already HOT sequence.