Port FSTAsyncQueryListener to C++ (#2620)

* Test that shows a leak.

* Add EventListener to match Android.

* Use AsyncEventListener in FSTQueryListenerTests

* Remove FSTAsyncQueryListener

* Use EventListener throughout

Author: wilhuff

Firestore/Example/Tests/Core/FSTEventManagerTests.mm

@@ -35,25 +35,26 @@
 #include "Firestore/core/src/firebase/firestore/util/statusor.h"
 #include "Firestore/core/test/firebase/firestore/testutil/xcgmock.h"
 
+using firebase::firestore::core::EventListener;
 using firebase::firestore::core::ListenOptions;
 using firebase::firestore::core::ViewSnapshot;
-using firebase::firestore::core::ViewSnapshotHandler;
 using firebase::firestore::model::DocumentKeySet;
 using firebase::firestore::model::DocumentSet;
 using firebase::firestore::model::OnlineState;
 using firebase::firestore::util::StatusOr;
+using firebase::firestore::util::StatusOrCallback;
 using testing::ElementsAre;
 
 NS_ASSUME_NONNULL_BEGIN
 
 namespace {
 
-ViewSnapshotHandler NoopViewSnapshotHandler() {
-  return [](const StatusOr<ViewSnapshot> &) {};
+ViewSnapshot::Listener NoopViewSnapshotHandler() {
+  return EventListener<ViewSnapshot>::Create([](const StatusOr<ViewSnapshot> &) {});
 }
 
 std::shared_ptr<QueryListener> NoopQueryListener(FSTQuery *query) {
-  return QueryListener::Create(query, NoopViewSnapshotHandler());
+  return QueryListener::Create(query, ListenOptions::DefaultOptions(), NoopViewSnapshotHandler());
 }
 
 }  // namespace
@@ -112,17 +113,14 @@ - (void)testNotifiesListenersInTheRightOrder {
   FSTQuery *query2 = FSTTestQuery("bar/baz");
   NSMutableArray *eventOrder = [NSMutableArray array];
 
-  auto listener1 = QueryListener::Create(query1, [eventOrder](const StatusOr<ViewSnapshot> &) {
-    [eventOrder addObject:@"listener1"];
-  });
+  auto listener1 = QueryListener::Create(
+      query1, [eventOrder](StatusOr<ViewSnapshot>) { [eventOrder addObject:@"listener1"]; });
 
-  auto listener2 = QueryListener::Create(query2, [eventOrder](const StatusOr<ViewSnapshot> &) {
-    [eventOrder addObject:@"listener2"];
-  });
+  auto listener2 = QueryListener::Create(
+      query2, [eventOrder](StatusOr<ViewSnapshot>) { [eventOrder addObject:@"listener2"]; });
 
-  auto listener3 = QueryListener::Create(query1, [eventOrder](const StatusOr<ViewSnapshot> &) {
-    [eventOrder addObject:@"listener3"];
-  });
+  auto listener3 = QueryListener::Create(
+      query1, [eventOrder](StatusOr<ViewSnapshot>) { [eventOrder addObject:@"listener3"]; });
 
   FSTSyncEngine *syncEngineMock = OCMClassMock([FSTSyncEngine class]);
   FSTEventManager *eventManager = [FSTEventManager eventManagerWithSyncEngine:syncEngineMock];

Firestore/Example/Tests/Core/FSTQueryListenerTests.mm

@@ -24,11 +24,11 @@
 #import "Firestore/Source/Core/FSTQuery.h"
 #import "Firestore/Source/Core/FSTView.h"
 #import "Firestore/Source/Model/FSTDocument.h"
-#import "Firestore/Source/Util/FSTAsyncQueryListener.h"
 
 #import "Firestore/Example/Tests/Util/FSTHelpers.h"
 
 #include "Firestore/core/include/firebase/firestore/firestore_errors.h"
+#include "Firestore/core/src/firebase/firestore/core/event_listener.h"
 #include "Firestore/core/src/firebase/firestore/core/view_snapshot.h"
 #include "Firestore/core/src/firebase/firestore/model/document_set.h"
 #include "Firestore/core/src/firebase/firestore/model/types.h"
@@ -40,10 +40,12 @@
 #include "Firestore/core/test/firebase/firestore/testutil/xcgmock.h"
 
 using firebase::firestore::FirestoreErrorCode;
+using firebase::firestore::core::AsyncEventListener;
+using firebase::firestore::core::EventListener;
 using firebase::firestore::core::DocumentViewChange;
+using firebase::firestore::core::EventListener;
 using firebase::firestore::core::ListenOptions;
 using firebase::firestore::core::ViewSnapshot;
-using firebase::firestore::core::ViewSnapshotHandler;
 using firebase::firestore::model::DocumentKeySet;
 using firebase::firestore::model::DocumentSet;
 using firebase::firestore::model::OnlineState;
@@ -72,10 +74,11 @@ ViewSnapshot ExcludingMetadataChanges(const ViewSnapshot &snapshot) {
   };
 }
 
-ViewSnapshotHandler Accumulating(std::vector<ViewSnapshot> *values) {
-  return [values](const StatusOr<ViewSnapshot> &maybe_snapshot) {
-    values->push_back(maybe_snapshot.ValueOrDie());
-  };
+ViewSnapshot::Listener Accumulating(std::vector<ViewSnapshot> *values) {
+  return EventListener<ViewSnapshot>::Create(
+      [values](const StatusOr<ViewSnapshot> &maybe_snapshot) {
+        values->push_back(maybe_snapshot.ValueOrDie());
+      });
 }
 
 }  // namespace
@@ -166,26 +169,26 @@ - (void)testRaisesEventForEmptyCollectionAfterSync {
 }
 
 - (void)testMutingAsyncListenerPreventsAllSubsequentEvents {
-  __block std::vector<ViewSnapshot> accum;
+  std::vector<ViewSnapshot> accum;
 
   FSTQuery *query = FSTTestQuery("rooms/Eros");
   FSTDocument *doc1 = FSTTestDoc("rooms/Eros", 3, @{@"name" : @"Eros"}, FSTDocumentStateSynced);
   FSTDocument *doc2 = FSTTestDoc("rooms/Eros", 4, @{@"name" : @"Eros2"}, FSTDocumentStateSynced);
 
-  __block FSTAsyncQueryListener *listener = [[FSTAsyncQueryListener alloc]
-      initWithExecutor:_executor.get()
-       snapshotHandler:^(const StatusOr<ViewSnapshot> &maybe_snapshot) {
-         accum.push_back(maybe_snapshot.ValueOrDie());
-         [listener mute];
-       }];
+  std::shared_ptr<AsyncEventListener<ViewSnapshot>> listener =
+      AsyncEventListener<ViewSnapshot>::Create(
+          _executor.get(), EventListener<ViewSnapshot>::Create(
+                               [&accum, &listener](const StatusOr<ViewSnapshot> &maybe_snapshot) {
+                                 accum.push_back(maybe_snapshot.ValueOrDie());
+                                 listener->Mute();
+                               }));
 
   FSTView *view = [[FSTView alloc] initWithQuery:query remoteDocuments:DocumentKeySet{}];
   ViewSnapshot viewSnapshot1 = FSTTestApplyChanges(view, @[ doc1 ], absl::nullopt).value();
   ViewSnapshot viewSnapshot2 = FSTTestApplyChanges(view, @[ doc2 ], absl::nullopt).value();
 
-  ViewSnapshotHandler handler = listener.asyncSnapshotHandler;
-  handler(viewSnapshot1);
-  handler(viewSnapshot2);
+  listener->OnEvent(viewSnapshot1);
+  listener->OnEvent(viewSnapshot2);
 
   // Drain queue
   XCTestExpectation *expectation = [self expectationWithDescription:@"Queue drained"];

Firestore/Example/Tests/Integration/API/FIRFirestoreSourceTests.mm

@@ -69,6 +69,30 @@ - (void)testGetCollectionWhileOnlineWithDefaultSource {
                         ]));
 }
 
+- (void)testGetDocumentError {
+  FIRDocumentReference *doc = [self.db documentWithPath:@"foo/__invalid__"];
+
+  XCTestExpectation *completed = [self expectationWithDescription:@"get completed"];
+  [doc getDocumentWithCompletion:^(FIRDocumentSnapshot *snapshot, NSError *error) {
+    XCTAssertNotNil(error);
+    [completed fulfill];
+  }];
+
+  [self awaitExpectations];
+}
+
+- (void)testGetCollectionError {
+  FIRCollectionReference *col = [self.db collectionWithPath:@"__invalid__"];
+
+  XCTestExpectation *completed = [self expectationWithDescription:@"get completed"];
+  [col getDocumentsWithCompletion:^(FIRQuerySnapshot *snapshot, NSError *error) {
+    XCTAssertNotNil(error);
+    [completed fulfill];
+  }];
+
+  [self awaitExpectations];
+}
+
 - (void)testGetDocumentWhileOfflineWithDefaultSource {
   FIRDocumentReference *doc = [self documentRef];
 

Firestore/Example/Tests/SpecTests/FSTSyncEngineTestDriver.mm

@@ -351,7 +351,7 @@ - (TargetId)addUserListenerWithQuery:(FSTQuery *)query {
   // TODO(dimond): Allow customizing listen options in spec tests
   // TODO(dimond): Change spec tests to verify isFromCache on snapshots
   ListenOptions options = ListenOptions::FromIncludeMetadataChanges(true);
-  auto listener = std::make_shared<QueryListener>(
+  auto listener = QueryListener::Create(
       query, options, [self, query](const StatusOr<ViewSnapshot> &maybe_snapshot) {
         FSTQueryEvent *event = [[FSTQueryEvent alloc] init];
         event.query = query;

Firestore/Source/API/FIRDocumentReference.mm

@@ -34,6 +34,7 @@
 
 #include "Firestore/core/src/firebase/firestore/api/document_reference.h"
 #include "Firestore/core/src/firebase/firestore/api/document_snapshot.h"
+#include "Firestore/core/src/firebase/firestore/core/event_listener.h"
 #include "Firestore/core/src/firebase/firestore/model/document_key.h"
 #include "Firestore/core/src/firebase/firestore/model/document_set.h"
 #include "Firestore/core/src/firebase/firestore/model/precondition.h"
@@ -48,6 +49,7 @@
 using firebase::firestore::api::DocumentReference;
 using firebase::firestore::api::DocumentSnapshot;
 using firebase::firestore::api::Firestore;
+using firebase::firestore::core::EventListener;
 using firebase::firestore::core::ListenOptions;
 using firebase::firestore::core::ParsedSetData;
 using firebase::firestore::core::ParsedUpdateData;
@@ -213,21 +215,30 @@ - (void)getDocumentWithSource:(FIRFirestoreSource)source
                                                              listener:(FIRDocumentSnapshotBlock)
                                                                           listener {
   ListenerRegistration result = _documentReference.AddSnapshotListener(
-      [self wrapDocumentSnapshotBlock:listener], std::move(internalOptions));
+      std::move(internalOptions), [self wrapDocumentSnapshotBlock:listener]);
   return [[FSTListenerRegistration alloc] initWithRegistration:std::move(result)];
 }
 
-- (StatusOrCallback<DocumentSnapshot>)wrapDocumentSnapshotBlock:(FIRDocumentSnapshotBlock)block {
-  FIRFirestore *firestore = self.firestore;
-  return [block, firestore](StatusOr<DocumentSnapshot> maybe_snapshot) {
-    if (maybe_snapshot.ok()) {
-      FIRDocumentSnapshot *result =
-          [[FIRDocumentSnapshot alloc] initWithSnapshot:std::move(maybe_snapshot).ValueOrDie()];
-      block(result, nil);
-    } else {
-      block(nil, util::MakeNSError(maybe_snapshot.status()));
+- (DocumentSnapshot::Listener)wrapDocumentSnapshotBlock:(FIRDocumentSnapshotBlock)block {
+  class Converter : public EventListener<DocumentSnapshot> {
+   public:
+    explicit Converter(FIRDocumentSnapshotBlock block) : block_(block) {
     }
+
+    void OnEvent(StatusOr<DocumentSnapshot> maybe_snapshot) override {
+      if (maybe_snapshot.ok()) {
+        FIRDocumentSnapshot *result =
+            [[FIRDocumentSnapshot alloc] initWithSnapshot:std::move(maybe_snapshot).ValueOrDie()];
+        block_(result, nil);
+      } else {
+        block_(nil, util::MakeNSError(maybe_snapshot.status()));
+      }
+    }
+
+   private:
+    FIRDocumentSnapshotBlock block_;
   };
+  return absl::make_unique<Converter>(block);
 }
 
 @end

Firestore/Source/API/FIRQuery.mm

@@ -38,7 +38,6 @@
 #import "Firestore/Source/Core/FSTQuery.h"
 #import "Firestore/Source/Model/FSTDocument.h"
 #import "Firestore/Source/Model/FSTFieldValue.h"
-#import "Firestore/Source/Util/FSTAsyncQueryListener.h"
 #import "Firestore/Source/Util/FSTUsageValidation.h"
 
 #include "Firestore/core/src/firebase/firestore/model/document_key.h"
@@ -50,8 +49,9 @@
 #include "Firestore/core/src/firebase/firestore/util/string_apple.h"
 
 namespace util = firebase::firestore::util;
+using firebase::firestore::core::AsyncEventListener;
+using firebase::firestore::core::EventListener;
 using firebase::firestore::core::ViewSnapshot;
-using firebase::firestore::core::ViewSnapshotHandler;
 using firebase::firestore::model::DocumentKey;
 using firebase::firestore::model::FieldPath;
 using firebase::firestore::model::ResourcePath;
@@ -175,34 +175,34 @@ ListenOptions listenOptions(
   Firestore *firestore = self.firestore.wrapped;
   FSTQuery *query = self.query;
 
-  ViewSnapshotHandler snapshotHandler = [listener, firestore,
-                                         query](const StatusOr<ViewSnapshot> &maybe_snapshot) {
-    if (!maybe_snapshot.status().ok()) {
-      listener(nil, MakeNSError(maybe_snapshot.status()));
-      return;
-    }
-    ViewSnapshot snapshot = maybe_snapshot.ValueOrDie();
-    SnapshotMetadata metadata(snapshot.has_pending_writes(), snapshot.from_cache());
-
-    listener([[FIRQuerySnapshot alloc] initWithFirestore:firestore
-                                           originalQuery:query
-                                                snapshot:std::move(snapshot)
-                                                metadata:std::move(metadata)],
-             nil);
-  };
+  // Convert from ViewSnapshots to QuerySnapshots.
+  auto view_listener = EventListener<ViewSnapshot>::Create(
+      [listener, firestore, query](StatusOr<ViewSnapshot> maybe_snapshot) {
+        if (!maybe_snapshot.status().ok()) {
+          listener(nil, MakeNSError(maybe_snapshot.status()));
+          return;
+        }
+
+        ViewSnapshot snapshot = std::move(maybe_snapshot).ValueOrDie();
+        SnapshotMetadata metadata(snapshot.has_pending_writes(), snapshot.from_cache());
+
+        listener([[FIRQuerySnapshot alloc] initWithFirestore:firestore
+                                               originalQuery:query
+                                                    snapshot:std::move(snapshot)
+                                                    metadata:std::move(metadata)],
+                 nil);
+      });
 
-  FSTAsyncQueryListener *asyncListener =
-      [[FSTAsyncQueryListener alloc] initWithExecutor:self.firestore.client.userExecutor
-                                      snapshotHandler:std::move(snapshotHandler)];
+  // Call the view_listener on the user Executor.
+  auto async_listener = AsyncEventListener<ViewSnapshot>::Create(firestore->client().userExecutor,
+                                                                 std::move(view_listener));
 
-  std::shared_ptr<QueryListener> internalListener =
-      [firestore->client() listenToQuery:query
-                                 options:internalOptions
-                     viewSnapshotHandler:[asyncListener asyncSnapshotHandler]];
+  std::shared_ptr<QueryListener> query_listener =
+      [firestore->client() listenToQuery:query options:internalOptions listener:async_listener];
 
   return [[FSTListenerRegistration alloc]
-      initWithRegistration:ListenerRegistration(firestore->client(), asyncListener,
-                                                std::move(internalListener))];
+      initWithRegistration:ListenerRegistration(firestore->client(), std::move(async_listener),
+                                                std::move(query_listener))];
 }
 
 - (FIRQuery *)queryWhereField:(NSString *)field isEqualTo:(id)value {

Firestore/Source/Core/FSTEventManager.mm

@@ -36,7 +36,6 @@
 namespace objc = firebase::firestore::util::objc;
 using firebase::firestore::core::DocumentViewChange;
 using firebase::firestore::core::ViewSnapshot;
-using firebase::firestore::core::ViewSnapshotHandler;
 using firebase::firestore::model::OnlineState;
 using firebase::firestore::model::TargetId;
 using firebase::firestore::util::MakeStatus;

Firestore/Source/Core/FSTFirestoreClient.h

@@ -47,8 +47,10 @@
 
 NS_ASSUME_NONNULL_BEGIN
 
+using firebase::firestore::api::DocumentSnapshot;
 using firebase::firestore::core::ListenOptions;
 using firebase::firestore::core::QueryListener;
+using firebase::firestore::core::ViewSnapshot;
 
 /**
  * FirestoreClient is a top-level class that constructs and owns all of the pieces of the client
@@ -84,8 +86,7 @@ using firebase::firestore::core::QueryListener;
 /** Starts listening to a query. */
 - (std::shared_ptr<QueryListener>)listenToQuery:(FSTQuery *)query
                                         options:(ListenOptions)options
-                            viewSnapshotHandler:(firebase::firestore::core::ViewSnapshotHandler &&)
-                                                    viewSnapshotHandler;
+                                       listener:(ViewSnapshot::SharedListener &&)listener;
 
 /** Stops listening to a query previously listened to. */
 - (void)removeListener:(const std::shared_ptr<QueryListener> &)listener;
@@ -95,8 +96,7 @@ using firebase::firestore::core::QueryListener;
  * doesn't exist, an error will be sent to the completion.
  */
 - (void)getDocumentFromLocalCache:(const firebase::firestore::api::DocumentReference &)doc
-                       completion:(firebase::firestore::util::StatusOrCallback<
-                                      firebase::firestore::api::DocumentSnapshot> &&)completion;
+                       completion:(DocumentSnapshot::Listener &&)completion;
 
 /**
  * Retrieves a (possibly empty) set of documents from the cache via the

Firestore/Source/Core/FSTFirestoreClient.mm

@@ -64,7 +64,6 @@
 using firebase::firestore::auth::User;
 using firebase::firestore::core::DatabaseInfo;
 using firebase::firestore::core::ViewSnapshot;
-using firebase::firestore::core::ViewSnapshotHandler;
 using firebase::firestore::local::LruParams;
 using firebase::firestore::model::DatabaseId;
 using firebase::firestore::model::DocumentKeySet;
@@ -314,22 +313,22 @@ - (void)shutdownWithCompletion:(nullable FSTVoidErrorBlock)completion {
 
 - (std::shared_ptr<QueryListener>)listenToQuery:(FSTQuery *)query
                                         options:(ListenOptions)options
-                            viewSnapshotHandler:(ViewSnapshotHandler &&)viewSnapshotHandler {
-  auto listener =
-      std::make_shared<QueryListener>(query, std::move(options), std::move(viewSnapshotHandler));
+                                       listener:(ViewSnapshot::SharedListener &&)listener {
+  auto query_listener = QueryListener::Create(query, std::move(options), std::move(listener));
 
-  _workerQueue->Enqueue([self, listener] { [self.eventManager addListener:listener]; });
+  _workerQueue->Enqueue([self, query_listener] { [self.eventManager addListener:query_listener]; });
 
-  return listener;
+  return query_listener;
 }
 
 - (void)removeListener:(const std::shared_ptr<QueryListener> &)listener {
   _workerQueue->Enqueue([self, listener] { [self.eventManager removeListener:listener]; });
 }
 
 - (void)getDocumentFromLocalCache:(const DocumentReference &)doc
-                       completion:(StatusOrCallback<DocumentSnapshot> &&)completion {
-  _workerQueue->Enqueue([self, doc, completion] {
+                       completion:(DocumentSnapshot::Listener &&)completion {
+  auto shared_completion = absl::ShareUniquePtr(std::move(completion));
+  _workerQueue->Enqueue([self, doc, shared_completion] {
     FSTMaybeDocument *maybeDoc = [self.localStore readDocument:doc.key()];
     StatusOr<DocumentSnapshot> maybe_snapshot;
 
@@ -349,8 +348,8 @@ - (void)getDocumentFromLocalCache:(const DocumentReference &)doc
                               "FIRFirestoreSourceCache to attempt to retrieve the document "};
     }
 
-    if (completion) {
-      self->_userExecutor->Execute([=] { completion(std::move(maybe_snapshot)); });
+    if (shared_completion) {
+      self->_userExecutor->Execute([=] { shared_completion->OnEvent(std::move(maybe_snapshot)); });
     }
   });
 }