ReactiveCommand and ConnectableObservable

Question

I'd like to wire a ReactiveCommand up to a ConnectableObservable, since the observable should be connectable by multiple subscribers.

The problem is that ReactiveCommand will only call the Subscribe method on the ConnectableObservable, rather than the Connect method as I would like.

The code below demonstrates what I'm trying to achieve. Notice the StartOrConnectService method that is called by service code.

public class Foo
{
    public IConnectableObservable<string> Connection { get; }

    public Foo(BarService barService)
    {
        Connection = Observable.Create<string>(observer =>
        {
            disp = barService.BarStream().Subscribe(
              bar => { 
                Console.WriteLine($"{bar}");
                observable.onNext("Connected");
              },
              ex => observable.onNext("Errored"),
              () => observable.onNext("Disconnected")
            );
            return disp;
        }).Publish();
    }

    // This method is called by service code.
    public IDisposable StartOrConnectService()
    {
        // bunch of other stuff going on here, but essentially calling connect
        return Connection.Connect();
    }

}

public sealed class FooViewModel : ReactiveObject
{
    public ReactiveCommand<Unit, string> ConnectCommand { get; }

    public FooViewModel(Foo foo)
    {
        ConnectCommand = ReactiveCommand
            .CreateFromObservable(() => foo.Connection);
    }
}     

Is there some way of adapting or wrapping a ConnectableObservable to a regular Observable so the ConnectableObservable.Connect method is called when the ReactiveCommand executes?

Answer 1

Aha, turns out I was looking for the RefCount extension method (introtorx), that converts an IConnectableObservable back into an IObservable, but magically implements the Connect semantics. #loveRX