#include "bsp_net_mqtt_hzluat.h"

#include "sr_config.h"
#include "sr_log.h"
#include "stddef.h"
#include "sr_osal.h"


//原厂头文件
#include "luat_network_adapter.h"
#include "libemqtt.h"
#include "luat_mqtt.h"




#define MQTT_DEMO_SSL 			0
#define MQTT_DEMO_AUTOCON 		0


static void PublishCallbackDefault(SR_MQTT_CLIENT_T *mqttc, SR_MQTT_MSG_T *mqttMsg)
{

  BSP_LOG(DEBUG,"[CB]recv:pub_topic: %.*s\n",mqttMsg->topicLen,mqttMsg->topic);
  BSP_LOG(DEBUG,"pub_msg: %.*s\n",mqttMsg->payloadLen,mqttMsg->payload);

}

	


static void luat_mqtt_cb_default(luat_mqtt_ctrl_t *luat_mqtt_ctrl, uint16_t event)
{
	int i=0;

	SR_MQTT_MSG_T mqttMsg ={0};
	SR_MQTT_CLIENT_T *mqttc = (SR_MQTT_CLIENT_T *)luat_mqtt_ctrl->userdata;  //luat_mqtt_ctrl->userdata  看有没有被使用（没有，可以用来传递mqttc）

    uint8_t needSendQueue =0;

	BSP_LOG(DEBUG,"[mqtt] BSP,luat cb,event=%d,mqttc=0x%x\n",event,mqttc);
	
	switch (event)
	{
	case MQTT_MSG_CONNACK:{
		BSP_LOG(DEBUG,"[mqtt] BSP,CONNACK OK\n");
		  
		mqttMsg.msgType = SR_MQTT_MSG_TYPE_CONNACK;
        needSendQueue =1;  
	
		if(mqttc && mqttc->eventCallback)  mqttc->eventCallback(mqttc,SR_MQTT_EVENT_CONN_OK,SR_NULL);//CONN_OK事件回调
			
		break;
	}
	case MQTT_MSG_PUBLISH : 
	{
        char topic[MQTT_TOPIC_MAX_LEN]={0};
		mqttMsg.msgType = SR_MQTT_MSG_TYPE_PUBLISH;

		mqttMsg.topicLen = mqtt_parse_pub_topic_ptr(luat_mqtt_ctrl->mqtt_packet_buffer, &mqttMsg.topic);
		BSP_LOG(DEBUG,"[mqtt]pub_topic: %.*s\n",mqttMsg.topicLen,mqttMsg.topic);
		mqttMsg.payloadLen = mqtt_parse_pub_msg_ptr(luat_mqtt_ctrl->mqtt_packet_buffer, &mqttMsg.payload);
		BSP_LOG(DEBUG,"pub_msg: %.*s\n",mqttMsg.payloadLen,mqttMsg.payload);
		mqttMsg.packetId = mqtt_parse_msg_id(luat_mqtt_ctrl->mqtt_packet_buffer);


		if(mqttc && mqttc->eventCallback)  mqttc->eventCallback(mqttc,SR_MQTT_EVENT_RECV_PUBLISH,&mqttMsg);//下行PUBLISH事件回调
		if((mqttc->tableSize>0) && (mqttc->table !=SR_NULL)) //遍历订阅列表，执行订阅处理函数
        {           
		   memcpy(topic,mqttMsg.topic,mqttMsg.topicLen);
		   topic[mqttMsg.topicLen]=0;
		   
		   for(i=0;i<mqttc->tableSize;i++)
		   {
		   	  
              if(SR_NULL != strstr(topic,mqttc->table[i].topicFilter))
              {
                 mqttc->table[i].topicCb(mqttc,&mqttMsg);
                 break;
			  }
		   }
	   
		}


		break;
	}
	case MQTT_MSG_PUBACK :  
	case MQTT_MSG_PUBCOMP : 
	   {
		 BSP_LOG(DEBUG,"[mqtt] PUBACK/PUBCOMP msg_id: %d",mqtt_parse_msg_id(luat_mqtt_ctrl->mqtt_packet_buffer));
		 break;
	   }
	case MQTT_MSG_SUBACK :
		{
#if  0
		  //要改luat_mqtt_client.c的luat_mqtt_msg_cb改，SDK底层没通知	  
		  mqttMsg.msgType = SR_MQTT_MSG_TYPE_SUBACK;
		  mqttMsg.packetId = mqtt_parse_msg_id(luat_mqtt_ctrl->mqtt_packet_buffer);
		  BSP_LOG(DEBUG,"[mqtt] SUBACK,pkgID=%d\n",mqttMsg.packetId);
#endif
		  break;
	    }
	case MQTT_MSG_UNSUBACK : 
		{
#if  0
			//要改luat_mqtt_client.c的luat_mqtt_msg_cb改，SDK底层没通知
			mqttMsg.msgType = SR_MQTT_MSG_TYPE_UNSUBACK;
		    mqttMsg.packetId = mqtt_parse_msg_id(luat_mqtt_ctrl->mqtt_packet_buffer);
		    BSP_LOG(DEBUG,"[mqtt] SUBACK,pkgID=%d\n",mqttMsg.packetId);
#endif
			break;
	    }
	case MQTT_MSG_RELEASE : {
		BSP_LOG(DEBUG,"[mqtt]luat_mqtt_cb mqtt release");
		break;
	}
	case MQTT_MSG_DISCONNECT : { // mqtt 断开(只要有断开就会上报,无论是否重连)
		BSP_LOG(DEBUG,"[mqtt]luat_mqtt_cb mqtt disconnect");
	
		if(mqttc && mqttc->eventCallback)  mqttc->eventCallback(mqttc,SR_MQTT_EVENT_DISCONNECT,SR_NULL);//DISCONNECT事件回调
	
		break;
	}
	case MQTT_MSG_TIMER_PING : {
		luat_mqtt_ping(luat_mqtt_ctrl);
		break;
	}
	case MQTT_MSG_RECONNECT : {
		if (MQTT_DEMO_AUTOCON == 1)
		{
			luat_mqtt_reconnect(luat_mqtt_ctrl);
		}
		break;
	}
	case MQTT_MSG_CLOSE : { // mqtt 关闭(不会再重连)  注意：一定注意和MQTT_MSG_DISCONNECT区别，如果要做手动重连处理推荐在这里 */
		BSP_LOG(DEBUG,"[mqtt]luat_mqtt_cb mqtt close");
		#if 0
		if (MQTT_DEMO_AUTOCON == 0){
	        ret = luat_mqtt_connect(luat_mqtt_ctrl);
			if (ret) {
				LUAT_DEBUG_PRINT("mqtt connect ret=%d\n", ret);
				luat_mqtt_close_socket(luat_mqtt_ctrl);
				return;
			}
         }
		#endif
		
		break;
	}
	default:
		break;
	}

    //luat_mqtt_ctrl->userdata  看有没有被原厂使用（没有，可以用来传递mqttc）
    if(needSendQueue == 1)
         luat_rtos_queue_send(mqttc->msgQueue, &mqttMsg, sizeof(SR_MQTT_MSG_T), 1000);
	
	return;
}


int _LuatMqttcInit(SR_MQTT_CLIENT_T *mqttc,SR_MQTT_INIT_PARAMS_T *initParams)
{	
    int ret = -1;

	ret = luat_rtos_queue_create(&mqttc->msgQueue, 3, sizeof(SR_MQTT_MSG_T));
	if(ret)
	{
        
       BSP_LOG(DEBUG,"[BSP]os queue create fail,ret=%d\n",ret);
	   return -1;
	}

	
	//创建底层平台需要的mqtt结构体
    luat_mqtt_ctrl_t *luat_mqtt_ctrl = (luat_mqtt_ctrl_t *)SR_OsMalloc(sizeof(luat_mqtt_ctrl_t));
    if(SR_NULL == luat_mqtt_ctrl)
    {
       goto _MALLOC_FAIL;
	}
	
	mqttc->thirdData = luat_mqtt_ctrl;

    ret = luat_mqtt_init(luat_mqtt_ctrl, NW_ADAPTER_INDEX_LWIP_GPRS);
	if (ret) {
		BSP_LOG(DEBUG,"[BSP]mqtt init FAIL,ret %d", ret);
		goto _INIT_FAIL;
	}
	
	luat_mqtt_ctrl->ip_addr.type = 0xff;
	if (MQTT_DEMO_AUTOCON == 1)
	{
		luat_mqtt_ctrl->reconnect = 1;
		luat_mqtt_ctrl->reconnect_time = 3000;
	}
	luat_mqtt_ctrl->broker.clean_session = 1;
	luat_mqtt_ctrl->keepalive = 60;
	luat_mqtt_set_cb(luat_mqtt_ctrl,luat_mqtt_cb_default);  //注册原厂SDK mqtt通知回调

	luat_mqtt_connopts_t opts = {0};

#if (MQTT_DEMO_SSL == 1)
	opts.is_tls = 1;
	opts.server_cert = testCaCrt;
	opts.server_cert_len = strlen(testCaCrt);
	opts.client_cert = testclientCert;
	opts.client_cert_len = strlen(testclientCert);
	opts.client_key = testclientPk;
	opts.client_key_len = strlen(testclientPk);
#else
	opts.is_tls = 0;
#endif 
	opts.host = initParams->serverIP;
	opts.port = initParams->serverPort;
	ret = luat_mqtt_set_connopts(luat_mqtt_ctrl, &opts);
	if (ret) {
		BSP_LOG(DEBUG,"[BSP]mqtt set connopts fail, ret %d", ret);
		goto _SET_CONNOPTS_FAIL;
	}

	luat_mqtt_ctrl->netc->is_debug = 1;  // debug信息
	
	luat_mqtt_ctrl->userdata = mqttc;  
	
	BSP_LOG(DEBUG,"[BSP]mqtt init ok\n");	
	return 0;


_SET_CONNOPTS_FAIL:
	luat_mqtt_release_socket(luat_mqtt_ctrl);
_INIT_FAIL:   
    SR_OsFree(luat_mqtt_ctrl);
_MALLOC_FAIL:
    if(mqttc->msgQueue) luat_rtos_queue_delete(mqttc->msgQueue);
    return ret;

}

int _LuatMqttcDeinit(SR_MQTT_CLIENT_T *mqttc)
{
    luat_mqtt_ctrl_t *luat_mqtt_ctrl = (luat_mqtt_ctrl_t *) mqttc->thirdData;
	
	if(mqttc->msgQueue)
		 luat_rtos_queue_delete(mqttc->msgQueue);
	
	BSP_LOG(DEBUG,"[BSP]mqttc_deinit queue_delete\n");

	luat_mqtt_release_socket(luat_mqtt_ctrl);
	BSP_LOG(DEBUG,"[BSP]mqttc_deinit release_socket\n");
	
	SR_OsFree(luat_mqtt_ctrl);
	BSP_LOG(DEBUG,"[BSP]mqttc_deinit free ctrl\r\n");
	
	return 0;

}


int _LuatMqttcConnect(SR_MQTT_CLIENT_T *mqttc,SR_MQTT_CONN_PARAMS_T *connParams)
{ 
    int ret = -1;
	SR_MQTT_MSG_T mqttMsg;

	luat_mqtt_ctrl_t *luat_mqtt_ctrl = ( luat_mqtt_ctrl_t *)mqttc->thirdData;

    mqtt_init(&(luat_mqtt_ctrl->broker), connParams->clientID);
	mqtt_init_auth(&(luat_mqtt_ctrl->broker), connParams->username, connParams->password);

	luat_mqtt_ctrl->broker.clean_session = connParams->cleansession;
	luat_mqtt_ctrl->keepalive = connParams->keepAlive;  //保活时间  60
	//will遗嘱设置
	if(connParams->willFlag)
	{
	  luat_mqtt_set_will(luat_mqtt_ctrl, connParams->willTopic, connParams->willMessage, strlen(connParams->willMessage), connParams->willQos, connParams->willRetained); 
	}


    
    ret = luat_mqtt_connect(luat_mqtt_ctrl);
	if (ret) {
		BSP_LOG(DEBUG,"[BSP] luat_mqtt_connect fail, ret=%d\n", ret);
		luat_mqtt_close_socket(luat_mqtt_ctrl);
		return -1;
	}


	//超时等待connect ok通知
	ret = luat_rtos_queue_recv(mqttc->msgQueue, &mqttMsg, sizeof(SR_MQTT_MSG_T), 6000);
	if(ret == 0 &&(mqttMsg.msgType == SR_MQTT_MSG_TYPE_CONNACK))
	{
       BSP_LOG(DEBUG,"[BSP] api connect ok\n");
	   return 0;
	}
	else
	{
        BSP_LOG(DEBUG,"[BSP] api wait CONNACK fail, ret=%d\n", ret);
		luat_mqtt_close_socket(luat_mqtt_ctrl);
		return -1;
	}



	
}

int _LuatMqttcDisconnect(SR_MQTT_CLIENT_T *mqttc)
{
  int ret =-1;
  luat_mqtt_ctrl_t *luat_mqtt_ctrl = ( luat_mqtt_ctrl_t *)mqttc->thirdData;

  ret = mqtt_disconnect(&(luat_mqtt_ctrl->broker));

  luat_mqtt_close_socket(luat_mqtt_ctrl);

  return ret==1?0:-1;

}


int _LuatMqttcPublish(SR_MQTT_CLIENT_T *mqttc, uint8_t qos, char *topic, char *msg,uint16_t msgLen)
{
   int ret = -1;
   uint16_t message_id;
   luat_mqtt_ctrl_t *luat_mqtt_ctrl = ( luat_mqtt_ctrl_t *)mqttc->thirdData;


   if (qos == QOS1 || qos == QOS2)
    message_id = _GetNextPacketId(mqttc);
   else
    message_id = 0;
   
   ret = mqtt_publish_with_qos(&(luat_mqtt_ctrl->broker), topic, msg, msgLen, 0, qos, &message_id);

   return ret==1?0:-1;
}


int _LuatMqttcSubscribe(SR_MQTT_CLIENT_T *mqttc, char *topic,uint8_t qos)
{
   int ret = -1;
   uint16_t message_id;
   luat_mqtt_ctrl_t *luat_mqtt_ctrl = ( luat_mqtt_ctrl_t *)mqttc->thirdData;


   message_id = _GetNextPacketId(mqttc);


   ret = mqtt_subscribe(&(luat_mqtt_ctrl->broker), topic, &message_id, qos);
   if (ret != 1) 
	{
		BSP_LOG(DEBUG,"mqtt subscribe fail, ret=%d\n", ret);
		return -1;
   }


   
   //超时等待SUBACK响应
  #if 0
	ret = luat_rtos_queue_recv(mqttc->msgQueue, &mqttMsg, sizeof(SR_MQTT_MSG_T), 3000);
	if(ret ==0 &&(mqttMsg.msgType == SR_MQTT_MSG_TYPE_SUBACK))
	{
         BSP_LOG(DEBUG,"subscribe ok\n");
	    return 0;
	}
	else
	{
        BSP_LOG(DEBUG,"wait suback fail, ret=%d\n", ret);
		return -1;
	}
  #endif

 return 0;

}												

int _LuatMqttcUnsubscribe(SR_MQTT_CLIENT_T *mqttc, char *topic)
{

  int ret = -1;
  uint16_t message_id;
  luat_mqtt_ctrl_t *luat_mqtt_ctrl = ( luat_mqtt_ctrl_t *)mqttc->thirdData;
  
  message_id = _GetNextPacketId(mqttc);

  ret =  mqtt_unsubscribe(&(luat_mqtt_ctrl->broker), topic, &message_id);
  if (ret != 1) 
	{
		BSP_LOG(DEBUG,"mqtt unsubscribe fail, ret=%d\n", ret);
		return -1;
   }

  //超时等待UNSUBACK响应
  #if 0
	ret = luat_rtos_queue_recv(mqttc->msgQueue, &mqttMsg, sizeof(SR_MQTT_MSG_T), 3000);
	if(ret ==0 &&(mqttMsg.msgType == SR_MQTT_MSG_TYPE_UNSUBACK))
	{
        BSP_LOG(DEBUG,"unsubscribe ok\n");
	    return 0;
	}
	else
	{
        BSP_LOG(DEBUG,"wait unsuback fail, ret=%d\n", ret);
		return -1;
	}
  #endif

  return 0;
}