7

我正在开发一个 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 应用程序与它自己的服务器通信，该服务器不能被另一个应用程序实例（内置设备）访问。

4

1 回答 1

2

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

这是我所做的：

  1. 重写createSseSource函数，返回类型为void
  2. 来自 SSE 的数据不是返回 observable，而是馈送到主题/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 事件被馈送到主题/操作，因此连接是否丢失并不重要，因为至少最后一个事件保留在主题或存储中。然后可以静默尝试重新连接，并且当发送新数据时，会无缝处理。

于 2020-05-28T05:55:15.933 回答