#if 1
#include <sys/select.h>
#include <sys/time.h>
#include <sys/types.h>
#include <unistd.h>
#include <sys/syscall.h>
#include <stdlib.h>
#include <stdio.h>
#include <string.h>
#include <sys/stat.h>
#include <sys/types.h>
#include <errno.h>

#include "azure_common.h"

/* Paste in the your iothub device connection string  */
//static const char* connectionString = "HostName=LNXALLIoT.azure-devices.net;DeviceId=987654321;SharedAccessKey=re0oGuRIvVl6vpc/3i/hSbR9KWGab/2eG6jzaKXzdh8=";
static const char* connectionString = "HostName=lnxall.azure-devices.net;DeviceId=device_test;SharedAccessKey=fjGDOINKdCgi098nk5EwSC9kBLIMsrApPJt+jZxJ46Q=";
#define DOWORK_LOOP_NUM     3

azure_var_t gazure_var;

typedef struct MAKER_TAG
{
    char* makerName;
    char* style;
    int year;
} Maker;

typedef struct GEO_TAG
{
    double longitude;
    double latitude;
} Geo;

typedef struct CAR_STATE_TAG
{
    int32_t softwareVersion;        // reported property
    uint8_t reported_maxSpeed;      // reported property
    char* vanityPlate;              // reported property
} CarState;

typedef struct CAR_SETTINGS_TAG
{
    uint8_t desired_maxSpeed;       // desired property
    Geo location;                   // desired property
} CarSettings;

typedef struct CAR_TAG
{
    char* lastOilChangeDate;        // reported property
    char* changeOilReminder;        // desired property
    Maker maker;                    // reported property
    CarState state;                 // reported property
    CarSettings settings;           // desired property
} Car;

//  Converts the Car object into a JSON blob with reported properties that is ready to be sent across the wire as a twin.
static char* serializeToJson(Car* car)
{
    char* result;

    JSON_Value* root_value = json_value_init_object();
    JSON_Object* root_object = json_value_get_object(root_value);

    // Only reported properties:
    (void)json_object_set_string(root_object, "lastOilChangeDate", car->lastOilChangeDate);
    (void)json_object_dotset_string(root_object, "maker.makerName", car->maker.makerName);
    (void)json_object_dotset_string(root_object, "maker.style", car->maker.style);
    (void)json_object_dotset_number(root_object, "maker.year", car->maker.year);
    (void)json_object_dotset_number(root_object, "state.reported_maxSpeed", car->state.reported_maxSpeed);
    (void)json_object_dotset_number(root_object, "state.softwareVersion", car->state.softwareVersion);
    (void)json_object_dotset_string(root_object, "state.vanityPlate", car->state.vanityPlate);

	(void)json_object_dotset_number(root_object, "temperature", 28.8);
	(void)json_object_dotset_number(root_object, "humidity", 68.1);

    result = json_serialize_to_string(root_value);

    json_value_free(root_value);

    return result;
}

//  Converts the desired properties of the Device Twin JSON blob received from IoT Hub into a Car object.
static Car* parseFromJson(const char* json, DEVICE_TWIN_UPDATE_STATE update_state)
{
    Car* car = malloc(sizeof(Car));
    JSON_Value* root_value = NULL;
    JSON_Object* root_object = NULL;

    if (NULL == car)
    {
        dy_syslog(LOG_ERR, "ERROR: Failed to allocate memory");
    }

    else
    {
        (void)memset(car, 0, sizeof(Car));

        root_value = json_parse_string(json);
        root_object = json_value_get_object(root_value);

        // Only desired properties:
        JSON_Value* changeOilReminder;
        JSON_Value* desired_maxSpeed;
        JSON_Value* latitude;
        JSON_Value* longitude;

        if (update_state == DEVICE_TWIN_UPDATE_COMPLETE)
        {
            changeOilReminder = json_object_dotget_value(root_object, "desired.changeOilReminder");
            desired_maxSpeed = json_object_dotget_value(root_object, "desired.settings.desired_maxSpeed");
            latitude = json_object_dotget_value(root_object, "desired.settings.location.latitude");
            longitude = json_object_dotget_value(root_object, "desired.settings.location.longitude");
        }
        else
        {
            changeOilReminder = json_object_dotget_value(root_object, "changeOilReminder");
            desired_maxSpeed = json_object_dotget_value(root_object, "settings.desired_maxSpeed");
            latitude = json_object_dotget_value(root_object, "settings.location.latitude");
            longitude = json_object_dotget_value(root_object, "settings.location.longitude");
        }

        if (changeOilReminder != NULL)
        {
            const char* data = json_value_get_string(changeOilReminder);

            if (data != NULL)
            {
                car->changeOilReminder = malloc(strlen(data) + 1);
                if (NULL != car->changeOilReminder)
                {
                    (void)strcpy(car->changeOilReminder, data);
                }
            }
        }

        if (desired_maxSpeed != NULL)
        {
            car->settings.desired_maxSpeed = (uint8_t)json_value_get_number(desired_maxSpeed);
        }

        if (latitude != NULL)
        {
            car->settings.location.latitude = json_value_get_number(latitude);
        }

        if (longitude != NULL)
        {
            car->settings.location.longitude = json_value_get_number(longitude);
        }
        json_value_free(root_value);
    }

    return car;
}

static int deviceMethodCallback(const char* method_name, const unsigned char* payload, size_t size, unsigned char** response, size_t* response_size, void* userContextCallback)
{
    (void)userContextCallback;
    (void)payload;
    (void)size;

    int result;

    if (strcmp("getCarVIN", method_name) == 0)
    {
        const char deviceMethodResponse[] = "{ \"Response\": \"1HGCM82633A004352\" }";
        *response_size = sizeof(deviceMethodResponse)-1;
        *response = malloc(*response_size);
        (void)memcpy(*response, deviceMethodResponse, *response_size);
        result = 200;
    }
    else
    {
        // All other entries are ignored.
        const char deviceMethodResponse[] = "{ }";
        *response_size = sizeof(deviceMethodResponse)-1;
        *response = malloc(*response_size);
        (void)memcpy(*response, deviceMethodResponse, *response_size);
        result = -1;
    }

    return result;
}

static void getCompleteDeviceTwinOnDemandCallback(DEVICE_TWIN_UPDATE_STATE update_state, const unsigned char* payLoad, size_t size, void* userContextCallback)
{
    (void)update_state;
    (void)userContextCallback;
    dy_syslog(LOG_DEBUG, "GetTwinAsync result:\r\n%.*s", (int)size, payLoad);
}

static void deviceTwinCallback(DEVICE_TWIN_UPDATE_STATE update_state, const unsigned char* payLoad, size_t size, void* userContextCallback)
{
    (void)update_state;
    (void)size;

	azure_dev_info_t *azure_dev = (azure_dev_info_t *)userContextCallback;

	dy_syslog(LOG_DEBUG, "payLoad:%s update_state:%d", payLoad, update_state);
#if 0
    Car* oldCar = (Car*)userContextCallback;
    Car* newCar = parseFromJson((const char*)payLoad, update_state);

    if (NULL == newCar)
    {
        dy_syslog(LOG_ERR, "ERROR: parseFromJson returned NULL");
    }
    else
    {
        if (newCar->changeOilReminder != NULL)
        {
            if ((oldCar->changeOilReminder != NULL) && (strcmp(oldCar->changeOilReminder, newCar->changeOilReminder) != 0))
            {
                free(oldCar->changeOilReminder);
            }
            
            if (oldCar->changeOilReminder == NULL)
            {
                dy_syslog(LOG_DEBUG, "Received a new changeOilReminder = %s", newCar->changeOilReminder);
                if ( NULL != (oldCar->changeOilReminder = malloc(strlen(newCar->changeOilReminder) + 1)))
                {
                    (void)strcpy(oldCar->changeOilReminder, newCar->changeOilReminder);
                    free(newCar->changeOilReminder);
                }
            }
        }

        if (newCar->settings.desired_maxSpeed != 0)
        {
            if (newCar->settings.desired_maxSpeed != oldCar->settings.desired_maxSpeed)
            {
                dy_syslog(LOG_DEBUG, "Received a new desired_maxSpeed = %" PRIu8, newCar->settings.desired_maxSpeed);
                oldCar->settings.desired_maxSpeed = newCar->settings.desired_maxSpeed;
            }
        }

        if (newCar->settings.location.latitude != 0)
        {
            if (newCar->settings.location.latitude != oldCar->settings.location.latitude)
            {
                dy_syslog(LOG_DEBUG, "Received a new latitude = %f", newCar->settings.location.latitude);
                oldCar->settings.location.latitude = newCar->settings.location.latitude;
            }
        }

        if (newCar->settings.location.longitude != 0)
        {
            if (newCar->settings.location.longitude != oldCar->settings.location.longitude)
            {
                dy_syslog(LOG_DEBUG, "Received a new longitude = %f", newCar->settings.location.longitude);
                oldCar->settings.location.longitude = newCar->settings.location.longitude;
            }
        }
        
        free(newCar);
    }
#endif
}

static void reportedStateCallback(int status_code, void* userContextCallback)
{
    (void)userContextCallback;
    dy_syslog(LOG_DEBUG, "Device Twin reported properties update completed with result: %d", status_code);
}


static void iothub_client_device_twin_and_methods_sample_run(void)
{
#if 0
    IOTHUB_CLIENT_TRANSPORT_PROVIDER protocol;
    IOTHUB_DEVICE_CLIENT_HANDLE iotHubClientHandle;

    // Select the Protocol to use with the connection
    protocol = MQTT_Protocol;

    if (IoTHub_Init() != 0)
    {
        (void)printf("Failed to initialize the platform.\r\n");
    }
    else
    {
        if ((iotHubClientHandle = IoTHubDeviceClient_CreateFromConnectionString(connectionString, protocol)) == NULL)
        {
            (void)printf("ERROR: iotHubClientHandle is NULL!\r\n");
        }
        else
        {
            // Uncomment the following lines to enable verbose logging (e.g., for debugging).
            //bool traceOn = true;
            //(void)IoTHubDeviceClient_SetOption(iotHubClientHandle, OPTION_LOG_TRACE, &traceOn);

#ifdef SET_TRUSTED_CERT_IN_SAMPLES
			(void)printf("IoTHubDeviceClient_SetOption\r\n");
            // For mbed add the certificate information
            if (IoTHubDeviceClient_SetOption(iotHubClientHandle, "TrustedCerts", certificates) != IOTHUB_CLIENT_OK)
            {
                (void)printf("failure to set option \"TrustedCerts\"\r\n");
            }
#endif // SET_TRUSTED_CERT_IN_SAMPLES

            Car car;
            memset(&car, 0, sizeof(Car));
            car.lastOilChangeDate = "2016";
            car.maker.makerName = "Fabrikam";
            car.maker.style = "sedan";
            car.maker.year = 2014;
            car.state.reported_maxSpeed = 100;
            car.state.softwareVersion = 1;
            car.state.vanityPlate = "1I1";

            char* reportedProperties = serializeToJson(&car);

            (void)IoTHubDeviceClient_GetTwinAsync(iotHubClientHandle, getCompleteDeviceTwinOnDemandCallback, NULL);
            (void)IoTHubDeviceClient_SendReportedState(iotHubClientHandle, (const unsigned char*)reportedProperties, strlen(reportedProperties), reportedStateCallback, NULL);
            (void)IoTHubDeviceClient_SetDeviceMethodCallback(iotHubClientHandle, deviceMethodCallback, NULL);
            (void)IoTHubDeviceClient_SetDeviceTwinCallback(iotHubClientHandle, deviceTwinCallback, &car);

            (void)getchar();

            IoTHubDeviceClient_Destroy(iotHubClientHandle);
            free(reportedProperties);
            free(car.changeOilReminder);
        }

        IoTHub_Deinit();
    }
#endif
}
#if 0
int main(int argc, char *argv[])
{
	printf("\n\
            |********************************************|\n\
            |           azure start  X_X         |\n\
            |********************************************|\n");
    iothub_client_device_twin_and_methods_sample_run();

    return 0;
}
#endif

void azure_message_arrive(void *pcontext, void *pclient, void *msg)
{
#if 0
    iotx_mqtt_topic_info_t     *topic_info = (iotx_mqtt_topic_info_pt) msg->msg;

    switch (msg->event_type) {
        case IOTX_MQTT_EVENT_PUBLISH_RECEIVED:
            /* print topic name and topic message */
            dy_syslog(LOG_DEBUG,"Message Arrived:");
            dy_syslog(LOG_DEBUG,"Topic  : %.*s", topic_info->topic_len, topic_info->ptopic);
            dy_syslog(LOG_DEBUG,"Payload: %.*s", topic_info->payload_len, topic_info->payload);

			if (strstr(topic_info->ptopic,"_reply"))
			{
				dy_syslog(LOG_DEBUG,"reply message!!!\n");
				break;
			}
			
			ipc_msg_t *mqtt_message = NULL;
			char cmd_identifier[64] = {0};
			char topic[TOPIC_MAX_LEN] = {0};
			char method[256] = {0};
			char* identifier = NULL;
			
			cJSON* root=cJSON_Parse(topic_info->payload);
	        if (!root)
			{
				dy_syslog(LOG_WARNING, "cJSON_Parse payload failed");
				return -1;
	        }
	        cJSON* nodes = cJSON_GetObjectItem(root,"params");
	        if (!nodes)
			{
				dy_syslog(LOG_WARNING, "cJSON_GetObjectItem params failed");
				return -1;
	        }

			GET_JSON_VALUE_STRING(root,"method",method);
			identifier = strstr(method, IDENTIFIER_FLAG);
			if(identifier)
			{
				strncpy(cmd_identifier, identifier+strlen(IDENTIFIER_FLAG), sizeof(cmd_identifier));
				cJSON_AddStringToObject(nodes,"identifier",cmd_identifier);
			}
			cJSON_AddStringToObject(nodes,"sn",pcontext);

			char *payload = cJSON_PrintUnformatted(nodes);
			snprintf(topic, TOPIC_MAX_LEN, "ipc/%s/%s/device/%s/data/%s", gazure_var.sn_str, "azure", pcontext, TOPIC_EVT_SET_RGLT);
			dy_syslog(LOG_DEBUG, "topic:%s payload %d %s", topic,strlen(payload),payload);
			ipc_session_publish(&gazure_var.session, topic, payload, strlen(payload));
            break;
        default:
            break;
    }
#endif
}

int azure_subscribe(void *handle, char* product_key, char* device_name, char* pcontext, const char *fmt)
{
    
    return 0;
}

static IOTHUBMESSAGE_DISPOSITION_RESULT receive_msg_callback(IOTHUB_MESSAGE_HANDLE message, void* user_context)
{
    (void)user_context;
    const char* messageId;
    const char* correlationId;

    // Message properties
    if ((messageId = IoTHubMessage_GetMessageId(message)) == NULL)
    {
        messageId = "<unavailable>";
    }

    if ((correlationId = IoTHubMessage_GetCorrelationId(message)) == NULL)
    {
        correlationId = "<unavailable>";
    }

    IOTHUBMESSAGE_CONTENT_TYPE content_type = IoTHubMessage_GetContentType(message);
    if (content_type == IOTHUBMESSAGE_BYTEARRAY)
    {
        const unsigned char* buff_msg;
        size_t buff_len;

        if (IoTHubMessage_GetByteArray(message, &buff_msg, &buff_len) != IOTHUB_MESSAGE_OK)
        {
            dy_syslog(LOG_DEBUG, "Failure retrieving byte array message\r\n");
        }
        else
        {
            dy_syslog(LOG_DEBUG, "Received Binary message Message ID: %s Correlation ID: %s Data: <<<%.*s>>> & Size=%d", messageId, correlationId, (int)buff_len, buff_msg, (int)buff_len);
        }
    }
    else
    {
        const char* string_msg = IoTHubMessage_GetString(message);
        if (string_msg == NULL)
        {
            dy_syslog(LOG_DEBUG, "Failure retrieving byte array message\r\n");
        }
        else
        {
            dy_syslog(LOG_DEBUG, "Received String Message\r\nMessage ID: %s\r\n Correlation ID: %s\r\n Data: <<<%s>>>\r\n", messageId, correlationId, string_msg);
        }
    }
    return IOTHUBMESSAGE_ACCEPTED;
}

static void connection_status_callback(IOTHUB_CLIENT_CONNECTION_STATUS result, IOTHUB_CLIENT_CONNECTION_STATUS_REASON reason, void* user_context)
{
    (void)reason;
    (void)user_context;
    // This sample DOES NOT take into consideration network outages.
    dy_syslog(LOG_DEBUG, "result %d, IOTHUB_CLIENT_CONNECTION_AUTHENTICATED %d", result, IOTHUB_CLIENT_CONNECTION_AUTHENTICATED);
    if (result == IOTHUB_CLIENT_CONNECTION_AUTHENTICATED)
    {
        dy_syslog(LOG_DEBUG, "The device client is connected to iothub\r\n");
    }
    else
    {
        dy_syslog(LOG_DEBUG, "The device client has been disconnected\r\n");
    }
}

static int device_method_callback(const char* method_name, const unsigned char* payload, size_t size, unsigned char** response, size_t* resp_size, void* userContextCallback)
{
    const char* SetTelemetryIntervalMethod = "SetTelemetryInterval";
	azure_dev_info_t *azure_dev = (azure_dev_info_t *)userContextCallback;
    char* end = NULL;
    int newInterval;

    int status = 501;
    const char* RESPONSE_STRING = "{ \"Response\": \"Unknown method requested.\" }";

    dy_syslog(LOG_DEBUG, "Device Method called for device %s", azure_dev->sn);
    dy_syslog(LOG_DEBUG, "Device Method name:    %s", method_name);
    dy_syslog(LOG_DEBUG, "Device Method payload: %.*s", (int)size, (const char*)payload);

    if (strcmp(method_name, SetTelemetryIntervalMethod) == 0)
    {
        if (payload)
        {
            newInterval = (int)strtol((char*)payload, &end, 10);

            // Interval must be greater than zero.
            if (newInterval > 0)
            {
                // expect sec and covert to ms
                //g_interval = 1000 * (int)strtol((char*)payload, &end, 10);
                status = 200;
                RESPONSE_STRING = "{ \"Response\": \"Telemetry reporting interval updated.\" }";
            }
            else
            {
                status = 500;
                RESPONSE_STRING = "{ \"Response\": \"Invalid telemetry reporting interval.\" }";
            }
        }
    }

    dy_syslog(LOG_DEBUG, "Response status: %d", status);
    dy_syslog(LOG_DEBUG, "Response payload: %s", RESPONSE_STRING);

    *resp_size = strlen(RESPONSE_STRING);
    if ((*response = malloc(*resp_size)) == NULL)
    {
        status = -1;
    }
    else
    {
        memcpy(*response, RESPONSE_STRING, *resp_size);
    }

	cJSON* root=cJSON_Parse(payload);
    if (!root)
	{
		dy_syslog(LOG_WARNING, "cJSON_Parse payload failed");
    }
	else
	{
		char topic[TOPIC_MAX_LEN] = {0};

		cJSON_AddStringToObject(root, "sn", azure_dev->sn);
		cJSON_AddNumberToObject(root, "mi", 12345);
		cJSON_AddStringToObject(root, "identifier", method_name);
			
		char *azure_payload = cJSON_PrintUnformatted(root);
		snprintf(topic, TOPIC_MAX_LEN, "ipc/%s/%s/device/%s/data/%s", gazure_var.sn_str, "azure", azure_dev->sn, TOPIC_EVT_SET_RGLT);
		dy_syslog(LOG_DEBUG, "topic:%s payload %d %s", topic,strlen(azure_payload),azure_payload);
		ipc_session_publish(gazure_var.session, topic, azure_payload, strlen(azure_payload));
	}
    
    return status;
}

static void send_confirm_callback(IOTHUB_CLIENT_CONFIRMATION_RESULT result, void* userContextCallback)
{
    (void)userContextCallback;
	static size_t g_message_count_send_confirmations = 0;
    // When a message is sent this callback will get invoked
    g_message_count_send_confirmations++;
    dy_syslog(LOG_DEBUG, "Confirmation callback received for message %lu with result %s", (unsigned long)g_message_count_send_confirmations, MU_ENUM_TO_STRING(IOTHUB_CLIENT_CONFIRMATION_RESULT, result));
}

int azure_publish(azure_dev_info_t *azure_dev, int type, char *params)
{
	dy_syslog(LOG_DEBUG, "type %d azure_publish:%s", type, params);
	if(type == 0)
	{
    	IoTHubDeviceClient_SendReportedState(azure_dev->iotHubClientHandle, (const unsigned char*)params, strlen(params), reportedStateCallback, azure_dev);
	}
	else if(type == 1 || type == 2)
	{
		IOTHUB_MESSAGE_HANDLE message_handle;
		message_handle = IoTHubMessage_CreateFromString(params);

        // Set Message property
        IoTHubMessage_SetMessageId(message_handle, "MSG_ID");
        IoTHubMessage_SetCorrelationId(message_handle, "CORE_ID");
        IoTHubMessage_SetContentTypeSystemProperty(message_handle, "application%2fjson");
        IoTHubMessage_SetContentEncodingSystemProperty(message_handle, "utf-8");

        // Add custom properties to message
        //IoTHubMessage_SetProperty(message_handle, "property_key", "property_value");
		
		IoTHubDeviceClient_SendEventAsync(azure_dev->iotHubClientHandle, message_handle, send_confirm_callback, azure_dev);
	}
	
    return 0;
}

void azure_event_handle(void *pcontext, void *pclient, void *msg)
{
    //dy_syslog(LOG_DEBUG, "msg->event_type : %d pcontext : %s", msg->event_type,pcontext);
}

int azure_mqtt_connect(azure_dev_info_t *azure_dev)
{
    int res = 0;

	if ((azure_dev->iotHubClientHandle = IoTHubDeviceClient_CreateFromConnectionString(azure_dev->connectionString, azure_dev->protocol)) == NULL)
    {
        dy_syslog(LOG_ERR, "ERROR: iotHubClientHandle is NULL");
    }
    else
    {
        IoTHubDeviceClient_GetTwinAsync(azure_dev->iotHubClientHandle, getCompleteDeviceTwinOnDemandCallback, azure_dev);
        //IoTHubDeviceClient_SendReportedState(azure_dev->iotHubClientHandle, (const unsigned char*)reportedProperties, strlen(reportedProperties), reportedStateCallback, NULL);
        IoTHubDeviceClient_SetDeviceMethodCallback(azure_dev->iotHubClientHandle, device_method_callback, azure_dev);
        IoTHubDeviceClient_SetDeviceTwinCallback(azure_dev->iotHubClientHandle, deviceTwinCallback, azure_dev);
		
		// Setting message callback to get C2D messages
        IoTHubDeviceClient_SetMessageCallback(azure_dev->iotHubClientHandle, receive_msg_callback, azure_dev);
        // Setting connection status callback to get indication of connection to iothub
        IoTHubDeviceClient_SetConnectionStatusCallback(azure_dev->iotHubClientHandle, connection_status_callback, azure_dev);
    }
    azure_dev->connected = 1;

    return 0;
}

int azure_get_message(azure_var_t* var)
{
	TIMER_CONFIRM(var->get_msg_timer);

	if (var->data_flag)
		return 0;
#if 0
    int ret;
	static int null_cnt = 0;
	
	azure_info_list_t *azure = NULL;
    list_for_each_entry(azure, &var->node_list,list)
    {
    	if(!azure->meta_info.construct)
			continue;
		
    	if(!azure->meta_info.pclient)
    	{
    		null_cnt++;
    		dy_syslog(LOG_WARNING, "get_msg device_name %s pclient is %p null_cnt(%d %d) %d\n",azure->meta_info.device_name,azure->meta_info.pclient,null_cnt,var->link_node_cnt,var->nodes_cfg_table->node_cnt);
			if(null_cnt > var->link_node_cnt)
			{
				dy_syslog(LOG_ERR,"%s exit(1)\n",__FUNCTION__);
				exit(1);
			}
			continue;
    	}
        //ret = azure_subscribe_t(azure->meta_info.pclient, azure->meta_info.product_key, azure->meta_info.device_name, azure->meta_info.alias_sn, "/%s/%s/user/get");
		ret |= azure_subscribe(azure->meta_info.pclient, azure->meta_info.product_key, azure->meta_info.device_name, azure->meta_info.alias_sn, "/sys/%s/%s/thing/service/#");
		if (ret < 0) {
			azure->meta_info.communication_fail++;
			if(azure->meta_info.communication_fail > 2)
			{
				ret=DY_IOT_MQTT_Destroy(&azure->meta_info.pclient);
				dy_syslog(LOG_WARNING,"DY_IOT_MQTT_Destroy ret %d\n",ret);
				azure_mqtt_connect(&azure->meta_info);
			}
    	}
		else
			azure->meta_info.communication_fail = 0;
    }
#endif
}

static int azure_node_connect(azure_var_t *var)
{
	int i;

	if (IoTHub_Init() != 0)
    {
        dy_syslog(LOG_ERR, "Failed to initialize the platform..");
		return 0;
    }
	
	var->link_node_cnt = 0;
	for(i=0; i<var->nodes_cfg_table->node_cnt; i++)
    {
    	if(strlen(var->nodes_cfg_table->node[i].connection_string))
		{
			azure_info_list_t* azure = calloc(sizeof(azure_info_list_t), 1);

			strncpy(azure->azure_dev.sn, var->nodes_cfg_table->node[i].sn, sizeof(azure->azure_dev.sn));
			strncpy(azure->azure_dev.connectionString, var->nodes_cfg_table->node[i].connection_string, sizeof(azure->azure_dev.connectionString));
			azure->azure_dev.protocol = MQTT_Protocol;

			var->link_node_cnt++;
	        dy_syslog(LOG_DEBUG,"===%s %d sn %s connectionString %s)===\n",__FUNCTION__,__LINE__,azure->azure_dev.sn,azure->azure_dev.connectionString);
	        azure_mqtt_connect(&azure->azure_dev);
	        list_add_tail(&azure->list,&var->node_list);
		}
    }
}

static int mqtt_msg_process(azure_var_t *var, int type, ipc_msg_t *mqtt_msg)
{
	int ret;
	static int null_cnt = 0;
	tag_table_t tag;

	//先检查参数
	ret = get_tag_data_from_str(&tag, mqtt_msg->payload);
    if (ret < 0)
    {
        dy_syslog(LOG_ERR, "get data structure failed");
        return -1;
    }
	
	azure_info_list_t *azure = NULL;
	list_for_each_entry(azure, &var->node_list, list)
	{
		if(strcmp(azure->azure_dev.sn, tag.sn) == 0)
		{
			char topic[256] = {0};
			char method[256] = {0};
			char *params = tag.tag_node;
			
			if (params == NULL)
			{
		        dy_syslog(LOG_WARNING,"type %d params is NULL!!!",type);
		        break;
		    }
			if(type == 0)
			{
				azure_publish(&azure->azure_dev, type, params);
			}
			else if(type == 1)
			{
				azure_publish(&azure->azure_dev, type, params);
			}
			else if(type == 2)
			{
				azure_publish(&azure->azure_dev, type, params);
			}
			break;
		}
	}
}

void msg_mqtt_recv(azure_var_t *var, int type, ipc_msg_t *mqtt_msg)
{
    int ret,i;

	data_list_t* data = calloc(1,sizeof(data_list_t)+mqtt_msg->payloadLen+1);	
	if(data)
	{
		strncpy(data->mqtt.topic, mqtt_msg->topic, sizeof(data->mqtt.topic));
		data->type = type;
		data->mqtt.payloadLen = mqtt_msg->payloadLen;
		memcpy(data->mqtt.payload, mqtt_msg->payload, data->mqtt.payloadLen);
		var->data_cnt++;
		dy_syslog(LOG_DEBUG, "++%s topic %s data_cnt %d\n",__FUNCTION__,data->mqtt.topic,var->data_cnt);
    	list_add_tail(&data->list,&var->data_list);
	}

	var->data_flag = 1;
}

static int mqtt_msg_process_loop(azure_var_t *var)
{
	data_list_t* data = NULL;	
	list_for_each_entry(data, &var->data_list, list)
	{
		dy_syslog(LOG_DEBUG, "var->data_cnt %d", var->data_cnt);
		var->data_cnt--;
		mqtt_msg_process(var, data->type, &data->mqtt);
		list_del(&data->list);
		break;
	}

	var->data_cnt = 0;
	var->data_flag = 0;
}

static void azure_loop(azure_var_t *var)
{
	int ret = -1, maxfd = 0, i;
    fd_set rset;
    struct timeval timeout;
	
    while (1)
    {
    	SELECT_INIT();
        SELECT_ADD_FD(var->get_msg_timer);

		if(var->data_flag)
        {
			timeout.tv_usec = 50000;
        	timeout.tv_sec = 0;
        }
		else
		{
			timeout.tv_usec = 0;
        	timeout.tv_sec = 2;
		}

        ret = select(maxfd + 1, &rset, 0, 0, &timeout);
        if (ret < 0)
        {
            dy_syslog(LOG_INFO, "errno %d\n", errno);

            if (errno == EINTR)
            {
                continue;
            }
            else
            {
                break;
            }
        }
        else if (ret > 0)
        {
            if (var->get_msg_timer > 0 && FD_ISSET(var->get_msg_timer, &rset))
            {
                FD_CLR(var->get_msg_timer, &rset);
                azure_get_message(var);
            }
        }

		mqtt_msg_process_loop(var);
    }
}

static void azure_subscribe_all(azure_var_t *var)
{
    ipc_session_t *ipc_session = var->session;
    char topic[TOPIC_MAX_LEN] = {0};

    snprintf(topic, TOPIC_MAX_LEN, "ipc/%s/+/device/+/data_filtered/property/+", var->sn_str);
    ipc_session_subscribe(ipc_session, topic);

    snprintf(topic, TOPIC_MAX_LEN, "ipc/%s/+/device/+/data_filtered/event/+", var->sn_str);
    ipc_session_subscribe(ipc_session, topic);

    snprintf(topic, TOPIC_MAX_LEN, "ipc/%s/+/device/+/data_filtered/service/+", var->sn_str);
    ipc_session_subscribe(ipc_session, topic);
}

static int azure_mqtt_handle_recv_msg(void *obj, ipc_msg_t *mqtt_msg)
{
    azure_var_t *var = (azure_var_t *)obj;
    bool matched;
    int ret;
    dy_syslog(LOG_DEBUG, " MQTT client: received MQTT topic:%s payload length:%d",
            mqtt_msg->topic, mqtt_msg->payloadLen);

    ret = mosquitto_topic_matches_sub("ipc/+/+/device/+/data_filtered/property/+", mqtt_msg->topic, &matched);
    if (ret == 0 && matched)
    {
        //实时数据
        msg_mqtt_recv(var, 0, mqtt_msg);
    }
    else
    {
        //事件
        ret = mosquitto_topic_matches_sub("ipc/+/+/device/+/data_filtered/event/+", mqtt_msg->topic, &matched);
        if (ret == 0 && matched)
        {
            msg_mqtt_recv(var, 1, mqtt_msg);
        }
        else
        {
            ret = mosquitto_topic_matches_sub("ipc/+/+/device/+/data_filtered/service/+", mqtt_msg->topic, &matched);
            if (ret == 0 && matched)
            {
                msg_mqtt_recv(var, 2, mqtt_msg);
            }
        }
    }
}

// 建立与内部broker之间的MQTT连接
int azure_mqtt_client_init(azure_var_t *var)
{
    char clientId[MAX_CLIENT_ID_LEN] = {0};

    snprintf(clientId, MAX_CLIENT_ID_LEN, "INT_azure_%s", var->sn_str);
    var->session = ipc_session_new(clientId, (void*)var, IPC_DEFAULT);
    if (var->session == NULL) return -1;
    ipc_session_set_callbacks(var->session, azure_mqtt_handle_recv_msg, NULL);
    azure_subscribe_all(var);
    ipc_session_start(var->session);
}

int azure_init(azure_var_t *var)
{
    int i, j;
    const char *board_name = NULL;

    check_make_dir(NODES_CACHE);
    check_make_dir(NODES_CFG);
    get_board_sn(var->sn_str);
    get_process_name(var->proc_name);
	
	INIT_LIST_HEAD(&var->node_list);
	INIT_LIST_HEAD(&var->data_list);
	
    board_name = get_board_name();

    dy_syslog(LOG_INFO, "board name:%s", board_name);
    dy_syslog(LOG_INFO, "board SN:%s", var->sn_str);

    if (load_nodes_cfg(&var->nodes_cfg_table, NODES_CFG_PATH) == -1)
    {
        dy_syslog(LOG_ERR, "load nodes cfg fail");
    }
	
    if (load_templates_cfg(&var->template_table, TEMPLATES_CFG_PATH) == -1)
    {
        dy_syslog(LOG_ERR, "load template cfg fail");
    }
	
    kv_array_init(&var->identifier_backup, 32);
    //azure_load_config(var);
	
	//if(strlen(var->response_event) == 0)
		//strncpy(var->response_event, RESPONSE_EVENT, sizeof(RESPONSE_EVENT));
	
	//dy_syslog(LOG_DEBUG,"%s response_event:%s \n", __FUNCTION__,var->response_event);
    azure_mqtt_client_init(var);
	azure_node_connect(var);
	
	var->get_msg_timer = my_timer_create();
    if (var->get_msg_timer > 0)
    {
        my_timer_set(var->get_msg_timer, 1, 1000);
    }
	
    return 0;
}


int main(int argc, char *argv[])
{
    azure_var_t *var = &gazure_var;

	dy_syslog(LOG_DEBUG, "\n\
            |********************************************|\n\
            |           azure start  X_X         |\n\
            |********************************************|\n");
    memset(var, 0, sizeof(azure_var_t));
    azure_init(var);
    azure_loop(var);

    return 0;
}
#endif