#include "MqttClient.h" #include #include MQTTAsync_responseOptions CMqttSensor::opts = MQTTAsync_responseOptions_initializer; MQTTAsync CMqttSensor::m_mqttClient; bool CMqttSensor::_run = false; char** CMqttSensor::subTopic = NULL; std::string CMqttSensor::_Esn; CMqttSensor::CMqttSensor(CMessageQueue* q) :_message(q) { } void CMqttSensor::SetSensorMQTT(MQTTAsync mqClient, std::string Esn) { m_mqttClient = mqClient; _Esn = Esn; } void CMqttSensor::onConnectCallCBack(void* context, char* cause) //���ӳɹ�callback { if(m_mqttClient != NULL) { std::cout << "Connection successful" << std::endl; int rc; const int qos[3] = { 0,1,2}; const int ROWS = 3; const int COLUMNS = 64; subTopic = new char* [ROWS]; for (int row = 0; row < ROWS; row++) { subTopic[row] = new char[COLUMNS]; } sprintf((char*)*(subTopic + 0), "v2x/v1/grasp/%s/info/up/ack", (char*)context); sprintf((char*)*(subTopic + 1), "v2x/v1/grasp/%s/status/up/ack", (char*)context); sprintf((char*)*(subTopic + 2), "v2x/v1/grasp/%s/stop/down/", (char*)context); opts = MQTTAsync_responseOptions_initializer; //char* const subTopic[2] = { "hello" ,"hello1" }; //�������� opts.context = m_mqttClient; if ((rc = MQTTAsync_subscribeMany(m_mqttClient, 3 , subTopic, qos, &opts)) != MQTTASYNC_SUCCESS) //subTopic { std::cout << "MQTTAsync_subscribe() fail, error code: " << rc << std::endl; } _run = true; } //sendMessage("11111",1, "hello"); } void CMqttSensor::onConnectFailure(void* context, MQTTAsync_failureData* response) //����ʧ��callback { _run = false; printf("connection fail, error code: %d, error message:%s\n", response ? response->code : 0, response ? response->message : 0); } void CMqttSensor::Disconnect(void* context, char* cause) //���ӶϿ�callback { _run = false; std::cout << "disconnect, cause: " << cause << std::endl; } void CMqttSensor::sendMessage(char* data, int qos , const char * pubTopic) { if (_run && (m_mqttClient != NULL)) { int rc; MQTTAsync_message pubmsg = MQTTAsync_message_initializer; MQTTAsync_responseOptions opts = MQTTAsync_responseOptions_initializer; opts.context = m_mqttClient; pubmsg.payload = data; pubmsg.payloadlen = strlen(data); pubmsg.qos = qos; //std::cout << "mqtt send: " << pubmsg.payload << std::endl; if ((rc = MQTTAsync_sendMessage(m_mqttClient, pubTopic, &pubmsg, &opts)) != MQTTASYNC_SUCCESS) { std::cout << "MQTTAsync_sendMessage() fail, error code: " << rc << std::endl; } } } int CMqttSensor::RecevieMessage(void* context, char* pubTopic, int topicLen, MQTTAsync_message* message) { if (_run) { char buf[message->payloadlen + 1]; memcpy(buf, message->payload, message->payloadlen); //printf("mqtt recv: %s\n", buf); //std::cout << "mqtt recv: " << buf << std::endl; MQTTAsync_freeMessage(&message); MQTTAsync_free(pubTopic); } return 1; } void CMqttSensor::Stop(const char* _Esn) { MQTTAsync_disconnectOptions optDis = MQTTAsync_disconnectOptions_initializer; if(_run) { MQTTAsync_unsubscribeMany(m_mqttClient, 3, subTopic, &opts); MQTTAsync_disconnect(&m_mqttClient, &optDis); const int ROWS = 3; for (int row = 0; row < ROWS; row++) { delete [] subTopic[row]; } delete [] subTopic; } _run = false; m_mqttClient = NULL; MQTTAsync_destroy(&m_mqttClient); }