Implement Client::ReadObject() using streaming reads (#938)

Use the streaming read classes to read object media. I removed
some tests because they would have become tests for "does iostream
works". This fixes #819.

* Fix tests in CentOS.

The old curl library in CentOS seems to be less forgiving and
caught a bug we had.  Fixed.

Author: coryan

google/cloud/storage/CMakeLists.txt

@@ -258,7 +258,6 @@ set(storage_client_unit_tests
     list_objects_reader_test.cc
     object_access_control_test.cc
     object_metadata_test.cc
-    object_stream_test.cc
     object_test.cc
     retry_policy_test.cc
     storage_client_options_test.cc

google/cloud/storage/client.h

@@ -81,7 +81,7 @@ class Client {
   }
 
   /**
-   * Fetch the bucket metadata and return it.
+   * Fetch the bucket metadata and return it.c
    *
    * @param bucket_name query metadata information about this bucket.
    * @param options a list of optional query parameters and/or request headers.
@@ -190,11 +190,12 @@ class Client {
    * @snippet storage_object_samples.cc read object
    */
   template <typename... Options>
-  ObjectReadStream Read(std::string const& bucket_name,
-                        std::string const& object_name, Options&&... options) {
+  ObjectReadStream ReadObject(std::string const& bucket_name,
+                              std::string const& object_name,
+                              Options&&... options) {
     internal::ReadObjectRangeRequest request(bucket_name, object_name);
     request.set_multiple_options(std::forward<Options>(options)...);
-    return ObjectReadStream(raw_client_, request);
+    return ObjectReadStream(raw_client_->ReadObject(request).second);
   }
 
   /**

google/cloud/storage/examples/storage_object_samples.cc

@@ -80,24 +80,27 @@ void GetObjectMetadata(gcs::Client client, int& argc, char* argv[]) {
   (std::move(client), bucket_name, object_name);
 }
 
-//! [read object]
 void ReadObject(gcs::Client client, int& argc, char* argv[]) {
   if (argc < 2) {
-    throw Usage{
-        "insert-object <bucket-name> <object-name> <object-contents (string)>"};
+    throw Usage{"read-object <bucket-name> <object-name>"};
   }
   auto bucket_name = ConsumeArg(argc, argv);
   auto object_name = ConsumeArg(argc, argv);
-  auto stream = client.Read(bucket_name, object_name);
-  int count = 0;
-  while (not stream.eof()) {
-    std::string line;
-    std::getline(stream, line, '\n');
-    ++count;
+  //! [read object]
+  [](google::cloud::storage::Client client, std::string bucket_name,
+     std::string object_name) {
+    auto stream = client.ReadObject(bucket_name, object_name);
+    int count = 0;
+    while (not stream.eof()) {
+      std::string line;
+      std::getline(stream, line, '\n');
+      ++count;
+    }
+    std::cout << "The object has " << count << " lines" << std::endl;
   }
-  std::cout << "The object has " << count << " lines" << std::endl;
+  //! [read object]
+  (std::move(client), bucket_name, object_name);
 }
-//! [read object]
 
 //! [delete object]
 void DeleteObject(gcs::Client client, int& argc, char* argv[]) {

google/cloud/storage/internal/curl_client.cc

@@ -95,31 +95,19 @@ std::pair<Status, ObjectMetadata> CurlClient::GetObjectMetadata(
                         ObjectMetadata::ParseFromString(payload.payload));
 }
 
-std::pair<Status, ReadObjectRangeResponse> CurlClient::ReadObjectRangeMedia(
+std::pair<Status, std::unique_ptr<ObjectReadStreambuf>> CurlClient::ReadObject(
     ReadObjectRangeRequest const& request) {
   // Assume the bucket name is validated by the caller.
   CurlRequestBuilder builder(storage_endpoint_ + "/b/" + request.bucket_name() +
                              "/o/" + request.object_name());
   builder.SetDebugLogging(options_.enable_http_tracing());
   builder.AddHeader(options_.credentials()->AuthorizationHeader());
   builder.AddQueryParameter("alt", "media");
-  // For the moment, we are using range reads to read the objects (see #727)
-  // disable decompression because range reads do not work in that case:
-  //   https://cloud.google.com/storage/docs/transcoding#range
-  // and
-  //   https://cloud.google.com/storage/docs/transcoding#decompressive_transcoding
-  builder.AddHeader("Cache-Control: no-transform");
-  builder.AddHeader("Range: bytes=" + std::to_string(request.begin()) + '-' +
-                    std::to_string(request.end()));
-  auto payload = builder.BuildRequest(std::string{}).MakeRequest();
-  if (payload.status_code >= 300) {
-    return std::make_pair(
-        Status{payload.status_code, std::move(payload.payload)},
-        internal::ReadObjectRangeResponse{});
-  }
-  return std::make_pair(
-      Status(),
-      internal::ReadObjectRangeResponse::FromHttpResponse(std::move(payload)));
+  // TODO(#937) - use client options to configure buffer size.
+  std::unique_ptr<CurlReadStreambuf> buf(new CurlReadStreambuf(
+      builder.BuildDownloadRequest(std::string{}), 128 * 1024));
+  return std::make_pair(Status(),
+                        std::unique_ptr<ObjectReadStreambuf>(std::move(buf)));
 }
 
 std::pair<Status, std::unique_ptr<ObjectWriteStreambuf>>
@@ -132,6 +120,7 @@ CurlClient::WriteObject(InsertObjectStreamingRequest const& request) {
   builder.AddQueryParameter("uploadType", "media");
   builder.AddQueryParameter("name", request.object_name());
   builder.AddHeader("Content-Type: application/octet-stream");
+  // TODO(#937) - use client options to configure buffer size.
   std::unique_ptr<internal::CurlStreambuf> buf(
       new internal::CurlStreambuf(builder.BuildUpload(), 128 * 1024));
   return std::make_pair(

google/cloud/storage/internal/curl_client.h

@@ -48,8 +48,8 @@ class CurlClient : public RawClient {
       InsertObjectMediaRequest const& request) override;
   std::pair<Status, ObjectMetadata> GetObjectMetadata(
       GetObjectMetadataRequest const& request) override;
-  std::pair<Status, ReadObjectRangeResponse> ReadObjectRangeMedia(
-      ReadObjectRangeRequest const& request) override;
+  std::pair<Status, std::unique_ptr<ObjectReadStreambuf>> ReadObject(
+      ReadObjectRangeRequest const&) override;
   std::pair<Status, std::unique_ptr<ObjectWriteStreambuf>> WriteObject(
       InsertObjectStreamingRequest const&) override;
   std::pair<Status, ListObjectsResponse> ListObjects(

google/cloud/storage/internal/curl_download_request.cc

@@ -30,29 +30,54 @@ namespace internal {
 CurlDownloadRequest::CurlDownloadRequest(std::size_t initial_buffer_size)
     : headers_(nullptr, &curl_slist_free_all),
       multi_(nullptr, &curl_multi_cleanup),
+      closing_(false),
       curl_closed_(false),
       initial_buffer_size_(initial_buffer_size) {
   buffer_.reserve(initial_buffer_size);
 }
 
+HttpResponse CurlDownloadRequest::Close() {
+  // Set the the closing_ flag to trigger a return 0 from the next read
+  // callback, see the comments in the header file for more details.
+  closing_ = true;
+  // Block until that callback is made.
+  Wait([this] { return curl_closed_; });
+
+  // Now remove the handle from the CURLM* interface and wait for the response.
+  auto error = curl_multi_remove_handle(multi_.get(), handle_.handle_.get());
+  RaiseOnError(__func__, error);
+
+  long http_code = handle_.GetResponseCode();
+  return HttpResponse{http_code, std::string{}, std::move(received_headers_)};
+}
+
 HttpResponse CurlDownloadRequest::GetMore(std::string& buffer) {
   handle_.FlushDebug(__func__);
-  GCP_LOG(DEBUG) << __func__ << "(), curl.size=" << buffer_.size();
   Wait([this] {
     return curl_closed_ or buffer_.size() >= initial_buffer_size_;
   });
+  GCP_LOG(DEBUG) << __func__ << "(), curl.size=" << buffer_.size()
+                 << ", closing=" << closing_ << ", closed=" << curl_closed_;
   if (curl_closed_) {
     // Remove the handle from the CURLM* interface and wait for the response.
     auto error = curl_multi_remove_handle(multi_.get(), handle_.handle_.get());
     RaiseOnError(__func__, error);
 
-    buffer.swap(buffer_);
+    buffer_.swap(buffer);
+    buffer_.clear();
     long http_code = handle_.GetResponseCode();
+    GCP_LOG(DEBUG) << __func__ << "(), size=" << buffer.size()
+                   << ", closing=" << closing_ << ", closed=" << curl_closed_
+                   << ", code=" << http_code;
     return HttpResponse{http_code, std::string{}, std::move(received_headers_)};
   }
   buffer_.swap(buffer);
+  buffer_.clear();
   buffer_.reserve(initial_buffer_size_);
   handle_.EasyPause(CURLPAUSE_RECV_CONT);
+  GCP_LOG(DEBUG) << __func__ << "(), size=" << buffer.size()
+                 << ", closing=" << closing_ << ", closed=" << curl_closed_
+                 << ", code=100";
   return HttpResponse{100, {}, {}};
 }
 
@@ -91,11 +116,18 @@ std::size_t CurlDownloadRequest::WriteCallback(void* ptr, std::size_t size,
   handle_.FlushDebug(__func__);
   GCP_LOG(DEBUG) << __func__ << "() size=" << size << ", nmemb=" << nmemb
                  << ", buffer.size=" << buffer_.size();
-
-  buffer_.append(static_cast<char const*>(ptr), size * nmemb);
-  if (buffer_.size() > 128 * 1024) {
+  // This transfer is closing, just return zero, that will make libcurl finish
+  // any pending work, and will return the handle_ pointer from
+  // curl_multi_info_read() in PerformWork(). That is the point where
+  // `curl_closed_` is set.
+  if (closing_) {
+    return 0;
+  }
+  if (buffer_.size() >= initial_buffer_size_) {
     return CURL_READFUNC_PAUSE;
   }
+
+  buffer_.append(static_cast<char const*>(ptr), size * nmemb);
   return size * nmemb;
 }
 

google/cloud/storage/internal/curl_download_request.h

@@ -46,6 +46,7 @@ class CurlDownloadRequest {
         logging_enabled_(rhs.logging_enabled_),
         handle_(std::move(rhs.handle_)),
         multi_(std::move(rhs.multi_)),
+        closing_(rhs.closing_),
         curl_closed_(rhs.curl_closed_),
         initial_buffer_size_(rhs.initial_buffer_size_) {
     ResetOptions();
@@ -59,12 +60,16 @@ class CurlDownloadRequest {
     logging_enabled_ = rhs.logging_enabled_;
     handle_ = std::move(rhs.handle_);
     multi_ = std::move(rhs.multi_);
+    closing_ = rhs.closing_;
     curl_closed_ = rhs.curl_closed_;
     initial_buffer_size_ = rhs.initial_buffer_size_;
     ResetOptions();
     return *this;
   }
 
+  bool IsOpen() const { return not curl_closed_; }
+  HttpResponse Close();
+
   /**
    * Wait for additional data or the end of the transfer.
    *
@@ -130,7 +135,16 @@ class CurlDownloadRequest {
   CurlMulti multi_;
 
   std::string buffer_;
-  // The curl_closed_ flag is set if the transfer is done.
+  // Closing the handle happens in two steps.
+  // 1. First the application (or higher-level class), calls Close(). This class
+  //    needs to notify libcurl that the transfer is terminated by returning 0
+  //    from the callback.
+  // 2. Once that callback returns 0, this class needs to know
+  //
+  // The closing_ flag is set when we enter step 1.
+  bool closing_;
+  // The curl_closed_ flag is set when we enter step 2, or when the transfer
+  // completes.
   bool curl_closed_;
 
   std::size_t initial_buffer_size_;

google/cloud/storage/internal/curl_streambuf.cc

@@ -20,6 +20,49 @@ namespace storage {
 inline namespace STORAGE_CLIENT_NS {
 namespace internal {
 
+CurlReadStreambuf::CurlReadStreambuf(CurlDownloadRequest&& download,
+                                     std::size_t target_buffer_size)
+    : download_(std::move(download)), target_buffer_size_(target_buffer_size) {
+  // Start with an empty read area, to force an underflow() on the first
+  // extraction.
+  current_ios_buffer_.push_back('\0');
+  char* data = &current_ios_buffer_[0];
+  setg(data, data + 1, data + 1);
+}
+
+bool CurlReadStreambuf::IsOpen() const { return download_.IsOpen(); }
+
+HttpResponse CurlReadStreambuf::Close() { return download_.Close(); }
+
+CurlReadStreambuf::int_type CurlReadStreambuf::underflow() {
+  if (not IsOpen()) {
+    current_ios_buffer_.clear();
+    current_ios_buffer_.push_back('\0');
+    char* data = &current_ios_buffer_[0];
+    setg(data, data + 1, data + 1);
+    return traits_type::eof();
+  }
+
+  current_ios_buffer_.reserve(target_buffer_size_);
+  auto response = download_.GetMore(current_ios_buffer_);
+  if (response.status_code >= 300) {
+    std::ostringstream os;
+    os << "CurlDownloadRequest reports error: " << response.status_code
+       << ", payload=" << response.payload;
+    google::cloud::internal::RaiseRuntimeError(os.str());
+  }
+
+  if (not current_ios_buffer_.empty()) {
+    char* data = &current_ios_buffer_[0];
+    setg(data, data, data + current_ios_buffer_.size());
+    return traits_type::to_int_type(*data);
+  }
+  current_ios_buffer_.push_back('\0');
+  char* data = &current_ios_buffer_[0];
+  setg(data, data + 1, data + 1);
+  return traits_type::eof();
+}
+
 CurlStreambuf::CurlStreambuf(CurlUploadRequest&& upload,
                              std::size_t max_buffer_size)
     : upload_(std::move(upload)), max_buffer_size_(max_buffer_size) {

google/cloud/storage/internal/curl_streambuf.h

@@ -15,6 +15,7 @@
 #ifndef GOOGLE_CLOUD_CPP_GOOGLE_CLOUD_STORAGE_INTERNAL_CURL_STREAMBUF_H_
 #define GOOGLE_CLOUD_CPP_GOOGLE_CLOUD_STORAGE_INTERNAL_CURL_STREAMBUF_H_
 
+#include "google/cloud/storage/internal/curl_download_request.h"
 #include "google/cloud/storage/internal/curl_upload_request.h"
 #include "google/cloud/storage/internal/object_streambuf.h"
 
@@ -23,6 +24,28 @@ namespace cloud {
 namespace storage {
 inline namespace STORAGE_CLIENT_NS {
 namespace internal {
+/**
+ * Implement a wrapper for libcurl-based streaming downloads.
+ */
+class CurlReadStreambuf : public ObjectReadStreambuf {
+ public:
+  explicit CurlReadStreambuf(CurlDownloadRequest&& download,
+                             std::size_t target_buffer_size);
+
+  ~CurlReadStreambuf() override = default;
+
+  HttpResponse Close() override;
+  bool IsOpen() const override;
+
+ protected:
+  int_type underflow() override;
+
+ private:
+  CurlDownloadRequest download_;
+  std::string current_ios_buffer_;
+  std::size_t target_buffer_size_;
+};
+
 /**
  * Implement a wrapper for libcurl-based streaming uploads.
  */

google/cloud/storage/internal/logging_client.cc

@@ -103,10 +103,10 @@ std::pair<Status, ObjectMetadata> LoggingClient::GetObjectMetadata(
   return MakeCall(*client_, &RawClient::GetObjectMetadata, request, __func__);
 }
 
-std::pair<Status, ReadObjectRangeResponse> LoggingClient::ReadObjectRangeMedia(
-    ReadObjectRangeRequest const& request) {
-  return MakeCall(*client_, &RawClient::ReadObjectRangeMedia, request,
-                  __func__);
+std::pair<Status, std::unique_ptr<ObjectReadStreambuf>>
+LoggingClient::ReadObject(ReadObjectRangeRequest const& request) {
+  return MakeCallNoResponseLogging(*client_, &RawClient::ReadObject, request,
+                                   __func__);
 }
 
 std::pair<Status, std::unique_ptr<ObjectWriteStreambuf>>