C++ migration: port watch stream-related part of FSTRemoteStore

Firestore/Example/Tests/Integration/FSTDatastoreTests.mm

@@ -55,6 +55,7 @@
 using firebase::firestore::model::DocumentKey;
 using firebase::firestore::model::DocumentKeySet;
 using firebase::firestore::model::Precondition;
+using firebase::firestore::model::OnlineState;
 using firebase::firestore::model::TargetId;
 using firebase::firestore::remote::Datastore;
 using firebase::firestore::remote::GrpcConnection;
@@ -186,7 +187,8 @@ - (void)setUp {
 
   _remoteStore = [[FSTRemoteStore alloc] initWithLocalStore:_localStore
                                                   datastore:_datastore
-                                                workerQueue:_testWorkerQueue.get()];
+                                                workerQueue:_testWorkerQueue.get()
+                                         onlineStateHandler:[](OnlineState) {}];
 
   _testWorkerQueue->Enqueue([=] { [_remoteStore start]; });
 }

Firestore/Example/Tests/Local/FSTLocalStoreTests.mm

@@ -19,6 +19,8 @@
 #import <FirebaseFirestore/FIRTimestamp.h>
 #import <XCTest/XCTest.h>
 
+#include <vector>
+
 #import "Firestore/Source/Core/FSTQuery.h"
 #import "Firestore/Source/Local/FSTLocalWriteResult.h"
 #import "Firestore/Source/Local/FSTPersistence.h"
@@ -51,6 +53,7 @@
 using firebase::firestore::model::SnapshotVersion;
 using firebase::firestore::model::TargetId;
 using firebase::firestore::remote::RemoteEvent;
+using firebase::firestore::remote::TestTargetMetadataProvider;
 using firebase::firestore::remote::WatchChangeAggregator;
 using firebase::firestore::remote::WatchTargetChange;
 using firebase::firestore::remote::WatchTargetChangeState;
@@ -906,9 +909,9 @@ - (void)testPersistsResumeTokens {
   NSData *resumeToken = FSTTestResumeTokenFromSnapshotVersion(1000);
 
   WatchTargetChange watchChange{WatchTargetChangeState::Current, {targetID}, resumeToken};
-  WatchChangeAggregator aggregator{[FSTTestTargetMetadataProvider
-      providerWithSingleResultForKey:testutil::Key("foo/bar")
-                             targets:{targetID}]};
+  auto metadataProvider = TestTargetMetadataProvider::CreateSingleResultProvider(
+      testutil::Key("foo/bar"), std::vector<TargetId>{targetID});
+  WatchChangeAggregator aggregator{&metadataProvider};
   aggregator.HandleTargetChange(watchChange);
   RemoteEvent remoteEvent = aggregator.CreateRemoteEvent(testutil::Version(1000));
   [self applyRemoteEvent:remoteEvent];

Firestore/Example/Tests/Remote/FSTRemoteEventTests.mm

@@ -46,6 +46,7 @@
 using firebase::firestore::remote::ExistenceFilterWatchChange;
 using firebase::firestore::remote::RemoteEvent;
 using firebase::firestore::remote::TargetChange;
+using firebase::firestore::remote::TestTargetMetadataProvider;
 using firebase::firestore::remote::WatchChange;
 using firebase::firestore::remote::WatchChangeAggregator;
 using firebase::firestore::remote::WatchTargetChange;
@@ -92,13 +93,12 @@ @interface FSTRemoteEventTests : XCTestCase
 
 @implementation FSTRemoteEventTests {
   NSData *_resumeToken1;
-  FSTTestTargetMetadataProvider *_targetMetadataProvider;
+  TestTargetMetadataProvider _targetMetadataProvider;
   std::unordered_map<TargetId, int> _noOutstandingResponses;
 }
 
 - (void)setUp {
   _resumeToken1 = [@"resume1" dataUsingEncoding:NSUTF8StringEncoding];
-  _targetMetadataProvider = [FSTTestTargetMetadataProvider new];
 }
 
 /**
@@ -145,7 +145,7 @@ - (void)setUp {
  * considered active, or `_noOutstandingResponses` if all targets are already active.
  * @param existingKeys The set of documents that are considered synced with the test targets as
  * part of a previous listen. To modify this set during test execution, invoke
- * `[_targetMetadataProvider setSyncedKeys:forQueryData:]`.
+ * `_targetMetadataProvider.SetSyncedKeys()`.
  * @param watchChanges The watch changes to apply before returning the aggregator. Supported
  * changes are `DocumentWatchChange` and `WatchTargetChange`.
  */
@@ -154,15 +154,15 @@ - (void)setUp {
        outstandingResponses:(const std::unordered_map<TargetId, int> &)outstandingResponses
                existingKeys:(DocumentKeySet)existingKeys
                     changes:(const std::vector<std::unique_ptr<WatchChange>> &)watchChanges {
-  WatchChangeAggregator aggregator{_targetMetadataProvider};
+  WatchChangeAggregator aggregator{&_targetMetadataProvider};
 
   std::vector<TargetId> targetIDs;
   for (const auto &kv : targetMap) {
     TargetId targetID = kv.first;
     FSTQueryData *queryData = kv.second;
 
     targetIDs.push_back(targetID);
-    [_targetMetadataProvider setSyncedKeys:existingKeys forQueryData:queryData];
+    _targetMetadataProvider.SetSyncedKeys(existingKeys, queryData);
   };
 
   for (const auto &kv : outstandingResponses) {
@@ -223,7 +223,7 @@ - (void)setUp {
 
 - (void)testWillAccumulateDocumentAddedAndRemovedEvents {
   // The target map that contains an entry for every target in this test. If a target ID is
-  // omitted, the target is considered inactive and FSTTestTargetMetadataProvider will fail on
+  // omitted, the target is considered inactive and `TestTargetMetadataProvider` will fail on
   // access.
   std::unordered_map<TargetId, FSTQueryData *> targetMap{
       [self queryDataForTargets:{1, 2, 3, 4, 5, 6}]};
@@ -614,8 +614,7 @@ - (void)testDocumentUpdate {
   XCTAssertEqualObjects(event.document_updates().at(doc1.key), doc1);
   XCTAssertEqualObjects(event.document_updates().at(doc2.key), doc2);
 
-  [_targetMetadataProvider setSyncedKeys:DocumentKeySet{doc1.key, doc2.key}
-                            forQueryData:targetMap[1]];
+  _targetMetadataProvider.SetSyncedKeys(DocumentKeySet{doc1.key, doc2.key}, targetMap[1]);
 
   FSTDeletedDocument *deletedDoc1 = [FSTDeletedDocument documentWithKey:doc1.key
                                                                 version:testutil::Version(3)

Firestore/Example/Tests/SpecTests/FSTMockDatastore.h

@@ -39,7 +39,7 @@ class MockDatastore : public Datastore {
                 util::AsyncQueue* worker_queue,
                 auth::CredentialsProvider* credentials);
 
-  std::shared_ptr<WatchStream> CreateWatchStream(id<FSTWatchStreamDelegate> delegate) override;
+  std::shared_ptr<WatchStream> CreateWatchStream(WatchStreamCallback* callback) override;
   std::shared_ptr<WriteStream> CreateWriteStream(id<FSTWriteStreamDelegate> delegate) override;
 
   /**

Firestore/Example/Tests/SpecTests/FSTMockDatastore.mm

@@ -70,11 +70,11 @@
                   CredentialsProvider* credentials_provider,
                   FSTSerializerBeta* serializer,
                   GrpcConnection* grpc_connection,
-                  id<FSTWatchStreamDelegate> delegate,
+                  WatchStreamCallback* callback,
                   MockDatastore* datastore)
-      : WatchStream{worker_queue, credentials_provider, serializer, grpc_connection, delegate},
+      : WatchStream{worker_queue, credentials_provider, serializer, grpc_connection, callback},
         datastore_{datastore},
-        delegate_{delegate} {
+        callback_{callback} {
   }
 
   const std::unordered_map<TargetId, FSTQueryData*>& ActiveTargets() const {
@@ -84,7 +84,7 @@
   void Start() override {
     HARD_ASSERT(!open_, "Trying to start already started watch stream");
     open_ = true;
-    [delegate_ watchStreamDidOpen];
+    callback_->OnWatchStreamOpen();
   }
 
   void Stop() override {
@@ -118,7 +118,7 @@ void UnwatchTargetId(model::TargetId target_id) override {
 
   void FailStream(const Status& error) {
     open_ = false;
-    [delegate_ watchStreamWasInterruptedWithError:error];
+    callback_->OnWatchStreamClose(error);
   }
 
   void WriteWatchChange(const WatchChange& change, SnapshotVersion snap) {
@@ -145,14 +145,14 @@ void WriteWatchChange(const WatchChange& change, SnapshotVersion snap) {
       }
     }
 
-    [delegate_ watchStreamDidChange:change snapshotVersion:snap];
+    callback_->OnWatchStreamChange(change, snap);
   }
 
  private:
   bool open_ = false;
   std::unordered_map<TargetId, FSTQueryData*> active_targets_;
   MockDatastore* datastore_ = nullptr;
-  id<FSTWatchStreamDelegate> delegate_ = nullptr;
+  WatchStreamCallback* callback_ = nullptr;
 };
 
 class MockWriteStream : public WriteStream {
@@ -248,11 +248,11 @@ int sent_mutations_count() const {
       credentials_{credentials} {
 }
 
-std::shared_ptr<WatchStream> MockDatastore::CreateWatchStream(id<FSTWatchStreamDelegate> delegate) {
+std::shared_ptr<WatchStream> MockDatastore::CreateWatchStream(WatchStreamCallback* callback) {
   watch_stream_ = std::make_shared<MockWatchStream>(
       worker_queue_, credentials_,
       [[FSTSerializerBeta alloc] initWithDatabaseID:&database_info_->database_id()],
-      grpc_connection(), delegate, this);
+      grpc_connection(), callback, this);
 
   return watch_stream_;
 }

Firestore/Example/Tests/SpecTests/FSTSyncEngineTestDriver.h

@@ -86,7 +86,7 @@ typedef std::unordered_map<firebase::firestore::auth::User,
  *
  * Each method on the driver injects a different event into the system.
  */
-@interface FSTSyncEngineTestDriver : NSObject <FSTOnlineStateDelegate>
+@interface FSTSyncEngineTestDriver : NSObject
 
 /**
  * Initializes the underlying FSTSyncEngine with the given local persistence implementation and

Firestore/Example/Tests/SpecTests/FSTSyncEngineTestDriver.mm

@@ -153,18 +153,21 @@ - (instancetype)initWithPersistence:(id<FSTPersistence>)persistence
 
     _datastore =
         std::make_shared<MockDatastore>(_databaseInfo, _workerQueue.get(), &_credentialProvider);
-    _remoteStore = [[FSTRemoteStore alloc] initWithLocalStore:_localStore
-                                                    datastore:_datastore
-                                                  workerQueue:_workerQueue.get()];
+    _remoteStore =
+        [[FSTRemoteStore alloc] initWithLocalStore:_localStore
+                                         datastore:_datastore
+                                       workerQueue:_workerQueue.get()
+                                onlineStateHandler:[self](OnlineState onlineState) {
+                                  [self.syncEngine applyChangedOnlineState:onlineState];
+                                  [self.eventManager applyChangedOnlineState:onlineState];
+                                }];
 
     _syncEngine = [[FSTSyncEngine alloc] initWithLocalStore:_localStore
                                                 remoteStore:_remoteStore
                                                 initialUser:initialUser];
     _remoteStore.syncEngine = _syncEngine;
     _eventManager = [FSTEventManager eventManagerWithSyncEngine:_syncEngine];
 
-    _remoteStore.onlineStateDelegate = self;
-
     // Set up internal event tracking for the spec tests.
     NSMutableArray<FSTQueryEvent *> *events = [NSMutableArray array];
     _eventHandler = ^(FSTQueryEvent *e) {
@@ -203,11 +206,6 @@ - (void)drainQueue {
   return _currentUser;
 }
 
-- (void)applyChangedOnlineState:(OnlineState)onlineState {
-  [self.syncEngine applyChangedOnlineState:onlineState];
-  [self.eventManager applyChangedOnlineState:onlineState];
-}
-
 - (void)start {
   _workerQueue->EnqueueBlocking([&] {
     [self.localStore start];

Firestore/Example/Tests/Util/FSTHelpers.h

@@ -17,6 +17,7 @@
 #import <Foundation/Foundation.h>
 
 #include <string>
+#include <unordered_map>
 #include <vector>
 
 #import "Firestore/Source/Model/FSTDocument.h"
@@ -141,51 +142,57 @@ inline NSString *FSTRemoveExceptionPrefix(NSString *exception) {
   } while (0)
 
 /**
- * An implementation of FSTTargetMetadataProvider that provides controlled access to the
- * `FSTTargetMetadataProvider` callbacks. Any target accessed via these callbacks must be
+ * An implementation of `TargetMetadataProvider` that provides controlled access to the
+ * `TargetMetadataProvider` callbacks. Any target accessed via these callbacks must be
  * registered beforehand via the factory methods or via `setSyncedKeys:forQueryData:`.
  */
-@interface FSTTestTargetMetadataProvider : NSObject <FSTTargetMetadataProvider>
-
-/**
- * Creates an FSTTestTargetMetadataProvider that behaves as if there's an established listen for
- * each of the given targets, where each target has previously seen query results containing just
- * the given documentKey.
- *
- * Internally this means that the `remoteKeysForTarget` callback for these targets will return just
- * the documentKey and that the provided targets will be returned as active from the
- * `queryDataForTarget` target.
- */
-+ (instancetype)
-    providerWithSingleResultForKey:(firebase::firestore::model::DocumentKey)documentKey
-                           targets:
-                               (const std::vector<firebase::firestore::model::TargetId> &)targets;
-
-+ (instancetype)
-    providerWithSingleResultForKey:(firebase::firestore::model::DocumentKey)documentKey
-                     listenTargets:
-                         (const std::vector<firebase::firestore::model::TargetId> &)listenTargets
-                      limboTargets:
-                          (const std::vector<firebase::firestore::model::TargetId> &)limboTargets;
-
-/**
- * Creates an FSTTestTargetMetadataProvider that behaves as if there's an established listen for
- * each of the given targets, where each target has not seen any previous document.
- *
- * Internally this means that the `remoteKeysForTarget` callback for these targets will return an
- * empty set of document keys and that the provided targets will be returned as active from the
- * `queryDataForTarget` target.
- */
-+ (instancetype)
-    providerWithEmptyResultForKey:(firebase::firestore::model::DocumentKey)documentKey
-                          targets:
-                              (const std::vector<firebase::firestore::model::TargetId> &)targets;
+namespace firebase {
+namespace firestore {
+namespace remote {
 
-/** Sets or replaces the local state for the provided query data. */
-- (void)setSyncedKeys:(firebase::firestore::model::DocumentKeySet)keys
-         forQueryData:(FSTQueryData *)queryData;
+class TestTargetMetadataProvider : public TargetMetadataProvider {
+ public:
+  /**
+   * Creates a `TestTargetMetadataProvider` that behaves as if there's an established listen for
+   * each of the given targets, where each target has previously seen query results containing just
+   * the given `document_key`.
+   *
+   * Internally this means that the `GetRemoteKeysForTarget` callback for these targets will return
+   * just the `document_key` and that the provided targets will be returned as active from the
+   * `GetQueryDataForTarget` target.
+   */
+  static TestTargetMetadataProvider CreateSingleResultProvider(
+      model::DocumentKey document_key, const std::vector<model::TargetId> &targets);
+  static TestTargetMetadataProvider CreateSingleResultProvider(
+      model::DocumentKey document_key,
+      const std::vector<model::TargetId> &targets,
+      const std::vector<model::TargetId> &limbo_targets);
+
+  /**
+   * Creates an `TestTargetMetadataProvider` that behaves as if there's an established listen for
+   * each of the given targets, where each target has not seen any previous document.
+   *
+   * Internally this means that the `GetRemoteKeysForTarget` callback for these targets will return
+   * an empty set of document keys and that the provided targets will be returned as active from the
+   * `GetQueryDataForTarget` target.
+   */
+  static TestTargetMetadataProvider CreateEmptyResultProvider(
+      const model::DocumentKey &document_key, const std::vector<model::TargetId> &targets);
+
+  /** Sets or replaces the local state for the provided query data. */
+  void SetSyncedKeys(model::DocumentKeySet keys, FSTQueryData *query_data);
+
+  model::DocumentKeySet GetRemoteKeysForTarget(model::TargetId target_id) const override;
+  FSTQueryData *GetQueryDataForTarget(model::TargetId target_id) const override;
+
+ private:
+  std::unordered_map<model::TargetId, model::DocumentKeySet> synced_keys_;
+  std::unordered_map<model::TargetId, FSTQueryData *> query_data_;
+};
 
-@end
+}  // namespace remote
+}  // namespace firestore
+}  // namespace firebase
 
 /** Creates a new FIRTimestamp from components. Note that year, month, and day are all one-based. */
 FIRTimestamp *FSTTestTimestamp(int year, int month, int day, int hour, int minute, int second);