title: Execute logic when RxJS 5 refCount() connects to or unsubscribes from source
body: <p>According to the RxJS 5 manual's section on <a href="http://reactivex.io/rxjs/manual/overview.html#multicasted-observables" rel="nofollow noreferrer">Multicasting</a></p>
  
  <blockquote>
    <p>...we can use ConnectableObservable's refCount() method (reference counting), which returns an Observable that keeps track of how many subscribers it has. When the number of subscribers increases from 0 to 1, it will call connect() for us, which starts the shared execution. Only when the number of subscribers decreases from 1 to 0 will it be fully unsubscribed, stopping further execution.</p>
  </blockquote>
  
  <p>I'd like to understand whether it's possible to hook into each of these events and execute some logic, ideally before the source observable's <code>connect()</code> or <code>unsubscribe()</code> occurs, but even after the fact would be acceptable. </p>
  
  <p>If there's no way to do this when using the <code>refCount()</code> operator, it'd be much appreciated if you could provide an example how one could achieve this with a custom operator.</p>
  
  <p>I thought maybe I could somehow use the completeFn from <code>do(nextFn,errFn,completeFn)</code> to hook into this, but doesn't seem to work as shown by the below snippet.</p>
  
  <p><div class="snippet" data-lang="js" data-hide="false" data-console="true" data-babel="true">
  <div class="snippet-code">
  <pre class="snippet-code-js lang-js prettyprint-override"><code>var source = Rx.Observable.interval(500)
    .do(
      (x) =&gt; console.log('SOURCE emitted ' + x),
      (err) =&gt; console.log('SOURCE erred ' + err),
      () =&gt; console.log('SOURCE completed ')
    );
  var subject = new Rx.Subject();
  var refCounted = source.multicast(subject).refCount();
  var subscription1, subscription2, subscriptionConnect;
  
  // This calls `connect()`, because
  // it is the first subscriber to `refCounted`
  console.log('observerA subscribed');
  subscription1 = refCounted.subscribe({
    next: (v) =&gt; console.log('observerA: ' + v)
  });
  
  setTimeout(() =&gt; {
    console.log('observerB subscribed');
    subscription2 = refCounted.subscribe({
      next: (v) =&gt; console.log('observerB: ' + v)
    });
  }, 600);
  
  setTimeout(() =&gt; {
    console.log('observerA unsubscribed');
    subscription1.unsubscribe();
  }, 1200);
  
  // This is when the shared Observable execution will stop, because
  // `refCounted` would have no more subscribers after this
  setTimeout(() =&gt; {
    console.log('observerB unsubscribed');
    subscription2.unsubscribe();
  }, 2000);</code></pre>
  <pre class="snippet-code-html lang-html prettyprint-override"><code>&lt;script src="https://unpkg.com/@reactivex/rxjs@5.0.3/dist/global/Rx.js"&gt;&lt;/script&gt;</code></pre>
  </div>
  </div>
  </p>