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()按照这个答案实现功能，但我无法让它工作。然后我放弃了这个reconnectOnError()功能，这似乎是一件更好的事情。不要尝试关闭和重新连接，也不要将错误传播到 observable。只需坐等，当服务器再次运行时，它会自动重新连接。

问题是，这真的是最好的做法吗？需要提及的重要一点是，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 事件被馈送到主题/操作，因此连接是否丢失并不重要，因为至少最后一个事件保留在主题或存储中。然后，重新连接的尝试可以悄无声息地进行，并且当发送新数据时，会得到无缝处理。