Verwenden von RxJs und Angular 2, um mit serverseitigen Ereignissen umzugehen

Question

Ich versuche servergesendete Ereignisse in einer eckigen 2/RxJs App anzuzeigen.

Das Back-End sendet regelmäßig einzelne Zeichenfolgen über serverseitige Ereignisse an den Client.

Ich bin nicht sicher, wie mit den abgerufenen Werten auf der eckigen 2/RxJs Seite umzugehen ist.

Hier ist mein Kunde (a ng Komponente):

import {Component, OnInit} from 'angular2/core'; 
import {Http, Response} from 'angular2/http'; 
import 'rxjs/Rx'; 
import {Observable}  from 'rxjs/Observable'; 

@Component({ 
    selector: 'my-app', 
    template: `<h1>My second Angular 2 App</h1> 
    <ul> 
     <li *ngFor="#s of someStrings | async"> 
      a string: {{ s }} 
     </li> 
    </ul> 
    ` 
}) 
export class AppComponent implements OnInit { 

    constructor(private http:Http) { 
    } 

    errorMessage:string; 
    someStrings:string[]; 

    ngOnInit() { 
     this.getSomeStrings() 
      .subscribe(
       aString => this.someStrings.push(aString), 
       error => this.errorMessage = <any>error); 
    } 

    private getSomeStrings():Observable<string> { 
     return this.http.get('interval-sse-observable') 
      .map(this.extractData) 
      .catch(this.handleError); 
    } 

    private extractData(res:Response) { 
     if (res.status < 200 || res.status >= 300) { 
      throw new Error('Bad response status: ' + res.status); 
     } 
     let body = res.json(); 
     return body || {}; 
    } 

    private handleError(error:any) { 
     // In a real world app, we might send the error to remote logging infrastructure 
     let errMsg = error.message || 'Server error'; 
     console.error(errMsg); // log to console instead 
     return Observable.throw(errMsg); 
    } 
} 

Die Backend-Methode ist wie folgt (und verwendet RxJava):

@ResponseStatus(HttpStatus.OK) 
    @RequestMapping(method = RequestMethod.GET, path = "interval-sse-observable") 
    public SseEmitter tickSseObservable() { 
     return RxResponse.sse(
       Observable.interval(5, TimeUnit.SECONDS, Schedulers.io()) 
         .map(tick -> randomUUID().toString()) 
     ); 
    } 

Ich habe gerade bemerkt, dass der App auf Antrag hängt und dass auf der Seite nichts angezeigt wird.

Ich vermute, es gibt ein Problem mit meiner Verwendung der Kartenmethode, d. H. .map(this.extractData).

Ich möchte nur die eingehenden Strings dem Array hinzufügen und in der Vorlage dieses Array angezeigt werden, die als die Strings kommen aktualisieren würde.

Kann jemand bitte helfen?

bearbeiten: Hier ist eine Arbeitslösung (dank Thierry Antwort unten):

import {Component, OnInit} from 'angular2/core'; 
import 'rxjs/Rx'; 

@Component({ 
    selector: 'my-app', 
    template: `<h1>My second Angular 2 App</h1> 
    <ul> 
     <li *ngFor="#s of someStrings"> 
      a string: {{ s }} 
     </li> 
    </ul> 
    ` 
}) 
export class AppComponent implements OnInit { 

    someStrings:string[] = []; 

    ngOnInit() { 
     let source = new EventSource('/interval-sse-observable'); 
     source.addEventListener('message', aString => this.someStrings.push(aString.data), false); 
    } 
} 

Answer 1

Sie können nicht die Http Klasse von Angular2 verwenden serverseitige Ereignisse zu behandeln, da es auf dem XHR Objekt basiert .

Sie könnten das Objekt Eventsource nutzen:

var source = new EventSource('/...'); 
source.addListener('message', (event) => { 
    (...) 
}); 

diese Artikel Siehe:

• https://dzone.com/articles/html5-server-sent-events
• http://www.nurkiewicz.com/2015/07/server-sent-events-with-rxjava-and.html?m=1

Answer 2

Hier ist ein funktionierendes Beispiel:

SseService

import {Injectable} from '@angular/core'; 
import {Observable} from 'rxjs/Observable'; 

declare var EventSource; 

@Injectable() 
export class SseService { 

    constructor() { 
    } 

    observeMessages(sseUrl: string): Observable<string> { 
     return new Observable<string>(obs => { 
      const es = new EventSource(sseUrl); 
      es.addEventListener('message', (evt) => { 
       console.log(evt.data); 
       obs.next(evt.data); 
      }); 
      return() => es.close(); 
     }); 
    } 
} 

AppComponent

import {Component, OnDestroy, OnInit} from '@angular/core'; 
import {SseService} from './shared/services/sse/sse.service'; 
import {Observable, Subscription} from 'rxjs/Rx'; 

@Component({ 
    selector: 'my-app', 
    template: `<h1>Angular Server-Sent Events</h1> 
    <ul> 
     <li *ngFor="let message of messages"> 
      {{ message }} 
     </li> 
    </ul> 
    ` 
}) 
export class AppComponent implements OnInit, OnDestroy { 
    private sseStream: Subscription; 
    messages:Array<string> = []; 

    constructor(private sseService: SseService){ 
    } 

    ngOnInit() { 
     this.sseStream = this.sseService.observeMessages('https://server.com/mysse') 
         .subscribe(message => { 
          messages.push(message); 
         }); 
    } 

    ngOnDestroy() { 
     if (this.sseStream) { 
      this.sseStream.unsubscribe(); 
     } 
    } 
} 

Answer 3

Um zu Thierry's answer, durch Vorgabe der Ereignistyp ist 'message' hinzuzufügen. Der Ereignistyp könnte jedoch basierend auf der serverseitigen Implementierung etwas wie ('chat', 'log' usw.) sein. In meinem Fall waren die ersten beiden Ereignisse vom Server "Nachricht" und der Rest von ihnen waren "Protokoll". Mein Code sieht wie unten

var source = new EventSource('/...'); source.addListener('message', message => { (...) }); source.addListener('log', log => { (...) });