Angular 8-处理SSE出错时重新连接

Question

我正在研究Angular 8（使用Electron 6和Ionic 4）项目，现在我们处于评估阶段，在此阶段，我们决定是使用SSE（服务器发送的事件）还是Web套接字替换轮询。我的工作是研究SSE。

我创建了一个小型快递应用程序，该应用程序生成随机数，并且一切正常。唯一困扰我的是在服务器错误时重新连接的正确方法。

我的实现看起来像这样：

  private createSseSource(): Observable<MessageEvent> {
    return Observable.create(observer => {
      this.eventSource = new EventSource(SSE_URL);
      this.eventSource.onmessage = (event) => {
        this.zone.run(() => observer.next(event));
      };

    this.eventSource.onopen = (event) => {
      console.log('connection open');
    };

    this.eventSource.onerror = (error) => {
      console.log('looks like the best thing to do is to do nothing');
      // this.zone.run(() => observer.error(error));
      // this.closeSseConnection();
      // this.reconnectOnError();
    };
  });
}

我试图在此reconnectOnError()之后实现answer功能，但是我无法使其工作。然后我放弃了reconnectOnError()函数，这似乎是一个更好的选择。不要尝试关闭并重新连接，也不要将错误传播到可观察到的地方。只需坐下等待，当服务器再次运行时，它将自动重新连接。

问题是，这真的是最好的方法吗？要提及的重要一点是，FE应用程序与其自己的服务器进行通信，而该应用程序的另一个实例（内置设备无法访问） ）。

Answer 1

我看到我的问题受到关注，所以我决定发布解决方案。要回答我的问题：“ 这真的是最好的方法，忽略重新连接功能吗？”我不知道:)。但是该解决方案对我有用，并且已在生产中得到证明，它提供了在某种程度上实际控制SSE重新连接的方式。

这是我所做的：

1. 已重写的createSseSource函数，因此返回类型为void
2. 而不是返回可观察的结果，而是将来自SSE的数据馈送到主题/ NgRx动作
3. 添加了公共openSseChannel和私有reconnectOnError功能以更好地控制
4. 添加了专用功能processSseEvent以处理自定义消息类型

由于我在此项目上使用NgRx，所以每条SSE消息都会调度相应的操作，但是可以将其替换为ReplaySubject并公开为observable。

// Public function, initializes connection, returns true if successful
openSseChannel(): boolean {
  this.createSseEventSource();
  return !!this.eventSource;
}

// Creates SSE event source, handles SSE events
protected createSseEventSource(): void {
  // Close event source if current instance of SSE service has some
  if (this.eventSource) {
    this.closeSseConnection();
    this.eventSource = null;
  }
  // Open new channel, create new EventSource
  this.eventSource = new EventSource(this.sseChannelUrl);

  // Process default event
  this.eventSource.onmessage = (event: MessageEvent) => {
    this.zone.run(() => this.processSseEvent(event));
  };

  // Add custom events
  Object.keys(SSE_EVENTS).forEach(key => {
    this.eventSource.addEventListener(SSE_EVENTS[key], event => {
      this.zone.run(() => this.processSseEvent(event));
    });
  });

  // Process connection opened
  this.eventSource.onopen = () => {
    this.reconnectFrequencySec = 1;
  };

  // Process error
  this.eventSource.onerror = (error: any) => {
    this.reconnectOnError();
  };
}

// Processes custom event types
private processSseEvent(sseEvent: MessageEvent): void {
  const parsed = sseEvent.data ? JSON.parse(sseEvent.data) : {};
  switch (sseEvent.type) {
    case SSE_EVENTS.STATUS: {
      this.store.dispatch(StatusActions.setStatus({ status: parsed }));
      // or
      // this.someReplaySubject.next(parsed);
      break;
    }
    // Add others if neccessary
    default: {
      console.error('Unknown event:', sseEvent.type);
      break;
    }
  }
}

// Handles reconnect attempts when the connection fails for some reason.
// const SSE_RECONNECT_UPPER_LIMIT = 64;
private reconnectOnError(): void {
  const self = this;
  this.closeSseConnection();
  clearTimeout(this.reconnectTimeout);
  this.reconnectTimeout = setTimeout(() => {
    self.openSseChannel();
    self.reconnectFrequencySec *= 2;
    if (self.reconnectFrequencySec >= SSE_RECONNECT_UPPER_LIMIT) {
      self.reconnectFrequencySec = SSE_RECONNECT_UPPER_LIMIT;
    }
  }, this.reconnectFrequencySec * 1000);
}

由于SSE事件被馈送给主题/动作，因此，连接是否丢失并不重要，因为至少最后一个事件保留在主题或商店中。然后，尝试重新连接可能会静默进行，并且在发送新数据时，将进行无缝处理。