/*
 *
 * Copyright 2014, Google Inc.
 * All rights reserved.
 *
 * Redistribution and use in source and binary forms, with or without
 * modification, are permitted provided that the following conditions are
 * met:
 *
 *     * Redistributions of source code must retain the above copyright
 * notice, this list of conditions and the following disclaimer.
 *     * Redistributions in binary form must reproduce the above
 * copyright notice, this list of conditions and the following disclaimer
 * in the documentation and/or other materials provided with the
 * distribution.
 *     * Neither the name of Google Inc. nor the names of its
 * contributors may be used to endorse or promote products derived from
 * this software without specific prior written permission.
 *
 * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
 * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
 * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
 * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
 * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
 * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
 * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
 * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
 * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
 * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
 * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
 *
 */

#include "src/cpp/stream/stream_context.h"

#include <grpc/grpc.h>
#include <grpc/support/log.h>
#include "src/cpp/rpc_method.h"
#include "src/cpp/proto/proto_utils.h"
#include "src/cpp/util/time.h"
#include <grpc++/client_context.h>
#include <grpc++/config.h>
#include <google/protobuf/message.h>

namespace grpc {

// Client only ctor
StreamContext::StreamContext(const RpcMethod& method, ClientContext* context,
                             const google::protobuf::Message* request,
                             google::protobuf::Message* result)
    : is_client_(true),
      method_(&method),
      context_(context),
      request_(request),
      result_(result),
      invoke_ev_(nullptr),
      read_ev_(nullptr),
      write_ev_(nullptr),
      reading_(false),
      writing_(false),
      got_read_(false),
      got_write_(false),
      peer_halfclosed_(false),
      self_halfclosed_(false),
      stream_finished_(false),
      waiting_(false) {
  GPR_ASSERT(method_->method_type() != RpcMethod::RpcType::NORMAL_RPC);
}

StreamContext::~StreamContext() { cq_poller_.join(); }

void StreamContext::PollingLoop() {
  grpc_event* ev = nullptr;
  gpr_timespec absolute_deadline;
  AbsoluteDeadlineTimepoint2Timespec(context_->absolute_deadline(),
                                     &absolute_deadline);
  std::condition_variable* cv_to_notify = nullptr;
  std::unique_lock<std::mutex> lock(mu_, std::defer_lock);
  while (1) {
    cv_to_notify = nullptr;
    lock.lock();
    if (stream_finished_ && !reading_ && !writing_) {
      return;
    }
    lock.unlock();
    ev = grpc_completion_queue_next(context_->cq(), absolute_deadline);
    lock.lock();
    if (!ev) {
      stream_finished_ = true;
      final_status_ = Status(StatusCode::DEADLINE_EXCEEDED);
      std::condition_variable* cvs[3] = {reading_ ? &read_cv_ : nullptr,
                                         writing_ ? &write_cv_ : nullptr,
                                         waiting_ ? &finish_cv_ : nullptr};
      got_read_ = reading_;
      got_write_ = writing_;
      read_ev_ = nullptr;
      write_ev_ = nullptr;
      lock.unlock();
      for (int i = 0; i < 3; i++) {
        if (cvs[i]) cvs[i]->notify_one();
      }
      break;
    }
    switch (ev->type) {
      case GRPC_READ:
        GPR_ASSERT(reading_);
        got_read_ = true;
        read_ev_ = ev;
        cv_to_notify = &read_cv_;
        reading_ = false;
        break;
      case GRPC_FINISH_ACCEPTED:
      case GRPC_WRITE_ACCEPTED:
        got_write_ = true;
        write_ev_ = ev;
        cv_to_notify = &write_cv_;
        writing_ = false;
        break;
      case GRPC_FINISHED: {
        grpc::string error_details(
            ev->data.finished.details ? ev->data.finished.details : "");
        final_status_ = Status(static_cast<StatusCode>(ev->data.finished.code),
                               error_details);
        grpc_event_finish(ev);
        stream_finished_ = true;
        if (waiting_) {
          cv_to_notify = &finish_cv_;
        }
        break;
      }
      case GRPC_INVOKE_ACCEPTED:
        invoke_ev_ = ev;
        cv_to_notify = &invoke_cv_;
        break;
      case GRPC_CLIENT_METADATA_READ:
        grpc_event_finish(ev);
        break;
      default:
        grpc_event_finish(ev);
        // not handling other types now
        gpr_log(GPR_ERROR, "unknown event type");
        abort();
    }
    lock.unlock();
    if (cv_to_notify) {
      cv_to_notify->notify_one();
    }
  }
}

void StreamContext::Start(bool buffered) {
  // TODO(yangg) handle metadata send path
  int flag = buffered ? GRPC_WRITE_BUFFER_HINT : 0;
  grpc_call_error error = grpc_call_start_invoke(
      context_->call(), context_->cq(), this, this, this, flag);
  GPR_ASSERT(GRPC_CALL_OK == error);
  // kicks off the poller thread
  cq_poller_ = std::thread(&StreamContext::PollingLoop, this);
  std::unique_lock<std::mutex> lock(mu_);
  while (!invoke_ev_) {
    invoke_cv_.wait(lock);
  }
  lock.unlock();
  GPR_ASSERT(invoke_ev_->data.invoke_accepted == GRPC_OP_OK);
  grpc_event_finish(invoke_ev_);
}

namespace {
// Wait for got_event with event_cv protected by mu, return event.
grpc_event* WaitForEvent(bool* got_event, std::condition_variable* event_cv,
                         std::mutex* mu, grpc_event** event) {
  std::unique_lock<std::mutex> lock(*mu);
  while (*got_event == false) {
    event_cv->wait(lock);
  }
  *got_event = false;
  return *event;
}
}  // namespace

bool StreamContext::Read(google::protobuf::Message* msg) {
  std::unique_lock<std::mutex> lock(mu_);
  if (stream_finished_) {
    peer_halfclosed_ = true;
    return false;
  }
  reading_ = true;
  lock.unlock();

  grpc_call_error err = grpc_call_start_read(context_->call(), this);
  GPR_ASSERT(err == GRPC_CALL_OK);

  grpc_event* ev = WaitForEvent(&got_read_, &read_cv_, &mu_, &read_ev_);
  if (!ev) {
    return false;
  }
  GPR_ASSERT(ev->type == GRPC_READ);
  bool ret = true;
  if (ev->data.read) {
    if (!DeserializeProto(ev->data.read, msg)) {
      ret = false;  // parse error
                    // TODO(yangg) cancel the stream.
    }
  } else {
    ret = false;
    peer_halfclosed_ = true;
  }
  grpc_event_finish(ev);
  return ret;
}

bool StreamContext::Write(const google::protobuf::Message* msg, bool is_last) {
  bool ret = true;
  grpc_event* ev = nullptr;

  std::unique_lock<std::mutex> lock(mu_);
  if (stream_finished_) {
    self_halfclosed_ = true;
    return false;
  }
  writing_ = true;
  lock.unlock();

  if (msg) {
    grpc_byte_buffer* out_buf = nullptr;
    if (!SerializeProto(*msg, &out_buf)) {
      FinishStream(Status(StatusCode::INVALID_ARGUMENT,
                          "Failed to serialize request proto"),
                   true);
      return false;
    }
    int flag = is_last ? GRPC_WRITE_BUFFER_HINT : 0;
    grpc_call_error err =
        grpc_call_start_write(context_->call(), out_buf, this, flag);
    grpc_byte_buffer_destroy(out_buf);
    GPR_ASSERT(err == GRPC_CALL_OK);

    ev = WaitForEvent(&got_write_, &write_cv_, &mu_, &write_ev_);
    if (!ev) {
      return false;
    }
    GPR_ASSERT(ev->type == GRPC_WRITE_ACCEPTED);

    ret = ev->data.write_accepted == GRPC_OP_OK;
    grpc_event_finish(ev);
  }
  if (is_last) {
    grpc_call_error err = grpc_call_writes_done(context_->call(), this);
    GPR_ASSERT(err == GRPC_CALL_OK);
    ev = WaitForEvent(&got_write_, &write_cv_, &mu_, &write_ev_);
    if (!ev) {
      return false;
    }
    GPR_ASSERT(ev->type == GRPC_FINISH_ACCEPTED);
    grpc_event_finish(ev);
    self_halfclosed_ = true;
  }
  return ret;
}

const Status& StreamContext::Wait() {
  std::unique_lock<std::mutex> lock(mu_);
  // TODO(yangg) if not halfclosed cancel the stream
  GPR_ASSERT(self_halfclosed_);
  GPR_ASSERT(peer_halfclosed_);
  GPR_ASSERT(!waiting_);
  waiting_ = true;
  while (!stream_finished_) {
    finish_cv_.wait(lock);
  }
  return final_status_;
}

void StreamContext::FinishStream(const Status& status, bool send) { return; }

}  // namespace grpc