为什么基于Angular observable的SseService只工作一次，接收事件却先产生事件？

Question

我已经使用这个模式来创建ServerSentEvent服务。Angular和服务器发送事件这里是服务代码。

import { Injectable, NgZone } from "@angular/core";
import { Observable } from "rxjs";

@Injectable({
  providedIn: "root"
})
export class SseService {
  constructor(private _zone: NgZone) { }

  getServerSentEvent(url: string): Observable<any> {
    return Observable.create(observer => {
      const eventSource = this.getEventSource(url);
      eventSource.onmessage = event => {
        this._zone.run(() => {
          observer.next(event.data);
          console.log(event); // Here I see every received  event
        });
      };
      eventSource.onerror = error => {
        this._zone.run(() => {
          observer.error(error);
          console.log(error);
        });
      };
      // return () => eventSource.close();
    });
  }

  private getEventSource(url: string): EventSource {
    return new EventSource(url);
  }
}

然后我在主组件中调用这个ss-events流。

ngOnInit() { //This subscribe works just for the first event no more
  this.sseService.getServerSentEvent("http://localhost:8080/ssevents")
    .subscribe(
      data => {
        this.eventmessage = data;
      },
      error => this.error = error
    );

  // But direct next EventSource onmessage() call works here: 
  this.source = new EventSource("http://localhost:8080/ssevents");
  this.source.onmessage = event => {
    this.zone.run(() => {
      this.eventmessage = event.data;
      console.log(event);
    });
  };
  this.source.onerror = error => {
    this.zone.run(() => {
      this.error = error.type;
      console.log(error);
    });
  };
}

为什么ssservice的内部观测器不能工作？只有第一个选项，所以我需要使用第二个选项。所以我需要使用第二个选项--但是在那里我收到了永久错误的EventTarget（每一秒或半秒），除了Type=error，EventPhase=2之外，没有错误的原因描述。但是，在插入到mongodb之后，服务器就收到了由其产生的事件。

Answer 1

可能有几个问题

1. 尝试用addEventListener代替onmessage。
2. 确认信息格式 联系
3. 浏览器会中断连接，如果你有移动然后6连接到同一主机。联系

所以，当你在测试时，你可以很快地超过它。

    @Injectable()
    export class SseService {
      private eventSource: EventSource;

      constructor(private zone: NgZone) {}

      getServerSentEvent(url: string): Observable<MessageEvent> {
        return Observable.create(observer => {
          const eventSource = this.getEventSource(url);
          eventSource.onopen = (ev) => {
            console.log('Connection to server opened.', ev);
          };
          eventSource.onerror = (ev) => {
            console.log('EventSource failed.', ev);
          };
          eventSource.addEventListener('message', event => {
            this.zone.run(() => {
              observer.next(event);
            });
          });
        });
      }

      private getEventSource(url: string): EventSource {
        if (this.eventSource) {
          console.log('EventSource closed.');
          this.eventSource.close();
        }
        this.eventSource = new EventSource(url);
        return this.eventSource;
      }
    }

我的试验 例子