#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include "mqtt_api.h"
#include "com_api.h"
#include "kk_product.h"
#include "kk_hal.h"
#include "cJSON.h"
static const char* OPT_SEND = "MQTTAsync_sendMessage";
static const char* OPT_SUB = "MQTTAsync_subscribe";
static const char* OPT_UNSUB = "MQTTAsync_unsubscribe";
static MQTTAsync s_Client;
static int s_cloudStatus = DEVICE_OFFLINE;

static char *kk_will_message_create(char *ccuid,cJSON*payload)
{
	char msgId[64] = {0};
	char timestamp[16] = {0};
	sprintf(msgId,"6c0645c4-f6d6-b11e-%lld",HAL_GetTime_ms());
	sprintf(timestamp,"%lld",HAL_GetTime_ms());
	cJSON_AddStringToObject(payload, "msgId", msgId);
	cJSON_AddStringToObject(payload, "version", "1.0");
	cJSON_AddStringToObject(payload, "time", timestamp);
	cJSON_AddStringToObject(payload, "method", "thing.status.offline");
	cJSON* param = cJSON_CreateObject();
	cJSON_AddStringToObject(param, "deviceCode", ccuid);	
	cJSON_AddStringToObject(param, "ccuCode", ccuid);
	cJSON_AddItemToObject(payload, "params", param);
	return cJSON_Print(payload);
}
int KK_Send_CloudState(int state);
int kk_get_cloud_status(void){
	return s_cloudStatus;
}
void kk_write_disconnect_flag(void)
{
    if((access("/tmp/CLOUD_STATUS_FAIL",F_OK)) != 0){
        system("touch /tmp/CLOUD_STATUS_FAIL");
    }
}
void kk_remove_disconnect_flag(void){
	if((access("/tmp/CLOUD_STATUS_FAIL",F_OK)) == 0){
        system("rm /tmp/CLOUD_STATUS_FAIL");
    }
}
/*Connect lost callback*/
static void connlost(void *context, char *cause)
{
	if(context == NULL || cause == NULL)
	{
		ERROR_PRINT("PARAM ERROR\n");
		s_cloudStatus = DEVICE_OFFLINE;
		kk_write_disconnect_flag();
		KK_Send_CloudState(s_cloudStatus);
		return;
	}
	char topicBuf[256] = {0};
	char *message = NULL;	
	MQTTAsync client = (MQTTAsync)context;
	MQTTAsync_connectOptions conn_opts = MQTTAsync_connectOptions_initializer;
	MQTTAsync_willOptions will_opts = MQTTAsync_willOptions_initializer;
	int rc;
	char ccuid[32] = {0};
	char pid[32] = {0};
	KK_Get_ccuid(ccuid);	
	kk_cloud_get_pid(pid);
	sprintf(topicBuf, "biz/newkk/%s/%s/thing/status/offline",pid, ccuid);
	INFO_PRINT("\nConnection lost\n");
	INFO_PRINT("cause: %s\n", cause);
	s_cloudStatus = DEVICE_OFFLINE;
	kk_write_disconnect_flag();
	KK_Send_CloudState(s_cloudStatus);
	cJSON* willPayload = cJSON_CreateObject();
	message = kk_will_message_create(ccuid,willPayload);
	conn_opts.keepAliveInterval = 20;
	conn_opts.cleansession = 0;
	will_opts.retained = 0;  //retained = 1 时， broker会一直保留消息，这里不需要，使用默认的0就行
    will_opts.topicName = topicBuf;
    will_opts.message = message;
    conn_opts.will = &will_opts;	
	if ((rc = MQTTAsync_connect(client, &conn_opts)) != MQTTASYNC_SUCCESS)
	{
		ERROR_PRINT("Failed to start connect, return code %d\n", rc);
	}
	free(message);
	cJSON_Delete(willPayload);
}

void onDisconnectFailure(void* context, MQTTAsync_failureData* response)
{
	WARNING_PRINT("Disconnect failed\n");
}

void onDisconnect(void* context, MQTTAsync_successData* response)
{
	WARNING_PRINT("Successful disconnection\n");
}

void onSendFailure(void* context, MQTTAsync_failureData* response)
{
	if(context == NULL || response == NULL)
	{
		ERROR_PRINT("PARAM ERROR\n");
		return;
	}

	MQTTAsync client = (MQTTAsync)context;
	MQTTAsync_disconnectOptions opts = MQTTAsync_disconnectOptions_initializer;
	int rc;

	WARNING_PRINT("Message send failed token %d error code %d\n", response->token, response->code);
	opts.onSuccess = onDisconnect;
	opts.onFailure = onDisconnectFailure;
	opts.context = client;
	if ((rc = MQTTAsync_disconnect(client, &opts)) != MQTTASYNC_SUCCESS)
	{
		ERROR_PRINT("Failed to start disconnect, return code %d\n", rc);
	}
}

void onSend(void* context, MQTTAsync_successData* response)
{
	if(response == NULL)
	{
		ERROR_PRINT("PARAM ERROR\n");
		return;
	}
	INFO_PRINT("Message with token value %d delivery confirmed\n", response->token);
}

extern void KK_reset_sub_flag(void);
void onConnectFailure(void* context, MQTTAsync_failureData* response)
{
	if(response == NULL)
	{
		ERROR_PRINT("PARAM ERROR\n");
		return;
	}
	KK_reset_sub_flag();
	WARNING_PRINT("Connect failed, rc %d\n", response ? response->code : 0);
}

void onConnect(void* context, MQTTAsync_successData* response)
{
	INFO_PRINT("-----------Successful connection\n");
	kk_remove_disconnect_flag();
	if(s_cloudStatus != DEVICE_ONLINE){
		s_cloudStatus = DEVICE_ONLINE;
		
		KK_Send_CloudState(s_cloudStatus);
	}	
}


int messageArrived(void* context, char* topicName, int topicLen, MQTTAsync_message* message)
{
	/* not expecting any messages */
	//INFO_PRINT("onMessageArrived topic:%s,message length:%d.\n",topicName,message->payloadlen);
	//INFO_PRINT("payload:%s,\n",message->payload);
	KK_MQTT_RecvMsg(topicName,message->payload);
	MQTTAsync_freeMessage(&message);
	MQTTAsync_free(topicName);
	return 1;
}

static void mqttTraceCallback(enum MQTTASYNC_TRACE_LEVELS level, char *message)
{
	//printf("mqttTraceCallback level:%d,msg:%s.\n",level,message);
}
static void onDeliveryComplete(void* context, MQTTAsync_token token)
{
	//INFO_PRINT("onDeliveryComplete,token:%d \n",token);
}
static void onConnectBuild(void *context, char *cause)
{
	INFO_PRINT("----onConnectBuild:%s \n",cause);
	//rc = KK_Client_Gateway_Subscribe();
	//if(rc != 0)
	//{
		//ERROR_PRINT("KK_MQTT_SubTopic ERROR rc = %d\n",rc);
	//}
	kk_remove_disconnect_flag();
	if(s_cloudStatus != DEVICE_ONLINE){
		s_cloudStatus = DEVICE_ONLINE;
		INFO_PRINT("--11--onConnectBuild:%s \n",cause);
		KK_Send_CloudState(s_cloudStatus);
	}
}
static void onDisConnected(void *context, MQTTProperties* properties,enum MQTTReasonCodes reasonCode)
{
	INFO_PRINT("onDisConnected,maybe kicked by broker.\n");
}

static void onOptSuccess(void* context, MQTTAsync_successData* response)
{
	if(strcmp((char *)context,OPT_SEND)==0)
	{
		INFO_PRINT("MQTTAsync_sendMessage success,return token:%d,msg length:%d \n",
			response->token,response->alt.pub.message.payloadlen);
	}
	else if(strcmp((char *)context,OPT_SUB)==0)
	{
		INFO_PRINT("MQTTAsync_subscribe success,return token:%d \n",response->token);
	}
	else if(strcmp((char *)context,OPT_UNSUB)==0)
	{
		INFO_PRINT("MQTTAsync_unsubscribe success,return token:%d \n",response->token);
	}
}
static void onOptFail(void* context,  MQTTAsync_failureData* response)
{
	if(strcmp((char *)context,OPT_SEND)==0)
	{
		WARNING_PRINT("MQTTAsync_sendMessage fail,token:%d,code:%d,msg:%s \n",
					response->token,response->code,response->message);
	}
	else if(strcmp((char *)context,OPT_SUB)==0)
	{
		WARNING_PRINT("MQTTAsync_subscribe fail,return token:%d \n",response->token);
	}
	else if(strcmp((char *)context,OPT_UNSUB)==0)
	{
		WARNING_PRINT("MQTTAsync_unsubscribe fail,return token:%d \n",response->token);
	}
}
static void mqtt_set_callbacks(void)
{
	MQTTAsync_setConnectionLostCallback(s_Client,NULL,connlost);
	MQTTAsync_setMessageArrivedCallback(s_Client,NULL,messageArrived);
	MQTTAsync_setDeliveryCompleteCallback(s_Client,NULL,onDeliveryComplete);
	MQTTAsync_setConnected(s_Client,NULL,onConnectBuild);
	MQTTAsync_setDisconnected(s_Client,NULL,onDisConnected);
}
void KK_Get_MqttClient(MQTTAsync *pClient)
{
	if(pClient != NULL)
	{
		*pClient = s_Client;
	}
}
extern char g_clientId[64];
extern char s_ServerIp[16];

MQTTAsync KK_MQTT_Connect(void)
{
	int rc = 0;
	char temp_address[128] = {0};
	FILE *fp; 	
	char topicBuf[256] = {0};
	char *message = NULL;
	
	MQTTAsync_createOptions opts  = MQTTAsync_createOptions_initializer;
	MQTTAsync_connectOptions conn_opts = MQTTAsync_connectOptions_initializer;
	MQTTAsync_willOptions will_opts = MQTTAsync_willOptions_initializer;
	MQTTAsync_setTraceCallback(mqttTraceCallback);
	opts.MQTTVersion = MQTTVERSION_3_1_1;
	char token[512] = {0};
	char usrname[128] = {0};
	fp = fopen(KK_TOKEN_PATH, "r");
	if(fp != NULL){
		fread(token,1,sizeof(token),fp);
		fclose(fp);
	}	
	char ccuid[32] = {0};
	char pid[32] = {0};
	KK_Get_ccuid(ccuid);	
	kk_cloud_get_pid(pid);
	sprintf(usrname, "%s.%s", ccuid,pid);
	sprintf(topicBuf, "biz/newkk/%s/%s/thing/status/offline",pid, ccuid);
	
	INFO_PRINT("topicBuf:%s\n",topicBuf);
	INFO_PRINT("cliendid:%s,usrname:%s\n",g_clientId,usrname);
	INFO_PRINT("------------>token:%s\n",token);
	sprintf(temp_address,"tcp://%s:1983",s_ServerIp);
	INFO_PRINT("------------>temp_address:%s\n",temp_address);
	if ((rc = MQTTAsync_createWithOptions(&s_Client, temp_address, g_clientId, MQTTCLIENT_PERSISTENCE_NONE, NULL,&opts)) != MQTTASYNC_SUCCESS)
	{
		ERROR_PRINT("Failed to create client object, return code %d\n", rc);
		return NULL;
	}
	/*Set the mqtt callback*/
	mqtt_set_callbacks();
	cJSON* willPayload = cJSON_CreateObject();
	message = kk_will_message_create(ccuid,willPayload);
	conn_opts.keepAliveInterval = 30;
	conn_opts.connectTimeout = CONNECT_TIMEOUT;
	conn_opts.automaticReconnect = AUTO_CONN;
	conn_opts.minRetryInterval = 1;
	conn_opts.maxRetryInterval = 32;
	//conn_opts.username = USRNAME;
	//conn_opts.password = PASSWORD;	
	conn_opts.username = usrname;
	conn_opts.password = token;
	conn_opts.cleansession = 0;
	conn_opts.onSuccess = onConnect;
	conn_opts.onFailure = onConnectFailure;
	conn_opts.context = s_Client;
	will_opts.retained = 0;  //retained = 1 时， broker会一直保留消息，这里不需要，使用默认的0就行
    will_opts.topicName = topicBuf;
    will_opts.message = message;
    conn_opts.will = &will_opts;
	if ((rc = MQTTAsync_connect(s_Client, &conn_opts)) != MQTTASYNC_SUCCESS)
	{
		ERROR_PRINT("Failed to start connect, return code %d\n", rc);
		free(message);
		cJSON_Delete(willPayload);
		return NULL;
	}
	cJSON_Delete(willPayload);
	free(message);
	return s_Client;
}
MQTTAsync KK_MQTT_Connect_ex(char *usrname,char*pwd,char*cliendId,char*host,char*port)
{
	int rc = 0;
	char temp_address[256] = {0};
	FILE *fp; 	
	char topicBuf[256] = {0};
	char *message = NULL;
	char ccuid[32] = {0};
	char pid[32] = {0};
	KK_Get_ccuid(ccuid);	
	kk_cloud_get_pid(pid);
	sprintf(topicBuf, "biz/newkk/%s/%s/thing/status/offline",pid, ccuid);	
	INFO_PRINT("topicBuf:%s\n",topicBuf);	
	MQTTAsync_createOptions opts  = MQTTAsync_createOptions_initializer;
	MQTTAsync_connectOptions conn_opts = MQTTAsync_connectOptions_initializer;
	MQTTAsync_willOptions will_opts = MQTTAsync_willOptions_initializer;	
	MQTTAsync_setTraceCallback(mqttTraceCallback);
	opts.MQTTVersion = MQTTVERSION_3_1_1;
	sprintf(temp_address,"tcp://%s:%s",host,port);
	INFO_PRINT("------------>temp_address:%s\n",temp_address);
	if ((rc = MQTTAsync_createWithOptions(&s_Client, temp_address, cliendId, MQTTCLIENT_PERSISTENCE_NONE, NULL,&opts)) != MQTTASYNC_SUCCESS)
	{
		ERROR_PRINT("Failed to create client object, return code %d\n", rc);
		return NULL;
	}
	/*Set the mqtt callback*/
	mqtt_set_callbacks();
	cJSON* willPayload = cJSON_CreateObject();
	message = kk_will_message_create(ccuid,willPayload);
	INFO_PRINT("will message:%s\n",message);
	conn_opts.keepAliveInterval = 30;
	conn_opts.connectTimeout = CONNECT_TIMEOUT;
	conn_opts.automaticReconnect = AUTO_CONN;
	conn_opts.minRetryInterval = 1;
	conn_opts.maxRetryInterval = 32;
	//conn_opts.username = USRNAME;
	//conn_opts.password = PASSWORD;	
	conn_opts.username = usrname;
	conn_opts.password = pwd;
	conn_opts.cleansession = 0;
	conn_opts.onSuccess = onConnect;
	conn_opts.onFailure = onConnectFailure;
	conn_opts.context = s_Client;
	will_opts.retained = 0;  //retained = 1 时， broker会一直保留消息，这里不需要，使用默认的0就行
    will_opts.topicName = topicBuf;
    will_opts.message = message;
    conn_opts.will = &will_opts;	
	if ((rc = MQTTAsync_connect(s_Client, &conn_opts)) != MQTTASYNC_SUCCESS)
	{
		ERROR_PRINT("Failed to start connect, return code %d\n", rc);
		free(message);
		cJSON_Delete(willPayload);		
		return NULL;
	}
	free(message);
	cJSON_Delete(willPayload);	
	return s_Client;
}
int KK_MQTT_SubTopic(char *topicName)
{
	INFO_PRINT("to subtopic:%s \n",topicName);

	MQTTAsync_responseOptions opts = MQTTAsync_responseOptions_initializer;
	int rc;
	opts.onSuccess = onOptSuccess;
	opts.onFailure = onOptFail;
	opts.context = (void*)OPT_SUB;

	if(s_cloudStatus == DEVICE_OFFLINE){
		ERROR_PRINT("Failed to KK_MQTT_SubTopic, s_cloudStatus == DEVICE_OFFLINE\n");
		return -1;
	}
	if ((rc = MQTTAsync_subscribe(s_Client,topicName, QOS, &opts)) != MQTTASYNC_SUCCESS)
	{
		ERROR_PRINT("Failed to start subscribe, return code:%d.\n", rc);
		return -1;
	}
	return 0;
}

int KK_MQTT_SendMsg(char *topicName,const char *payload)
{
	MQTTAsync_responseOptions opts = MQTTAsync_responseOptions_initializer;
	MQTTAsync_message pubmsg = MQTTAsync_message_initializer;
	int rc;
	cJSON_Minify((char*)payload);
	INFO_PRINT("mqtt send payload :%s.\n",payload);
	opts.onSuccess = onOptSuccess;
	opts.onFailure = onOptFail;
	opts.context =  (void*)OPT_SEND;

	pubmsg.payload = (void*)payload;
	pubmsg.payloadlen = strlen(payload);
	pubmsg.qos = QOS;
	pubmsg.retained = 0;

	if(s_cloudStatus == DEVICE_OFFLINE){
		ERROR_PRINT("Failed to start sendMessage, s_cloudStatus == DEVICE_OFFLINE\n");
		return -1;
	}
	if ((rc = MQTTAsync_sendMessage(s_Client, topicName, &pubmsg, &opts)) != MQTTASYNC_SUCCESS)
	{
		ERROR_PRINT("Failed to start sendMessage, return code:%d.\n", rc);
		return -1;
	}
	return rc;
}
extern void KK_Sendto_DevData(const char *topic,const char *data);
int KK_MQTT_RecvMsg(const char *topicName,const char *payload)
{
	if(topicName == NULL || payload ==NULL)
	{
		ERROR_PRINT("PARAM ERROR\n");
		return -1;
	}
	KK_Sendto_DevData(topicName,payload);
	return 0;
}

int KK_MQTT_UnsubTopic(const char *topicName)
{
	INFO_PRINT("to unsubtopic:%s \n",topicName);

	MQTTAsync_responseOptions opts = MQTTAsync_responseOptions_initializer;
	int rc;

	opts.onSuccess = onOptSuccess;
	opts.onFailure = onOptFail;
	opts.context = (void*)OPT_UNSUB;

	if(s_cloudStatus == DEVICE_OFFLINE){
		ERROR_PRINT("Failed to KK_MQTT_UnsubTopic, s_cloudStatus == DEVICE_OFFLINE\n");
		return -1;
	}
	if ((rc = MQTTAsync_unsubscribe(s_Client,topicName,&opts)) != MQTTASYNC_SUCCESS)
	{
		ERROR_PRINT("Failed to start unubscribe, return code:%d.\n", rc);
		return -1;
	}
	return rc;
}



