title: Creating an RxJS Observable from a (server sent) EventSource
body: <p>I would like to create a RxJs Observable from an EventSource (server sent events).</p>
  
  <p>I tried the following:</p>
  
  <pre><code>import {Component, OnInit} from 'angular2/core';
  import {Subject, Observable}  from 'rxjs/Rx';
  
  @Component({
      selector: 'my-app',
      template: `&lt;h1&gt;My second Angular 2 App&lt;/h1&gt;
      &lt;ul&gt;
          &lt;li *ngFor="#s of someStrings"&gt;
             a string: {{ s }}
          &lt;/li&gt;
      &lt;/ul&gt;
      `
  })
  export class AppComponent implements OnInit {
  
      someStrings:string[] = [];
  
      ngOnInit() {
          let eventSource = new EventSource('/interval-sse-observable');
          let observable = Observable.create(eventSource);
          observable.subscribe({
              next: aString =&gt; this.someStrings.push(aString.data),
              error: err =&gt; console.error('something wrong occurred: ' + err)
          });
      }
  }
  </code></pre>
  
  <p>But I get the following exception:</p>
  
  <pre><code>EXCEPTION: Error: Uncaught (in promise): EXCEPTION: TypeError: this._subscribe is not a function in [null]
  ORIGINAL EXCEPTION: TypeError: this._subscribe is not a function
  ORIGINAL STACKTRACE:
  TypeError: this._subscribe is not a function
      at Observable.subscribe (https://cdnjs.cloudflare.com/ajax/libs/angular.js/2.0.0-beta.15/Rx.js:11210:29)
      at AppComponent.ngOnInit (http://localhost:8080/scripts/app.component.ts!transpiled:30:28)
      at AbstractChangeDetector.ChangeDetector_HostAppComponent_0.detectChangesInRecordsInternal (viewFactory_HostAppComponent:21:99)
      at AbstractChangeDetector.detectChangesInRecords (https://code.angularjs.org/2.0.0-beta.15/angular2.dev.js:9689:14)
      at AbstractChangeDetector.runDetectChanges (https://code.angularjs.org/2.0.0-beta.15/angular2.dev.js:9672:12)
      at AbstractChangeDetector.detectChanges (https://code.angularjs.org/2.0.0-beta.15/angular2.dev.js:9661:12)
      at ChangeDetectorRef_.detectChanges (https://code.angularjs.org/2.0.0-beta.15/angular2.dev.js:5280:16)
      at https://code.angularjs.org/2.0.0-beta.15/angular2.dev.js:13048:27
      at Array.forEach (native)
      at ApplicationRef_.tick (https://code.angularjs.org/2.0.0-beta.15/angular2.dev.js:13047:34)
  </code></pre>
  
  <p>For completeness' sake, here is the contents of my index.html:</p>
  
  <pre><code>&lt;!DOCTYPE html&gt;
  &lt;html lang="en"&gt;
  &lt;head&gt;
      &lt;meta charset="UTF-8"&gt;
      &lt;title&gt;sse demo&lt;/title&gt;
      &lt;!-- 1. Load libraries --&gt;
      &lt;script src="https://cdnjs.cloudflare.com/ajax/libs/angular.js/2.0.0-beta.15/angular2-polyfills.js"&gt;&lt;/script&gt;
      &lt;script src="https://code.angularjs.org/tools/system.js"&gt;&lt;/script&gt;
      &lt;script src="https://code.angularjs.org/tools/typescript.js"&gt;&lt;/script&gt;
      &lt;script src="https://cdnjs.cloudflare.com/ajax/libs/angular.js/2.0.0-beta.15/Rx.js"&gt;&lt;/script&gt;
      &lt;script src="https://code.angularjs.org/2.0.0-beta.15/angular2.dev.js"&gt;&lt;/script&gt;
      &lt;script src="https://cdnjs.cloudflare.com/ajax/libs/angular.js/2.0.0-beta.15/http.dev.js"&gt;&lt;/script&gt;
  
      &lt;!-- 2. Configure SystemJS --&gt;
      &lt;script&gt;
          System.config({
              transpiler: 'typescript',
              typescriptOptions: { emitDecoratorMetadata: true }
          });
          System.import('./scripts/app.ts')
                  .then(null, console.error.bind(console));
      &lt;/script&gt;
  &lt;/head&gt;
  &lt;body&gt;
  
  &lt;my-app&gt;Loading...&lt;/my-app&gt;
  
  &lt;/body&gt;
  &lt;/html&gt;
  </code></pre>
  
  <p>Can someone please help?</p>
  
  <p><strong>edit 1</strong>: Following Yurzui's advice, I modified my code as follows:</p>
  
  <pre><code>ngOnInit() {
      let observable = Observable.create(observer =&gt; {
          const eventSource = new EventSource('/interval-sse-observable');
          eventSource.onmessage = x =&gt; observer.next(console.log(x));
          eventSource.onerror = x =&gt; observer.error(console.log('EventSource failed'));
  
          return () =&gt; {
              eventSource.close();
          };
      });
      observable.subscribe({
          next: aString =&gt; this.someStrings.push(aString.data),
          error: err =&gt; console.error('something wrong occurred: ' + err)
      });
  }
  </code></pre>
  
  <p>It does log the first message in the console as follows:</p>
  
  <pre><code>MessageEvent {isTrusted: true, data: "c374a15b-b37d-498e-8ab0-49643b79c1bb", origin: "http://localhost:8080", lastEventId: "", source: null…}bubbles: falsecancelBubble: falsecancelable: falsecurrentTarget: EventSourcedata: "c374a15b-b37d-498e-8ab0-49643b79c1bb"defaultPrevented: falseeventPhase: 0isTrusted: trueisTrusted: truelastEventId: ""origin: "http://localhost:8080"path: Array[0]ports: nullreturnValue: truesource: nullsrcElement: EventSourcetarget: EventSourcetimeStamp: 6257.125type: "message"__proto__: MessageEvent
  Rx.js:10982 Uncaught TypeError: Cannot read property 'data' of undefinedSystem.register.exports_1.execute.AppComponent.ngOnInit.observable.subscribe.next @ app.component.ts:29SafeSubscriber.__tryOrUnsub @ Rx.js:10979SafeSubscriber.next @ Rx.js:10934Subscriber._next @ Rx.js:10894Subscriber.next @ Rx.js:10871System.register.exports_1.execute.AppComponent.ngOnInit.Rx_1.Observable.create.eventSource.onmessage @ app.component.ts:21
  </code></pre>
  
  <p>Now if instead of logging the <code>x</code> variable in the console I just pass it to the next method as follows:</p>
  
  <pre><code>eventSource.onmessage = x =&gt; observer.next(x);
  </code></pre>
  
  <p>The server sent events are retrieved by the client (I see them in the chrome dev tools) but nothing is displayed in the template indicating the array of strings is not populated...</p>
  
  <p>By the way I had to remove the <code>JSON.parse(x.data)</code> as it was causing an error.</p>