Partially migrate stream tokens from NSData to ByteString (#3643)

* Convert stream tokens from NSData to ByteString

Note that this persists in using protobuf's conflation of empty and null
strings.

* Rename remote_event.mm to .cc

Author: wilhuff

Firestore/Example/Tests/Local/FSTLocalSerializerTests.mm

@@ -215,7 +215,7 @@ - (void)testEncodesQueryData {
   expected.targetId = targetID;
   expected.lastListenSequenceNumber = 10;
   expected.snapshotVersion.nanos = 1039000;
-  expected.resumeToken = MakeNSData(resumeToken);
+  expected.resumeToken = MakeNullableNSData(resumeToken);
   expected.query.parent = queryTarget.parent;
   expected.query.structuredQuery = queryTarget.structuredQuery;
 

Firestore/Example/Tests/Local/FSTMutationQueueTests.mm

@@ -44,6 +44,7 @@
 using firebase::firestore::model::kBatchIdUnknown;
 using firebase::firestore::model::Mutation;
 using firebase::firestore::model::SetMutation;
+using firebase::firestore::nanopb::ByteString;
 using firebase::firestore::testutil::Key;
 using firebase::firestore::testutil::Query;
 
@@ -103,11 +104,11 @@ - (void)testAcknowledgeBatchID {
 
     XCTAssertEqual([self batchCount], 3);
 
-    self.mutationQueue->AcknowledgeBatch(batch1, nil);
+    self.mutationQueue->AcknowledgeBatch(batch1, {});
     self.mutationQueue->RemoveMutationBatch(batch1);
     XCTAssertEqual([self batchCount], 2);
 
-    self.mutationQueue->AcknowledgeBatch(batch2, nil);
+    self.mutationQueue->AcknowledgeBatch(batch2, {});
     XCTAssertEqual([self batchCount], 2);
 
     self.mutationQueue->RemoveMutationBatch(batch2);
@@ -124,7 +125,7 @@ - (void)testAcknowledgeThenRemove {
   self.persistence.run("testAcknowledgeThenRemove", [&]() {
     FSTMutationBatch *batch1 = [self addMutationBatch];
 
-    self.mutationQueue->AcknowledgeBatch(batch1, nil);
+    self.mutationQueue->AcknowledgeBatch(batch1, {});
     self.mutationQueue->RemoveMutationBatch(batch1);
 
     XCTAssertEqual([self batchCount], 0);
@@ -377,19 +378,19 @@ - (void)testRemoveMutationBatches {
 - (void)testStreamToken {
   if ([self isTestBaseClass]) return;
 
-  NSData *streamToken1 = [@"token1" dataUsingEncoding:NSUTF8StringEncoding];
-  NSData *streamToken2 = [@"token2" dataUsingEncoding:NSUTF8StringEncoding];
+  ByteString streamToken1("token1");
+  ByteString streamToken2("token2");
 
   self.persistence.run("testStreamToken", [&]() {
     self.mutationQueue->SetLastStreamToken(streamToken1);
 
     FSTMutationBatch *batch1 = [self addMutationBatch];
     [self addMutationBatch];
 
-    XCTAssertEqualObjects(self.mutationQueue->GetLastStreamToken(), streamToken1);
+    XCTAssertEqual(self.mutationQueue->GetLastStreamToken(), streamToken1);
 
     self.mutationQueue->AcknowledgeBatch(batch1, streamToken2);
-    XCTAssertEqualObjects(self.mutationQueue->GetLastStreamToken(), streamToken2);
+    XCTAssertEqual(self.mutationQueue->GetLastStreamToken(), streamToken2);
   });
 }
 

Firestore/Source/Local/FSTLocalSerializer.mm

@@ -238,7 +238,7 @@ - (FSTPBTarget *)encodedQueryData:(const QueryData &)queryData {
   proto.targetId = queryData.target_id();
   proto.lastListenSequenceNumber = queryData.sequence_number();
   proto.snapshotVersion = [remoteSerializer encodedVersion:queryData.snapshot_version()];
-  proto.resumeToken = MakeNSData(queryData.resume_token());
+  proto.resumeToken = MakeNullableNSData(queryData.resume_token());
 
   const Query &query = queryData.query();
   if (query.IsDocumentQuery()) {

Firestore/Source/Local/FSTLocalStore.h

@@ -30,6 +30,7 @@
 #include "Firestore/core/src/firebase/firestore/model/mutation.h"
 #include "Firestore/core/src/firebase/firestore/model/snapshot_version.h"
 #include "Firestore/core/src/firebase/firestore/model/types.h"
+#include "Firestore/core/src/firebase/firestore/nanopb/byte_string.h"
 
 namespace firebase {
 namespace firestore {
@@ -51,6 +52,7 @@ namespace auth = firebase::firestore::auth;
 namespace core = firebase::firestore::core;
 namespace local = firebase::firestore::local;
 namespace model = firebase::firestore::model;
+namespace nanopb = firebase::firestore::nanopb;
 namespace remote = firebase::firestore::remote;
 
 NS_ASSUME_NONNULL_BEGIN
@@ -139,14 +141,14 @@ NS_ASSUME_NONNULL_BEGIN
 - (model::MaybeDocumentMap)rejectBatchID:(model::BatchId)batchID;
 
 /** Returns the last recorded stream token for the current user. */
-- (nullable NSData *)lastStreamToken;
+- (nanopb::ByteString)lastStreamToken;
 
 /**
  * Sets the stream token for the current user without acknowledging any mutation batch. This is
  * usually only useful after a stream handshake or in response to an error that requires clearing
  * the stream token.
  */
-- (void)setLastStreamToken:(nullable NSData *)streamToken;
+- (void)setLastStreamToken:(const nanopb::ByteString &)streamToken;
 
 /**
  * Returns the last consistent snapshot processed (used by the RemoteStore to determine whether to

Firestore/Source/Local/FSTLocalStore.mm

@@ -43,6 +43,7 @@
 #include "Firestore/core/src/firebase/firestore/model/document_map.h"
 #include "Firestore/core/src/firebase/firestore/model/patch_mutation.h"
 #include "Firestore/core/src/firebase/firestore/model/snapshot_version.h"
+#include "Firestore/core/src/firebase/firestore/nanopb/nanopb_util.h"
 #include "Firestore/core/src/firebase/firestore/remote/remote_event.h"
 #include "Firestore/core/src/firebase/firestore/util/hard_assert.h"
 #include "Firestore/core/src/firebase/firestore/util/log.h"
@@ -224,7 +225,7 @@ - (LocalWriteResult)locallyWriteMutations:(std::vector<Mutation> &&)mutations {
 - (MaybeDocumentMap)acknowledgeBatchWithResult:(FSTMutationBatchResult *)batchResult {
   return self.persistence.run("Acknowledge batch", [&] {
     FSTMutationBatch *batch = batchResult.batch;
-    _mutationQueue->AcknowledgeBatch(batch, batchResult.streamToken);
+    _mutationQueue->AcknowledgeBatch(batch, nanopb::MakeByteString(batchResult.streamToken));
     [self applyBatchResult:batchResult];
     _mutationQueue->PerformConsistencyCheck();
 
@@ -244,11 +245,11 @@ - (MaybeDocumentMap)rejectBatchID:(BatchId)batchID {
   });
 }
 
-- (nullable NSData *)lastStreamToken {
+- (ByteString)lastStreamToken {
   return _mutationQueue->GetLastStreamToken();
 }
 
-- (void)setLastStreamToken:(nullable NSData *)streamToken {
+- (void)setLastStreamToken:(const ByteString &)streamToken {
   self.persistence.run("Set stream token",
                        [&]() { _mutationQueue->SetLastStreamToken(streamToken); });
 }

Firestore/Source/Remote/FSTSerializerBeta.mm

@@ -788,9 +788,7 @@ - (GCFSTarget *)encodedTarget:(const QueryData &)queryData {
   }
 
   result.targetId = queryData.target_id();
-  if (!queryData.resume_token().empty()) {
-    result.resumeToken = MakeNSData(queryData.resume_token());
-  }
+  result.resumeToken = MakeNullableNSData(queryData.resume_token());
 
   return result;
 }

Firestore/core/src/firebase/firestore/local/leveldb_mutation_queue.h

@@ -65,7 +65,7 @@ class LevelDbMutationQueue : public MutationQueue {
   bool IsEmpty() override;
 
   void AcknowledgeBatch(FSTMutationBatch* batch,
-                        NSData* _Nullable stream_token) override;
+                        const nanopb::ByteString& stream_token) override;
 
   FSTMutationBatch* AddMutationBatch(
       const Timestamp& local_write_time,
@@ -95,9 +95,9 @@ class LevelDbMutationQueue : public MutationQueue {
 
   void PerformConsistencyCheck() override;
 
-  NSData* _Nullable GetLastStreamToken() override;
+  nanopb::ByteString GetLastStreamToken() override;
 
-  void SetLastStreamToken(NSData* _Nullable stream_token) override;
+  void SetLastStreamToken(const nanopb::ByteString& stream_token) override;
 
  private:
   /**

Firestore/core/src/firebase/firestore/local/leveldb_mutation_queue.mm

@@ -27,6 +27,7 @@
 #include "Firestore/core/src/firebase/firestore/local/leveldb_util.h"
 #include "Firestore/core/src/firebase/firestore/model/mutation_batch.h"
 #include "Firestore/core/src/firebase/firestore/model/resource_path.h"
+#include "Firestore/core/src/firebase/firestore/nanopb/nanopb_util.h"
 #include "Firestore/core/src/firebase/firestore/util/string_util.h"
 #include "Firestore/core/src/firebase/firestore/util/to_string.h"
 #include "absl/strings/match.h"
@@ -48,6 +49,9 @@
 using model::kBatchIdUnknown;
 using model::Mutation;
 using model::ResourcePath;
+using nanopb::ByteString;
+using nanopb::MakeByteString;
+using nanopb::MakeNSData;
 
 BatchId LoadNextBatchIdFromDb(DB* db) {
   // TODO(gsoltis): implement Prev() and SeekToLast() on
@@ -151,7 +155,7 @@ BatchId LoadNextBatchIdFromDb(DB* db) {
 }
 
 void LevelDbMutationQueue::AcknowledgeBatch(FSTMutationBatch* batch,
-                                            NSData* _Nullable stream_token) {
+                                            const ByteString& stream_token) {
   SetLastStreamToken(stream_token);
 }
 
@@ -423,12 +427,12 @@ BatchId LoadNextBatchIdFromDb(DB* db) {
       util::ToString(dangling_mutation_references));
 }
 
-NSData* _Nullable LevelDbMutationQueue::GetLastStreamToken() {
-  return metadata_.lastStreamToken;
+ByteString LevelDbMutationQueue::GetLastStreamToken() {
+  return MakeByteString(metadata_.lastStreamToken);
 }
 
-void LevelDbMutationQueue::SetLastStreamToken(NSData* _Nullable stream_token) {
-  metadata_.lastStreamToken = stream_token;
+void LevelDbMutationQueue::SetLastStreamToken(const ByteString& stream_token) {
+  metadata_.lastStreamToken = MakeNullableNSData(stream_token);
 
   db_.currentTransaction->Put(mutation_queue_key(), metadata_);
 }

Firestore/core/src/firebase/firestore/local/memory_mutation_queue.h

@@ -53,7 +53,7 @@ class MemoryMutationQueue : public MutationQueue {
   bool IsEmpty() override;
 
   void AcknowledgeBatch(FSTMutationBatch* batch,
-                        NSData* _Nullable stream_token) override;
+                        const nanopb::ByteString& stream_token) override;
 
   FSTMutationBatch* AddMutationBatch(
       const Timestamp& local_write_time,
@@ -89,8 +89,8 @@ class MemoryMutationQueue : public MutationQueue {
 
   size_t CalculateByteSize(FSTLocalSerializer* serializer);
 
-  NSData* _Nullable GetLastStreamToken() override;
-  void SetLastStreamToken(NSData* _Nullable token) override;
+  nanopb::ByteString GetLastStreamToken() override;
+  void SetLastStreamToken(const nanopb::ByteString& token) override;
 
  private:
   using DocumentKeyReferenceSet =
@@ -143,7 +143,7 @@ class MemoryMutationQueue : public MutationQueue {
    * responses the client has processed. Stream tokens are opaque checkpoint
    * markers whose only real value is their inclusion in the next request.
    */
-  NSData* _Nullable last_stream_token_;
+  nanopb::ByteString last_stream_token_;
 
   /** An ordered mapping between documents and the mutation batch IDs. */
   DocumentKeyReferenceSet batches_by_document_key_;

Firestore/core/src/firebase/firestore/local/memory_mutation_queue.mm

@@ -41,6 +41,7 @@
 using model::kBatchIdUnknown;
 using model::Mutation;
 using model::ResourcePath;
+using nanopb::ByteString;
 
 MemoryMutationQueue::MemoryMutationQueue(FSTMemoryPersistence* persistence)
     : persistence_(persistence) {
@@ -53,7 +54,7 @@
 }
 
 void MemoryMutationQueue::AcknowledgeBatch(FSTMutationBatch* batch,
-                                           NSData* _Nullable stream_token) {
+                                           const ByteString& stream_token) {
   HARD_ASSERT(!queue_.empty(), "Cannot acknowledge batch on an empty queue");
 
   // Guaranteed to exist, due to above assert
@@ -262,11 +263,11 @@
   return count;
 }
 
-NSData* _Nullable MemoryMutationQueue::GetLastStreamToken() {
+ByteString MemoryMutationQueue::GetLastStreamToken() {
   return last_stream_token_;
 }
 
-void MemoryMutationQueue::SetLastStreamToken(NSData* _Nullable token) {
+void MemoryMutationQueue::SetLastStreamToken(const ByteString& token) {
   last_stream_token_ = token;
 }