如何检测 Firebase Firestore 数据库中的实时监听器错误？

Question

Firestore listeners 将在一段时间后随机关闭（可能是由于不活动），并且在 python 中没有简单的方法来捕获它们抛出的错误，因为它们将它们抛出到单独的线程中。就我而言，我想保持一个持久的监听器，它永远不会因为不活动或服务器端错误而关闭。

我尝试将所有内容都包装在 try - except 中，然后将所有内容包装在 while(True) 循环中，但这并没有捕获错误，因为错误是在单独的线程中引发的。

错误发生在 Linux 和 Windows 设备上 10 分钟 - 24 小时不活动（我不确定是否是这种情况，它可能是随机的，但我发现的最短时间间隔是启动后 10 分钟）后.我没有尝试过 Mac 或任何其他设备，但我怀疑它是特定于设备的。

查看 gRPC（侦听器用于在客户端和服务器之间进行通信的东西）规范，python api 没有默认超时（超时并不能解释为什么它会在不同的时间后断开连接），也没有超时它设置在 Firestores 监听器代码中的任何位置。

发生的具体错误是：

google.api_core.exceptions.InternalServerError: 500 Received RST_STREAM with error code 0

有时

google.api_core.exceptions.InternalServerError: 500 Received RST_STREAM with error code 2

显示问题的最少代码（在名为 info 的虚拟集合上运行，该集合中暂时只有一个文档）：

class TestWatchInfo():
  def __init__(self):
    self.query_watch = db.collection(u'info').on_snapshot(self.on_snapshot)

  def on_snapshot(self, col_snapshot, changes, read_time):
    try:
      for change in changes:
        pass
    except Exception as err:
      print(err)
      print("Error occurred at " + str(time.ctime()))
      traceback.print_exc()

if __name__ == '__main__':
  try:
    test_object = TestWatchInfo()
    while(True):
      time.sleep(60)
  except Exception as err:
    print(err)
    print("Error occurred at " + str(time.ctime()))
    traceback.print_exc()

理想情况下，我将能够捕获主 python 线程中发生的实际错误，但据我所知，由于我不是产生线程的人，因此我无法将线程/gRPC 特定代码添加到抓住那个错误。或者，我希望能够在 gRPC 连接因服务器端而关闭后自动重启。

实际上，Firestore 侦听器只是在它创建的线程中引发错误并关闭侦听器。

Answer 1

我想出了一种替代方法来检测侦听器错误并在服务器端关闭后重新启动侦听器。我不知道如何捕捉实际错误，但我想出了如何检测 Firestore 何时随机关闭侦听器连接。

在 Firebase 侦听器代码中，它们会跟踪一个私有变量“_closed”，如果连接因任何原因关闭，该变量就会变为真。因此，如果我们定期检查这一点，我们就可以重新启动我们的侦听器并继续前进。

使用之前的代码，我添加了一个新方法 start_snapshot 以便在出错时重新启动失败的侦听器表达式，在我长时间运行的代码中，我添加了对侦听器的检查以查看它是否已关闭，然后重新启动它如果是的话。

class TestWatchInfo():
  def __init__(self):
    self.start_snapshot()

  def start_snapshot(self):
    self.query_watch = db.collection(u'info').on_snapshot(self.on_snapshot)

  def on_snapshot(self, col_snapshot, changes, read_time):
    try:
      for change in changes:
        pass
    except Exception as err:
      print(err)
      print("Error occurred at " + str(time.ctime()))
      traceback.print_exc()

if __name__ == '__main__':
  try:
    test_object = TestWatchInfo()
    while(True):
      if test_object.query_watch._closed:
        test_object.start_snapshot()
      # code here
  except Exception as err:
    print(err)
    print("Error occurred at " + str(time.ctime()))
    traceback.print_exc()