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

Author: wu-hui

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

@@ -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/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/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

@@ -42,10 +42,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_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,17 +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()
@@ -50,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

Firestore/core/src/api/stages.cc

@@ -53,7 +53,6 @@ google_firestore_v1_Pipeline_Stage CollectionSource::to_proto() const {
   result.args = nanopb::MakeArray<google_firestore_v1_Value>(1);
   result.args[0].which_value_type =
       google_firestore_v1_Value_reference_value_tag;
-  // TODO: use EncodeResourceName instead
   result.args[0].reference_value =
       nanopb::MakeBytesArray(this->path_.CanonicalString());
 
@@ -105,10 +104,12 @@ google_firestore_v1_Pipeline_Stage DocumentsSource::to_proto() const {
   result.args_count = documents_.size();
   result.args = nanopb::MakeArray<google_firestore_v1_Value>(result.args_count);
 
-  for (size_t i = 0; i < documents_.size(); ++i) {
+  size_t i = 0;
+  for (const auto& document : documents_) {
     result.args[i].which_value_type =
         google_firestore_v1_Value_string_value_tag;
-    result.args[i].string_value = nanopb::MakeBytesArray(documents_[i]);
+    result.args[i].string_value = nanopb::MakeBytesArray(document);
+    ++i;
   }
 
   result.options_count = 0;
@@ -359,6 +360,19 @@ model::PipelineInputOutputVector DatabaseSource::Evaluate(
   return results;
 }
 
+model::PipelineInputOutputVector DocumentsSource::Evaluate(
+    const EvaluateContext& /*context*/,
+    const model::PipelineInputOutputVector& inputs) const {
+  model::PipelineInputOutputVector results;
+  for (const model::PipelineInputOutput& input : inputs) {
+    if (input.is_found_document() &&
+        documents_.count(input.key().path().CanonicalString()) > 0) {
+      results.push_back(input);
+    }
+  }
+  return results;
+}
+
 model::PipelineInputOutputVector Where::Evaluate(
     const EvaluateContext& context,
     const model::PipelineInputOutputVector& inputs) const {
@@ -380,16 +394,29 @@ model::PipelineInputOutputVector Where::Evaluate(
 model::PipelineInputOutputVector LimitStage::Evaluate(
     const EvaluateContext& /*context*/,
     const model::PipelineInputOutputVector& inputs) const {
+  model::PipelineInputOutputVector::const_iterator begin;
+  model::PipelineInputOutputVector::const_iterator end;
+  size_t count;
+
   if (limit_ < 0) {
-    // Or handle as error? Assuming non-negative limit.
-    return {};
-  }
-  size_t count = static_cast<size_t>(limit_);
-  if (count > inputs.size()) {
-    count = inputs.size();
+    // if limit_ is negative, we treat it as limit to last, returns the last
+    // limit_ documents.
+    count = static_cast<size_t>(-limit_);
+    if (count > inputs.size()) {
+      count = inputs.size();
+    }
+    begin = inputs.end() - count;
+    end = inputs.end();
+  } else {
+    count = static_cast<size_t>(limit_);
+    if (count > inputs.size()) {
+      count = inputs.size();
+    }
+    begin = inputs.begin();
+    end = inputs.begin() + count;
   }
-  return model::PipelineInputOutputVector(inputs.begin(),
-                                          inputs.begin() + count);
+
+  return model::PipelineInputOutputVector(begin, end);
 }
 
 model::PipelineInputOutputVector SortStage::Evaluate(