C++ migration: port write stream-related part of FSTRemoteStore (#2335)

Author: var-const

Firestore/Example/Tests/Local/FSTLocalStoreTests.mm

@@ -153,7 +153,7 @@ - (void)acknowledgeMutationWithVersion:(FSTTestSnapshotVersion)documentVersion {
                                                                 transformResults:nil];
   FSTMutationBatchResult *result = [FSTMutationBatchResult resultWithBatch:batch
                                                              commitVersion:version
-                                                           mutationResults:@[ mutationResult ]
+                                                           mutationResults:{mutationResult}
                                                                streamToken:nil];
   _lastChanges = [self.localStore acknowledgeBatchWithResult:result];
 }

Firestore/Example/Tests/SpecTests/FSTMockDatastore.h

@@ -18,6 +18,7 @@
 
 #include <memory>
 #include <unordered_map>
+#include <vector>
 
 #include "Firestore/core/src/firebase/firestore/model/snapshot_version.h"
 #include "Firestore/core/src/firebase/firestore/model/types.h"
@@ -40,7 +41,7 @@ class MockDatastore : public Datastore {
                 auth::CredentialsProvider* credentials);
 
   std::shared_ptr<WatchStream> CreateWatchStream(WatchStreamCallback* callback) override;
-  std::shared_ptr<WriteStream> CreateWriteStream(id<FSTWriteStreamDelegate> delegate) override;
+  std::shared_ptr<WriteStream> CreateWriteStream(WriteStreamCallback* callback) override;
 
   /**
    * A count of the total number of requests sent to the watch stream since the beginning of the
@@ -82,7 +83,7 @@ class MockDatastore : public Datastore {
   int WritesSent() const;
 
   /** Injects a write ack as though it had come from the backend in response to a write. */
-  void AckWrite(const model::SnapshotVersion& version, NSArray<FSTMutationResult*>* results);
+  void AckWrite(const model::SnapshotVersion& version, std::vector<FSTMutationResult*> results);
 
   /** Injects a stream failure as though it had come from the backend. */
   void FailWrite(const util::Status& error);

Firestore/Example/Tests/SpecTests/FSTMockDatastore.mm

@@ -25,7 +25,6 @@
 #import "Firestore/Source/Local/FSTQueryData.h"
 #import "Firestore/Source/Model/FSTMutation.h"
 #import "Firestore/Source/Remote/FSTSerializerBeta.h"
-#import "Firestore/Source/Remote/FSTStream.h"
 
 #include "Firestore/core/src/firebase/firestore/auth/credentials_provider.h"
 #include "Firestore/core/src/firebase/firestore/auth/empty_credentials_provider.h"
@@ -161,18 +160,18 @@ void WriteWatchChange(const WatchChange& change, SnapshotVersion snap) {
                   CredentialsProvider* credentials_provider,
                   FSTSerializerBeta* serializer,
                   GrpcConnection* grpc_connection,
-                  id<FSTWriteStreamDelegate> delegate,
+                  WriteStreamCallback* callback,
                   MockDatastore* datastore)
-      : WriteStream{worker_queue, credentials_provider, serializer, grpc_connection, delegate},
+      : WriteStream{worker_queue, credentials_provider, serializer, grpc_connection, callback},
         datastore_{datastore},
-        delegate_{delegate} {
+        callback_{callback} {
   }
 
   void Start() override {
     HARD_ASSERT(!open_, "Trying to start already started write stream");
     open_ = true;
     sent_mutations_ = {};
-    [delegate_ writeStreamDidOpen];
+    callback_->OnWriteStreamOpen();
   }
 
   void Stop() override {
@@ -194,7 +193,7 @@ bool IsOpen() const override {
   void WriteHandshake() override {
     datastore_->IncrementWriteStreamRequests();
     SetHandshakeComplete();
-    [delegate_ writeStreamDidCompleteHandshake];
+    callback_->OnWriteStreamHandshakeComplete();
   }
 
   void WriteMutations(NSArray<FSTMutation*>* mutations) override {
@@ -203,14 +202,14 @@ void WriteMutations(NSArray<FSTMutation*>* mutations) override {
   }
 
   /** Injects a write ack as though it had come from the backend in response to a write. */
-  void AckWrite(const SnapshotVersion& commitVersion, NSArray<FSTMutationResult*>* results) {
-    [delegate_ writeStreamDidReceiveResponseWithVersion:commitVersion mutationResults:results];
+  void AckWrite(const SnapshotVersion& commitVersion, std::vector<FSTMutationResult*> results) {
+    callback_->OnWriteStreamMutationResult(commitVersion, std::move(results));
   }
 
   /** Injects a failed write response as though it had come from the backend. */
   void FailStream(const Status& error) {
     open_ = false;
-    [delegate_ writeStreamWasInterruptedWithError:error];
+    callback_->OnWriteStreamClose(error);
   }
 
   /**
@@ -236,7 +235,7 @@ int sent_mutations_count() const {
   bool open_ = false;
   std::queue<NSArray<FSTMutation*>*> sent_mutations_;
   MockDatastore* datastore_ = nullptr;
-  id<FSTWriteStreamDelegate> delegate_ = nullptr;
+  WriteStreamCallback* callback_ = nullptr;
 };
 
 MockDatastore::MockDatastore(const core::DatabaseInfo& database_info,
@@ -257,11 +256,11 @@ int sent_mutations_count() const {
   return watch_stream_;
 }
 
-std::shared_ptr<WriteStream> MockDatastore::CreateWriteStream(id<FSTWriteStreamDelegate> delegate) {
+std::shared_ptr<WriteStream> MockDatastore::CreateWriteStream(WriteStreamCallback* callback) {
   write_stream_ = std::make_shared<MockWriteStream>(
       worker_queue_, credentials_,
       [[FSTSerializerBeta alloc] initWithDatabaseID:&database_info_->database_id()],
-      grpc_connection(), delegate, this);
+      grpc_connection(), callback, this);
 
   return write_stream_;
 }
@@ -290,8 +289,9 @@ int sent_mutations_count() const {
   return write_stream_->sent_mutations_count();
 }
 
-void MockDatastore::AckWrite(const SnapshotVersion& version, NSArray<FSTMutationResult*>* results) {
-  write_stream_->AckWrite(version, results);
+void MockDatastore::AckWrite(const SnapshotVersion& version,
+                             std::vector<FSTMutationResult*> results) {
+  write_stream_->AckWrite(version, std::move(results));
 }
 
 void MockDatastore::FailWrite(const Status& error) {

Firestore/Example/Tests/SpecTests/FSTSpecTests.mm

@@ -363,7 +363,7 @@ - (void)doWriteAck:(NSDictionary *)spec {
 
   FSTMutationResult *mutationResult = [[FSTMutationResult alloc] initWithVersion:version
                                                                 transformResults:nil];
-  [self.driver receiveWriteAckWithVersion:version mutationResults:@[ mutationResult ]];
+  [self.driver receiveWriteAckWithVersion:version mutationResults:{mutationResult}];
 }
 
 - (void)doFailWrite:(NSDictionary *)spec {

Firestore/Example/Tests/SpecTests/FSTSyncEngineTestDriver.h

@@ -18,6 +18,7 @@
 
 #include <map>
 #include <unordered_map>
+#include <vector>
 
 #import "Firestore/Source/Remote/FSTRemoteStore.h"
 
@@ -195,9 +196,9 @@ typedef std::unordered_map<firebase::firestore::auth::User,
  *     the mutation. Snapshot versions must be monotonically increasing.
  * @param mutationResults The mutation results for the write that is being acked.
  */
-- (FSTOutstandingWrite *)receiveWriteAckWithVersion:
-                             (const firebase::firestore::model::SnapshotVersion &)commitVersion
-                                    mutationResults:(NSArray<FSTMutationResult *> *)mutationResults;
+- (FSTOutstandingWrite *)
+    receiveWriteAckWithVersion:(const firebase::firestore::model::SnapshotVersion &)commitVersion
+               mutationResults:(std::vector<FSTMutationResult *>)mutationResults;
 
 /**
  * A count of the mutations written to the write stream by the FSTSyncEngine, but not yet

Firestore/Example/Tests/SpecTests/FSTSyncEngineTestDriver.mm

@@ -273,12 +273,13 @@ - (void)changeUser:(const User &)user {
 
 - (FSTOutstandingWrite *)receiveWriteAckWithVersion:(const SnapshotVersion &)commitVersion
                                     mutationResults:
-                                        (NSArray<FSTMutationResult *> *)mutationResults {
+                                        (std::vector<FSTMutationResult *>)mutationResults {
   FSTOutstandingWrite *write = [self currentOutstandingWrites].firstObject;
   [[self currentOutstandingWrites] removeObjectAtIndex:0];
   [self validateNextWriteSent:write.write];
 
-  _workerQueue->EnqueueBlocking([&] { _datastore->AckWrite(commitVersion, mutationResults); });
+  _workerQueue->EnqueueBlocking(
+      [&] { _datastore->AckWrite(commitVersion, std::move(mutationResults)); });
 
   return write;
 }

Firestore/Source/Model/FSTMutationBatch.h

@@ -17,6 +17,7 @@
 #import <Foundation/Foundation.h>
 
 #include <unordered_map>
+#include <vector>
 
 #include "Firestore/core/src/firebase/firestore/model/document_key.h"
 #include "Firestore/core/src/firebase/firestore/model/document_key_set.h"
@@ -102,13 +103,13 @@ NS_ASSUME_NONNULL_BEGIN
  */
 + (instancetype)resultWithBatch:(FSTMutationBatch *)batch
                   commitVersion:(firebase::firestore::model::SnapshotVersion)commitVersion
-                mutationResults:(NSArray<FSTMutationResult *> *)mutationResults
+                mutationResults:(std::vector<FSTMutationResult *>)mutationResults
                     streamToken:(nullable NSData *)streamToken;
 
 - (const firebase::firestore::model::SnapshotVersion &)commitVersion;
+- (const std::vector<FSTMutationResult *> &)mutationResults;
 
 @property(nonatomic, strong, readonly) FSTMutationBatch *batch;
-@property(nonatomic, strong, readonly) NSArray<FSTMutationResult *> *mutationResults;
 @property(nonatomic, strong, readonly, nullable) NSData *streamToken;
 
 - (const firebase::firestore::model::DocumentVersionMap &)docVersions;

Firestore/Source/Model/FSTMutationBatch.mm

@@ -82,9 +82,9 @@ - (FSTMaybeDocument *_Nullable)applyToRemoteDocument:(FSTMaybeDocument *_Nullabl
               "applyTo: key %s doesn't match maybeDoc key %s", documentKey.ToString(),
               maybeDoc.key.ToString());
 
-  HARD_ASSERT(mutationBatchResult.mutationResults.count == self.mutations.count,
+  HARD_ASSERT(mutationBatchResult.mutationResults.size() == self.mutations.count,
               "Mismatch between mutations length (%s) and results length (%s)",
-              self.mutations.count, mutationBatchResult.mutationResults.count);
+              self.mutations.count, mutationBatchResult.mutationResults.size());
 
   for (NSUInteger i = 0; i < self.mutations.count; i++) {
     FSTMutation *mutation = self.mutations[i];
@@ -130,25 +130,26 @@ - (DocumentKeySet)keys {
 @interface FSTMutationBatchResult ()
 - (instancetype)initWithBatch:(FSTMutationBatch *)batch
                 commitVersion:(SnapshotVersion)commitVersion
-              mutationResults:(NSArray<FSTMutationResult *> *)mutationResults
+              mutationResults:(std::vector<FSTMutationResult *>)mutationResults
                   streamToken:(nullable NSData *)streamToken
                   docVersions:(DocumentVersionMap)docVersions NS_DESIGNATED_INITIALIZER;
 @end
 
 @implementation FSTMutationBatchResult {
   SnapshotVersion _commitVersion;
+  std::vector<FSTMutationResult *> _mutationResults;
   DocumentVersionMap _docVersions;
 }
 
 - (instancetype)initWithBatch:(FSTMutationBatch *)batch
                 commitVersion:(SnapshotVersion)commitVersion
-              mutationResults:(NSArray<FSTMutationResult *> *)mutationResults
+              mutationResults:(std::vector<FSTMutationResult *>)mutationResults
                   streamToken:(nullable NSData *)streamToken
                   docVersions:(DocumentVersionMap)docVersions {
   if (self = [super init]) {
     _batch = batch;
     _commitVersion = std::move(commitVersion);
-    _mutationResults = mutationResults;
+    _mutationResults = std::move(mutationResults);
     _streamToken = streamToken;
     _docVersions = std::move(docVersions);
   }
@@ -159,17 +160,21 @@ - (instancetype)initWithBatch:(FSTMutationBatch *)batch
   return _commitVersion;
 }
 
+- (const std::vector<FSTMutationResult *> &)mutationResults {
+  return _mutationResults;
+}
+
 - (const DocumentVersionMap &)docVersions {
   return _docVersions;
 }
 
 + (instancetype)resultWithBatch:(FSTMutationBatch *)batch
                   commitVersion:(SnapshotVersion)commitVersion
-                mutationResults:(NSArray<FSTMutationResult *> *)mutationResults
+                mutationResults:(std::vector<FSTMutationResult *>)mutationResults
                     streamToken:(nullable NSData *)streamToken {
-  HARD_ASSERT(batch.mutations.count == mutationResults.count,
+  HARD_ASSERT(batch.mutations.count == mutationResults.size(),
               "Mutations sent %s must equal results received %s", batch.mutations.count,
-              mutationResults.count);
+              mutationResults.size());
 
   DocumentVersionMap docVersions;
   NSArray<FSTMutation *> *mutations = batch.mutations;
@@ -186,7 +191,7 @@ + (instancetype)resultWithBatch:(FSTMutationBatch *)batch
 
   return [[FSTMutationBatchResult alloc] initWithBatch:batch
                                          commitVersion:std::move(commitVersion)
-                                       mutationResults:mutationResults
+                                       mutationResults:std::move(mutationResults)
                                            streamToken:streamToken
                                            docVersions:std::move(docVersions)];
 }