/**
 * Copyright (c) 2021 OceanBase
 * OceanBase CE is licensed under Mulan PubL v2.
 * You can use this software according to the terms and conditions of the Mulan PubL v2.
 * You may obtain a copy of Mulan PubL v2 at:
 *          http://license.coscl.org.cn/MulanPubL-2.0
 * THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND,
 * EITHER EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT,
 * MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE.
 * See the Mulan PubL v2 for more details.
 */

#ifndef OCEANBASE_SQL_EXECUTOR_RPC_IMPL_
#define OCEANBASE_SQL_EXECUTOR_RPC_IMPL_

#include "share/ob_define.h"
#include "lib/container/ob_array.h"
#include "lib/allocator/ob_allocator.h"
#include "rpc/obrpc/ob_rpc_proxy.h"
#include "share/ob_scanner.h"
#include "share/rpc/ob_batch_rpc.h"
#include "sql/executor/ob_task.h"
#include "sql/executor/ob_task_info.h"
#include "sql/executor/ob_slice_id.h"
#include "sql/executor/ob_executor_rpc_proxy.h"
#include "lib/ob_define.h"

namespace oceanbase
{
namespace obrpc
{
class ObExecutorRpcProxy;
}

namespace sql
{
class ObQueryRetryInfo;

/* MySteamHandler生命周期为整个SQL语句, 与ObResultSet相同，本质为栈上变量。参考:obmp_query.cpp */
template <obrpc::ObRpcPacketCode pcode>
class MyStreamHandle
{
public:
  typedef typename obrpc::ObExecutorRpcProxy::SSHandle<pcode> MyHandle;
  typedef common::ObScanner MyResult;
  explicit MyStreamHandle(const char *label, uint64_t tenant_id)
    : result_(label, NULL, common::ObScanner::DEFAULT_MAX_SERIALIZE_SIZE, tenant_id),
      rc_(common::OB_SUCCESS)
  {
  }
  virtual ~MyStreamHandle() {}
  void reset()
  {
    result_.reset();
    rc_ = common::OB_SUCCESS;
  }
  const common::ObAddr &get_dst_addr() const
  {
    return handle_.get_dst_addr();
  }
  MyHandle &get_handle() { return handle_; }
  MyResult *get_result()
  {
    MyResult *ret_result = NULL;
    if (!result_.is_inited()) {
      SQL_EXE_LOG_RET(ERROR, common::OB_NOT_INIT, "result_ is not inited");
    } else {
      ret_result = &result_;
    }
    return ret_result;
  }

  int reset_and_init_result()
  {
    int ret = common::OB_SUCCESS;
    result_.reset();
    if (!result_.is_inited() && OB_FAIL(result_.init())) {
      SQL_EXE_LOG(WARN, "fail to init result", K(ret));
    }
    return ret;
  }

  void set_result_code(int code) { rc_ = code; }
  int get_result_code() { return rc_; }

  void set_task_id(const ObTaskID &task_id) { task_id_ = task_id; }
  const ObTaskID &get_task_id() const { return task_id_; }
private:
  ObTaskID task_id_;
  MyHandle handle_;
  MyResult result_;
  int rc_;
};

typedef MyStreamHandle<obrpc::OB_REMOTE_EXECUTE> RemoteStreamHandle;
typedef MyStreamHandle<obrpc::OB_REMOTE_SYNC_EXECUTE> RemoteStreamHandleV2;
typedef MyStreamHandle<obrpc::OB_TASK_FETCH_RESULT> FetchResultStreamHandle;

class RemoteExecuteStreamHandle
{
public:
  RemoteExecuteStreamHandle(const char *label, uint64_t tenant_id) :
    use_remote_protocol_v2_(false),
    sync_stream_handle_(label, tenant_id),
    sync_stream_handle_v2_(label, tenant_id)
  {
  }
  ~RemoteExecuteStreamHandle() = default;
  void set_use_remote_protocol_v2() { use_remote_protocol_v2_ = true; }
  void reset()
  {
    if (use_remote_protocol_v2_) {
      sync_stream_handle_v2_.reset();
    } else {
      sync_stream_handle_.reset();
    }
  }
  const common::ObAddr &get_dst_addr() const
  {
    const common::ObAddr *dst_addr = nullptr;
    if (use_remote_protocol_v2_) {
      dst_addr = &sync_stream_handle_v2_.get_dst_addr();
    } else {
      dst_addr = &sync_stream_handle_.get_dst_addr();
    }
    return *dst_addr;
  }

  int reset_and_init_result()
  {
    int ret = common::OB_SUCCESS;
    if (use_remote_protocol_v2_) {
      ret = sync_stream_handle_v2_.reset_and_init_result();
    } else {
      ret = sync_stream_handle_.reset_and_init_result();
    }
    return ret;
  }

  void set_result_code(int code)
  {
    if (use_remote_protocol_v2_) {
      sync_stream_handle_v2_.set_result_code(code);
    } else {
      sync_stream_handle_.set_result_code(code);
    }
  }
  int get_result_code()
  {
    int ret = common::OB_SUCCESS;
    if (use_remote_protocol_v2_) {
      ret = sync_stream_handle_v2_.get_result_code();
    } else {
      ret = sync_stream_handle_.get_result_code();
    }
    return ret;
  }

  void set_task_id(const ObTaskID &task_id)
  {
    if (use_remote_protocol_v2_) {
      sync_stream_handle_v2_.set_task_id(task_id);
    } else {
      sync_stream_handle_.set_task_id(task_id);
    }
  }
  const ObTaskID &get_task_id() const
  {
    const ObTaskID *task_id = nullptr;
    if (use_remote_protocol_v2_) {
      task_id = &(sync_stream_handle_v2_.get_task_id());
    } else {
      task_id = &(sync_stream_handle_.get_task_id());
    }
    return *task_id;
  }
  common::ObScanner *get_result()
  {
    common::ObScanner *result = nullptr;
    if (use_remote_protocol_v2_) {
      result = sync_stream_handle_v2_.get_result();
    } else {
      result = sync_stream_handle_.get_result();
    }
    return result;
  }
  bool has_more()
  {
    bool bret = false;
    if (use_remote_protocol_v2_) {
      bret = sync_stream_handle_v2_.get_handle().has_more();
    } else {
      bret = sync_stream_handle_.get_handle().has_more();
    }
    return bret;
  }
  int abort()
  {
    int ret = common::OB_SUCCESS;
    if (use_remote_protocol_v2_) {
      ret = sync_stream_handle_v2_.get_handle().abort();
    } else {
      ret = sync_stream_handle_.get_handle().abort();
    }
    return ret;
  }
  int get_more(common::ObScanner &result)
  {
    int ret = common::OB_SUCCESS;
    if (use_remote_protocol_v2_) {
      ret = sync_stream_handle_v2_.get_handle().get_more(result);
    } else {
      ret = sync_stream_handle_.get_handle().get_more(result);
    }
    return ret;
  }
  RemoteStreamHandle &get_remote_stream_handle() { return sync_stream_handle_; }
  RemoteStreamHandleV2 &get_remote_stream_handle_v2() { return sync_stream_handle_v2_; }
private:
  bool use_remote_protocol_v2_;
  RemoteStreamHandle sync_stream_handle_;
  RemoteStreamHandleV2 sync_stream_handle_v2_;
};

class ObExecutorRpcCtx
{
public:
  //FIXME qianfu 仅用于兼容，1.4.0之后去掉
  static const uint64_t INVALID_CLUSTER_VERSION = 0;
public:
  ObExecutorRpcCtx(uint64_t rpc_tenant_id,
                   int64_t timeout_timestamp,
                   uint64_t min_cluster_version,
                   ObQueryRetryInfo *retry_info,
                   ObSQLSessionInfo *session,
                   bool is_plain_select,
                   int32_t group_id)
    : rpc_tenant_id_(rpc_tenant_id),
      timeout_timestamp_(timeout_timestamp),
      min_cluster_version_(min_cluster_version),
      retry_info_(retry_info),
      session_(session),
      is_plain_select_(is_plain_select),
      group_id_(group_id)
  {
  }
  ~ObExecutorRpcCtx() {}

  uint64_t get_rpc_tenant_id() const { return rpc_tenant_id_; }
  inline int64_t get_timeout_timestamp() const { return timeout_timestamp_; }
  // The timeout provided to the storage layer will be reduced by 100ms
  // The timeout here needs to be aligned.
  inline int64_t get_ps_timeout_timestamp() const { return timeout_timestamp_ - ESTIMATE_PS_RESERVE_TIME; }
  // 等于INVALID_CLUSTER_VERSION说明是从远端的旧observer上序列化过来的
  inline bool min_cluster_version_is_valid() const
  {
    return INVALID_CLUSTER_VERSION != min_cluster_version_;
  }
  inline uint64_t get_min_cluster_version() const { return min_cluster_version_; }
  inline const ObQueryRetryInfo *get_retry_info() const { return retry_info_; }
  inline ObQueryRetryInfo *get_retry_info_for_update() const { return retry_info_; }
  bool is_retry_for_rpc_timeout() const { return is_plain_select_; }
  int check_status() const;
  int32_t get_group_id() const { return group_id_; }
  TO_STRING_KV(K_(rpc_tenant_id),
               K_(timeout_timestamp),
               K_(min_cluster_version),
               K_(retry_info),
               K_(is_plain_select),
               K_(group_id));
private:
  uint64_t rpc_tenant_id_;
  int64_t timeout_timestamp_;
  uint64_t min_cluster_version_;
  // retry_info_ == NULL表示本次rpc不用给重试模块反馈信息
  ObQueryRetryInfo *retry_info_;
  const ObSQLSessionInfo *session_;//该类中的变量会并发访问，注意session成功并发访问是否正确
  bool is_plain_select_;//stmt_type == T_SELECT && not select...for update
  int32_t group_id_;
private:
  DISALLOW_COPY_AND_ASSIGN(ObExecutorRpcCtx);
};


#define OB_SQL_REMOTE_TASK_TYPE 1
#define OB_SQL_REMOTE_RESULT_TYPE 2

// 所有调用rpc的地方都要使用to函数，以便支持并发调用
class ObExecutorRpcImpl
{
public:
  ObExecutorRpcImpl() : proxy_(NULL), batch_rpc_(nullptr) { }
  virtual ~ObExecutorRpcImpl() {}
  /*
   * 设置rpc proxy
   * */
  int init(obrpc::ObExecutorRpcProxy *rpc_proxy, obrpc::ObBatchRpc *batch_rpc);
  /*
   * 发送一个task并阻塞等待，直到对端返回执行状态
   * 将执行句柄保存在handler中, 随后可以通过handler收取数据
   * */
  virtual int task_execute(ObExecutorRpcCtx &rpc_ctx,
                           ObTask &task,
                           const common::ObAddr &svr,
                           RemoteExecuteStreamHandle &handler,
                           bool &has_sent_task,
                           bool &has_transfer_err);
  virtual int task_execute_v2(ObExecutorRpcCtx &rpc_ctx,
                              ObRemoteTask &task,
                              const common::ObAddr &svr,
                              RemoteExecuteStreamHandle &handler,
                              bool &has_sent_task,
                              bool &has_transfer_err);
  int remote_task_batch_submit(const uint64_t tenant_id,
                               const common::ObAddr &server,
                               const int64_t cluster_id,
                               const ObRemoteTask &task,
                               bool &has_sent_task);
  int remote_batch_post_result(const uint64_t tenant_id,
                               const common::ObAddr &server,
                               const int64_t cluster_id,
                               const ObRemoteResult &result,
                               bool &has_sent_result);
  /*
   * used to terminate remote streaming
   * */
  virtual int task_kill(
      ObExecutorRpcCtx &rpc_ctx,
      const ObTaskID &task_id,
      const common::ObAddr &svr);

  obrpc::ObExecutorRpcProxy *get_proxy() { return proxy_; }

private:
  void deal_with_rpc_timeout_err(ObExecutorRpcCtx &rpc_ctx,
                                 int &err,
                                 const common::ObAddr &dist_server) const;
  int get_sql_batch_req_type(int64_t execution_id) const;
private:
  /* functions */
  /* variables */
  obrpc::ObExecutorRpcProxy *proxy_;
  obrpc::ObBatchRpc *batch_rpc_;
  /* other */
  DISALLOW_COPY_AND_ASSIGN(ObExecutorRpcImpl);
};

}
}
#endif /* OCEANBASE_SQL_EXECUTOR_RPC_IMPL_ */
//// end of header file