<commit_before>// Copyright 2022 Google LLC
//
// 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
//
//      https://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.

#include "google/cloud/storage/internal/async_accumulate_read_object.h"

namespace google {
namespace cloud {
namespace storage_internal {
GOOGLE_CLOUD_CPP_INLINE_NAMESPACE_BEGIN
namespace {

/**
 * Keeps the state for `AsyncAccumulateReadObjectPartial()`.
 *
 * This is analogous to a C++20 coroutine handle. It keeps the state for
 * `AsyncAccumulateReadObjectPartial()`, that is, all the function call
 * arguments, as well as any local variables.
 *
 * Whereas in C++20 coroutines we would use `co_await` to suspend execution,
 * here we need to use a callback, so code like:
 *
 * @code
 *   auto x = co_await Foo();
 * @endcode
 *
 * becomes
 *
 * @code
 *   Foo().then([self = shared_from_this()](auto f) { self->OnFoo(f.get()); });
 * @endcode
 *
 * An extra complication is that we use timeouts, so while a naive
 * implementation of this code would say:
 *
 * @code
 *   auto read = co_await stream->Read();
 * @endcode
 *
 * here we launch two coroutines, one to run the timer, and one to actually
 * make the call subject to a timeout. The trick is to set the timeout to cancel
 * the full streaming RPC if it expires successfully, and to cancel the timer
 * if the operation completes:
 *
 * @code
 *   auto tm = cq.MakeRelativeTimer(timeout).then([&stream](auto f) {
 *     if (!f.get().ok()) return false; // timer canceled, nothing to do
 *     stream->Cancel();
 *     return true;
 *   });
 *   auto read = co_await stream->Read();
 *   tm.cancel();
 *   if (co_await tm) { // handle the timeout
 *   // success!
 * @endcode
 *
 * Without coroutines this requires passing the timer future to the `OnRead()`
 * callback. See below for the details.
 */
class AsyncAccumulateReadObjectPartialHandle
    : public std::enable_shared_from_this<
          AsyncAccumulateReadObjectPartialHandle> {
 public:
  using Response = ::google::storage::v2::ReadObjectResponse;
  using Stream = ::google::cloud::internal::AsyncStreamingReadRpc<Response>;
  using StreamingRpcMetadata = ::google::cloud::internal::StreamingRpcMetadata;
  using Result = AsyncAccumulateReadObjectResult;

  AsyncAccumulateReadObjectPartialHandle(CompletionQueue cq,
                                         std::unique_ptr<Stream> stream,
                                         std::chrono::milliseconds timeout)
      : cq_(std::move(cq)), stream_(std::move(stream)), timeout_(timeout) {}

  future<Result> Invoke() {
    struct ByMove {
      std::shared_ptr<AsyncAccumulateReadObjectPartialHandle> self;
      future<bool> tm;
      void operator()(future<bool> f) { self->OnStart(std::move(tm), f.get()); }
    };
    auto tm = MakeTimeout();
    stream_->Start().then(ByMove{shared_from_this(), std::move(tm)});
    return promise_.get_future();
  }

 private:
  void OnStart(future<bool> tm, bool ok) {
    tm.cancel();
    if (tm.get()) return OnTimeout("Start()");
    if (!ok) return Finish();
    Read();
  }

  void Read() {
    struct ByMove {
      std::shared_ptr<AsyncAccumulateReadObjectPartialHandle> self;
      future<bool> tm;
      void operator()(future<absl::optional<Response>> f) {
        self->OnRead(std::move(tm), f.get());
      }
    };
    auto tm = MakeTimeout();
    stream_->Read().then(ByMove{shared_from_this(), std::move(tm)});
  }

  void OnRead(future<bool> tm, absl::optional<Response> response) {
    tm.cancel();
    if (tm.get()) return OnTimeout("Read()");
    if (!response.has_value()) return Finish();
    accumulator_.push_back(*std::move(response));
    Read();
  }

  void Finish() {
    struct ByMove {
      std::shared_ptr<AsyncAccumulateReadObjectPartialHandle> self;
      future<bool> tm;
      void operator()(future<Status> f) {
        self->OnFinish(std::move(tm), f.get());
      }
    };
    auto tm = MakeTimeout();
    stream_->Finish().then(ByMove{shared_from_this(), std::move(tm)});
  }

  void OnFinish(future<bool> tm, Status status) {
    tm.cancel();
    promise_.set_value(Result{std::move(accumulator_),
                              stream_->GetRequestMetadata(),
                              std::move(status)});
  }

  future<bool> MakeTimeout() {
    auto self = shared_from_this();
    return cq_.MakeRelativeTimer(timeout_).then(
        [self](future<StatusOr<std::chrono::system_clock::time_point>> f) {
          if (!f.get().ok()) return false;
          self->stream_->Cancel();
          return true;
        });
  }
  void OnTimeout(char const* where) {
    auto finish = stream_->Finish();
    finish.then(WaitForFinish{std::move(stream_)});
    promise_.set_value(<commit_msg>fix(GCS+gRPC): all `co_await` should be callbacks (#9291)<commit_after>// Copyright 2022 Google LLC
//
// 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
//
//      https://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.

#include "google/cloud/storage/internal/async_accumulate_read_object.h"

namespace google {
namespace cloud {
namespace storage_internal {
GOOGLE_CLOUD_CPP_INLINE_NAMESPACE_BEGIN
namespace {

/**
 * Keeps the state for `AsyncAccumulateReadObjectPartial()`.
 *
 * This is analogous to a C++20 coroutine handle. It keeps the state for
 * `AsyncAccumulateReadObjectPartial()`, that is, all the function call
 * arguments, as well as any local variables.
 *
 * Whereas in C++20 coroutines we would use `co_await` to suspend execution,
 * here we need to use a callback, so code like:
 *
 * @code
 *   auto x = co_await Foo();
 * @endcode
 *
 * becomes
 *
 * @code
 *   Foo().then([self = shared_from_this()](auto f) { self->OnFoo(f.get()); });
 * @endcode
 *
 * An extra complication is that we use timeouts, so while a naive
 * implementation of this code would say:
 *
 * @code
 *   auto read = co_await stream->Read();
 * @endcode
 *
 * here we launch two coroutines, one to run the timer, and one to actually
 * make the call subject to a timeout. The trick is to set the timeout to cancel
 * the full streaming RPC if it expires successfully, and to cancel the timer
 * if the operation completes:
 *
 * @code
 *   auto tm = cq.MakeRelativeTimer(timeout).then([&stream](auto f) {
 *     if (!f.get().ok()) return false; // timer canceled, nothing to do
 *     stream->Cancel();
 *     return true;
 *   });
 *   auto read = co_await stream->Read();
 *   tm.cancel();
 *   if (co_await tm) { // handle the timeout
 *   // success!
 * @endcode
 *
 * Without coroutines this requires passing the timer future to the `OnRead()`
 * callback. See below for the details.
 */
class AsyncAccumulateReadObjectPartialHandle
    : public std::enable_shared_from_this<
          AsyncAccumulateReadObjectPartialHandle> {
 public:
  using Response = ::google::storage::v2::ReadObjectResponse;
  using Stream = ::google::cloud::internal::AsyncStreamingReadRpc<Response>;
  using StreamingRpcMetadata = ::google::cloud::internal::StreamingRpcMetadata;
  using Result = AsyncAccumulateReadObjectResult;

  AsyncAccumulateReadObjectPartialHandle(CompletionQueue cq,
                                         std::unique_ptr<Stream> stream,
                                         std::chrono::milliseconds timeout)
      : cq_(std::move(cq)), stream_(std::move(stream)), timeout_(timeout) {}

  future<Result> Invoke() {
    struct ByMove {
      std::shared_ptr<AsyncAccumulateReadObjectPartialHandle> self;
      future<bool> tm;
      void operator()(future<bool> f) { self->OnStart(std::move(tm), f.get()); }
    };
    auto tm = MakeTimeout();
    stream_->Start().then(ByMove{shared_from_this(), std::move(tm)});
    return promise_.get_future();
  }

 private:
  void OnStart(future<bool> tm, bool read_ok) {
    tm.cancel();
    auto self = shared_from_this();
    tm.then([self, read_ok](future<bool> f) {
      self->OnStartTimer(f.get(), read_ok);
    });
  }

  void OnStartTimer(bool timer_ok, bool read_ok) {
    if (timer_ok) return OnTimeout("Start()");
    if (!read_ok) return Finish();
    Read();
  }

  void Read() {
    struct ByMove {
      std::shared_ptr<AsyncAccumulateReadObjectPartialHandle> self;
      future<bool> tm;
      void operator()(future<absl::optional<Response>> f) {
        self->OnRead(std::move(tm), f.get());
      }
    };
    auto tm = MakeTimeout();
    stream_->Read().then(ByMove{shared_from_this(), std::move(tm)});
  }

  void OnRead(future<bool> tm, absl::optional<Response> response) {
    tm.cancel();
    struct ByMove {
      std::shared_ptr<AsyncAccumulateReadObjectPartialHandle> self;
      absl::optional<Response> response;
      void operator()(future<bool> f) {
        self->OnReadTimer(f.get(), std::move(response));
      }
    };
    tm.then(ByMove{shared_from_this(), std::move(response)});
  }

  void OnReadTimer(bool timer_ok, absl::optional<Response> response) {
    if (timer_ok) return OnTimeout("Read()");
    if (!response.has_value()) return Finish();
    accumulator_.push_back(*std::move(response));
    Read();
  }

  void Finish() {
    struct ByMove {
      std::shared_ptr<AsyncAccumulateReadObjectPartialHandle> self;
      future<bool> tm;
      void operator()(future<Status> f) {
        self->OnFinish(std::move(tm), f.get());
      }
    };
    auto tm = MakeTimeout();
    stream_->Finish().then(ByMove{shared_from_this(), std::move(tm)});
  }

  void OnFinish(future<bool> tm, Status status) {
    tm.cancel();
    promise_.set_value(Result{std::move(accumulator_),
                              stream_->GetRequestMetadata(),
                              std::move(status)});
  }

  future<bool> MakeTimeout() {
    auto self = shared_from_this();
    return cq_.MakeRelativeTimer(timeout_).then(
        [self](future<StatusOr<std::chrono::system_clock::time_point>> f) {
          if (!f.get().ok()) return false;
          self->stream_->Cancel();
          return true;
        });
  }

  void OnTimeout(char const* where) {
    auto finish = stream_->Finish();
    finish.then(WaitForFinish{std::move(stream_)});
    promise_.set_value(
{'fraction_non_alphanumeric': 0.13295537673738111, 'fraction_numerical': 0.0027432333577176297, 'mean_word_length': 3.3923694779116467, 'pattern_counts': {'":': 0, '<': 50, '<?xml version=': 0, '>': 80, 'https://': 2, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 8, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '4938490', 'n_tokens_mistral': 3290, 'n_tokens_neox': 3256, 'n_words': 888}