[realppl 7] realppl integration with remote/local and unit tests

Author: wu-hui

Firestore/Example/Firestore.xcodeproj/project.pbxproj

@@ -30,6 +30,7 @@
 #import "Firestore/Source/API/FIRQuerySnapshot+Internal.h"
 #import "Firestore/Source/API/FIRSnapshotMetadata+Internal.h"
 
+#include "Firestore/core/src/core/pipeline_util.h"
 #include "Firestore/core/src/core/query.h"
 #include "Firestore/core/src/core/view_snapshot.h"
 #include "Firestore/core/src/model/document.h"
@@ -101,7 +102,8 @@ - (void)testIncludeMetadataChanges {
 
   std::shared_ptr<Firestore> firestore = FSTTestFirestore().wrapped;
   core::Query query = Query("foo");
-  ViewSnapshot viewSnapshot(query, newDocuments, oldDocuments, std::move(documentChanges),
+  ViewSnapshot viewSnapshot(core::QueryOrPipeline(query), newDocuments, oldDocuments,
+                            std::move(documentChanges),
                             /*mutated_keys=*/DocumentKeySet(),
                             /*from_cache=*/false,
                             /*sync_state_changed=*/true,

Firestore/Example/Tests/API/FIRQuerySnapshotTests.mm

@@ -33,6 +33,7 @@
 #import "Firestore/Source/API/FIRSnapshotMetadata+Internal.h"
 #import "Firestore/Source/API/FSTUserDataReader.h"
 
+#include "Firestore/core/src/core/pipeline_util.h"
 #include "Firestore/core/src/core/view_snapshot.h"
 #include "Firestore/core/src/model/document.h"
 #include "Firestore/core/src/model/document_set.h"
@@ -148,7 +149,7 @@
     }
     newDocuments = newDocuments.insert(doc);
   }
-  ViewSnapshot viewSnapshot{Query(path),
+  ViewSnapshot viewSnapshot{core::QueryOrPipeline(Query(path)),
                             newDocuments,
                             oldDocuments,
                             std::move(documentChanges),

Firestore/Example/Tests/API/FSTAPIHelpers.mm

@@ -109,7 +109,7 @@ bool IsOpen() const override {
   }
 
   void WatchQuery(const TargetData& query) override {
-    LOG_DEBUG("WatchQuery: %s: %s, %s", query.target_id(), query.target().ToString(),
+    LOG_DEBUG("WatchQuery: %s: %s, %s", query.target_id(), query.target_or_pipeline().ToString(),
               query.resume_token().ToString());
 
     // Snapshot version is ignored on the wire

Firestore/Example/Tests/SpecTests/FSTMockDatastore.mm

@@ -842,36 +842,36 @@ - (void)validateExpectedState:(nullable NSDictionary *)expectedState {
     }
     if (expectedState[@"activeTargets"]) {
       __block ActiveTargetMap expectedActiveTargets;
-      [expectedState[@"activeTargets"]
-          enumerateKeysAndObjectsUsingBlock:^(NSString *targetIDString, NSDictionary *queryData,
-                                              BOOL *) {
-            TargetId targetID = [targetIDString intValue];
-            NSArray *queriesJson = queryData[@"queries"];
-            std::vector<TargetData> queries;
-            for (id queryJson in queriesJson) {
-              Query query = [self parseQuery:queryJson];
-
-              QueryPurpose purpose = QueryPurpose::Listen;
-              if ([queryData objectForKey:@"targetPurpose"] != nil) {
-                purpose = [self parseQueryPurpose:queryData[@"targetPurpose"]];
-              }
-
-              TargetData target_data(query.ToTarget(), targetID, 0, purpose);
-              if ([queryData objectForKey:@"resumeToken"] != nil) {
-                target_data = target_data.WithResumeToken(
-                    MakeResumeToken(queryData[@"resumeToken"]), SnapshotVersion::None());
-              } else {
-                target_data = target_data.WithResumeToken(
-                    ByteString(), [self parseVersion:queryData[@"readTime"]]);
-              }
-
-              if ([queryData objectForKey:@"expectedCount"] != nil) {
-                target_data = target_data.WithExpectedCount([queryData[@"expectedCount"] intValue]);
-              }
-              queries.push_back(std::move(target_data));
-            }
-            expectedActiveTargets[targetID] = std::move(queries);
-          }];
+      [expectedState[@"activeTargets"] enumerateKeysAndObjectsUsingBlock:^(NSString *targetIDString,
+                                                                           NSDictionary *queryData,
+                                                                           BOOL *) {
+        TargetId targetID = [targetIDString intValue];
+        NSArray *queriesJson = queryData[@"queries"];
+        std::vector<TargetData> queries;
+        for (id queryJson in queriesJson) {
+          Query query = [self parseQuery:queryJson];
+
+          QueryPurpose purpose = QueryPurpose::Listen;
+          if ([queryData objectForKey:@"targetPurpose"] != nil) {
+            purpose = [self parseQueryPurpose:queryData[@"targetPurpose"]];
+          }
+
+          TargetData target_data(core::TargetOrPipeline(query.ToTarget()), targetID, 0, purpose);
+          if ([queryData objectForKey:@"resumeToken"] != nil) {
+            target_data = target_data.WithResumeToken(MakeResumeToken(queryData[@"resumeToken"]),
+                                                      SnapshotVersion::None());
+          } else {
+            target_data = target_data.WithResumeToken(ByteString(),
+                                                      [self parseVersion:queryData[@"readTime"]]);
+          }
+
+          if ([queryData objectForKey:@"expectedCount"] != nil) {
+            target_data = target_data.WithExpectedCount([queryData[@"expectedCount"] intValue]);
+          }
+          queries.push_back(std::move(target_data));
+        }
+        expectedActiveTargets[targetID] = std::move(queries);
+      }];
       [self.driver setExpectedActiveTargets:std::move(expectedActiveTargets)];
     }
   }
@@ -982,7 +982,7 @@ - (void)validateActiveTargets {
     const TargetData &actual = found->second;
 
     XCTAssertEqual(actual.purpose(), targetData.purpose());
-    XCTAssertEqual(actual.target(), targetData.target());
+    XCTAssertEqual(actual.target_or_pipeline(), targetData.target_or_pipeline());
     XCTAssertEqual(actual.target_id(), targetData.target_id());
     XCTAssertEqual(actual.snapshot_version(), targetData.snapshot_version());
     XCTAssertEqual(actual.resume_token(), targetData.resume_token());

Firestore/Example/Tests/SpecTests/FSTSpecTests.mm

@@ -478,18 +478,19 @@ - (FSTOutstandingWrite *)receiveWriteError:(int)errorCode
 
 - (TargetId)addUserListenerWithQuery:(Query)query options:(ListenOptions)options {
   // TODO(dimond): Change spec tests to verify isFromCache on snapshots
-  auto listener = QueryListener::Create(
-      query, options, [self, query](const StatusOr<ViewSnapshot> &maybe_snapshot) {
-        FSTQueryEvent *event = [[FSTQueryEvent alloc] init];
-        event.query = query;
-        if (maybe_snapshot.ok()) {
-          [event setViewSnapshot:maybe_snapshot.ValueOrDie()];
-        } else {
-          event.error = MakeNSError(maybe_snapshot.status());
-        }
-
-        [self.events addObject:event];
-      });
+  auto listener =
+      QueryListener::Create(core::QueryOrPipeline(query), options,
+                            [self, query](const StatusOr<ViewSnapshot> &maybe_snapshot) {
+                              FSTQueryEvent *event = [[FSTQueryEvent alloc] init];
+                              event.query = query;
+                              if (maybe_snapshot.ok()) {
+                                [event setViewSnapshot:maybe_snapshot.ValueOrDie()];
+                              } else {
+                                event.error = MakeNSError(maybe_snapshot.status());
+                              }
+
+                              [self.events addObject:event];
+                            });
   _queryListeners[query] = listener;
   TargetId targetID;
   _workerQueue->EnqueueBlocking([&] { targetID = _eventManager->AddQueryListener(listener); });

Firestore/Example/Tests/SpecTests/FSTSyncEngineTestDriver.mm

@@ -51,6 +51,7 @@
 #include "Firestore/core/src/core/firestore_client.h"
 #include "Firestore/core/src/core/listen_options.h"
 #include "Firestore/core/src/core/order_by.h"
+#include "Firestore/core/src/core/pipeline_util.h"
 #include "Firestore/core/src/core/query.h"
 #include "Firestore/core/src/model/document_key.h"
 #include "Firestore/core/src/model/field_path.h"
@@ -228,8 +229,8 @@ - (void)getDocumentsWithSource:(FIRFirestoreSource)publicSource
   auto async_listener = AsyncEventListener<ViewSnapshot>::Create(
       firestore->client()->user_executor(), std::move(view_listener));
 
-  std::shared_ptr<QueryListener> query_listener =
-      firestore->client()->ListenToQuery(query, internalOptions, async_listener);
+  std::shared_ptr<QueryListener> query_listener = firestore->client()->ListenToQuery(
+      core::QueryOrPipeline(query), internalOptions, async_listener);
 
   return [[FSTListenerRegistration alloc]
       initWithRegistration:absl::make_unique<QueryListenerRegistration>(firestore->client(),

Firestore/Source/API/FIRQuery.mm

@@ -46,6 +46,7 @@ class Firestore;
 class ListenerRegistration;
 class Pipeline;
 class PipelineSnapshot;
+class RealtimePipeline;
 class Query;
 class QuerySnapshot;
 class Settings;

Firestore/core/src/api/api_fwd.h

@@ -238,8 +238,8 @@ std::unique_ptr<ListenerRegistration> DocumentReference::AddSnapshotListener(
 
   core::Query query(key_.path());
   std::shared_ptr<QueryListener> query_listener =
-      firestore_->client()->ListenToQuery(std::move(query), options,
-                                          async_listener);
+      firestore_->client()->ListenToQuery(
+          core::QueryOrPipeline(std::move(query)), options, async_listener);
 
   return absl::make_unique<QueryListenerRegistration>(
       firestore_->client(), std::move(async_listener),

Firestore/core/src/api/document_reference.cc

@@ -50,6 +50,10 @@ google_firestore_v1_Value Constant::to_proto() const {
   return *model::DeepClone(*value_).release();
 }
 
+const google_firestore_v1_Value& Constant::value() const {
+  return *value_;
+}
+
 std::unique_ptr<core::EvaluableExpr> Constant::ToEvaluable() const {
   return std::make_unique<core::CoreConstant>(
       std::make_unique<Constant>(*this));