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

Firestore/Example/Tests/API/FIRQuerySnapshotTests.mm

@@ -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/FSTAPIHelpers.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/SpecTests/FSTMockDatastore.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/FSTSpecTests.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/FSTSyncEngineTestDriver.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/Source/API/FIRQuery.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/core/src/api/api_fwd.h

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

Firestore/core/src/api/document_reference.cc

@@ -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/expressions.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));

Firestore/core/src/api/expressions.h

@@ -80,6 +80,8 @@ class Constant : public Expr {
   }
   google_firestore_v1_Value to_proto() const override;
 
+  const google_firestore_v1_Value& value() const;
+
   std::unique_ptr<core::EvaluableExpr> ToEvaluable() const override;
 
  private:

Firestore/core/src/api/ordering.h

@@ -49,10 +49,18 @@ class Ordering {
     return expr_.get();
   }
 
+  const std::shared_ptr<Expr> expr_shared() const {
+    return expr_;
+  }
+
   Direction direction() const {
     return direction_;
   }
 
+  Ordering WithReversedDirection() const {
+    return Ordering(expr_, direction_ == ASCENDING ? DESCENDING : ASCENDING);
+  }
+
   google_firestore_v1_Value to_proto() const;
 
  private:

Firestore/core/src/api/query_core.cc

@@ -221,8 +221,8 @@ std::unique_ptr<ListenerRegistration> Query::AddSnapshotListener(
       firestore_->client()->user_executor(), std::move(view_listener));
 
   std::shared_ptr<QueryListener> query_listener =
-      firestore_->client()->ListenToQuery(this->query(), options,
-                                          async_listener);
+      firestore_->client()->ListenToQuery(core::QueryOrPipeline(this->query()),
+                                          options, async_listener);
 
   return absl::make_unique<QueryListenerRegistration>(
       firestore_->client(), std::move(async_listener),

Firestore/core/src/api/query_snapshot.cc

@@ -110,7 +110,8 @@ void QuerySnapshot::ForEachChange(
     // Special case the first snapshot because index calculation is easy and
     // fast. Also all changes on the first snapshot are adds so there are also
     // no metadata-only changes to filter out.
-    DocumentComparator doc_comparator = snapshot_.query().Comparator();
+    DocumentComparator doc_comparator =
+        snapshot_.query_or_pipeline().Comparator();
     absl::optional<Document> last_document;
     size_t index = 0;
     for (const DocumentViewChange& change : snapshot_.document_changes()) {

Firestore/core/src/api/realtime_pipeline.cc

@@ -19,6 +19,7 @@
 #include <memory>
 #include <utility>
 
+#include "Firestore/core/src/core/pipeline_util.h"
 #include "Firestore/core/src/remote/serializer.h"
 
 namespace firebase {
@@ -27,16 +28,35 @@ namespace api {
 
 RealtimePipeline::RealtimePipeline(
     std::vector<std::shared_ptr<EvaluableStage>> stages,
-    remote::Serializer serializer)
-    : stages_(std::move(stages)), serializer_(serializer) {
+    std::unique_ptr<remote::Serializer> serializer)
+    : stages_(std::move(stages)), serializer_(std::move(serializer)) {
+  this->rewritten_stages_ = core::RewriteStages(this->stages());
+}
+
+RealtimePipeline::RealtimePipeline(const RealtimePipeline& other)
+    : stages_(other.stages_),
+      rewritten_stages_(other.rewritten_stages_),
+      serializer_(std::make_unique<remote::Serializer>(
+          other.serializer_->database_id())) {
+}
+
+RealtimePipeline& RealtimePipeline::operator=(const RealtimePipeline& other) {
+  if (this != &other) {
+    stages_ = other.stages_;
+    rewritten_stages_ = other.rewritten_stages_;
+    serializer_ =
+        std::make_unique<remote::Serializer>(other.serializer_->database_id());
+  }
+  return *this;
 }
 
 RealtimePipeline RealtimePipeline::AddingStage(
     std::shared_ptr<EvaluableStage> stage) {
   auto copy = std::vector<std::shared_ptr<EvaluableStage>>(this->stages_);
   copy.push_back(stage);
 
-  return {copy, serializer_};
+  return {copy,
+          std::make_unique<remote::Serializer>(serializer_->database_id())};
 }
 
 const std::vector<std::shared_ptr<EvaluableStage>>& RealtimePipeline::stages()
@@ -49,13 +69,8 @@ RealtimePipeline::rewritten_stages() const {
   return this->rewritten_stages_;
 }
 
-void RealtimePipeline::SetRewrittentStages(
-    std::vector<std::shared_ptr<EvaluableStage>> stages) {
-  this->rewritten_stages_ = std::move(stages);
-}
-
-EvaluateContext RealtimePipeline::evaluate_context() {
-  return EvaluateContext(&serializer_);
+EvaluateContext RealtimePipeline::evaluate_context() const {
+  return EvaluateContext(serializer_.get());
 }
 
 }  // namespace api

Firestore/core/src/api/realtime_pipeline.h

@@ -18,36 +18,37 @@
 #define FIRESTORE_CORE_SRC_API_REALTIME_PIPELINE_H_
 
 #include <memory>
-#include <utility>
 #include <vector>
 
-#include "Firestore/core/src/api/firestore.h"
-#include "Firestore/core/src/api/pipeline_snapshot.h"
 #include "Firestore/core/src/api/stages.h"
-#include "Firestore/core/src/remote/serializer.h"
 
 namespace firebase {
 namespace firestore {
+namespace remote {
+class Serializer;
+}  // namespace remote
+
 namespace api {
 
 class RealtimePipeline {
  public:
   RealtimePipeline(std::vector<std::shared_ptr<EvaluableStage>> stages,
-                   remote::Serializer serializer);
+                   std::unique_ptr<remote::Serializer> serializer);
+
+  RealtimePipeline(const RealtimePipeline& other);
+  RealtimePipeline& operator=(const RealtimePipeline& other);
 
   RealtimePipeline AddingStage(std::shared_ptr<EvaluableStage> stage);
 
   const std::vector<std::shared_ptr<EvaluableStage>>& stages() const;
   const std::vector<std::shared_ptr<EvaluableStage>>& rewritten_stages() const;
 
-  void SetRewrittentStages(std::vector<std::shared_ptr<EvaluableStage>>);
-
-  EvaluateContext evaluate_context();
+  EvaluateContext evaluate_context() const;
 
  private:
   std::vector<std::shared_ptr<EvaluableStage>> stages_;
   std::vector<std::shared_ptr<EvaluableStage>> rewritten_stages_;
-  remote::Serializer serializer_;
+  std::unique_ptr<remote::Serializer> serializer_;
 };
 
 }  // namespace api