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

Question

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

我试过尝试将所有内容包装起来-除外，然后将所有内容包装在while（True）循环中，但这并不能捕获错误，因为错误是在单独的线程中抛出的。

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

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

发生的特定错误是：

google.api_core.exceptions.InternalServerError：500收到了RST_STREAM，错误代码为0

有时

google.api_core.exceptions.InternalServerError：500收到RST_STREAM，错误代码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”，如果由于某种原因连接被关闭，则该变量变为true。因此，如果我们定期进行检查，则可以重新启动侦听器并保持愉快的状态。

使用之前的代码，我添加了一个新方法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()