<|start_filename|>SaveMoney/Pods/FirebaseFirestore/Firestore/core/src/firebase/firestore/remote/grpc_streaming_reader.h<|end_filename|>
/*
 * 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_GRPC_STREAMING_READER_H_
#define FIRESTORE_CORE_SRC_FIREBASE_FIRESTORE_REMOTE_GRPC_STREAMING_READER_H_

#include <functional>
#include <map>
#include <memory>
#include <vector>

#include "Firestore/core/src/firebase/firestore/remote/grpc_stream.h"
#include "Firestore/core/src/firebase/firestore/remote/grpc_stream_observer.h"
#include "Firestore/core/src/firebase/firestore/util/status.h"
#include "Firestore/core/src/firebase/firestore/util/statusor.h"
#include "grpcpp/client_context.h"
#include "grpcpp/generic/generic_stub.h"
#include "grpcpp/support/byte_buffer.h"

namespace firebase {
namespace firestore {
namespace remote {

/**
 * Sends a single request to the server, reads one or more streaming server
 * responses, and invokes the given callback with the accumulated responses.
 */
class GrpcStreamingReader : public GrpcStreamObserver {
 public:
  using MetadataT = GrpcStream::MetadataT;
  using ResponsesT = std::vector<grpc::ByteBuffer>;
  using CallbackT = std::function<void(const util::StatusOr<ResponsesT>&)>;

  GrpcStreamingReader(
      std::unique_ptr<grpc::ClientContext> context,
      std::unique_ptr<grpc::GenericClientAsyncReaderWriter> call,
      util::AsyncQueue* worker_queue,
      const grpc::ByteBuffer& request);

  /**
   * Starts the call; the given `callback` will be invoked with the accumulated
   * results of the call. If the call fails, the `callback` will be invoked with
   * a non-ok status.
   */
  void Start(CallbackT&& callback);

  /**
   * If the call is in progress, attempts to cancel the call; otherwise, it's
   * a no-op. Cancellation is done on best-effort basis; however:
   * - the call is guaranteed to be finished when this function returns;
   * - this function is blocking but should finish very fast (order of
   *   milliseconds).
   *
   * If this function succeeds in cancelling the call, the callback will not be
   * invoked.
   */
  void Cancel();

  /**
   * Returns the metadata received from the server.
   *
   * Can only be called once the `GrpcStreamingReader` has started.
   */
  MetadataT GetResponseHeaders() const {
    return stream_->GetResponseHeaders();
  }

  void OnStreamStart() override;
  void OnStreamRead(const grpc::ByteBuffer& message) override;
  void OnStreamFinish(const util::Status& status) override;

 private:
  std::unique_ptr<GrpcStream> stream_;
  grpc::ByteBuffer request_;

  CallbackT callback_;
  ResponsesT responses_;
};

}  // namespace remote
}  // namespace firestore
}  // namespace firebase

#endif  // FIRESTORE_CORE_SRC_FIREBASE_FIRESTORE_REMOTE_GRPC_STREAMING_READER_H_


<|start_filename|>SaveMoney/Pods/FirebaseFirestore/Firestore/core/src/firebase/firestore/remote/grpc_completion.cc<|end_filename|>
/*
 * 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.
 */

#include "Firestore/core/src/firebase/firestore/remote/grpc_completion.h"

#include <utility>

namespace firebase {
namespace firestore {
namespace remote {

using util::AsyncQueue;

GrpcCompletion::GrpcCompletion(AsyncQueue* worker_queue, Callback&& callback)
    : worker_queue_{worker_queue}, callback_{std::move(callback)} {
}

void GrpcCompletion::Cancel() {
  worker_queue_->VerifyIsCurrentQueue();
  callback_ = {};
}

void GrpcCompletion::WaitUntilOffQueue() {
  worker_queue_->VerifyIsCurrentQueue();

  EnsureValidFuture();
  return off_queue_future_.wait();
}

std::future_status GrpcCompletion::WaitUntilOffQueue(
    std::chrono::milliseconds timeout) {
  worker_queue_->VerifyIsCurrentQueue();

  EnsureValidFuture();
  return off_queue_future_.wait_for(timeout);
}

void GrpcCompletion::EnsureValidFuture() {
  if (!off_queue_future_.valid()) {
    off_queue_future_ = off_queue_.get_future();
  }
}

void GrpcCompletion::Complete(bool ok) {
  // This mechanism allows `GrpcStream` to know when the completion is off the
  // gRPC completion queue (and thus no longer requires the underlying gRPC
  // objects to be valid).
  off_queue_.set_value();

  worker_queue_->Enqueue([this, ok] {
    if (callback_) {
      callback_(ok, this);
    }
    delete this;
  });
}

}  // namespace remote
}  // namespace firestore
}  // namespace firebase
