feat(storage): Add bidiread and bidi appendable write to the async client.

ci/cloudbuild/builds/check-api.sh

@@ -124,7 +124,7 @@ function check_abi() {
       # characters in the string "internal", and it should again be followed
       # by some other number indicating the length of the symbol within the
       # "internal" namespace. See: https://en.wikipedia.org/wiki/Name_mangling
-      -skip-internal-symbols "(8internal|_internal|4absl|4grpc|6google8protobuf|6google3rpc)\d"
+      -skip-internal-symbols "(8internal|_internal|4absl|4grpc|6google8protobuf|6google3rpc|20storage_experimental)\d"
       # We ignore the raw gRPC Stub class. The generated gRPC headers that
       # contain these classes are installed alongside our headers. When a new
       # RPC is added to a service, these classes gain a pure virtual method. Our

google/cloud/storage/async/client.cc

@@ -49,6 +49,26 @@ future<StatusOr<google::storage::v2::Object>> AsyncClient::InsertObject(
        internal::MergeOptions(std::move(opts), connection_->options())});
 }
 
+future<StatusOr<ObjectDescriptor>> AsyncClient::Open(
+    BucketName const& bucket_name, std::string object_name, Options opts) {
+  auto spec = google::storage::v2::BidiReadObjectSpec{};
+  spec.set_bucket(bucket_name.FullName());
+  spec.set_object(std::move(object_name));
+  return Open(std::move(spec), std::move(opts));
+}
+
+future<StatusOr<ObjectDescriptor>> AsyncClient::Open(
+    google::storage::v2::BidiReadObjectSpec spec, Options opts) {
+  return connection_
+      ->Open({std::move(spec),
+              internal::MergeOptions(std::move(opts), connection_->options())})
+      .then([](auto f) -> StatusOr<ObjectDescriptor> {
+        auto connection = f.get();
+        if (!connection) return std::move(connection).status();
+        return ObjectDescriptor(*std::move(connection));
+      });
+}
+
 future<StatusOr<std::pair<AsyncReader, AsyncToken>>> AsyncClient::ReadObject(
     BucketName const& bucket_name, std::string object_name, Options opts) {
   auto request = google::storage::v2::ReadObjectRequest{};
@@ -92,6 +112,65 @@ future<StatusOr<ReadPayload>> AsyncClient::ReadObjectRange(
        internal::MergeOptions(std::move(opts), connection_->options())});
 }
 
+future<StatusOr<std::pair<AsyncWriter, AsyncToken>>>
+AsyncClient::StartAppendableObjectUpload(BucketName const& bucket_name,
+                                         std::string object_name,
+                                         Options opts) {
+  auto request = google::storage::v2::BidiWriteObjectRequest{};
+  auto& resource = *request.mutable_write_object_spec()->mutable_resource();
+
+  resource.set_bucket(BucketName(bucket_name).FullName());
+  resource.set_name(std::move(object_name));
+  request.mutable_write_object_spec()->set_appendable(true);
+
+  return StartAppendableObjectUpload(std::move(request), std::move(opts));
+}
+
+future<StatusOr<std::pair<AsyncWriter, AsyncToken>>>
+AsyncClient::StartAppendableObjectUpload(
+    google::storage::v2::BidiWriteObjectRequest request, Options opts) {
+  return connection_
+      ->StartAppendableObjectUpload(
+          {std::move(request),
+           internal::MergeOptions(std::move(opts), connection_->options())})
+      .then([](auto f) -> StatusOr<std::pair<AsyncWriter, AsyncToken>> {
+        auto w = f.get();
+        if (!w) return std::move(w).status();
+        auto t = absl::holds_alternative<google::storage::v2::Object>(
+                     (*w)->PersistedState())
+                     ? AsyncToken()
+                     : storage_internal::MakeAsyncToken(w->get());
+        return std::make_pair(AsyncWriter(*std::move(w)), std::move(t));
+      });
+}
+
+future<StatusOr<std::pair<AsyncWriter, AsyncToken>>>
+AsyncClient::ResumeAppendableObjectUpload(BucketName const& bucket_name,
+                                          std::string object_name,
+                                          std::int64_t generation,
+                                          Options opts) {
+  auto request = google::storage::v2::BidiWriteObjectRequest{};
+  auto& append_object_spec = *request.mutable_append_object_spec();
+
+  append_object_spec.set_bucket(BucketName(bucket_name).FullName());
+  append_object_spec.set_object(std::move(object_name));
+  append_object_spec.set_generation(generation);
+
+  return connection_
+      ->ResumeAppendableObjectUpload(
+          {std::move(request),
+           internal::MergeOptions(std::move(opts), connection_->options())})
+      .then([](auto f) -> StatusOr<std::pair<AsyncWriter, AsyncToken>> {
+        auto w = f.get();
+        if (!w) return std::move(w).status();
+        auto t = absl::holds_alternative<google::storage::v2::Object>(
+                     (*w)->PersistedState())
+                     ? AsyncToken()
+                     : storage_internal::MakeAsyncToken(w->get());
+        return std::make_pair(AsyncWriter(*std::move(w)), std::move(t));
+      });
+}
+
 future<StatusOr<std::pair<AsyncWriter, AsyncToken>>>
 AsyncClient::StartBufferedUpload(BucketName const& bucket_name,
                                  std::string object_name, Options opts) {

google/cloud/storage/async/client.h

@@ -17,6 +17,7 @@
 
 #include "google/cloud/storage/async/bucket_name.h"
 #include "google/cloud/storage/async/connection.h"
+#include "google/cloud/storage/async/object_descriptor.h"
 #include "google/cloud/storage/async/reader.h"
 #include "google/cloud/storage/async/rewriter.h"
 #include "google/cloud/storage/async/token.h"
@@ -231,6 +232,41 @@ class AsyncClient {
       google::storage::v2::WriteObjectRequest request, WritePayload contents,
       Options opts = {});
 
+  /**
+   * Open an object descriptor to perform one or more ranged reads.
+   *
+   * @par Idempotency
+   * This is a read-only operation and is always idempotent. The operation will
+   * retry until the descriptor is successfully created. The descriptor itself
+   * will resume any incomplete ranged reads if the connection(s) are
+   * interrupted. Use `ResumePolicyOption` and `ResumePolicy` to control this.
+   *
+   * @param bucket_name the name of the bucket that contains the object.
+   * @param object_name the name of the object to be read.
+   * @param opts options controlling the behavior of this RPC, for example
+   *     the application may change the retry policy.
+   */
+  future<StatusOr<ObjectDescriptor>> Open(BucketName const& bucket_name,
+                                          std::string object_name,
+                                          Options opts = {});
+
+  /**
+   * Open an object descriptor to perform one or more ranged reads.
+   *
+   * @par Idempotency
+   * This is a read-only operation and is always idempotent. The operation will
+   * retry until the descriptor is successfully created. The descriptor itself
+   * will resume any incomplete ranged reads if the connection(s) are
+   * interrupted. Use `ResumePolicyOption` and `ResumePolicy` to control this.
+   *
+   * @param spec the BidiReadObjectSpec to use when retrieving the
+   * ObjectDescriptor.
+   * @param opts options controlling the behavior of this RPC, for example
+   *     the application may change the retry policy.
+   */
+  future<StatusOr<ObjectDescriptor>> Open(
+      google::storage::v2::BidiReadObjectSpec spec, Options opts = {});
+
   /**
    * A streaming download for the contents of an object.
    *
@@ -243,7 +279,7 @@ class AsyncClient {
    * @par Idempotency
    * This is a read-only operation and is always idempotent. Once the download
    * starts, this operation will automatically resume the download if is
-   * interrupted. Use `ResumePolicyOption` and `ResumePolicy` to control this
+   * interrupted. Use `ResumePolicyOption` and `ResumePolicy` to control this.
    *
    * @param bucket_name the name of the bucket that contains the object.
    * @param object_name the name of the object to be read.
@@ -334,10 +370,82 @@ class AsyncClient {
       google::storage::v2::ReadObjectRequest request, std::int64_t offset,
       std::int64_t limit, Options opts = {});
 
+  /*
+  [start-appendable-object-upload]
+  Initiates a [resumable upload][resumable-link] for an appendable object.
+
+  Appendable objects allow you to create an object and upload data to it
+  incrementally until it is finalized. This means you can start an upload
+  and append data to the object later.
+
+  You can finalize an appendable object in the first call itself by providing
+  all the data in the initial upload. You can also explicitly Flush to ensure
+  the data is persisted.
+
+  The recovery can be done from most transient errors, including an unexpected
+  closure of the streaming RPC used for the upload.
+
+  @par Example
+  @snippet storage_async_samples.cc start-appendable-object-upload
+
+  @par Idempotency
+  This function is always treated as idempotent, and the library will
+  automatically retry the function on transient errors.
+
+  [resumable-link]: https://cloud.google.com/storage/docs/resumable-uploads
+  [start-appendable-object-upload]
+  */
+
+  /**
+   * Starts a new resumable upload session for appendable objects and
+   * automatic recovery from transient failures.
+   *
+   * @snippet{doc} async/client.h start-appendable-object-upload
+   *
+   * @param bucket_name the name of the bucket that contains the object.
+   * @param object_name the name of the object to be read.
+   * @param opts options controlling the behavior of this RPC, for example
+   *     the application may change the retry policy.
+   */
+  future<StatusOr<std::pair<AsyncWriter, AsyncToken>>>
+  StartAppendableObjectUpload(BucketName const& bucket_name,
+                              std::string object_name, Options opts = {});
+
+  /**
+   * Starts a new resumable upload session for appendable objects and
+   * automatic recovery from transient failures.
+   *
+   * @snippet{doc} async/client.h start-appendable-object-upload
+   *
+   * @param request the request contents, it must include the bucket name and
+   *     object names. Many other fields are optional.
+   * @param opts options controlling the behavior of this RPC, for example
+   *     the application may change the retry policy.
+   */
+  future<StatusOr<std::pair<AsyncWriter, AsyncToken>>>
+  StartAppendableObjectUpload(
+      google::storage::v2::BidiWriteObjectRequest request, Options opts = {});
+
+  /**
+   * Resume a resumable upload session for appendable objects and automatic
+   * recovery from transient failures.
+   *
+   * @param bucket_name the name of the bucket that contains the object.
+   * @param object_name the name of the object to be uploaded.
+   * @param generation the object generation to be uploaded.
+   * @param opts options controlling the behaviour of this RPC, for example the
+   * application may change the retry policy.
+   */
+  future<StatusOr<std::pair<AsyncWriter, AsyncToken>>>
+  ResumeAppendableObjectUpload(BucketName const& bucket_name,
+                               std::string object_name, std::int64_t generation,
+                               Options opts = {});
+
   /*
   [start-buffered-upload-common]
   This function always uses [resumable uploads][resumable-link]. The objects
   returned by this function buffer data until it is persisted on the service.
+
   If the buffer becomes full, they stop accepting new data until the service
   has persisted enough data.