/*
 * .============.
 * // M A K E / \
 * // C++ DEV / \
 * // E A S Y / \/ \
 * ++ ----------. \/\ .
 * \\ \ \ /\ /
 * \\ \ \ /
 * \\ \ \ /
 * -============'
 *
 * Copyright (c) 2018 Hevake and contributors, all rights reserved.
 *
 * This file is part of cpp-tbox (https://github.com/cpp-main/cpp-tbox)
 * Use of this source code is governed by MIT license that can be found
 * in the LICENSE file in the root of the source tree. All contributing
 * project authors may be found in the CONTRIBUTORS.md file in the root
 * of the source tree.
 */
#include "client.h"

#include <thread>

#include <tbox/base/log.h>
#include <tbox/base/assert.h>
#include <tbox/base/lifetime_tag.hpp>
#include <tbox/base/wrapped_recorder.h>
#include <tbox/event/timer_event.h>
#include <tbox/event/fd_event.h>

#include <mosquitto.h>

namespace tbox {
namespace mqtt {

using namespace std;
using namespace event;

struct Client::Data {
 Loop *wp_loop = nullptr;

 Config config;
 Callbacks callbacks;

 mosquitto *sp_mosq = nullptr; //!< mosquitto 对象

 TimerEvent *sp_timer_ev = nullptr;
 FdEvent *sp_sock_read_ev = nullptr;
 FdEvent *sp_sock_write_ev = nullptr;

 State state = State::kNone;

 int cb_level = 0; //! 回调层级
 int reconnect_wait_remain_sec = 0; //! 重连等待剩余时长

 std::thread *sp_thread = nullptr;

 LifetimeTag alive_tag; //!< 对像存活标签
 /// Q1: 为什么需要定义它?
 ///
 /// 因为在下面使用了runInLoop()函数注册了延后执行的回调函数。在该函数中将this指针捕获了。
 /// 在它真正执行的时候如果 this 指针所指向的对象就已经被 delete 了,那么如果我们还继续
 /// 访问那么就一定会触发访问失效的内存地址,行为不可预知。
 ///
 /// 怎么解决呢?
 /// 方法一:在对象析构的时候,撤消之前通过 runInLoop() 注册的回调函数;
 /// 方法二:在回调函数真正执行的时候,判定一下 this 所指的对象是否有效。
 ///
 /// 这里,我们采用了方法二。
 /// 那怎么实现在后期能知晓 this 指针是否有效呢?
 /// 通用的方案是:将 this 指针用 std::shared_ptr 进行管理。传入回调函数中的是它的 weak_ptr
 /// 这样,在使用 this 指针的时候就能得知它是否有效了。
 ///
 /// 这么做有一个缺点:这个对象必须继续于 enable_this_shared_ptr 类,而且实例化它的时候必须
 /// 使用 std::make_shared<T>() 才有效。这样种会给使用者造成一定的负担。
 /// 除此之外,还有没有别的方法?于是,我想到了 alive_tag。
 /// 由于对象的成员的生命期与对象自身是一致的,判定一个对象是否有效,可以通过判定它的某个成
 /// 员生命期是否有效即可。内部附带一个任意类型的智能指针。于是,我在这里定义了一个 alive_tag
 /// 的智能指针。它的类型可以是任意的,这里就选定了bool。
 /// 无所谓它的类型,反正我们也不会往里写内容。
 /// 在 runInLoop() 注册回调函数的时候,就通过 alive_tag 生成对应的 std::weak_ptr 并传入。
 /// 在回调函数执行之前,检查一下这个 std::weak_ptr 是否过期就可以得知 this 指针是否有效。
 /// 进而避免继续访问到无效 this 指向的内容。

 int connect_fail_count = 0; //!< 单次连接失败

 static int _instance_count;
};

int Client::Data::_instance_count = 0;

Client::Client(Loop *wp_loop) :
 d_(new Data)
{
 TBOX_ASSERT(d_ != nullptr);
 d_->wp_loop = wp_loop;

 if (Data::_instance_count == 0)
 mosquitto_lib_init();
 ++Data::_instance_count;

 d_->sp_mosq = mosquitto_new(nullptr, true, this);

 d_->sp_timer_ev = wp_loop->newTimerEvent("mqtt::Client::sp_timer_ev");
 d_->sp_timer_ev->initialize(chrono::seconds(1), Event::Mode::kPersist);
 d_->sp_timer_ev->setCallback(std::bind(&Client::onTimerTick, this));

 d_->sp_sock_read_ev = wp_loop->newFdEvent("mqtt::Client::sp_sock_read_ev");
 d_->sp_sock_read_ev->setCallback(std::bind(&Client::onSocketRead, this));

 d_->sp_sock_write_ev = wp_loop->newFdEvent("mqtt::Client::sp_sock_write_ev");
 d_->sp_sock_write_ev->setCallback(std::bind(&Client::onSocketWrite, this));
}

Client::~Client()
{
 TBOX_ASSERT(d_->cb_level == 0);

 cleanup();

 delete d_->sp_sock_write_ev;
 delete d_->sp_sock_read_ev;
 delete d_->sp_timer_ev;
 mosquitto_destroy(d_->sp_mosq);
 delete d_;

 if (Data::_instance_count > 0) {
 --Data::_instance_count;
 if (Data::_instance_count == 0)
 mosquitto_lib_cleanup();
 }
}

bool Client::Config::isValid() const
{
 //! 检查base
 if (base.broker.domain.empty()) {
 LogWarn("broker domain invalid");
 return false;
 }

 if (base.broker.port == 0) {
 LogWarn("broker port == 0");
 return false;
 }

 if (base.keepalive < 0) {
 LogWarn("keeplive < 0");
 return false;
 }

 //! 检查tls
 if (tls.enabled) {
 if (tls.is_require_peer_cert) {
 //! 要验对方的证书,就必须得有CA。要么指定了文件,要么指定了路径
 if (tls.ca_file.empty() && tls.ca_path.empty()) {
 LogWarn("both CA file and path is empty");
 return false;
 }
 }

 if (!tls.cert_file.empty() && tls.key_file.empty()) {
 //! 如果指定了自己证书,但没有给出私钥,也是错的
 LogWarn("key_file is empty");
 return false;
 }
 }

 //! 检查will
 if (will.enabled) {
 if (will.topic.empty()) {
 LogWarn("will topic is empty");
 return false;
 }
 }

 return true;
}

bool Client::initialize(const Config &config, const Callbacks &callbacks)
{
 if (d_->state != State::kNone) {
 LogWarn("cleanup first");
 return false;
 }

 if (!config.isValid())
 return false;

 d_->config = config;
 d_->callbacks = callbacks;

 updateStateTo(State::kInited);
 return true;
}

void Client::cleanup()
{
 if (d_->state <= State::kNone)
 return;

 stop();

 d_->config = Config();
 d_->callbacks = Callbacks();
 updateStateTo(State::kNone);
}

bool Client::start()
{
 if (d_->state != State::kInited &&
 d_->state != State::kEnd) {
 LogWarn("state is not kInited or kEnd");
 return false;
 }

 RECORD_SCOPE();
 //! 基础设置
 const char *client_id = nullptr;
 if (!d_->config.base.client_id.empty())
 client_id = d_->config.base.client_id.c_str();

 int ret = mosquitto_reinitialise(d_->sp_mosq, client_id, true, this);
 if (ret == MOSQ_ERR_SUCCESS) {
 mosquitto_connect_callback_set(d_->sp_mosq, OnConnectWrapper);
 mosquitto_disconnect_callback_set(d_->sp_mosq, OnDisconnectWrapper);
 mosquitto_message_callback_set(d_->sp_mosq, OnMessageWrapper);
 mosquitto_publish_callback_set(d_->sp_mosq, OnPublishWrapper);
 mosquitto_subscribe_callback_set(d_->sp_mosq, OnSubscribeWrapper);
 mosquitto_unsubscribe_callback_set(d_->sp_mosq, OnUnsubscribeWrapper);
 mosquitto_log_callback_set(d_->sp_mosq, OnLogWrapper);
 } else {
 LogWarn("mosquitto_reinitialise() fail, ret:%s, %s", ret, mosquitto_strerror(ret));
 return false;
 }

 //! TLS 设置
 if (d_->config.tls.enabled) {
 const char *ca_file = nullptr;
 const char *ca_path = nullptr;
 const char *cert_file = nullptr;
 const char *key_file = nullptr;

 if (!d_->config.tls.ca_file.empty())
 ca_file = d_->config.tls.ca_file.c_str();

 if (!d_->config.tls.ca_path.empty())
 ca_path = d_->config.tls.ca_path.c_str();

 if (!d_->config.tls.cert_file.empty()) {
 cert_file = d_->config.tls.cert_file.c_str();
 key_file = d_->config.tls.key_file.c_str();
 }

 //! TODO: 没有完善 pw_callback 功能,将来需要再实现,暂时填nullptr
 mosquitto_tls_set(d_->sp_mosq, ca_file, ca_path, cert_file, key_file, nullptr);

 const char *ssl_version = nullptr;
 const char *ciphers = nullptr; //! TODO: 暂不设置

 if (!d_->config.tls.ssl_version.empty())
 ssl_version = d_->config.tls.ssl_version.c_str();

 mosquitto_tls_opts_set(d_->sp_mosq, d_->config.tls.is_require_peer_cert, ssl_version, ciphers);
 mosquitto_tls_insecure_set(d_->sp_mosq, d_->config.tls.is_insecure);
 }

 //! Will 设置
 if (d_->config.will.enabled) {
 mosquitto_will_set(d_->sp_mosq,
 d_->config.will.topic.c_str(),
 d_->config.will.payload.size(),
 d_->config.will.payload.data(),
 d_->config.will.qos,
 d_->config.will.retain);
 }

 const char *username = nullptr;
 const char *passwd = nullptr;

 if (!d_->config.base.username.empty())
 username = d_->config.base.username.c_str();

 if (!d_->config.base.passwd.empty())
 passwd = d_->config.base.passwd.c_str();

 if (username != nullptr)
 mosquitto_username_pw_set(d_->sp_mosq, username, passwd);

 updateStateTo(State::kConnecting);

 CHECK_DELETE_RESET_OBJ(d_->sp_thread);
 auto is_alive = d_->alive_tag.get(); //! 原理见Q1
 d_->connect_fail_count = 0;

 //! 由于 mosquitto_connect() 是阻塞函数,为了避免阻塞其它事件,特交给子线程去做
 d_->sp_thread = new thread(
 [this, is_alive] {
 RECORD_SCOPE();
 int ret = mosquitto_connect_async(d_->sp_mosq,
 d_->config.base.broker.domain.c_str(),
 d_->config.base.broker.port,
 d_->config.base.keepalive);
 d_->wp_loop->runInLoop(
 [this, is_alive, ret] {
 RECORD_SCOPE();
 if (!is_alive) { //!< 判定this指针是否有效
 LogWarn("object not alive");
 return;
 }

 onTcpConnectDone(ret);
 enableTimer();
 },
 "mqtt::Client::start, connect done"
 );
 }
 );

 return true;
}

void Client::stop()
{
 if (d_->state <= State::kInited ||
 d_->state == State::kEnd)
 return;

 RECORD_SCOPE();
 if (d_->state == State::kTcpConnected ||
 d_->state == State::kMqttConnected) {
 //! 如果已成功连接,则立即断开
 mosquitto_disconnect(d_->sp_mosq); //! 请求断开,后续工作在 onDisconnected() 中处理
 } else {
 //! 如果正在连接,则停止连接线程
 if (d_->sp_thread != nullptr) {
 d_->sp_thread->join();
 CHECK_DELETE_RESET_OBJ(d_->sp_thread);
 }
 }

 disableTimer();

 updateStateTo(State::kInited);
 return;
}

int Client::subscribe(const std::string &topic, int *p_mid, int qos)
{
 if (d_->state != State::kMqttConnected) {
 LogWarn("mqtt is not connected");
 return false;
 }

 RECORD_SCOPE();
 int mid = 0;
 int ret = mosquitto_subscribe(d_->sp_mosq, &mid, topic.c_str(), qos);
 if (p_mid != nullptr)
 *p_mid = mid;

 enableSocketWriteIfNeed();
 return ret == MOSQ_ERR_SUCCESS;
}

int Client::unsubscribe(const std::string &topic, int *p_mid)
{
 if (d_->state != State::kMqttConnected) {
 LogWarn("mqtt is not connected");
 return false;
 }

 RECORD_SCOPE();
 int mid = 0;
 int ret = mosquitto_unsubscribe(d_->sp_mosq, &mid, topic.c_str());
 if (p_mid != nullptr)
 *p_mid = mid;

 enableSocketWriteIfNeed();
 return ret == MOSQ_ERR_SUCCESS;
}

int Client::publish(const std::string &topic, const void *payload_ptr, size_t payload_size,
 int qos, bool retain, int *p_mid)
{
 if (d_->state != State::kMqttConnected) {
 LogWarn("mqtt is not connected");
 return false;
 }

 RECORD_SCOPE();
 int mid = 0;
 int ret = mosquitto_publish(d_->sp_mosq, &mid, topic.c_str(),
 payload_size, payload_ptr,
 qos, retain);
 if (p_mid != nullptr)
 *p_mid = mid;

 enableSocketWriteIfNeed();
 return ret == MOSQ_ERR_SUCCESS;
}

Client::State Client::getState() const
{
 return d_->state;
}

void Client::onTimerTick()
{
 if (d_->state == State::kTcpConnected ||
 d_->state == State::kMqttConnected) {
 mosquitto_loop_misc(d_->sp_mosq);

 if (mosquitto_socket(d_->sp_mosq) > 0)
 enableSocketWriteIfNeed();

 } else if (d_->state == State::kConnecting) {
 if (d_->sp_thread == nullptr) {
 auto is_alive = d_->alive_tag.get(); //! 原理见Q1
 d_->sp_thread = new thread(
 [this, is_alive] {
 RECORD_SCOPE();
 int ret = mosquitto_reconnect_async(d_->sp_mosq);
 d_->wp_loop->runInLoop(
 [this, is_alive, ret] {
 RECORD_SCOPE();
 if (!is_alive) { //!< 判定this指针是否有效
 LogWarn("object not alive");
 return;
 }
 onTcpConnectDone(ret);
 },
 "mqtt::Client::onTimerTick, reconnect done"
 );
 }
 );
 }

 } else if (d_->state == State::kReconnWaiting) {
 auto &remain_sec = d_->reconnect_wait_remain_sec;
 if (remain_sec > 0) {
 --remain_sec;
 if (remain_sec == 0) {
 updateStateTo(State::kConnecting);
 LogDbg("wait timeout, reconnect now");
 }
 }

 } else if (d_->state == State::kEnd) {
 disableTimer();
 }
}

void Client::onSocketRead()
{
 mosquitto_loop_read(d_->sp_mosq, 1);
 enableSocketWriteIfNeed();
}

void Client::onSocketWrite()
{
 mosquitto_loop_write(d_->sp_mosq, 1);
 enableSocketWriteIfNeed();
}

void Client::OnConnectWrapper(struct mosquitto *, void *userdata, int rc)
{
 Client *p_this = static_cast<Client*>(userdata);
 p_this->onConnected(rc);
}

void Client::OnDisconnectWrapper(struct mosquitto *, void *userdata, int rc)
{
 Client *p_this = static_cast<Client*>(userdata);
 p_this->onDisconnected(rc);
}

void Client::OnPublishWrapper(struct mosquitto *, void *userdata, int mid)
{
 Client *p_this = static_cast<Client*>(userdata);
 p_this->onPublish(mid);
}

void Client::OnMessageWrapper(struct mosquitto *, void *userdata, const struct mosquitto_message *msg)
{
 Client *p_this = static_cast<Client*>(userdata);
 p_this->onMessage(msg);
}

void Client::OnSubscribeWrapper(struct mosquitto *, void *userdata, int mid, int qos_count, const int *granted_qos)
{
 Client *p_this = static_cast<Client*>(userdata);
 p_this->onSubscribe(mid, qos_count, granted_qos);
}

void Client::OnUnsubscribeWrapper(struct mosquitto *, void *userdata, int mid)
{
 Client *p_this = static_cast<Client*>(userdata);
 p_this->onUnsubscribe(mid);
}

void Client::OnLogWrapper(struct mosquitto *, void *userdata, int level, const char *str)
{
 Client *p_this = static_cast<Client*>(userdata);
 p_this->onLog(level, str);
}

void Client::onConnected(int rc)
{
 if (rc != 0) {
 LogWarn("connect fail, rc:%d, %s", rc, mosquitto_strerror(rc));
 return;
 }

 LogInfo("connected");

 RECORD_SCOPE();
 if (d_->state != State::kMqttConnected) {
 updateStateTo(State::kMqttConnected);
 ++d_->cb_level;
 if (d_->callbacks.connected)
 d_->callbacks.connected();
 --d_->cb_level;
 }
}

void Client::onDisconnected(int rc)
{
 RECORD_SCOPE();
 LogNotice("disconnected, rc:%d, %s", rc, mosquitto_strerror(rc));
 handleDisconnectEvent();
}

void Client::onPublish(int mid)
{
 LogInfo("mid:%d", mid);

 RECORD_SCOPE();
 ++d_->cb_level;
 if (d_->callbacks.message_pub)
 d_->callbacks.message_pub(mid);
 --d_->cb_level;
}

void Client::onSubscribe(int mid, int qos, const int *granted_qos)
{
 LogInfo("mid:%d, qos:%d", mid, qos);

 RECORD_SCOPE();
 ++d_->cb_level;
 if (d_->callbacks.subscribed)
 d_->callbacks.subscribed(mid, qos, granted_qos);
 --d_->cb_level;
}

void Client::onUnsubscribe(int mid)
{
 LogInfo("mid:%d", mid);

 RECORD_SCOPE();
 ++d_->cb_level;
 if (d_->callbacks.unsubscribed)
 d_->callbacks.unsubscribed(mid);
 --d_->cb_level;
}

void Client::onMessage(const struct mosquitto_message *msg)
{
 if (msg == nullptr)
 return;

 LogInfo("mid:%d, topic:%s", msg->mid, msg->topic);

 RECORD_SCOPE();
 ++d_->cb_level;
 if (d_->callbacks.message_recv)
 d_->callbacks.message_recv(msg->mid,
 msg->topic,
 msg->payload,
 msg->payloadlen,
 msg->qos,
 msg->retain);
 --d_->cb_level;
}

void Client::onLog(int level, const char *str)
{
 auto new_level = TBOX_LOG_LEVEL_DEBUG;
 switch (level & 0x1F) {
 case MOSQ_LOG_ERR:
 new_level = TBOX_LOG_LEVEL_ERROR;
 break;
 case MOSQ_LOG_WARNING:
 new_level = TBOX_LOG_LEVEL_WARN;
 break;
 case MOSQ_LOG_NOTICE:
 new_level = TBOX_LOG_LEVEL_NOTICE;
 break;
 default:; //! regard MOSQ_LOG_INFO as TBOX_LOG_LEVEL_DEBUG
 }
 LogPrintfFunc("mosq", nullptr, nullptr, 0, new_level, 0, str);
}

void Client::onTcpConnectDone(int ret)
{
 RECORD_SCOPE();
 if (d_->sp_thread == nullptr) {
 LogWarn("sp_thread == nullptr");
 return;
 }

 d_->sp_thread->join();
 CHECK_DELETE_RESET_OBJ(d_->sp_thread);

 if (ret == MOSQ_ERR_SUCCESS) {
 d_->connect_fail_count = 0;
 LogDbg("connect success");
 enableSocketRead();
 enableSocketWriteIfNeed();
 updateStateTo(State::kTcpConnected);

 } else {
 ++d_->connect_fail_count;
 LogNotice("connect fail, rc:%d, %s, fail_count:%d",
 ret, mosquitto_strerror(ret), d_->connect_fail_count);

 tryReconnect();

 ++d_->cb_level;
 if (d_->callbacks.connect_fail)
 d_->callbacks.connect_fail();
 --d_->cb_level;
 }
}

void Client::enableSocketRead()
{
 if (mosquitto_socket(d_->sp_mosq) < 0)
 return;

 if (d_->sp_sock_read_ev->isEnabled())
 return;

 d_->sp_sock_read_ev->initialize(mosquitto_socket(d_->sp_mosq), FdEvent::kReadEvent, Event::Mode::kPersist);
 d_->sp_sock_read_ev->enable();
}

void Client::enableSocketWrite()
{
 if (mosquitto_socket(d_->sp_mosq) < 0)
 return;

 if (d_->sp_sock_write_ev->isEnabled())
 return;

 d_->sp_sock_write_ev->initialize(mosquitto_socket(d_->sp_mosq), FdEvent::kWriteEvent, Event::Mode::kOneshot);
 d_->sp_sock_write_ev->enable();
}

void Client::enableSocketWriteIfNeed()
{
 if (mosquitto_want_write(d_->sp_mosq))
 enableSocketWrite();
}

void Client::enableTimer()
{
 d_->sp_timer_ev->enable();
}

void Client::disableSocketRead()
{
 d_->sp_sock_read_ev->disable();
}

void Client::disableSocketWrite()
{
 d_->sp_sock_write_ev->disable();
}

void Client::disableTimer()
{
 d_->sp_timer_ev->disable();
}

void Client::tryReconnect()
{
 //! 如果开启了自动重连
 if (d_->config.auto_reconnect_enable) {
 auto wait_remain_sec = d_->config.auto_reconnect_wait_sec;

 //! 如果设置了 auto_reconnect_wait_sec_gen_func 函数
 //! 就则使用 auto_reconnect_wait_sec_gen_func 函数生成 wait_remain_sec
 if (d_->config.auto_reconnect_wait_sec_gen_func)
 wait_remain_sec = d_->config.auto_reconnect_wait_sec_gen_func(d_->connect_fail_count);

 if (wait_remain_sec > 0) {
 LogDbg("reconnect after %d sec", wait_remain_sec);
 updateStateTo(State::kReconnWaiting);

 } else {
 LogDbg("reconnect now");
 updateStateTo(State::kConnecting);
 }

 d_->reconnect_wait_remain_sec = wait_remain_sec;

 } else { //! 如果不需要自动重连
 LogDbg("no need reconnect, end");
 updateStateTo(State::kEnd);
 }
}

void Client::handleDisconnectEvent()
{
 disableSocketRead();
 disableSocketWrite();

 if (d_->state == State::kTcpConnected ||
 d_->state == State::kMqttConnected) {
 auto is_mqtt_disconnected = d_->state == State::kMqttConnected;
 //! 一定要先判定,因为 tryReconnect() 会改 d_->state

 tryReconnect();

 ++d_->cb_level;
 if (is_mqtt_disconnected) {
 if (d_->callbacks.disconnected)
 d_->callbacks.disconnected();
 } else {
 if (d_->callbacks.connect_fail)
 d_->callbacks.connect_fail();
 }
 --d_->cb_level;
 }
}

void Client::updateStateTo(State new_state)
{
 d_->state = new_state;
 LogDbg("new_state: %d", d_->state);

 ++d_->cb_level;
 if (d_->callbacks.state_changed)
 d_->callbacks.state_changed(new_state);
 --d_->cb_level;
}

}
}