MQTTClient
MQTTClient.cpp
- Committer:
- SomeRandomBloke
- Date:
- 2012-03-21
- Revision:
- 4:9e25b78e0352
- Parent:
- 3:ac326c5b065c
- Child:
- 7:c12d2bfe40e2
File content as of revision 4:9e25b78e0352:
/* MQTTClient.cpp Based on MQTTClient from http://ceit.uq.edu.au/content/mqttclient-mbed-version-20 A simple MQTT client for mbed, version 2.0 By Yilun FAN, @CEIT, @JAN 2011 Bug fixes and additions by Andrew Lindsay (andrew [at] thiseldo [dot] co [dot] uk) Permission is hereby granted, free of charge, to any person obtaining a copy of this software and associated documentation files (the "Software"), to deal in the Software without restriction, including without limitation the rights to use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the Software, and to permit persons to whom the Software is furnished to do so, subject to the following conditions: The above copyright notice and this permission notice shall be included in all copies or substantial portions of the Software. THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE. */ #include "MQTTClient.h" /** Default Constructor */ MQTTClient::MQTTClient() {} /** Default Destructor */ MQTTClient::~MQTTClient() {} /** Alternative Constructor with parameters * * Allow object to be constructed with minimum parameters. * * @param server The IP address of the server to connect to * @param port The TCP/IP port on the server to connect to * @param callback Callback function to handle subscription to topics */ MQTTClient::MQTTClient(IpAddr server, int port, void (*callback)(char*, char*)) { this->port = port; callback_server = callback; serverIp = server; connected = false; sessionOpened = false; timer.start(); } /** MQTT initialisation method * * Used when default constructor used and need to specify parameters at runtime * * @param server The IP address of the server to connect to * @param port The TCP/IP port on the server to connect to * @param callback Callback function to handle subscription to topics */ void MQTTClient::init(IpAddr *server, int port, void (*callback)(char*, char*)) { this->port = port; callback_server = callback; serverIp = *server; this->userName = NULL; this->password = NULL; connected = false; sessionOpened = false; timer.start(); } /** A brief description of the function foo * * More details about the function goes here * and here * * @param server The IP address of the server to connect to * @param port The TCP/IP port on the server to connect to * @param userName Pointer to username string, zero terminated. Null for not used * @param password Pointer to password string, zero terminated. Null for not used * @param callback Callback function to handle subscription to topics */ void MQTTClient::init(IpAddr *server, int port, char *userName, char *password, void (*callback)(char*, char*)) { this->port = port; callback_server = callback; serverIp = *server; this->userName = userName; this->password = password; connected = false; sessionOpened = false; timer.start(); } /** Send a message of specified size * * @param msg message string * @param size Size of the message to send * @returns value to indicate message sent successfully or not, -1 for error, 1 for success. */ int MQTTClient::send_data(const char* msg, int size) { int transLen = pTCPSocket->send(msg, size); /*Check send length matches the message length*/ if (transLen != size) { for (int i = 0; i < size; i++) { printf("%x, ", msg[i]); } printf("Error on send.\r\n"); return -1; } return 1; } /** Start a MQTT session, build CONNECT packet * * @param id The client name shown on MQTT server. * @returns -1 for error, 1 for success */ int MQTTClient::open_session(char* id) { /*variable header*/ char var_header[] = {0x00,0x06,0x4d,0x51,0x49,0x73,0x64,0x70,0x03,0x02,0x00,KEEPALIVE/500,0x00,strlen(id)}; /*fixed header: 2 bytes, big endian*/ char fixed_header[] = {MQTTCONNECT,12+strlen(id)+2}; short usernameLen = strlen( userName ); short passwordLen = strlen( password ); var_header[9] |= (usernameLen > 0 ? 0x80 : 0x00 ); var_header[9] |= (passwordLen > 0 ? 0x40 : 0x00 ); // printf("fixed %d, var %d id %d, username %d, password %d\n",sizeof(fixed_header), sizeof(var_header), sizeof(id), usernameLen, passwordLen ); char packet[sizeof(fixed_header) + sizeof(var_header) + sizeof(id) + usernameLen + (usernameLen > 0 ? 2 : 0) + passwordLen + (passwordLen > 0 ? 2 : 0) ]; memset(packet,0,sizeof(packet)); memcpy(packet,fixed_header,sizeof(fixed_header)); memcpy(packet+sizeof(fixed_header),var_header,sizeof(var_header)); memcpy(packet+sizeof(fixed_header)+sizeof(var_header), id, strlen(id)); if ( usernameLen > 0 ) { packet[sizeof(fixed_header)+sizeof(var_header)+strlen(id)] = 0x00; packet[sizeof(fixed_header)+sizeof(var_header)+strlen(id)+1] = usernameLen; memcpy(packet+sizeof(fixed_header)+sizeof(var_header)+strlen(id)+2, userName, usernameLen); packet[1] += (usernameLen + 2); } if ( passwordLen > 0 ) { packet[sizeof(fixed_header)+sizeof(var_header)+strlen(id) + usernameLen + (usernameLen > 0 ? 2 : 0) ] = 0x00; packet[sizeof(fixed_header)+sizeof(var_header)+strlen(id) + usernameLen + (usernameLen > 0 ? 2 : 0)+1] = passwordLen; memcpy(packet+sizeof(fixed_header)+sizeof(var_header)+strlen(id)+ usernameLen + (usernameLen > 0 ? 2 : 0)+2, password, passwordLen); packet[1] += (passwordLen + 2); } /* printf("Packet length %d\n", sizeof(packet)); for (int i = 0; i < sizeof(packet); i++) { printf("%x, ", packet[i]); } printf("\r\n"); */ if (!send_data(packet, sizeof(packet))) { return -1; } // printf("Sent\n"); return 1; } /** Open TCP port, connect to server on given IP address. * * @param id The client name shown on MQTT server. * @returns -1: If connect to server failed. -2: Failed to open session on server. 1: Connection accessed. */ int MQTTClient::connect(char* id) { clientId = id; /*Initial TCP socket*/ pTCPSocket = new TCPSocket; pTCPSocket->setOnEvent(this, &MQTTClient::onTCPSocketEvent); host.setPort(port); host.setIp(serverIp); host.setName("localhost"); /*Trying to connect to host*/ printf("Trying to connect to host..\r\n\r\n"); TCPSocketErr err = pTCPSocket->connect(host); Net::poll(); if (err) { printf("Error connecting to host [%d]\r\n", (int) err); return -1; } printf("Connect to host sucessed..\r\n\r\n"); /*Wait TCP connection with server to be established*/ int i = 0; while (!connected) { Net::poll(); wait(1); i++; printf("Wait for connections %d..\r\n", i); if (i == 35) {//If wait too long, give up. return -1; } } /*Send open session message to server*/ open_session(id); /*Wait server notice of open sesion*/ while (!sessionOpened) { Net::poll(); wait(1); if (!connected) { break; } printf("Wait for session..\r\n"); } if (!connected) { return -2; } lastActivity = timer.read_ms(); return 1; } /** Publish a message on a topic. * * @param pub_topic The topic name the massage will be publish on. * @param msg The massage to be published. * @returns -1: Failed to publish message. 1: Publish sucessed. */ int MQTTClient::publish(char* pub_topic, char* msg) { uint8_t var_header_pub[strlen(pub_topic)+3]; strcpy((char *)&var_header_pub[2], pub_topic); var_header_pub[0] = 0; var_header_pub[1] = strlen(pub_topic); var_header_pub[sizeof(var_header_pub)-1] = 0; uint8_t fixed_header_pub[] = {MQTTPUBLISH,sizeof(var_header_pub)+strlen(msg)}; uint8_t packet_pub[sizeof(fixed_header_pub)+sizeof(var_header_pub)+strlen(msg)]; memset(packet_pub,0,sizeof(packet_pub)); memcpy(packet_pub,fixed_header_pub,sizeof(fixed_header_pub)); memcpy(packet_pub+sizeof(fixed_header_pub),var_header_pub,sizeof(var_header_pub)); memcpy(packet_pub+sizeof(fixed_header_pub)+sizeof(var_header_pub),msg,strlen(msg)); if (!send_data((char*)packet_pub, sizeof(packet_pub))) { return -1; } return 1; } /** Disconnect from server */ void MQTTClient::disconnect() { char packet_224[] = {0xe0, 0x00}; send_data((char*)packet_224, 2); connected = false; } /** Read data from receive packet * Determine what needs to be done with packet. */ void MQTTClient::read_data() { char buffer[1024]; int len = 0, readLen; while ((readLen = pTCPSocket->recv(buffer, 1024)) != 0) { len += readLen; } buffer[len] = '\0'; printf("Read length: %d %d\r\n", len, readLen); for (int i = 0; i < len; i++) { printf("%2X ", buffer[i]); } printf("\r\n"); char type = buffer[0]>>4; if ( type == 2 ) { // CONNACK printf("CONNACK\n"); } else if (type == 3) { // PUBLISH if (callback_server) { short index = 1; short multiplier = 1; short value = 0; uint8_t digit; do { digit = buffer[index++]; value += (digit & 127) * multiplier; multiplier *= 128; } while ((digit & 128) != 0); printf( "variable length %d, index %d\n", value, index ); // uint8_t tl = (buffer[2]<<3)+buffer[3]; uint8_t tl = (buffer[index]<<3)+buffer[index+1]; printf("Topic len %d\n",tl); char topic[tl+1]; for (int i=0; i<tl; i++) { topic[i] = buffer[index+2+i]; } topic[tl] = 0; // ignore msgID - only support QoS 0 subs char *payload = buffer+index+2+tl; callback_server(topic,(char*)payload); } } else if (type == 12) { // PINGREG -- Ask for alive char packet_208[] = {0xd0, 0x00}; send_data((char*)packet_208, 2); lastActivity = timer.read_ms(); } } /** Check for session opened */ void MQTTClient::read_open_session() { char buffer[32]; int len = 0, readLen; while ((readLen = pTCPSocket->recv(buffer, 32)) != 0) { len += readLen; } if (len == 4 && buffer[3] == 0) { printf("Session opened\r\n"); sessionOpened = true; } } /** Subscribe to a topic * * @param topic The topic name to be subscribed. * @returns -1: Failed to subscribe to topic. 1: Subscribe sucessed. */ int MQTTClient::subscribe(char* topic) { if (connected) { uint8_t var_header_topic[] = {0,10}; uint8_t fixed_header_topic[] = {MQTTSUBSCRIBE,sizeof(var_header_topic)+strlen(topic)+3}; //utf topic uint8_t utf_topic[strlen(topic)+3]; strcpy((char *)&utf_topic[2], topic); utf_topic[0] = 0; utf_topic[1] = strlen(topic); utf_topic[sizeof(utf_topic)-1] = 0; char packet_topic[sizeof(var_header_topic)+sizeof(fixed_header_topic)+strlen(topic)+3]; memset(packet_topic,0,sizeof(packet_topic)); memcpy(packet_topic,fixed_header_topic,sizeof(fixed_header_topic)); memcpy(packet_topic+sizeof(fixed_header_topic),var_header_topic,sizeof(var_header_topic)); memcpy(packet_topic+sizeof(fixed_header_topic)+sizeof(var_header_topic),utf_topic,sizeof(utf_topic)); if (!send_data(packet_topic, sizeof(packet_topic))) { return -1; } return 1; } return -1; } /** Send heartbeat/keep-alive message * */ void MQTTClient::live() { if (connected) { int t = timer.read_ms(); if (t - lastActivity > KEEPALIVE) { //Send 192 0 to broker printf("Send 192\r\n"); char packet_192[] = {0xc0, 0x00}; send_data((char*)packet_192, 2); lastActivity = t; } } } /** TCP Socket event handling * * @param e Event object */ void MQTTClient::onTCPSocketEvent(TCPSocketEvent e) { switch (e) { case TCPSOCKET_ACCEPT: printf("New TCPSocketEvent: TCPSOCKET_ACCEPT\r\n"); break; case TCPSOCKET_CONNECTED: printf("New TCPSocketEvent: TCPSOCKET_CONNECTED\r\n"); connected = true; break; case TCPSOCKET_WRITEABLE: printf("New TCPSocketEvent: TCPSOCKET_WRITEABLE\r\n"); break; case TCPSOCKET_READABLE: printf("New TCPSocketEvent: TCPSOCKET_READABLE\r\n"); if (!sessionOpened) { read_open_session(); } read_data(); break; case TCPSOCKET_CONTIMEOUT: printf("New TCPSocketEvent: TCPSOCKET_CONTIMEOUT\r\n"); break; case TCPSOCKET_CONRST: printf("New TCPSocketEvent: TCPSOCKET_CONRST\r\n"); break; case TCPSOCKET_CONABRT: printf("New TCPSocketEvent: TCPSOCKET_CONABRT\r\n"); break; case TCPSOCKET_ERROR: printf("New TCPSocketEvent: TCPSOCKET_ERROR\r\n"); break; case TCPSOCKET_DISCONNECTED: printf("New TCPSocketEvent: TCPSOCKET_DISCONNECTED\r\n"); pTCPSocket->close(); connected = false; break; } }