#include <stdio.h>
#include <string.h>
#include <pthread.h>
#include "mosquitto.h"
#include "bmser_dev.h"
#include "bmser_mqtt.h"
#include "bmser_mqtt_upgrade.h"
#include "bmser_log.h"
#include "bmser_error_code.h"

#ifndef WIN32
#include <sys/time.h>
#include <time.h>
#else
#include <stdio.h>
#include <process.h>
#include <winsock2.h>
#endif

#define MAX_MOSQ_MQTT_CNT 256

#ifndef KEEP_ALIVE
	#define KEEP_ALIVE 10
#endif
#ifndef MQTT_QOS_LEVEL2 
	#define MQTT_QOS_LEVEL2 2
#endif

typedef struct
{
	bool valid;														//mqtt连接有效性
	struct mosquitto* mosq;											//mosq句柄
	BMSER_MOSQ_MQTT_MSG_CB_MGR_S mosq_cb_mgr;						//mqtt回调函数
	BMSER_MOSQ_MQTT_MSG_CB_MGR_NEW_S mosq_cb_mgr_new;				//新的mqtt回调函数
} BMSER_MOSQ_MQTT_NODE_S;

typedef struct
{
	int cnt;														//mqtt连接个数
	bool init;														//mqtt是否初始化
	pthread_mutex_t mutex;											//mqtt数据锁
	BMSER_MOSQ_MQTT_NODE_S nodes[MAX_MOSQ_MQTT_CNT];				//mqtt节点数据
} BMSER_MOSQ_MQTT_NODE_MGR_S;

static BMSER_MOSQ_MQTT_NODE_MGR_S g_mosq_mqtt_mgr = {0};

/**
 * @description: 找到mosq在数组中的位置下标 
 * @param {mosquitto*} mosq
 * @return {*} 位置
 */
static int get_index_from_mosq_list(struct mosquitto* mosq)
{
	if (NULL == mosq || 0 == g_mosq_mqtt_mgr.cnt) {
		return -1;
	}

	int i = 0;
	for (i = 0; i < g_mosq_mqtt_mgr.cnt; i++) {
		if (mosq == g_mosq_mqtt_mgr.nodes[i].mosq && true == g_mosq_mqtt_mgr.nodes[i].valid) {
			return i;
		}
	}

	return -1;
}

/**
 * @description: 初始化mqtt 
 * @return {*} 0: success, -1: fail
 */
static int bmser_mqtt_session_init()
{
	if (true == g_mosq_mqtt_mgr.init) {
		return 0;
	}

	int ret = pthread_mutex_init(&g_mosq_mqtt_mgr.mutex, NULL);
	if (0 == ret) {
		g_mosq_mqtt_mgr.init = true;
		return 0;
	}

	return -1;
}

/**
 * @description: mqtt日志回调函数
 * @param {mosquitto*} mosq
 * @param {void*} userdata
 * @param {int} level
 * @param {char*} str
 * @return {*} 0:success, others:fail.
 */
static void mqtt_log_callback(struct mosquitto* mosq, void* userdata, int level, const char* str)
{
	return;
}

/**
 * @description: mqtt连接回调函数
 * @param {mosquitto*} mosq 
 * @param {void*} userdata
 * @param {int} result
 * @return {*} 0:success, others:fail.
 */
static void mqtt_connect_callback(struct mosquitto* mosq, void* userdata, int result)
{
	if (NULL == mosq) {
		return;
	}
	pthread_mutex_lock(&g_mosq_mqtt_mgr.mutex);
	int index = get_index_from_mosq_list(mosq);
	pthread_mutex_unlock(&g_mosq_mqtt_mgr.mutex);
	if (index < 0) {
		return;
	}

	if (g_mosq_mqtt_mgr.nodes[index].mosq_cb_mgr.recv_conn_status_cb) {
		g_mosq_mqtt_mgr.nodes[index].mosq_cb_mgr.recv_conn_status_cb(result);
	}

	if (g_mosq_mqtt_mgr.nodes[index].mosq_cb_mgr_new.recv_conn_status_cb) {
		g_mosq_mqtt_mgr.nodes[index].mosq_cb_mgr_new.recv_conn_status_cb(mosq, result);
	}

	return;
}

/**
 * @description: 消息回调函数，收到订阅的消息后调用
 * @param {mosquitto*} mosq
 * @param {void*} userdata
 * @param {mosquitto_message*} message
 * @return {*}
 */
static void mqtt_message_callback(struct mosquitto* mosq, void* userdata, const struct mosquitto_message* message)
{
	if (NULL == mosq || NULL == message) {
		return;
	}
	pthread_mutex_lock(&g_mosq_mqtt_mgr.mutex);
	int index = get_index_from_mosq_list(mosq);
	pthread_mutex_unlock(&g_mosq_mqtt_mgr.mutex);
	if (index < 0) {
		return;
	}

	if (g_mosq_mqtt_mgr.nodes[index].mosq_cb_mgr.recv_msg_cb) {
		g_mosq_mqtt_mgr.nodes[index].mosq_cb_mgr.recv_msg_cb(userdata, message);
	}

	if (g_mosq_mqtt_mgr.nodes[index].mosq_cb_mgr_new.recv_msg_cb) {
		g_mosq_mqtt_mgr.nodes[index].mosq_cb_mgr_new.recv_msg_cb(mosq, userdata, message);
	}

	return;
}

/**
 * @description: 订阅消息回调函数
 * @param {mosquitto*} mosq
 * @param {void*} userdata
 * @param {int} mid
 * @param {int} qos_count
 * @param {int*} granted_qos
 * @return {*}
 */
static void mqtt_subscribe_callback(struct mosquitto* mosq, void* userdata, int mid, int qos_count, const int* granted_qos)
{
	return;
}

#ifdef WIN32
static void* mosquitto__thread_main(void* obj)
{
	struct mosquitto* mosq = (struct mosquitto*)obj;
	mosquitto_loop_forever(mosq, 1000 * KEEP_ALIVE, 1);
	pthread_exit(NULL);
}
#endif

/**
 * @description: 创建mqtt会话 
 * @param {char*} id
 * @param {char*} ip
 * @param {int} port
 * @param {BMSER_MOSQ_MQTT_RECV_MSG_CB} msg_recv_cb
 * @return {*} 非空句柄:success, NULL:fail.
 */
struct mosquitto* bmser_mqtt_session_create(const char* id, const char* ip, int port, BMSER_MOSQ_MQTT_RECV_MSG_CB msg_recv_cb, BMSER_MOSQ_MQTT_CONN_STATUS_CB msg_conn_cb)
{
	if (NULL == ip) {
		dy_syslog(LOG_ERR, "para is null.");
		return NULL;
	}

	if (!g_mosq_mqtt_mgr.init) {
		bmser_mqtt_session_init();
	}

	if (!g_mosq_mqtt_mgr.init || g_mosq_mqtt_mgr.cnt >= MAX_MOSQ_MQTT_CNT) {
		return NULL;
	}

	int rc = 0, i = 0;
	struct mosquitto* mosq = NULL;

	rc = mosquitto_lib_init();
	if (MOSQ_ERR_SUCCESS != rc) {
		return NULL;
	}

	//创建mqtt new session
	mosq = mosquitto_new(id, true, NULL);
	if (NULL == mosq) {
		mosquitto_lib_cleanup();
		return NULL;
	}

	//设置mqtt回调函数
	mosquitto_log_callback_set(mosq, mqtt_log_callback);
	mosquitto_connect_callback_set(mosq, mqtt_connect_callback);
	mosquitto_message_callback_set(mosq, mqtt_message_callback);
	mosquitto_subscribe_callback_set(mosq, mqtt_subscribe_callback);

	//mqtt连接
	rc = mosquitto_connect(mosq, ip, port <= 0 ? MQTT_CONNECT_PORT : port, MQTT_KEEP_ALIVE);
	if (rc != MOSQ_ERR_SUCCESS) {
		mosquitto_destroy(mosq);
		mosquitto_lib_cleanup();
		return NULL;
	}

	pthread_mutex_lock(&g_mosq_mqtt_mgr.mutex);
	//保存mosq句柄到数组中
	for (i = 0; i < MAX_MOSQ_MQTT_CNT; i++) {
		if (false == g_mosq_mqtt_mgr.nodes[i].valid) {
			break;
		}
	}

	//判断是否越界
	if (MAX_MOSQ_MQTT_CNT == i) {
		pthread_mutex_unlock(&g_mosq_mqtt_mgr.mutex);
		mosquitto_disconnect(mosq);
		mosquitto_destroy(mosq);
		mosquitto_lib_cleanup();
		return NULL;
	}

	g_mosq_mqtt_mgr.nodes[i].mosq = mosq;
	g_mosq_mqtt_mgr.nodes[i].valid = true;

	//这里直接指向，没有使用深拷贝
	g_mosq_mqtt_mgr.nodes[i].mosq_cb_mgr.recv_msg_cb = msg_recv_cb;
	g_mosq_mqtt_mgr.nodes[i].mosq_cb_mgr.recv_conn_status_cb = msg_conn_cb;
	g_mosq_mqtt_mgr.cnt++;
	pthread_mutex_unlock(&g_mosq_mqtt_mgr.mutex);

#ifdef WIN32
	//自带自动重连功能
	pthread_t thread_id;
	if (0 != pthread_create(&thread_id, NULL, mosquitto__thread_main, mosq)) {
		mosquitto_disconnect(mosq);
		mosquitto_destroy(mosq);
		mosquitto_lib_cleanup();
		return NULL;
	}
    //rc = mosquitto_loop_forever(mosq, 1000 * KEEP_ALIVE, 1);
#else
	//开启线程loop
	rc = mosquitto_loop_start(mosq);
	if (rc != MOSQ_ERR_SUCCESS) {
		mosquitto_disconnect(mosq);
		mosquitto_destroy(mosq);
		mosquitto_lib_cleanup();
		return NULL;
	}
#endif // WIN32
	
	return mosq;
}


/**
 * @description: 停止mqtt会话
 * @return {*} 0:success, others:fail.
 */
int bmser_mqtt_session_stop(struct mosquitto* mosq)
{
	if (NULL == mosq) {
		return OPRT_PARA_NULL;
	}
	
	pthread_mutex_lock(&g_mosq_mqtt_mgr.mutex);
	int index = get_index_from_mosq_list(mosq);
	if (index >= 0 && g_mosq_mqtt_mgr.cnt > 0) {
		g_mosq_mqtt_mgr.nodes[index].mosq = NULL;
		g_mosq_mqtt_mgr.nodes[index].valid = false;
		g_mosq_mqtt_mgr.cnt--;
	}
	pthread_mutex_unlock(&g_mosq_mqtt_mgr.mutex);

	mosquitto_loop_stop(mosq, true);
	
	mosquitto_disconnect(mosq);
	
	mosquitto_destroy(mosq);

	mosquitto_lib_cleanup();

	return 0;
}

/**
 * @description: 订阅mqtt topic
 * @param {char*} topic
 * @return {*} 0:success, others:fail.(注意错误返回值可能大于0)
 */
int bmser_mqtt_session_subscribe(struct mosquitto* mosq, const char* topic)
{
	if (NULL == mosq || NULL == topic) {
		return OPRT_PARA_NULL;
	}

	int ret = 0, i = 0, retry_cnt = 5; //增加订阅失败重试，默认重试5次
	for (i = 0; i < retry_cnt; i++) {
		ret = mosquitto_subscribe(mosq, NULL, topic, MQTT_QOS_LEVEL2);
		if (0 != ret) {
#ifdef WIN32
			uv_sleep(1); //延时1ms
#else
			usleep(1000); //延时1ms
#endif
		} else {
			return ret;
		}
	}
	dy_syslog(LOG_ERR, "subscribe fail, topic = %s, ret = %d.", topic, ret);
	return ret;
}

/**
 * @description: 发布mqtt消息，新业务使用bmser_mqtt_session_publish_raw_new接口
 * @param {mosquitto*} mosq
 * @param {char*} topic: MQTT主题
 * @param {void*} payload: MQTT payload
 * @param {int} len: MQTT payload 长度
 * @return {*} 0:success, others:fail. (注意错误返回值可能大于0)
 */
int bmser_mqtt_session_publish_raw(struct mosquitto* mosq, const char* topic, const void* payload, int len)
{
	if (NULL == mosq || NULL == topic || NULL == payload || len < 0) {
		dy_syslog(LOG_ERR, "para is not valid, mosq:%p, topic:%p, payload:%p, len:%d.", mosq, topic, payload, len);
		return OPRT_PARA_NULL;
	}

	return mosquitto_publish(mosq, NULL, topic, len, payload, MQTT_QOS_LEVEL2, 0);
}

/**
 * @description: 发布mqtt消息，新业务使用此接口，替换bmser_mqtt_session_publish_raw接口，增加了QOS级别和retain传参
 * @param {mosquitto*} mosq
 * @param {char*} topic: MQTT主题
 * @param {void*} payload: MQTT payload
 * @param {int} len: MQTT payload 长度
 * @param {int} qos: MQTT QOS
 * @param {bool} retain: 服务器是否保留此消息，当前业务默认传false
 * @return {*} 0:success, others:fail. (注意错误返回值可能大于0)
 */
int bmser_mqtt_session_publish_raw_new(struct mosquitto* mosq, const char* topic, const void* payload, int len, int qos, bool retain)
{
	if (NULL == mosq || NULL == topic || NULL == payload || len < 0) {
		dy_syslog(LOG_ERR, "para is not valid, mosq:%p, topic:%p, payload:%p, len:%d.", mosq, topic, payload, len);
		return OPRT_PARA_NULL;
	}

	return mosquitto_publish(mosq, NULL, topic, len, payload, qos, retain);
}

/**
 * @description: mqtt会话重连
 * @param {mosquitto*} mosq
 * @return {*} 0:success, others:fail.(注意错误返回值可能大于0)
 */
int bmser_mqtt_session_reconnect(struct mosquitto* mosq)
{
	if (NULL == mosq) {
		return OPRT_PARA_NULL;
	}
	
	return mosquitto_reconnect(mosq);
}

static int32_t set_new_callback(struct mosquitto* mosq, BMSER_MOSQ_MQTT_MSG_CB_MGR_NEW_S *mgr_cb)
{
	if (!mgr_cb) {
		dy_syslog(LOG_WARNING, "maybe not nedd callback.");
		return OPRT_OK;
	}

	int32_t index = 0;
	
	pthread_mutex_lock(&g_mosq_mqtt_mgr.mutex);
	index = get_index_from_mosq_list(mosq);
	if (index < 0) {
		pthread_mutex_unlock(&g_mosq_mqtt_mgr.mutex);
		return OPRT_NOT_FOUND;
	}
	memcpy(&g_mosq_mqtt_mgr.nodes[index].mosq_cb_mgr_new, mgr_cb, sizeof(BMSER_MOSQ_MQTT_MSG_CB_MGR_NEW_S));
	pthread_mutex_unlock(&g_mosq_mqtt_mgr.mutex);

	return OPRT_OK;
}

/**
 * @description: 创建mqtt会话新 新接口修改了传参方式，将回调函数封装成结构体传入，且回调函数中增加mosquitto句柄，可以通过回调函数知道是哪个连接
 * @param {char*} id
 * @param {char*} ip
 * @param {int} port
 * @param {BMSER_MOSQ_MQTT_MSG_CB_MGR_NEW_S} mgr_cb
 * @return {*} 非空句柄:success, NULL:fail.
 */
struct mosquitto* bmser_mqtt_session_create_new(const char* id, const char* ip, int port, BMSER_MOSQ_MQTT_MSG_CB_MGR_NEW_S *mgr_cb)
{
	if (!ip || port <= 0) {
		return NULL;
	}

	int32_t ret = 0;

	struct mosquitto* mosq = bmser_mqtt_session_create(id, ip, port, NULL, NULL);
	if (!mosq) {
		return NULL;
	}

	ret = set_new_callback(mosq, mgr_cb);
	if (0 != ret) {
		bmser_mqtt_session_stop(mosq);
		return NULL;		
	}

	return mosq;
}