WebSocketListenProvider will now properly handle errors and gracefully fail if

an error occurs while attempting to listen on a location.

PiperOrigin-RevId: 246583767

Author: amablue

database/src/desktop/core/listen_provider.h

@@ -16,6 +16,7 @@
 #define FIREBASE_DATABASE_CLIENT_CPP_SRC_DESKTOP_CORE_LISTEN_PROVIDER_H_
 
 #include "database/src/common/query_spec.h"
+#include "database/src/desktop/view/view.h"
 
 namespace firebase {
 namespace database {
@@ -30,7 +31,8 @@ class ListenProvider {
   // Begin listening on a location with a set of parameters given by the
   // QuerySpec. While listening, the server will send down updates which will be
   // parsed and passed along to the SyncTree to be cached locally.
-  virtual void StartListening(const QuerySpec& query_spec) = 0;
+  virtual void StartListening(const QuerySpec& query_spec,
+                              const View* view) = 0;
 
   // Stop listening on a location given by the QuerySpec.
   virtual void StopListening(const QuerySpec& query_spec) = 0;

database/src/desktop/core/repo.cc

@@ -391,11 +391,13 @@ void Repo::DeferredInitialization() {
   UniquePtr<PersistenceManager> persistence_manager =
       MakeUnique<PersistenceManager>(std::move(persistence_storage_engine),
                                      std::move(tracked_query_manager));
-  UniquePtr<ListenProvider> listen_provider =
-      MakeUnique<WebSocketListenProvider>(connection_.get());
+  UniquePtr<WebSocketListenProvider> listen_provider =
+      MakeUnique<WebSocketListenProvider>(this, connection_.get());
+  WebSocketListenProvider* listen_provider_ptr = listen_provider.get();
   server_sync_tree_ = MakeUnique<SyncTree>(std::move(pending_write_tree),
                                            std::move(persistence_manager),
                                            std::move(listen_provider));
+  listen_provider_ptr->set_sync_tree(server_sync_tree_.get());
 }
 
 void Repo::PostEvents(const std::vector<Event>& events) {

database/src/desktop/core/repo.h

@@ -107,6 +107,8 @@ class Repo : public connection::PersistentConnectionEventHandler {
 
   static scheduler::Scheduler& scheduler() { return s_scheduler_; }
 
+  ThisRef& this_ref() { return safe_this_; }
+
  private:
   WriteId GetNextWriteId();
 

database/src/desktop/core/sync_tree.cc

@@ -454,7 +454,7 @@ static QuerySpec QuerySpecForListening(const QuerySpec& query_spec) {
 
 void SyncTree::SetupListener(const QuerySpec& query_spec, const View* view) {
   const Path& path = query_spec.path;
-  listen_provider_->StartListening(QuerySpecForListening(query_spec));
+  listen_provider_->StartListening(QuerySpecForListening(query_spec), view);
 
   Tree<SyncPoint>* subtree = sync_point_tree_.GetChild(path);
 
@@ -557,7 +557,8 @@ std::vector<Event> SyncTree::RemoveEventRegistration(
           // Ok, we've collected all the listens we need. Set them up.
           for (const View* view : new_views) {
             QuerySpec new_query = view->query_spec();
-            listen_provider_->StartListening(QuerySpecForListening(new_query));
+            listen_provider_->StartListening(QuerySpecForListening(new_query),
+                                             view);
           }
         } else {
           // There's nothing below us, so nothing we need to start listening on

database/src/desktop/core/web_socket_listen_provider.cc

@@ -17,6 +17,7 @@
 #include "database/src/common/query_spec.h"
 #include "database/src/desktop/connection/persistent_connection.h"
 #include "database/src/desktop/core/listen_provider.h"
+#include "database/src/desktop/view/view.h"
 
 namespace firebase {
 namespace database {
@@ -25,8 +26,66 @@ namespace internal {
 using connection::PersistentConnection;
 using connection::ResponsePtr;
 
-void WebSocketListenProvider::StartListening(const QuerySpec& query_spec) {
-  connection_->Listen(query_spec, PersistentConnection::Tag(), ResponsePtr());
+class WebSocketListenResponse : public connection::Response {
+ public:
+  WebSocketListenResponse(const Response::ResponseCallback& callback,
+                          const Repo::ThisRef& repo_ref, SyncTree* sync_tree,
+                          const QuerySpec& query_spec, const View* view)
+      : connection::Response(callback),
+        repo_ref_(repo_ref),
+        sync_tree_(sync_tree),
+        query_spec_(query_spec),
+        view_(view) {}
+
+  Repo::ThisRef& repo_ref() { return repo_ref_; }
+  SyncTree* sync_tree() { return sync_tree_; }
+  const QuerySpec& query_spec() { return query_spec_; }
+  const View* view() { return view_; }
+
+ private:
+  Repo::ThisRef repo_ref_;
+  SyncTree* sync_tree_;
+  QuerySpec query_spec_;
+  const View* view_;
+};
+
+void WebSocketListenProvider::StartListening(const QuerySpec& query_spec,
+                                             const View* view) {
+  connection_->Listen(
+      query_spec, PersistentConnection::Tag(),
+      MakeShared<WebSocketListenResponse>(
+          [](const SharedPtr<connection::Response>& connection_response) {
+            WebSocketListenResponse* response =
+                static_cast<WebSocketListenResponse*>(
+                    connection_response.get());
+
+            Repo::ThisRefLock lock(&response->repo_ref());
+            Repo* repo = lock.GetReference();
+            if (repo == nullptr) {
+              // Repo was deleted, do not proceed.
+              return;
+            }
+
+            std::vector<Event> events;
+            if (!response->HasError()) {
+              const QuerySpec& query_spec = response->view()->query_spec();
+              events =
+                  response->sync_tree()->ApplyListenComplete(query_spec.path);
+            } else {
+              LogWarning("Listen at %s failed: %s",
+                         response->query_spec().path.c_str(),
+                         response->GetErrorMessage().c_str());
+
+              // If a listen failed, kill all of the listeners here, not just
+              // the one that triggered the error. Note that this may need to be
+              // scoped to just this listener if we change permissions on
+              // filtered children
+              events = response->sync_tree()->RemoveAllEventRegistrations(
+                  response->query_spec(), response->GetErrorCode());
+            }
+            repo->PostEvents(events);
+          },
+          repo_->this_ref(), sync_tree_, query_spec, view));
 }
 
 void WebSocketListenProvider::StopListening(const QuerySpec& query_spec) {

database/src/desktop/core/web_socket_listen_provider.h

@@ -18,23 +18,29 @@
 #include "database/src/common/query_spec.h"
 #include "database/src/desktop/connection/persistent_connection.h"
 #include "database/src/desktop/core/listen_provider.h"
+#include "database/src/desktop/core/repo.h"
 
 namespace firebase {
 namespace database {
 namespace internal {
 
 class WebSocketListenProvider : public ListenProvider {
  public:
-  WebSocketListenProvider(connection::PersistentConnection* connection)
-      : connection_(connection) {}
+  WebSocketListenProvider(Repo* repo,
+                          connection::PersistentConnection* connection)
+      : repo_(repo), sync_tree_(nullptr), connection_(connection) {}
 
-  virtual ~WebSocketListenProvider() {}
+  ~WebSocketListenProvider() override {}
 
-  void StartListening(const QuerySpec& query_spec) override;
+  void set_sync_tree(SyncTree* sync_tree) { sync_tree_ = sync_tree; }
+
+  void StartListening(const QuerySpec& query_spec, const View* view) override;
 
   void StopListening(const QuerySpec& query_spec) override;
 
  private:
+  Repo* repo_;
+  SyncTree* sync_tree_;
   connection::PersistentConnection* connection_;
 };