title: RxJs: take action on first and last subscriber to observable
body: <p>In my Angular 2.x application, I have a component which subscribes to an observable exposed by a service. The observable data service is currently implemented with polling HTTP requests but that will change to a reactive WS implementation. I'd like to keep the polling out of the components so the question is: how can I take an action, like HTTP polling via <code>setInterval</code>, once I have subscribers? And, how can I take an action, like <code>clearInterval</code> when there are no more subscribers?</p>
  
  <p>The view is asynchronously updated:</p>
  
  <pre><code>  &lt;div *ngFor="let headline of headlines$ | async"&gt;
      &lt;md-list-item (click)="onSelect(headline)"&gt;
        &lt;h4 md-line&gt;{{headline.title}}&lt;/h4&gt;
      &lt;/md-list-item&gt;
      &lt;md-divider&gt;&lt;/md-divider&gt;
    &lt;/div&gt;
  </code></pre>
  
  <p>The component gets the headline observable from the service:</p>
  
  <pre><code>export class HeadlinesComponent implements OnInit, OnDestroy {
    constructor(
        private _headlineService: HeadlineService,
    ) {
    }
  
    headlines$: Observable&lt;Headline[]&gt;;
  
    ngOnInit() {
      this.headlines$ = this._headlineService.headlines$;
    }
  }
  </code></pre>
  
  <p>And the service is implemented as an observable data store:</p>
  
  <pre><code>@Injectable()
  export class HeadlineService {
    constructor(
      private _http: Http) {
      _init();
    }
  
    private _headlines$: BehaviorSubject&lt;Headline[]&gt; = 
      new BehaviorSubject&lt;Headline[]&gt;([]);
  
    get headlines$(): Observable&lt;Headline[]&gt; {
      return this._headlines$.asObservable();
    }
  
    private _store: {
      headlines: Headline[]
    } = { headlines: [] };
  
    private _save(headlines: Headline[]) {
      this._store.headlines = headlines;
      this._headlines$.next(Object.assign({}, this._store).headlines);
    }
  }
  </code></pre>
  
  <p>and the store is updated via HTTP:</p>
  
  <pre><code>  private _interval;
  
    private _init() {
      let self = this;
      this._interval = setInterval(function(){ self._loadHeadlines();}, 5000);
    }
  
    private _loadHeadlines(): Observable&lt;Headline[]&gt; {
      let self = this;
      let observable: Observable&lt;Headline[]&gt; = this._http
        .get(url, options)
        .map(response =&gt; response.json()._embedded.headlines)
        .share();
  
      observable.subscribe(
        headlines =&gt; self._save(headlines)
        );    
  
      return observable;
    }
  </code></pre>
  
  <p>As implemented, the headline service starts polling on construction instead of delaying until a subscriber is present. To get the lifecycle correct, I could expose the service's <code>_loadHeadlines</code> functionality and start polling when the component's <code>ngOnInit</code> is triggered (and stop in <code>ngOnDestroy</code>) but that exposes the service implementation details to the component (these details will change when I move to a web-sockets implementation) and I'd need have an accurate count of the number of subscribers.</p>
  
  <p>So, is there a nice RxJs-ish way to solve this?</p>