Multiple Observable subscribers

Question

I am trying to replace an event pattern using Observables. The general idea is to be able to monitor changes to the localStorage. As seen below, for this I have created an Angular Injectable as a wrapper: StorageService. It works as expected: Items can be written and read. If an item gets updated, a reader gets the update as well. However, once multiple readers subscribe to the same key, only one is getting notified. Why is this happening? I am expecting every instance reading an item from a key using service.read('key') to end up with the same observable. With the test shown at the end, this does not seem to be the case.

StorageService:

import { Injectable } from '@angular/core';
import { Observable } from 'rxjs/Observable';
import 'rxjs/add/observable/concat';
import { Observer } from 'rxjs/Observer';

interface StorageItemsDictionary { [id: string]: any; }
interface ObserverObservableDictionary { [id: string]: { observer: Observer<any>, observable: Observable<any> }; };

@Injectable()
export class StorageService {

  cache: StorageItemsDictionary;
  ood: ObserverObservableDictionary;

  constructor() {
    this.cache = {};
    this.ood = {};
  }

  /**
   * Get item from current `key` and all further changes
   * @param key
   */
  public read<T>(key: string): Observable<T | null> {
    // assure there is an ObserverObservable pair for the key:
    this.assureOOPair(key);

    // create observable from the cache or localStorage if there is an item
    if (this.cache[key] === undefined) {
      const value = localStorage.getItem(key);
      const parsed = <T>JSON.parse(localStorage.getItem(key));
      this.cache[key] = parsed;
    }

    // one time observable for current item:
    const singleTimeObservable = new Observable(singleTimeObserver => {
      singleTimeObserver.next(this.cache[key] || null);
      singleTimeObserver.complete();
    });

    // merge it with the ood
    return <Observable<T | null>>Observable.concat(singleTimeObservable, this.ood[key].observable);
  }

  /**
   * Delete item from `key`
   * @param key
   */
  public delete(key: string): void {
    delete this.cache[key];
    this.ood[key].observer.next(null);
    localStorage.removeItem(key);
  }

  /**
   * Store `item` at `key` and notify all readers
   * @param key
   * @param item
   */
  public write<T>(key: string, item: T): void {
    this.assureOOPair(key);
    this.cache[key] = item;
    this.ood[key].observer.next(item); // notify others
    const stringified = JSON.stringify(item);
    localStorage.setItem(key, stringified);
  }

  /**
   * Create a ObserverObservable pair at `key` if it does not exist
   * @param key
   */
  private assureOOPair(key: string) {
    if (this.ood[key] === undefined) {
      this.ood[key] = {
        observable: null,
        observer: null
      };
      this.ood[key].observable = new Observable(observer => {
        this.ood[key].observer = observer;
      });
      this.ood[key].observable.subscribe(e => e); // to be able to apply next
    }
  }

}

Test:

  it('multiple reads', async(inject([StorageService], (service: StorageService) => {
    localStorage.removeItem('key');
    let counter = 0;
    let counter2 = 0;
    service.write('key', 'value1');

    service.read('key').forEach(v => {
      console.log('multiple reads: 1', v);
      switch (counter) {
        case 0:
          expect(v).toEqual('value1');
          break;
        case 1:
          expect(v).toEqual('value2fail'); // this case gets never called. As a result, the test passes, even though it should fail here…
          break;
      }
      counter++;
    });

    service.read('key').forEach(v => {
      console.log('multiple reads: 2', v);
      switch (counter2) {
        case 0:
          expect(v).toEqual('value1');
          break;
        case 1:
          expect(v).toEqual('value2');
          break;
      }
      counter2++;
    });

    service.write('key', 'value2');
  })));

Test's Log:

multiple reads: 1 value1
multiple reads: 2 value1
multiple reads: 2 value2

Answer 1

I think you are looking for Subject. Subjects are like observables but you are sharing the same data with all subscribers. observable will get executed for each subscription. I hope that helps.