title: RxJS Observable replace stream with new data
body: <p>I have a service which updates an observable's data upon initialization, and then upon a http search.</p>
  
  <p>Using next() adds records to the array in the observable, but how would I go about replacing those records with a new set? </p>
  
  <p>search.service.ts</p>
  
  <pre><code> private searchSource = new BehaviorSubject&lt;any&gt;(this.getInitialCollections());
  
      public currentState = this.searchSource.asObservable();
  
      public getInitialCollections() {
          return this.apiService.post(this.searchApi)
              .subscribe(data =&gt; this.changeSubject(data));
      }
  
      public search(phrase: string) {
          return this.apiService.post(this.searchApi, { phrase: phrase })
              .subscribe(data =&gt; this.changeSubject(data));
      }
  
      private changeSubject(results: object) {
          this.readCollectionResult(results).map(data =&gt; this.searchSource.next(data));
      }
  
      private readCollectionResult(result: any): Asset[] {
          return (result &amp;&amp; result.data) ? result.data.map(asset =&gt; asset) : [];
      }
  </code></pre>
  
  <p>component.ts</p>
  
  <pre><code>export class SearchDataSet {
      /** Stream that emits whenever the data has been modified. */
      dataChange: BehaviorSubject&lt;Asset[]&gt; = new BehaviorSubject&lt;Asset[]&gt;([]);
      get data(): Asset[] { return this.dataChange.value; }
  
      constructor(
          private _searchService: SearchService
      ) {
          this.initialize();
      }
  
      records = this.data.slice();
  
      initialize() {
          this._searchService.currentState
              .subscribe(item =&gt; !item.id ? null : this.addRecord(item));
      }
  
      /** Adds a new asset to the database. */
      addRecord(item: object) {
          this.records.push(item);
          this.dataChange.next(this.records);
      }
  }
  </code></pre>