ngДля не отображения данных из Observable Event Source

Question

В моем приложении Angular6 у меня проблема с отображением данных с помощью async ngfor.Я ожидаю, что некоторые серверные события отправлены из бэкэнда (просто объект с полем строки ответа).Хотя console.log показывает, что список ответов из службы содержит ответы, ngfor ничего не отображает.

Вот мой компонент:

import { Component } from '@angular/core';
import { Answer } from './answer';
import { AnswerReactiveService } from './answer-reactive.service';
import { Observable } from 'rxjs';

@Component({
  selector: 'app-answers',
  templateUrl: './answers.component.html',
  providers: [AnswerReactiveService],
  styleUrls: ['./answers.component.css']
})
export class AnswersComponent {
  answers: Observable<Answer[]>;

  constructor(private answerReactiveService: AnswerReactiveService) {
  }

  requestAnswerStream(): void {
    this.answers = this.answerReactiveService.getAnswerStream();
  }

}

Вот служба:

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

import { Answer } from './answer';

import { Observable } from 'rxjs';

@Injectable()
export class AnswerReactiveService {

  private answers: Answer[] = [];
  private url: string = 'http://localhost:8080/events/1';

  getAnswerStream(): Observable<Array<Answer>> {
    return Observable.create((observer) => {
      let url = this.url;
      let eventSource = new EventSource(url);
      eventSource.onmessage = (event) => {
        console.log('Received event: ', event);
        const json = JSON.parse(event.data);
        console.log(json);
        this.answers.push(new Answer(json['answer']));
        console.log(this.answers);
        observer.next(this.answers);
      };
      eventSource.onerror = (error) => {
        if (eventSource.readyState === 0) {
          console.log('The stream has been closed by the server.');
          eventSource.close();
          observer.complete();
        } else {
          observer.error('EventSource error: ' + error);
        }
      };
    });
  }
}

И HTML:

<div class="row">
  <div class="col-md-8">
    <p>
      <button (click)="requestAnswerStream()">Gimmie answers</button>
    </p>
    <ul>
      <li *ngFor="let answer of answers | async">{{answer.answer}}</li>
    </ul>
  </div>
</div>

Answer 1

Как только вы получите результат обслуживания, завершите последовательность, используя Наблюдатель.complete , после отправки значения с помощью Наблюдатель.next ()

eventSource.onmessage = (event) => {
        console.log('Received event: ', event);
        const json = JSON.parse(event.data);
        console.log(json);
        this.answers.push(new Answer(json['answer']));
        console.log(this.answers);
        observer.next(this.answers);
        observer.complete();
      };
   };

Внутренний компонент

export class AnswersComponent implements OnInit {
  answers: Observable<Answer[]>;

  constructor(private answerReactiveService: AnswerReactiveService) {
  }

  ngOnInit(): void {
        this.answerReactiveService.getAnswerStream().subscribe((response)=>{
        this.answers=response;
    });
  }
}

А затем в HTML вы можете просто перебрать ответы

 <li *ngFor="let answer of answers">{{answer.answer}}</li>

Answer 2

После некоторых исследований я достиг результата с помощью ngZone.Я нашел подобную проблему здесь: https://blog.octo.com/en/angular-2-sse-and-changes-detection/

, и теперь мой сервисный код выглядит так, и он работает правильно:

      eventSource.onmessage = (event) => {
        this.zone.run(() => {
          console.log('Received event: ', event);
          const json = JSON.parse(event.data);
          this.answers.push(new Answer(json['answer']));
          observer.next(this.answers);
        });
      }; 

Мне просто интересно, можно ли это решение воспринимать как полностью реактивное?Я не нашел другого способа сделать эту работу.

Answer 3

Это проблема синхронности.Observable.create работает синхронно и создает наблюдаемую.Обратный вызов из вашего EventSource действительно вызывается, но уже слишком поздно, чтобы создать созданную наблюдаемую.

Решение состоит в том, чтобы использовать Subject, который является членом класса обслуживания и будет сохраняться до тех пор, пока обратный вызов EventSource не сможет повлиятьЭто.Вам нужно будет «подключить» сервис к вашему компоненту, наблюдаемому в конструкторе компонента.Затем getAnswerStream() становится triggerAnswerStream(), который устанавливает события публикации событий в субъект.

Примерно так:

@Component({
  selector: 'app-answers',
  templateUrl: './answers.component.html',
  providers: [AnswerReactiveService],
  styleUrls: ['./answers.component.css']
})
export class AnswersComponent {
  answers: Observable<Answer[]>;

  constructor(private answerReactiveService: AnswerReactiveService) {
    this.answers = answerReactiveService.answerStreamSubject.asObservable();
  }

  requestAnswerStream(): void {
    this.answerReactiveService.getAnswerStream();
  }

}

и

@Injectable()
export class AnswerReactiveService {

  private answers: Answer[] = [];
  private url: string = 'http://localhost:8080/events/1';

  public answerStreamSubject: Subject<Array<Answer>>;

  triggerAnswerStream(): void {
      let url = this.url;
      let eventSource = new EventSource(url);
      eventSource.onmessage = (event) => {
        console.log('Received event: ', event);
        const json = JSON.parse(event.data);
        console.log(json);
        this.answers.push(new Answer(json['answer']));
        console.log(this.answers);
        this.answerStreamSubject.next(this.answers);
      };
      eventSource.onerror = (error) => {
        if (eventSource.readyState === 0) {
          console.log('The stream has been closed by the server.');
          eventSource.close();
          this.answerStreamSubject.complete();
        } else {
          this.answerStreamSubject.error('EventSource error: ' + error);
        }
      };
  }
}