Использование RxJs и Angular 2 для обработки событий, отправленных сервером

Question

Я пытаюсь отображать события, отправленные сервером, исходящие значения в угловом приложении 2/RxJs.

Бэкэнд регулярно отправляет отдельные строки клиенту через события, отправленные сервером.

Я не уверен, как бороться с полученными значениями на угловой стороне 2/RxJs.

Вот мой клиент (нг компонент):

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); 
    } 
} 

Метод Бэкэнд следующим образом (и использует 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()) 
     ); 
    } 

Я просто заметил, что приложение висит на запросе и что на странице ничего не отображается.

Я подозреваю, что есть проблема с моим использованием метода карты, то есть .map(this.extractData).

Я просто хотел бы добавить входящие строки в массив и вывести этот массив в шаблон, который будет обновлять, как струны бывают.

Может кто-нибудь, пожалуйста, помогите?

редактировать: Здесь рабочий раствор (благодаря ответу Тьерри ниже):

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

Вы не можете использовать класс Http из Angular2 для обработки на стороне сервера событий, поскольку он основан на объекте XHR ,

Вы можете использовать объект EventSource:

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

Смотрите эти статьи:

• 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

Вот рабочий пример:

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

Чтобы добавить Thierry's answer, По умолчанию тип события 'сообщение'. Однако тип события может быть что угодно («чат», «журнал» и т. Д.) На основе реализации на стороне сервера. В моем случае первые два события с сервера были «message», а остальные были «log». Мой код выглядит, как показано ниже

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