RXJS makes operator out of stream when subscription is called without a callback parameter

I am experimenting with a Redux-like pattern to maintain state in a service for a specific function. My intention is that a consumer can subscribe to separate data streams by calling getItem1()

or getItem2()

, and then call action methods such as initializeItem1()

or initializeItem2()

in a service as needed to dispatch actions that will change the state of the store. In a service, data can be retrieved either via an http call or something else, as shown in the provided plnkr.

https://plnkr.co/edit/f3HLp5C8ckIpjNsM8dTm?p=preview

In particular, I don't understand why, in my code example, the observable returned getItem1()

does not highlight the value unless I change this.service.initializeItem1().subscribe();

to this.service.initializeItem1().subscribe(() => {});

. Can someone please explain if this looks like a bug in rxjs, if there is a good / known explanation for why it behaves this way, or if I was just missing something? I appreciate any insight the community can provide me. Thank.

Component

//our root app component
import {Component, OnInit} from '@angular/core'
import {Observable} from 'rxjs/Rx'
import {SampleService} from './sample.service'

@Component({
  selector: 'my-app',
  template: `
    <div>
      <h2>Item 1 Value</h2>
      <div>{{ item1 }}</div>

      <h2>Item 2 Value</h2>
      <div>{{ item2 }}</div>
    </div>
  `,
})
export class App implements OnInit {
  name:string;
  item1: string;
  item2: string;
  constructor(private service: SampleService) { }

  ngOnInit() {
    this.item1 = this.service.getItem1().subscribe(item1 => this.item1 = item1);
    this.item2 = this.service.getItem2().subscribe(item2 => this.item2 = item2);

    /* broken */
    //Item 1 does not display
    //Item 2 displays as expected
    this.scenario1();

    /* works */
    //Item 1 displays as expected
    //Item 2 displays as expected
    // this.scenario2();
  }

  private scenario1() {
    this.service.initializeItem1().subscribe();
    this.service.initializeItem2().subscribe();
  }

  private scenario2() {
    this.service.initializeItem1().subscribe(() => {});
    this.service.initializeItem2().subscribe();
  }
}

      

        
        
        
      

    

Service

import {Injectable} from '@angular/core'
import {Observable, Subject} from 'rxjs/Rx'
import { Http, Headers } from '@angular/http';

@Injectable()
export class SampleService {
  constructor(private http: Http) { }
  private headers: Headers = new Headers({ 'Content-Type': 'application/json' });
  private _store: Subject<any> = new Subject<any>();
  private store = this._store.asObservable()
      .startWith({})
      .scan(reducer);

  public initializeItem1(): Observable<string> {
    return this.http.get('api/foobar')
      .map(response => response.json().item).do(data => {
        console.log('not hit in scenario 1, but hit in scenario 2:', data);

        this._store.next({
          type: 'INITIALIZE_1',
          payload: data
        });
      });
  }

  public initializeItem2(): Observable<string> {
    return Observable.of('foobar-2').do(data => {
      console.log('hit in scenario 1 and scenario 2:', data);

      this._store.next({
        type: 'INITIALIZE_2',
        payload: data
      });
    });
  }

  getItem1(): Observable<string> {
        return this.store.map(store => store.settings1);
    }

  getItem2(): Observable<string> {
      return this.store.map(store => store.settings2);
  }
}

function reducer(store, action) {
    switch(action.type) {
        case 'INITIALIZE_1':
            return Object.assign({ }, store, {
                initialSettings: action.payload,
                settings1: action.payload
            });
        case 'INITIALIZE_2':
            return Object.assign({ }, store, { settings2:action.payload });
        default:
            return store;
    }
}