diff options
Diffstat (limited to 'examples/network')
| -rw-r--r-- | examples/network/mqttclient.c | 560 |
1 files changed, 0 insertions, 560 deletions
diff --git a/examples/network/mqttclient.c b/examples/network/mqttclient.c deleted file mode 100644 index 17b049dd93..0000000000 --- a/examples/network/mqttclient.c +++ /dev/null @@ -1,560 +0,0 @@ - -#define OS_RTTHREAD 1 - -#if (OS_RTTHREAD == 0) -#include <sys/socket.h> -#include <sys/time.h> -#include <sys/select.h> -#include <netinet/in.h> -#include <netinet/tcp.h> -#include <arpa/inet.h> -#include <netdb.h> -#include <stdio.h> -#include <unistd.h> -#include <errno.h> -#include <fcntl.h> -#include <pthread.h> -#include <signal.h> -#include <stdlib.h> - -#define DEBUG printf -#define SLEEP(x) sleep(x) -#else -#include <rtthread.h> -#include <lwip/netdb.h> -#include <lwip/sockets.h> -#include <arch/sys_arch.h> -#include <lwip/sys.h> - -#define DEBUG rt_kprintf -#define SLEEP(x) rt_thread_delay((x)*RT_TICK_PER_SECOND) -#endif - -#include <string.h> -#include "MQTTPacket.h" - -#define HOSTNAME "m2m.eclipse.org" -#define HOSTPORT 1883 -#define USERNAME "testuser" -#define PASSWORD "testpassword" -#define TOPIC "test" - -#define KEEPALIVE_INTERVAL 20 - -typedef struct -{ - int qos; - unsigned char retained; - unsigned char dup; - unsigned short id; - int payloadlen; - unsigned char *payload; -}mqtt_msg_t; - -typedef struct -{ - int sockfd; - unsigned char *wbuf; // - int wbuflen; - unsigned char *rbuf; - int rbuflen; - int (*getfn)(unsigned char*, int); -}mqtt_client_t; - -static mqtt_client_t _cpub; -static mqtt_client_t _csub; - -static void *mqtt_ping_thread(void *param) -{ - int *sockfd = (int*)param; - unsigned char buf[2]; - int len; - - DEBUG("ping start\n"); - - while (*sockfd >= 0) - { - SLEEP(KEEPALIVE_INTERVAL-1); - len = MQTTSerialize_pingreq(buf, sizeof(buf)); - - send(*sockfd, buf, len, 0); - } - - DEBUG("ping exit\n"); - - return 0; -} - -void mqtt_ping_start(int *sockfd) -{ - #if (OS_RTTHREAD == 0) - pthread_t tid; - - pthread_create(&tid, NULL, mqtt_ping_thread, (void*)sockfd); - #else - sys_thread_new("ping", - mqtt_ping_thread, - (void*)sockfd, - 512, - 20); - #endif -} - -static int sub_read(unsigned char *buf, int len) -{ - int bytes = 0; - struct timeval interval; - int rc; - - interval.tv_sec = 3; - interval.tv_usec = 0; - - rc = setsockopt(_csub.sockfd, SOL_SOCKET, SO_RCVTIMEO, (char *)&interval, sizeof(struct timeval)); - - while (bytes < len) - { - rc = recv(_csub.sockfd, &buf[bytes], (size_t)(len - bytes), 0); - - if (rc == -1) - { - if (errno != ENOTCONN && errno != ECONNRESET) - { - bytes = -1; - break; - } - } - else - bytes += rc; - } - - return bytes; -} - -static int pub_read(unsigned char *buf, int len) -{ - int bytes = 0; - struct timeval interval; - int rc; - - interval.tv_sec = 3; - interval.tv_usec = 0; - - rc = setsockopt(_cpub.sockfd, SOL_SOCKET, SO_RCVTIMEO, (char *)&interval, sizeof(struct timeval)); - - while (bytes < len) - { - rc = recv(_cpub.sockfd, &buf[bytes], (size_t)(len - bytes), 0); - - if (rc == -1) - { - if (errno != ENOTCONN && errno != ECONNRESET) - { - bytes = -1; - break; - } - } - else - bytes += rc; - } - - return bytes; -} - -int mqtt_write(int sockfd, unsigned char *buf, int len) -{ - int rc; - struct timeval tv; - - tv.tv_sec = 2; - tv.tv_usec = 0; - - setsockopt(sockfd, SOL_SOCKET, SO_SNDTIMEO, (char *)&tv, sizeof(struct timeval)); - rc = send(sockfd, buf, len, 0); - if (rc == len) - rc = 0; - - return rc; -} - -int mqtt_subscribe(mqtt_client_t *c, char* topicstr, int qos) -{ - MQTTString topic = MQTTString_initializer; - int msgid = 1; - int len; - int rc = -1; - - topic.cstring = topicstr; - len = MQTTSerialize_subscribe(c->wbuf, c->wbuflen, 0, msgid, 1, &topic, &qos); - if (len <= 0) - goto exit; - - rc = mqtt_write(c->sockfd, c->wbuf, len); - if (rc < 0) - goto exit; - - if (MQTTPacket_read(c->rbuf, c->rbuflen, c->getfn) == SUBACK) - { - unsigned short submsgid; - int subcount; - int granted_qos; - - rc = MQTTDeserialize_suback(&submsgid, 1, &subcount, &granted_qos, c->rbuf, c->rbuflen); - if (granted_qos != 0) - { - DEBUG("granted qos != 0, %d\n", granted_qos); - rc = -1; - } - else - { - rc = 0; - } - } - else - { - rc = -1; - } - -exit: - return rc; -} - -int mqtt_publish(mqtt_client_t *c, char* topicstr, mqtt_msg_t *msg) -{ - int rc = -1; - MQTTString topic = MQTTString_initializer; - int len; - int pktype; - - topic.cstring = topicstr; - - len = MQTTSerialize_publish(c->wbuf, c->wbuflen, - msg->dup, msg->qos, msg->retained, msg->id, - topic, msg->payload, msg->payloadlen); - if (len <= 0) - goto exit; - - if ((rc = mqtt_write(c->sockfd, c->wbuf, len)) <= 0) - goto exit; - - pktype = MQTTPacket_read(c->rbuf, c->rbuflen, c->getfn); - - if (msg->qos == 1) - { - if (pktype == PUBACK) - { - unsigned short mypacketid; - unsigned char dup, type; - if (MQTTDeserialize_ack(&type, &dup, &mypacketid, c->rbuf, c->rbuflen) != 1) - rc = -1; - } - else - rc = -1; - } - else if (msg->qos == 2) - { - if (pktype == PUBCOMP) - { - unsigned short mypacketid; - unsigned char dup, type; - if (MQTTDeserialize_ack(&type, &dup, &mypacketid, c->rbuf, c->rbuflen) != 1) - rc = -1; - } - else - rc = -1; - } - -exit: - return rc; -} - -int mqtt_netconnect(char *addr, int port) -{ - struct hostent *host = 0; - struct sockaddr_in sockaddr; - int sock; - - host = gethostbyname(addr); - if (host == 0) - return -1; - - if ((sock = socket(AF_INET, SOCK_STREAM, 0)) == -1) - { - return -1; - } - - sockaddr.sin_family = AF_INET; - sockaddr.sin_port = htons(port); - sockaddr.sin_addr = *((struct in_addr *)host->h_addr); - memset(&(sockaddr.sin_zero), 0, sizeof(sockaddr.sin_zero)); - - if (connect(sock, (struct sockaddr *)&sockaddr, sizeof(struct sockaddr)) == -1) - { - #if (OS_RTTHREAD == 0) - close(sock); - #else - closesocket(sock); - #endif - return -2; - } - - return sock; -} - -void mqtt_netdisconnect(int *sockfd) -{ - #if (OS_RTTHREAD == 0) - close(*sockfd); - #else - closesocket(*sockfd); - #endif - *sockfd = -1; -} - -int mqtt_connect(mqtt_client_t *c, MQTTPacket_connectData *data) -{ - int rc = -1; - int len; - - len = MQTTSerialize_connect(c->wbuf, c->wbuflen, data); - if (len <= 0) - goto exit; - - rc = mqtt_write(c->sockfd, c->wbuf, len); - if (rc < 0) - goto exit; - - rc = MQTTPacket_read(c->rbuf, c->rbuflen, c->getfn); - if (rc < 0) - goto exit; - - if (rc == CONNACK) - { - unsigned char sessionPresent, connack_rc; - - if (MQTTDeserialize_connack(&sessionPresent, &connack_rc, c->rbuf, c->rbuflen) == 1) - { - rc = connack_rc; - } - else - { - rc = -1; - } - } - else - rc = -1; - -exit: - return rc; -} -int mqtt_disconnect(mqtt_client_t *c) -{ - int rc = -1; - int len; - - len = MQTTSerialize_disconnect(c->wbuf, c->wbuflen); - if (len > 0) - { - rc = mqtt_write(c->sockfd, c->wbuf, len); - } - - return rc; -} - -static void *mqtt_pub_thread(void *param) -{ - MQTTPacket_connectData condata = MQTTPacket_connectData_initializer; - int rc; - unsigned char rbuf[64]; - unsigned char wbuf[64]; - - DEBUG("pub thread start\n"); - /* */ - _cpub.rbuf = rbuf; - _cpub.rbuflen = sizeof(rbuf); - _cpub.wbuf = wbuf; - _cpub.wbuflen = sizeof(wbuf); - _cpub.getfn = pub_read; - - if ((_cpub.sockfd = mqtt_netconnect(HOSTNAME, HOSTPORT)) < 0) - { - DEBUG("pub netconnet fail\n"); - - return 0; - } - - DEBUG("pub connect to: %s %d\n", HOSTNAME, HOSTPORT); - - condata.clientID.cstring = "mqttpub"; - condata.keepAliveInterval = KEEPALIVE_INTERVAL; - condata.cleansession = 1; - condata.username.cstring = USERNAME; - condata.password.cstring = PASSWORD; - - rc = mqtt_connect(&_cpub, &condata); - if (rc < 0) - goto exit; - - DEBUG("pub connect ok\n"); - - mqtt_ping_start(&_cpub.sockfd); - - while (rc == 0) - { - mqtt_msg_t msg; - - SLEEP(5); - msg.dup = 0; - msg.id = 0; - msg.qos = 0; - msg.retained = 0; - msg.payload = (unsigned char*)"RT-Thread"; - msg.payloadlen = strlen((const char*)msg.payload); - - rc = mqtt_publish(&_cpub, TOPIC, &msg); - } - -exit: - mqtt_netdisconnect(&_cpub.sockfd); - DEBUG("pub thread exit\n"); - - return 0; -} - -void msgprocess(MQTTString *topic, mqtt_msg_t *msg) -{ - msg->payload[msg->payloadlen] = 0; - DEBUG("recv: size = %d, msg = %s\n", msg->payloadlen, msg->payload); -} - -static void *mqtt_sub_thread(void *param) -{ - MQTTPacket_connectData condata = MQTTPacket_connectData_initializer; - int pktype, rc, len; - int failcnt = 0; - unsigned char wbuf[64]; - unsigned char rbuf[64]; - - DEBUG("sub thread start\n"); - - _csub.wbuf = wbuf; - _csub.wbuflen = sizeof(wbuf); - _csub.rbuf = rbuf; - _csub.rbuflen = sizeof(rbuf); - _csub.getfn = sub_read; - - if ((_csub.sockfd = mqtt_netconnect(HOSTNAME, HOSTPORT)) < 0) - { - DEBUG("sub netconnect fail\n"); - - return 0; - } - - DEBUG("sub connect to: %s %d\n", HOSTNAME, HOSTPORT); - - condata.clientID.cstring = "mqttsub"; - condata.keepAliveInterval = KEEPALIVE_INTERVAL; - condata.cleansession = 1; - condata.username.cstring = USERNAME; - condata.password.cstring = PASSWORD; - - rc = mqtt_connect(&_csub, &condata); - if (rc < 0) - goto exit; - - DEBUG("sub connect ok\n"); - - rc = mqtt_subscribe(&_csub, TOPIC, 0); - if (rc < 0) - goto exit; - - DEBUG("sub topic: %s\n", TOPIC); - - mqtt_ping_start(&_csub.sockfd); - - while (1) - { - pktype = MQTTPacket_read(_csub.rbuf, _csub.rbuflen, sub_read); - - switch (pktype) - { - case CONNACK: - case PUBACK: - case SUBACK: - break; - case PUBLISH: - { - MQTTString topic; - mqtt_msg_t msg; - - if (MQTTDeserialize_publish(&msg.dup, &msg.qos, &msg.retained, &msg.id, &topic, - &msg.payload, &msg.payloadlen, _csub.rbuf, _csub.rbuflen) != 1) - goto exit; - - msgprocess(&topic, &msg); - - if (msg.qos != 0) - { - if (msg.qos == 1) - len = MQTTSerialize_ack(_csub.wbuf, _csub.wbuflen, PUBACK, 0, msg.id); - else if (msg.qos == 2) - len = MQTTSerialize_ack(_csub.wbuf, _csub.wbuflen, PUBREC, 0, msg.id); - - if (len <= 0) - rc = -1; - else - rc = mqtt_write(_csub.sockfd, _csub.wbuf, len); - - if (rc == -1) - goto exit; - } - } - break; - case PUBCOMP: - break; - case PINGRESP: - failcnt = 0; - break; - case -1: - if (++failcnt > KEEPALIVE_INTERVAL) - { - /* */ - goto exit; - } - break; - } - } - /* */ - mqtt_disconnect(&_csub); - -exit: - mqtt_netdisconnect(&_csub.sockfd); - DEBUG("sub thread exit\n"); - - return 0; -} - -void mqtt_client_init(void) -{ - #if (OS_RTTHREAD == 0) - pthread_t tid; - - pthread_create(&tid, NULL, mqtt_sub_thread, NULL); - pthread_create(&tid, NULL, mqtt_pub_thread, NULL); - #else - sys_thread_new("sub", - mqtt_sub_thread, - NULL, - 1024, - 20); - sys_thread_new("pub", - mqtt_pub_thread, - NULL, - 1024, - 20); - #endif -} - -#ifdef RT_USING_FINSH -#include <finsh.h> -FINSH_FUNCTION_EXPORT(mqtt_client_init, MQTT Pub/Sub Test); -#endif |