C++ migration: implement WriteStream (#1807)

This is a fairly close port of `FSTWriteStream`. Most of the comments are copied from the Web client.

Author: var-const

Firestore/core/src/firebase/firestore/remote/stream_objc_bridge.h

@@ -46,14 +46,14 @@ namespace bridge {
 bool IsLoggingEnabled();
 
 /**
- * This file contains operations in `WatchStream` that are still delegated to
+ * This file contains operations in remote/ folder that are still delegated to
  * Objective-C: proto parsing and delegates.
  *
  * The principle is that the C++ implementation can only take Objective-C
  * objects as parameters or return them, but never instantiate them or call any
  * methods on them -- if that is necessary, it's delegated to one of the bridge
- * classes. This allows easily identifying which parts of `WatchStream` still
- * rely on not-yet-ported code.
+ * classes. This allows easily identifying which parts of remote/ still rely on
+ * not-yet-ported code.
  */
 
 /**
@@ -68,7 +68,7 @@ class WatchStreamSerializer {
 
   GCFSListenRequest* CreateWatchRequest(FSTQueryData* query) const;
   GCFSListenRequest* CreateUnwatchRequest(model::TargetId target_id) const;
-  grpc::ByteBuffer ToByteBuffer(GCFSListenRequest* request) const;
+  static grpc::ByteBuffer ToByteBuffer(GCFSListenRequest* request);
 
   /**
    * If parsing fails, will return nil and write information on the error to
@@ -89,8 +89,52 @@ class WatchStreamSerializer {
 };
 
 /**
- * A C++ bridge that invokes methods on an `FSTWatchStreamDelegate`.
+ * A C++ bridge to `FSTSerializerBeta` that allows creating
+ * `GCFSWriteRequest`s and parsing `GCFSWriteResponse`s.
  */
+class WriteStreamSerializer {
+ public:
+  explicit WriteStreamSerializer(FSTSerializerBeta* serializer)
+      : serializer_{serializer} {
+  }
+
+  void UpdateLastStreamToken(GCFSWriteResponse* proto);
+  void SetLastStreamToken(NSData* token) {
+    last_stream_token_ = token;
+  }
+  NSData* GetLastStreamToken() const {
+    return last_stream_token_;
+  }
+
+  GCFSWriteRequest* CreateHandshake() const;
+  GCFSWriteRequest* CreateWriteMutationsRequest(
+      NSArray<FSTMutation*>* mutations) const;
+  GCFSWriteRequest* CreateEmptyMutationsList() {
+    return CreateWriteMutationsRequest(@[]);
+  }
+  static grpc::ByteBuffer ToByteBuffer(GCFSWriteRequest* request);
+
+  /**
+   * If parsing fails, will return nil and write information on the error to
+   * `out_status`. Otherwise, returns the parsed proto and sets `out_status` to
+   * ok.
+   */
+  GCFSWriteResponse* ParseResponse(const grpc::ByteBuffer& message,
+                                   util::Status* out_status) const;
+  model::SnapshotVersion ToCommitVersion(GCFSWriteResponse* proto) const;
+  NSArray<FSTMutationResult*>* ToMutationResults(
+      GCFSWriteResponse* proto) const;
+
+  /** Creates a pretty-printed description of the proto for debugging. */
+  static NSString* Describe(GCFSWriteRequest* request);
+  static NSString* Describe(GCFSWriteResponse* request);
+
+ private:
+  FSTSerializerBeta* serializer_;
+  NSData* last_stream_token_;
+};
+
+/** A C++ bridge that invokes methods on an `FSTWatchStreamDelegate`. */
 class WatchStreamDelegate {
  public:
   explicit WatchStreamDelegate(id<FSTWatchStreamDelegate> delegate)
@@ -106,6 +150,23 @@ class WatchStreamDelegate {
   id<FSTWatchStreamDelegate> delegate_;
 };
 
+/** A C++ bridge that invokes methods on an `FSTWriteStreamDelegate`. */
+class WriteStreamDelegate {
+ public:
+  explicit WriteStreamDelegate(id<FSTWriteStreamDelegate> delegate)
+      : delegate_{delegate} {
+  }
+
+  void NotifyDelegateOnOpen();
+  void NotifyDelegateOnHandshakeComplete();
+  void NotifyDelegateOnCommit(const model::SnapshotVersion& commit_version,
+                              NSArray<FSTMutationResult*>* results);
+  void NotifyDelegateOnClose(const util::Status& status);
+
+ private:
+  id<FSTWriteStreamDelegate> delegate_;
+};
+
 }  // namespace bridge
 }  // namespace remote
 }  // namespace firestore

Firestore/core/src/firebase/firestore/remote/stream_objc_bridge.mm

@@ -80,6 +80,7 @@
   return grpc::ByteBuffer{&slice, 1};
 }
 
+// Note: `StatusOr` cannot be used with ARC-managed objects.
 template <typename Proto>
 Proto* ToProto(const grpc::ByteBuffer& message, Status* out_status) {
   NSError* error = nil;
@@ -106,6 +107,8 @@ bool IsLoggingEnabled() {
   return [FIRFirestore isLoggingEnabled];
 }
 
+// WatchStreamSerializer
+
 GCFSListenRequest* WatchStreamSerializer::CreateWatchRequest(
     FSTQueryData* query) const {
   GCFSListenRequest* request = [GCFSListenRequest message];
@@ -124,16 +127,13 @@ bool IsLoggingEnabled() {
 }
 
 grpc::ByteBuffer WatchStreamSerializer::ToByteBuffer(
-    GCFSListenRequest* request) const {
+    GCFSListenRequest* request) {
   return ConvertToByteBuffer([request data]);
 }
 
-NSString* WatchStreamSerializer::Describe(GCFSListenRequest* request) {
-  return [request description];
-}
-
-NSString* WatchStreamSerializer::Describe(GCFSListenResponse* response) {
-  return [response description];
+GCFSListenResponse* WatchStreamSerializer::ParseResponse(
+    const grpc::ByteBuffer& message, Status* out_status) const {
+  return ToProto<GCFSListenResponse>(message, out_status);
 }
 
 FSTWatchChange* WatchStreamSerializer::ToWatchChange(
@@ -146,11 +146,78 @@ bool IsLoggingEnabled() {
   return [serializer_ versionFromListenResponse:proto];
 }
 
-GCFSListenResponse* WatchStreamSerializer::ParseResponse(
+NSString* WatchStreamSerializer::Describe(GCFSListenRequest* request) {
+  return [request description];
+}
+
+NSString* WatchStreamSerializer::Describe(GCFSListenResponse* response) {
+  return [response description];
+}
+
+// WriteStreamSerializer
+
+void WriteStreamSerializer::UpdateLastStreamToken(GCFSWriteResponse* proto) {
+  last_stream_token_ = proto.streamToken;
+}
+
+GCFSWriteRequest* WriteStreamSerializer::CreateHandshake() const {
+  // The initial request cannot contain mutations, but must contain a projectID.
+  GCFSWriteRequest* request = [GCFSWriteRequest message];
+  request.database = [serializer_ encodedDatabaseID];
+  return request;
+}
+
+GCFSWriteRequest* WriteStreamSerializer::CreateWriteMutationsRequest(
+    NSArray<FSTMutation*>* mutations) const {
+  NSMutableArray<GCFSWrite*>* protos =
+      [NSMutableArray arrayWithCapacity:mutations.count];
+  for (FSTMutation* mutation in mutations) {
+    [protos addObject:[serializer_ encodedMutation:mutation]];
+  };
+
+  GCFSWriteRequest* request = [GCFSWriteRequest message];
+  request.writesArray = protos;
+  request.streamToken = last_stream_token_;
+
+  return request;
+}
+
+grpc::ByteBuffer WriteStreamSerializer::ToByteBuffer(
+    GCFSWriteRequest* request) {
+  return ConvertToByteBuffer([request data]);
+}
+
+GCFSWriteResponse* WriteStreamSerializer::ParseResponse(
     const grpc::ByteBuffer& message, Status* out_status) const {
-  return ToProto<GCFSListenResponse>(message, out_status);
+  return ToProto<GCFSWriteResponse>(message, out_status);
+}
+
+model::SnapshotVersion WriteStreamSerializer::ToCommitVersion(
+    GCFSWriteResponse* proto) const {
+  return [serializer_ decodedVersion:proto.commitTime];
+}
+
+NSArray<FSTMutationResult*>* WriteStreamSerializer::ToMutationResults(
+    GCFSWriteResponse* response) const {
+  NSMutableArray<GCFSWriteResult*>* responses = response.writeResultsArray;
+  NSMutableArray<FSTMutationResult*>* results =
+      [NSMutableArray arrayWithCapacity:responses.count];
+  for (GCFSWriteResult* proto in responses) {
+    [results addObject:[serializer_ decodedMutationResult:proto]];
+  };
+  return results;
+}
+
+NSString* WriteStreamSerializer::Describe(GCFSWriteRequest* request) {
+  return [request description];
 }
 
+NSString* WriteStreamSerializer::Describe(GCFSWriteResponse* response) {
+  return [response description];
+}
+
+// WatchStreamDelegate
+
 void WatchStreamDelegate::NotifyDelegateOnOpen() {
   [delegate_ watchStreamDidOpen];
 }
@@ -164,6 +231,27 @@ bool IsLoggingEnabled() {
   [delegate_ watchStreamWasInterruptedWithError:MakeNSError(status)];
 }
 
+// WriteStreamDelegate
+
+void WriteStreamDelegate::NotifyDelegateOnOpen() {
+  [delegate_ writeStreamDidOpen];
+}
+
+void WriteStreamDelegate::NotifyDelegateOnHandshakeComplete() {
+  [delegate_ writeStreamDidCompleteHandshake];
+}
+
+void WriteStreamDelegate::NotifyDelegateOnCommit(
+    const SnapshotVersion& commit_version,
+    NSArray<FSTMutationResult*>* results) {
+  [delegate_ writeStreamDidReceiveResponseWithVersion:commit_version
+                                      mutationResults:results];
+}
+
+void WriteStreamDelegate::NotifyDelegateOnClose(const Status& status) {
+  [delegate_ writeStreamWasInterruptedWithError:MakeNSError(status)];
+}
+
 }  // namespace bridge
 }  // namespace remote
 }  // namespace firestore

Firestore/core/src/firebase/firestore/remote/write_stream.h

@@ -0,0 +1,117 @@
+/*
+ * Copyright 2018 Google
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+#ifndef FIRESTORE_CORE_SRC_FIREBASE_FIRESTORE_REMOTE_WRITE_STREAM_H_
+#define FIRESTORE_CORE_SRC_FIREBASE_FIRESTORE_REMOTE_WRITE_STREAM_H_
+
+#if !defined(__OBJC__)
+#error "This header only supports Objective-C++"
+#endif  // !defined(__OBJC__)
+
+#import <Foundation/Foundation.h>
+#include <memory>
+#include <string>
+
+#include "Firestore/core/src/firebase/firestore/remote/datastore.h"
+#include "Firestore/core/src/firebase/firestore/remote/stream.h"
+#include "Firestore/core/src/firebase/firestore/remote/stream_objc_bridge.h"
+#include "Firestore/core/src/firebase/firestore/util/async_queue.h"
+#include "Firestore/core/src/firebase/firestore/util/status.h"
+#include "absl/strings/string_view.h"
+#include "grpcpp/support/byte_buffer.h"
+
+#import "Firestore/Source/Core/FSTTypes.h"
+#import "Firestore/Source/Model/FSTMutation.h"
+#import "Firestore/Source/Remote/FSTSerializerBeta.h"
+
+namespace firebase {
+namespace firestore {
+namespace remote {
+
+/**
+ * A Stream that implements the Write RPC.
+ *
+ * The Write RPC requires the caller to maintain special stream token
+ * state in-between calls, to help the server understand which responses the
+ * client has processed by the time the next request is made. Every response
+ * will contain a stream token; this value must be passed to the next
+ * request.
+ *
+ * After calling `Start` on this stream, the next request must be a handshake,
+ * containing whatever stream token is on hand. Once a response to this
+ * request is received, all pending mutations may be submitted. When
+ * submitting multiple batches of mutations at the same time, it's
+ * okay to use the same stream token for the calls to `WriteMutations`.
+ */
+class WriteStream : public Stream {
+ public:
+  WriteStream(util::AsyncQueue* async_queue,
+              auth::CredentialsProvider* credentials_provider,
+              FSTSerializerBeta* serializer,
+              Datastore* datastore,
+              id<FSTWriteStreamDelegate> delegate);
+
+  void SetLastStreamToken(NSData* token);
+  /**
+   * The last received stream token from the server, used to acknowledge which
+   * responses the client has processed. Stream tokens are opaque checkpoint
+   * markers whose only real value is their inclusion in the next request.
+   *
+   * `WriteStream` manages propagating this value from responses to the
+   * next request.
+   */
+  NSData* GetLastStreamToken() const;
+
+  /**
+   * Tracks whether or not a handshake has been successfully exchanged and
+   * the stream is ready to accept mutations.
+   */
+  bool handshake_complete() const {
+    return handshake_complete_;
+  }
+
+  /**
+   * Sends an initial stream token to the server, performing the handshake
+   * required to make the StreamingWrite RPC work.
+   */
+  void WriteHandshake();
+
+  /** Sends a group of mutations to the Firestore backend to apply. */
+  void WriteMutations(NSArray<FSTMutation*>* mutations);
+
+ private:
+  std::unique_ptr<GrpcStream> CreateGrpcStream(
+      Datastore* datastore, const absl::string_view token) override;
+  void TearDown(GrpcStream* call) override;
+
+  void NotifyStreamOpen() override;
+  util::Status NotifyStreamResponse(const grpc::ByteBuffer& message) override;
+  void NotifyStreamClose(const util::Status& status) override;
+
+  std::string GetDebugName() const override {
+    return "WriteStream";
+  }
+
+  bridge::WriteStreamSerializer serializer_bridge_;
+  bridge::WriteStreamDelegate delegate_bridge_;
+  bool handshake_complete_ = false;
+};
+
+}  // namespace remote
+}  // namespace firestore
+}  // namespace firebase
+
+#endif  // FIRESTORE_CORE_SRC_FIREBASE_FIRESTORE_REMOTE_WRITE_STREAM_H_