diff --git a/lib/CMakeLists.txt b/lib/CMakeLists.txt index 7cb06035..d537e776 100644 --- a/lib/CMakeLists.txt +++ b/lib/CMakeLists.txt @@ -32,6 +32,9 @@ include_directories(${mosquitto_SOURCE_DIR} ${mosquitto_SOURCE_DIR}/lib link_directories(${mosquitto_SOURCE_DIR}/lib) set(C_SRC + actions.c + callbacks.c + connect.c handle_connack.c handle_ping.c handle_pubackcomp.c @@ -42,12 +45,14 @@ set(C_SRC handle_unsuback.c helpers.c logging_mosq.c logging_mosq.h + loop.c memory_mosq.c memory_mosq.h messages_mosq.c messages_mosq.h mosquitto.c mosquitto.h mosquitto_internal.h mqtt3_protocol.h net_mosq.c net_mosq.h + options.c packet_mosq.c packet_mosq.h read_handle.c read_handle.h send_connect.c diff --git a/lib/Makefile b/lib/Makefile index e2a71939..9e538195 100644 --- a/lib/Makefile +++ b/lib/Makefile @@ -3,6 +3,9 @@ include ../config.mk .PHONY : really clean install MOSQ_OBJS=mosquitto.o \ + actions.o \ + callbacks.o \ + connect.o \ handle_connack.o \ handle_ping.o \ handle_pubackcomp.o \ @@ -13,9 +16,11 @@ MOSQ_OBJS=mosquitto.o \ handle_unsuback.o \ helpers.o \ logging_mosq.o \ + loop.o \ memory_mosq.o \ messages_mosq.o \ net_mosq.o \ + options.o \ packet_mosq.o \ read_handle.o \ send_connect.o \ @@ -72,7 +77,16 @@ libmosquitto.so.${SOVERSION} : ${MOSQ_OBJS} libmosquitto.a : ${MOSQ_OBJS} ${CROSS_COMPILE}$(AR) cr $@ $^ -mosquitto.o : mosquitto.c mosquitto.h +mosquitto.o : mosquitto.c mosquitto.h mosquitto_internal.h + ${CROSS_COMPILE}$(CC) $(LIB_CFLAGS) -c $< -o $@ + +actions.o : actions.c mosquitto.h mosquitto_internal.h + ${CROSS_COMPILE}$(CC) $(LIB_CFLAGS) -c $< -o $@ + +callbacks.o : callbacks.c mosquitto.h mosquitto_internal.h + ${CROSS_COMPILE}$(CC) $(LIB_CFLAGS) -c $< -o $@ + +connect.o : connect.c mosquitto.h mosquitto_internal.h ${CROSS_COMPILE}$(CC) $(LIB_CFLAGS) -c $< -o $@ handle_connack.o : handle_connack.c read_handle.h @@ -105,6 +119,9 @@ helpers.o : helpers.c logging_mosq.o : logging_mosq.c logging_mosq.h ${CROSS_COMPILE}$(CC) $(LIB_CFLAGS) -c $< -o $@ +loop.o : loop.c mosquitto.h mosquitto_internal.h + ${CROSS_COMPILE}$(CC) $(LIB_CFLAGS) -c $< -o $@ + messages_mosq.o : messages_mosq.c messages_mosq.h ${CROSS_COMPILE}$(CC) $(LIB_CFLAGS) -c $< -o $@ @@ -114,6 +131,9 @@ memory_mosq.o : memory_mosq.c memory_mosq.h net_mosq.o : net_mosq.c net_mosq.h ${CROSS_COMPILE}$(CC) $(LIB_CFLAGS) -c $< -o $@ +options.o : options.c mosquitto.h mosquitto_internal.h + ${CROSS_COMPILE}$(CC) $(LIB_CFLAGS) -c $< -o $@ + packet_mosq.o : packet_mosq.c packet_mosq.h ${CROSS_COMPILE}$(CC) $(LIB_CFLAGS) -c $< -o $@ diff --git a/lib/actions.c b/lib/actions.c new file mode 100644 index 00000000..6a9a48ff --- /dev/null +++ b/lib/actions.c @@ -0,0 +1,120 @@ +/* +Copyright (c) 2010-2018 Roger Light + +All rights reserved. This program and the accompanying materials +are made available under the terms of the Eclipse Public License v1.0 +and Eclipse Distribution License v1.0 which accompany this distribution. + +The Eclipse Public License is available at + http://www.eclipse.org/legal/epl-v10.html +and the Eclipse Distribution License is available at + http://www.eclipse.org/org/documents/edl-v10.php. + +Contributors: + Roger Light - initial implementation and documentation. +*/ + +#include "config.h" + +#include "mosquitto.h" +#include "mosquitto_internal.h" +#include "memory_mosq.h" +#include "messages_mosq.h" +#include "mqtt3_protocol.h" +#include "net_mosq.h" +#include "send_mosq.h" +#include "util_mosq.h" + + +int mosquitto_publish(struct mosquitto *mosq, int *mid, const char *topic, int payloadlen, const void *payload, int qos, bool retain) +{ + struct mosquitto_message_all *message; + uint16_t local_mid; + int queue_status; + + if(!mosq || !topic || qos<0 || qos>2) return MOSQ_ERR_INVAL; + if(STREMPTY(topic)) return MOSQ_ERR_INVAL; + if(mosquitto_validate_utf8(topic, strlen(topic))) return MOSQ_ERR_MALFORMED_UTF8; + if(payloadlen < 0 || payloadlen > MQTT_MAX_PAYLOAD) return MOSQ_ERR_PAYLOAD_SIZE; + + if(mosquitto_pub_topic_check(topic) != MOSQ_ERR_SUCCESS){ + return MOSQ_ERR_INVAL; + } + + local_mid = mosquitto__mid_generate(mosq); + if(mid){ + *mid = local_mid; + } + + if(qos == 0){ + return send__publish(mosq, local_mid, topic, payloadlen, payload, qos, retain, false); + }else{ + message = mosquitto__calloc(1, sizeof(struct mosquitto_message_all)); + if(!message) return MOSQ_ERR_NOMEM; + + message->next = NULL; + message->timestamp = mosquitto_time(); + message->msg.mid = local_mid; + message->msg.topic = mosquitto__strdup(topic); + if(!message->msg.topic){ + message__cleanup(&message); + return MOSQ_ERR_NOMEM; + } + if(payloadlen){ + message->msg.payloadlen = payloadlen; + message->msg.payload = mosquitto__malloc(payloadlen*sizeof(uint8_t)); + if(!message->msg.payload){ + message__cleanup(&message); + return MOSQ_ERR_NOMEM; + } + memcpy(message->msg.payload, payload, payloadlen*sizeof(uint8_t)); + }else{ + message->msg.payloadlen = 0; + message->msg.payload = NULL; + } + message->msg.qos = qos; + message->msg.retain = retain; + message->dup = false; + + pthread_mutex_lock(&mosq->out_message_mutex); + queue_status = message__queue(mosq, message, mosq_md_out); + if(queue_status == 0){ + if(qos == 1){ + message->state = mosq_ms_wait_for_puback; + }else if(qos == 2){ + message->state = mosq_ms_wait_for_pubrec; + } + pthread_mutex_unlock(&mosq->out_message_mutex); + return send__publish(mosq, message->msg.mid, message->msg.topic, message->msg.payloadlen, message->msg.payload, message->msg.qos, message->msg.retain, message->dup); + }else{ + message->state = mosq_ms_invalid; + pthread_mutex_unlock(&mosq->out_message_mutex); + return MOSQ_ERR_SUCCESS; + } + } +} + + +int mosquitto_subscribe(struct mosquitto *mosq, int *mid, const char *sub, int qos) +{ + if(!mosq) return MOSQ_ERR_INVAL; + if(mosq->sock == INVALID_SOCKET) return MOSQ_ERR_NO_CONN; + + if(mosquitto_sub_topic_check(sub)) return MOSQ_ERR_INVAL; + if(mosquitto_validate_utf8(sub, strlen(sub))) return MOSQ_ERR_MALFORMED_UTF8; + + return send__subscribe(mosq, mid, sub, qos); +} + + +int mosquitto_unsubscribe(struct mosquitto *mosq, int *mid, const char *sub) +{ + if(!mosq) return MOSQ_ERR_INVAL; + if(mosq->sock == INVALID_SOCKET) return MOSQ_ERR_NO_CONN; + + if(mosquitto_sub_topic_check(sub)) return MOSQ_ERR_INVAL; + if(mosquitto_validate_utf8(sub, strlen(sub))) return MOSQ_ERR_MALFORMED_UTF8; + + return send__unsubscribe(mosq, mid, sub); +} + diff --git a/lib/callbacks.c b/lib/callbacks.c new file mode 100644 index 00000000..037ee82b --- /dev/null +++ b/lib/callbacks.c @@ -0,0 +1,78 @@ +/* +Copyright (c) 2010-2018 Roger Light + +All rights reserved. This program and the accompanying materials +are made available under the terms of the Eclipse Public License v1.0 +and Eclipse Distribution License v1.0 which accompany this distribution. + +The Eclipse Public License is available at + http://www.eclipse.org/legal/epl-v10.html +and the Eclipse Distribution License is available at + http://www.eclipse.org/org/documents/edl-v10.php. + +Contributors: + Roger Light - initial implementation and documentation. +*/ + +#include "config.h" + +#include "mosquitto.h" +#include "mosquitto_internal.h" + + +void mosquitto_connect_callback_set(struct mosquitto *mosq, void (*on_connect)(struct mosquitto *, void *, int)) +{ + pthread_mutex_lock(&mosq->callback_mutex); + mosq->on_connect = on_connect; + pthread_mutex_unlock(&mosq->callback_mutex); +} + +void mosquitto_connect_with_flags_callback_set(struct mosquitto *mosq, void (*on_connect)(struct mosquitto *, void *, int, int)) +{ + pthread_mutex_lock(&mosq->callback_mutex); + mosq->on_connect_with_flags = on_connect; + pthread_mutex_unlock(&mosq->callback_mutex); +} + +void mosquitto_disconnect_callback_set(struct mosquitto *mosq, void (*on_disconnect)(struct mosquitto *, void *, int)) +{ + pthread_mutex_lock(&mosq->callback_mutex); + mosq->on_disconnect = on_disconnect; + pthread_mutex_unlock(&mosq->callback_mutex); +} + +void mosquitto_publish_callback_set(struct mosquitto *mosq, void (*on_publish)(struct mosquitto *, void *, int)) +{ + pthread_mutex_lock(&mosq->callback_mutex); + mosq->on_publish = on_publish; + pthread_mutex_unlock(&mosq->callback_mutex); +} + +void mosquitto_message_callback_set(struct mosquitto *mosq, void (*on_message)(struct mosquitto *, void *, const struct mosquitto_message *)) +{ + pthread_mutex_lock(&mosq->callback_mutex); + mosq->on_message = on_message; + pthread_mutex_unlock(&mosq->callback_mutex); +} + +void mosquitto_subscribe_callback_set(struct mosquitto *mosq, void (*on_subscribe)(struct mosquitto *, void *, int, int, const int *)) +{ + pthread_mutex_lock(&mosq->callback_mutex); + mosq->on_subscribe = on_subscribe; + pthread_mutex_unlock(&mosq->callback_mutex); +} + +void mosquitto_unsubscribe_callback_set(struct mosquitto *mosq, void (*on_unsubscribe)(struct mosquitto *, void *, int)) +{ + pthread_mutex_lock(&mosq->callback_mutex); + mosq->on_unsubscribe = on_unsubscribe; + pthread_mutex_unlock(&mosq->callback_mutex); +} + +void mosquitto_log_callback_set(struct mosquitto *mosq, void (*on_log)(struct mosquitto *, void *, int, const char *)) +{ + pthread_mutex_lock(&mosq->log_callback_mutex); + mosq->on_log = on_log; + pthread_mutex_unlock(&mosq->log_callback_mutex); +} + diff --git a/lib/connect.c b/lib/connect.c new file mode 100644 index 00000000..ab14fd48 --- /dev/null +++ b/lib/connect.c @@ -0,0 +1,209 @@ +/* +Copyright (c) 2010-2018 Roger Light + +All rights reserved. This program and the accompanying materials +are made available under the terms of the Eclipse Public License v1.0 +and Eclipse Distribution License v1.0 which accompany this distribution. + +The Eclipse Public License is available at + http://www.eclipse.org/legal/epl-v10.html +and the Eclipse Distribution License is available at + http://www.eclipse.org/org/documents/edl-v10.php. + +Contributors: + Roger Light - initial implementation and documentation. +*/ + +#include "config.h" + +#include "mosquitto.h" +#include "mosquitto_internal.h" +#include "logging_mosq.h" +#include "messages_mosq.h" +#include "memory_mosq.h" +#include "packet_mosq.h" +#include "net_mosq.h" +#include "send_mosq.h" +#include "socks_mosq.h" + +static int mosquitto__reconnect(struct mosquitto *mosq, bool blocking); +static int mosquitto__connect_init(struct mosquitto *mosq, const char *host, int port, int keepalive, const char *bind_address); + + +static int mosquitto__connect_init(struct mosquitto *mosq, const char *host, int port, int keepalive, const char *bind_address) +{ + if(!mosq) return MOSQ_ERR_INVAL; + if(!host || port <= 0) return MOSQ_ERR_INVAL; + + mosquitto__free(mosq->host); + mosq->host = mosquitto__strdup(host); + if(!mosq->host) return MOSQ_ERR_NOMEM; + mosq->port = port; + + mosquitto__free(mosq->bind_address); + if(bind_address){ + mosq->bind_address = mosquitto__strdup(bind_address); + if(!mosq->bind_address) return MOSQ_ERR_NOMEM; + } + + mosq->keepalive = keepalive; + + if(mosq->sockpairR != INVALID_SOCKET){ + COMPAT_CLOSE(mosq->sockpairR); + mosq->sockpairR = INVALID_SOCKET; + } + if(mosq->sockpairW != INVALID_SOCKET){ + COMPAT_CLOSE(mosq->sockpairW); + mosq->sockpairW = INVALID_SOCKET; + } + + if(net__socketpair(&mosq->sockpairR, &mosq->sockpairW)){ + log__printf(mosq, MOSQ_LOG_WARNING, + "Warning: Unable to open socket pair, outgoing publish commands may be delayed."); + } + + return MOSQ_ERR_SUCCESS; +} + + +int mosquitto_connect(struct mosquitto *mosq, const char *host, int port, int keepalive) +{ + return mosquitto_connect_bind(mosq, host, port, keepalive, NULL); +} + + +int mosquitto_connect_bind(struct mosquitto *mosq, const char *host, int port, int keepalive, const char *bind_address) +{ + int rc; + rc = mosquitto__connect_init(mosq, host, port, keepalive, bind_address); + if(rc) return rc; + + pthread_mutex_lock(&mosq->state_mutex); + mosq->state = mosq_cs_new; + pthread_mutex_unlock(&mosq->state_mutex); + + return mosquitto__reconnect(mosq, true); +} + + +int mosquitto_connect_async(struct mosquitto *mosq, const char *host, int port, int keepalive) +{ + return mosquitto_connect_bind_async(mosq, host, port, keepalive, NULL); +} + + +int mosquitto_connect_bind_async(struct mosquitto *mosq, const char *host, int port, int keepalive, const char *bind_address) +{ + int rc = mosquitto__connect_init(mosq, host, port, keepalive, bind_address); + if(rc) return rc; + + pthread_mutex_lock(&mosq->state_mutex); + mosq->state = mosq_cs_connect_async; + pthread_mutex_unlock(&mosq->state_mutex); + + return mosquitto__reconnect(mosq, false); +} + + +int mosquitto_reconnect_async(struct mosquitto *mosq) +{ + return mosquitto__reconnect(mosq, false); +} + + +int mosquitto_reconnect(struct mosquitto *mosq) +{ + return mosquitto__reconnect(mosq, true); +} + + +static int mosquitto__reconnect(struct mosquitto *mosq, bool blocking) +{ + int rc; + struct mosquitto__packet *packet; + if(!mosq) return MOSQ_ERR_INVAL; + if(!mosq->host || mosq->port <= 0) return MOSQ_ERR_INVAL; + + pthread_mutex_lock(&mosq->state_mutex); +#ifdef WITH_SOCKS + if(mosq->socks5_host){ + mosq->state = mosq_cs_socks5_new; + }else +#endif + { + mosq->state = mosq_cs_new; + } + pthread_mutex_unlock(&mosq->state_mutex); + + pthread_mutex_lock(&mosq->msgtime_mutex); + mosq->last_msg_in = mosquitto_time(); + mosq->next_msg_out = mosq->last_msg_in + mosq->keepalive; + pthread_mutex_unlock(&mosq->msgtime_mutex); + + mosq->ping_t = 0; + + packet__cleanup(&mosq->in_packet); + + pthread_mutex_lock(&mosq->current_out_packet_mutex); + pthread_mutex_lock(&mosq->out_packet_mutex); + + if(mosq->out_packet && !mosq->current_out_packet){ + mosq->current_out_packet = mosq->out_packet; + mosq->out_packet = mosq->out_packet->next; + } + + while(mosq->current_out_packet){ + packet = mosq->current_out_packet; + /* Free data and reset values */ + mosq->current_out_packet = mosq->out_packet; + if(mosq->out_packet){ + mosq->out_packet = mosq->out_packet->next; + } + + packet__cleanup(packet); + mosquitto__free(packet); + } + pthread_mutex_unlock(&mosq->out_packet_mutex); + pthread_mutex_unlock(&mosq->current_out_packet_mutex); + + message__reconnect_reset(mosq); + + if(mosq->sock != INVALID_SOCKET){ + net__socket_close(mosq); //close socket + } + +#ifdef WITH_SOCKS + if(mosq->socks5_host){ + rc = net__socket_connect(mosq, mosq->socks5_host, mosq->socks5_port, mosq->bind_address, blocking); + }else +#endif + { + rc = net__socket_connect(mosq, mosq->host, mosq->port, mosq->bind_address, blocking); + } + if(rc>0){ + return rc; + } + +#ifdef WITH_SOCKS + if(mosq->socks5_host){ + return socks5__send(mosq); + }else +#endif + { + return send__connect(mosq, mosq->keepalive, mosq->clean_session); + } +} + + +int mosquitto_disconnect(struct mosquitto *mosq) +{ + if(!mosq) return MOSQ_ERR_INVAL; + + pthread_mutex_lock(&mosq->state_mutex); + mosq->state = mosq_cs_disconnecting; + pthread_mutex_unlock(&mosq->state_mutex); + + if(mosq->sock == INVALID_SOCKET) return MOSQ_ERR_NO_CONN; + return send__disconnect(mosq); +} + diff --git a/lib/loop.c b/lib/loop.c new file mode 100644 index 00000000..9c259790 --- /dev/null +++ b/lib/loop.c @@ -0,0 +1,400 @@ +/* +Copyright (c) 2010-2018 Roger Light + +All rights reserved. This program and the accompanying materials +are made available under the terms of the Eclipse Public License v1.0 +and Eclipse Distribution License v1.0 which accompany this distribution. + +The Eclipse Public License is available at + http://www.eclipse.org/legal/epl-v10.html +and the Eclipse Distribution License is available at + http://www.eclipse.org/org/documents/edl-v10.php. + +Contributors: + Roger Light - initial implementation and documentation. +*/ + +#include "config.h" + +#include + +#include "mosquitto.h" +#include "mosquitto_internal.h" +#include "net_mosq.h" +#include "packet_mosq.h" +#include "socks_mosq.h" +#include "tls_mosq.h" +#include "util_mosq.h" + +#if !defined(WIN32) && !defined(__SYMBIAN32__) +#define HAVE_PSELECT +#endif + +int mosquitto_loop(struct mosquitto *mosq, int timeout, int max_packets) +{ +#ifdef HAVE_PSELECT + struct timespec local_timeout; +#else + struct timeval local_timeout; +#endif + fd_set readfds, writefds; + int fdcount; + int rc; + char pairbuf; + int maxfd = 0; + time_t now; + + if(!mosq || max_packets < 1) return MOSQ_ERR_INVAL; +#ifndef WIN32 + if(mosq->sock >= FD_SETSIZE || mosq->sockpairR >= FD_SETSIZE){ + return MOSQ_ERR_INVAL; + } +#endif + + FD_ZERO(&readfds); + FD_ZERO(&writefds); + if(mosq->sock != INVALID_SOCKET){ + maxfd = mosq->sock; + FD_SET(mosq->sock, &readfds); + pthread_mutex_lock(&mosq->current_out_packet_mutex); + pthread_mutex_lock(&mosq->out_packet_mutex); + if(mosq->out_packet || mosq->current_out_packet){ + FD_SET(mosq->sock, &writefds); + } +#ifdef WITH_TLS + if(mosq->ssl){ + if(mosq->want_write){ + FD_SET(mosq->sock, &writefds); + }else if(mosq->want_connect){ + /* Remove possible FD_SET from above, we don't want to check + * for writing if we are still connecting, unless want_write is + * definitely set. The presence of outgoing packets does not + * matter yet. */ + FD_CLR(mosq->sock, &writefds); + } + } +#endif + pthread_mutex_unlock(&mosq->out_packet_mutex); + pthread_mutex_unlock(&mosq->current_out_packet_mutex); + }else{ +#ifdef WITH_SRV + if(mosq->achan){ + pthread_mutex_lock(&mosq->state_mutex); + if(mosq->state == mosq_cs_connect_srv){ + rc = ares_fds(mosq->achan, &readfds, &writefds); + if(rc > maxfd){ + maxfd = rc; + } + }else{ + pthread_mutex_unlock(&mosq->state_mutex); + return MOSQ_ERR_NO_CONN; + } + pthread_mutex_unlock(&mosq->state_mutex); + } +#else + return MOSQ_ERR_NO_CONN; +#endif + } + if(mosq->sockpairR != INVALID_SOCKET){ + /* sockpairR is used to break out of select() before the timeout, on a + * call to publish() etc. */ + FD_SET(mosq->sockpairR, &readfds); + if(mosq->sockpairR > maxfd){ + maxfd = mosq->sockpairR; + } + } + + if(timeout < 0){ + timeout = 1000; + } + + now = mosquitto_time(); + if(mosq->next_msg_out && now + timeout/1000 > mosq->next_msg_out){ + timeout = (mosq->next_msg_out - now)*1000; + } + + if(timeout < 0){ + /* There has been a delay somewhere which means we should have already + * sent a message. */ + timeout = 0; + } + + local_timeout.tv_sec = timeout/1000; +#ifdef HAVE_PSELECT + local_timeout.tv_nsec = (timeout-local_timeout.tv_sec*1000)*1e6; +#else + local_timeout.tv_usec = (timeout-local_timeout.tv_sec*1000)*1000; +#endif + +#ifdef HAVE_PSELECT + fdcount = pselect(maxfd+1, &readfds, &writefds, NULL, &local_timeout, NULL); +#else + fdcount = select(maxfd+1, &readfds, &writefds, NULL, &local_timeout); +#endif + if(fdcount == -1){ +#ifdef WIN32 + errno = WSAGetLastError(); +#endif + if(errno == EINTR){ + return MOSQ_ERR_SUCCESS; + }else{ + return MOSQ_ERR_ERRNO; + } + }else{ + if(mosq->sock != INVALID_SOCKET){ + if(FD_ISSET(mosq->sock, &readfds)){ +#ifdef WITH_TLS + if(mosq->want_connect){ + rc = net__socket_connect_tls(mosq); + if(rc) return rc; + }else +#endif + { + do{ + rc = mosquitto_loop_read(mosq, max_packets); + if(rc || mosq->sock == INVALID_SOCKET){ + return rc; + } + }while(SSL_DATA_PENDING(mosq)); + } + } + if(mosq->sockpairR != INVALID_SOCKET && FD_ISSET(mosq->sockpairR, &readfds)){ +#ifndef WIN32 + if(read(mosq->sockpairR, &pairbuf, 1) == 0){ + } +#else + recv(mosq->sockpairR, &pairbuf, 1, 0); +#endif + /* Fake write possible, to stimulate output write even though + * we didn't ask for it, because at that point the publish or + * other command wasn't present. */ + if(mosq->sock != INVALID_SOCKET) + FD_SET(mosq->sock, &writefds); + } + if(mosq->sock != INVALID_SOCKET && FD_ISSET(mosq->sock, &writefds)){ +#ifdef WITH_TLS + if(mosq->want_connect){ + rc = net__socket_connect_tls(mosq); + if(rc) return rc; + }else +#endif + { + rc = mosquitto_loop_write(mosq, max_packets); + if(rc || mosq->sock == INVALID_SOCKET){ + return rc; + } + } + } + } +#ifdef WITH_SRV + if(mosq->achan){ + ares_process(mosq->achan, &readfds, &writefds); + } +#endif + } + return mosquitto_loop_misc(mosq); +} + + +int mosquitto_loop_forever(struct mosquitto *mosq, int timeout, int max_packets) +{ + int run = 1; + int rc; + unsigned int reconnects = 0; + unsigned long reconnect_delay; + + if(!mosq) return MOSQ_ERR_INVAL; + + if(mosq->state == mosq_cs_connect_async){ + mosquitto_reconnect(mosq); + } + + while(run){ + do{ + rc = mosquitto_loop(mosq, timeout, max_packets); + if (reconnects !=0 && rc == MOSQ_ERR_SUCCESS){ + reconnects = 0; + } + }while(run && rc == MOSQ_ERR_SUCCESS); + /* Quit after fatal errors. */ + switch(rc){ + case MOSQ_ERR_NOMEM: + case MOSQ_ERR_PROTOCOL: + case MOSQ_ERR_INVAL: + case MOSQ_ERR_NOT_FOUND: + case MOSQ_ERR_TLS: + case MOSQ_ERR_PAYLOAD_SIZE: + case MOSQ_ERR_NOT_SUPPORTED: + case MOSQ_ERR_AUTH: + case MOSQ_ERR_ACL_DENIED: + case MOSQ_ERR_UNKNOWN: + case MOSQ_ERR_EAI: + case MOSQ_ERR_PROXY: + return rc; + case MOSQ_ERR_ERRNO: + break; + } + if(errno == EPROTO){ + return rc; + } + do{ + rc = MOSQ_ERR_SUCCESS; + pthread_mutex_lock(&mosq->state_mutex); + if(mosq->state == mosq_cs_disconnecting){ + run = 0; + pthread_mutex_unlock(&mosq->state_mutex); + }else{ + pthread_mutex_unlock(&mosq->state_mutex); + + if(mosq->reconnect_delay > 0 && mosq->reconnect_exponential_backoff){ + reconnect_delay = mosq->reconnect_delay*reconnects*reconnects; + }else{ + reconnect_delay = mosq->reconnect_delay; + } + + if(reconnect_delay > mosq->reconnect_delay_max){ + reconnect_delay = mosq->reconnect_delay_max; + }else{ + reconnects++; + } + +#ifdef WIN32 + Sleep(reconnect_delay*1000); +#else + sleep(reconnect_delay); +#endif + + pthread_mutex_lock(&mosq->state_mutex); + if(mosq->state == mosq_cs_disconnecting){ + run = 0; + pthread_mutex_unlock(&mosq->state_mutex); + }else{ + pthread_mutex_unlock(&mosq->state_mutex); + rc = mosquitto_reconnect(mosq); + } + } + }while(run && rc != MOSQ_ERR_SUCCESS); + } + return rc; +} + + +int mosquitto_loop_misc(struct mosquitto *mosq) +{ + time_t now; + int rc; + + if(!mosq) return MOSQ_ERR_INVAL; + if(mosq->sock == INVALID_SOCKET) return MOSQ_ERR_NO_CONN; + + mosquitto__check_keepalive(mosq); + now = mosquitto_time(); + if(mosq->ping_t && now - mosq->ping_t >= mosq->keepalive){ + /* mosq->ping_t != 0 means we are waiting for a pingresp. + * This hasn't happened in the keepalive time so we should disconnect. + */ + net__socket_close(mosq); + pthread_mutex_lock(&mosq->state_mutex); + if(mosq->state == mosq_cs_disconnecting){ + rc = MOSQ_ERR_SUCCESS; + }else{ + rc = 1; + } + pthread_mutex_unlock(&mosq->state_mutex); + pthread_mutex_lock(&mosq->callback_mutex); + if(mosq->on_disconnect){ + mosq->in_callback = true; + mosq->on_disconnect(mosq, mosq->userdata, rc); + mosq->in_callback = false; + } + pthread_mutex_unlock(&mosq->callback_mutex); + return MOSQ_ERR_CONN_LOST; + } + return MOSQ_ERR_SUCCESS; +} + + +static int mosquitto__loop_rc_handle(struct mosquitto *mosq, int rc) +{ + if(rc){ + net__socket_close(mosq); + pthread_mutex_lock(&mosq->state_mutex); + if(mosq->state == mosq_cs_disconnecting){ + rc = MOSQ_ERR_SUCCESS; + } + pthread_mutex_unlock(&mosq->state_mutex); + pthread_mutex_lock(&mosq->callback_mutex); + if(mosq->on_disconnect){ + mosq->in_callback = true; + mosq->on_disconnect(mosq, mosq->userdata, rc); + mosq->in_callback = false; + } + pthread_mutex_unlock(&mosq->callback_mutex); + return rc; + } + return rc; +} + + +int mosquitto_loop_read(struct mosquitto *mosq, int max_packets) +{ + int rc; + int i; + if(max_packets < 1) return MOSQ_ERR_INVAL; + + pthread_mutex_lock(&mosq->out_message_mutex); + max_packets = mosq->out_queue_len; + pthread_mutex_unlock(&mosq->out_message_mutex); + + pthread_mutex_lock(&mosq->in_message_mutex); + max_packets += mosq->in_queue_len; + pthread_mutex_unlock(&mosq->in_message_mutex); + + if(max_packets < 1) max_packets = 1; + /* Queue len here tells us how many messages are awaiting processing and + * have QoS > 0. We should try to deal with that many in this loop in order + * to keep up. */ + for(i=0; isocks5_host){ + rc = socks5__read(mosq); + }else +#endif + { + rc = packet__read(mosq); + } + if(rc || errno == EAGAIN || errno == COMPAT_EWOULDBLOCK){ + return mosquitto__loop_rc_handle(mosq, rc); + } + } + return rc; +} + + +int mosquitto_loop_write(struct mosquitto *mosq, int max_packets) +{ + int rc; + int i; + if(max_packets < 1) return MOSQ_ERR_INVAL; + + pthread_mutex_lock(&mosq->out_message_mutex); + max_packets = mosq->out_queue_len; + pthread_mutex_unlock(&mosq->out_message_mutex); + + pthread_mutex_lock(&mosq->in_message_mutex); + max_packets += mosq->in_queue_len; + pthread_mutex_unlock(&mosq->in_message_mutex); + + if(max_packets < 1) max_packets = 1; + /* Queue len here tells us how many messages are awaiting processing and + * have QoS > 0. We should try to deal with that many in this loop in order + * to keep up. */ + for(i=0; i +Copyright (c) 2010-2018 Roger Light All rights reserved. This program and the accompanying materials are made available under the terms of the Eclipse Public License v1.0 @@ -16,45 +16,19 @@ Contributors: #include "config.h" -#include #include #include -#include -#include -#ifndef WIN32 -#include -#include -#include -#include -#else -#include -#include -typedef int ssize_t; -#endif #include "mosquitto.h" #include "mosquitto_internal.h" -#include "logging_mosq.h" -#include "messages_mosq.h" #include "memory_mosq.h" -#include "mqtt3_protocol.h" +#include "messages_mosq.h" #include "net_mosq.h" #include "packet_mosq.h" -#include "read_handle.h" -#include "send_mosq.h" -#include "socks_mosq.h" -#include "time_mosq.h" -#include "tls_mosq.h" -#include "util_mosq.h" #include "will_mosq.h" -#if !defined(WIN32) && !defined(__SYMBIAN32__) -#define HAVE_PSELECT -#endif void mosquitto__destroy(struct mosquitto *mosq); -static int mosquitto__reconnect(struct mosquitto *mosq, bool blocking); -static int mosquitto__connect_init(struct mosquitto *mosq, const char *host, int port, int keepalive, const char *bind_address); int mosquitto_lib_version(int *major, int *minor, int *revision) { @@ -218,54 +192,6 @@ int mosquitto_reinitialise(struct mosquitto *mosq, const char *id, bool clean_se return MOSQ_ERR_SUCCESS; } -int mosquitto_will_set(struct mosquitto *mosq, const char *topic, int payloadlen, const void *payload, int qos, bool retain) -{ - if(!mosq) return MOSQ_ERR_INVAL; - return will__set(mosq, topic, payloadlen, payload, qos, retain); -} - -int mosquitto_will_clear(struct mosquitto *mosq) -{ - if(!mosq) return MOSQ_ERR_INVAL; - return will__clear(mosq); -} - -int mosquitto_username_pw_set(struct mosquitto *mosq, const char *username, const char *password) -{ - if(!mosq) return MOSQ_ERR_INVAL; - - mosquitto__free(mosq->username); - mosq->username = NULL; - - mosquitto__free(mosq->password); - mosq->password = NULL; - - if(username){ - mosq->username = mosquitto__strdup(username); - if(!mosq->username) return MOSQ_ERR_NOMEM; - if(password){ - mosq->password = mosquitto__strdup(password); - if(!mosq->password){ - mosquitto__free(mosq->username); - mosq->username = NULL; - return MOSQ_ERR_NOMEM; - } - } - } - return MOSQ_ERR_SUCCESS; -} - -int mosquitto_reconnect_delay_set(struct mosquitto *mosq, unsigned int reconnect_delay, unsigned int reconnect_delay_max, bool reconnect_exponential_backoff) -{ - if(!mosq) return MOSQ_ERR_INVAL; - - mosq->reconnect_delay = reconnect_delay; - mosq->reconnect_delay_max = reconnect_delay_max; - mosq->reconnect_exponential_backoff = reconnect_exponential_backoff; - - return MOSQ_ERR_SUCCESS; - -} void mosquitto__destroy(struct mosquitto *mosq) { @@ -377,804 +303,6 @@ int mosquitto_socket(struct mosquitto *mosq) return mosq->sock; } -static int mosquitto__connect_init(struct mosquitto *mosq, const char *host, int port, int keepalive, const char *bind_address) -{ - if(!mosq) return MOSQ_ERR_INVAL; - if(!host || port <= 0) return MOSQ_ERR_INVAL; - - mosquitto__free(mosq->host); - mosq->host = mosquitto__strdup(host); - if(!mosq->host) return MOSQ_ERR_NOMEM; - mosq->port = port; - - mosquitto__free(mosq->bind_address); - if(bind_address){ - mosq->bind_address = mosquitto__strdup(bind_address); - if(!mosq->bind_address) return MOSQ_ERR_NOMEM; - } - - mosq->keepalive = keepalive; - - if(mosq->sockpairR != INVALID_SOCKET){ - COMPAT_CLOSE(mosq->sockpairR); - mosq->sockpairR = INVALID_SOCKET; - } - if(mosq->sockpairW != INVALID_SOCKET){ - COMPAT_CLOSE(mosq->sockpairW); - mosq->sockpairW = INVALID_SOCKET; - } - - if(net__socketpair(&mosq->sockpairR, &mosq->sockpairW)){ - log__printf(mosq, MOSQ_LOG_WARNING, - "Warning: Unable to open socket pair, outgoing publish commands may be delayed."); - } - - return MOSQ_ERR_SUCCESS; -} - -int mosquitto_connect(struct mosquitto *mosq, const char *host, int port, int keepalive) -{ - return mosquitto_connect_bind(mosq, host, port, keepalive, NULL); -} - -int mosquitto_connect_bind(struct mosquitto *mosq, const char *host, int port, int keepalive, const char *bind_address) -{ - int rc; - rc = mosquitto__connect_init(mosq, host, port, keepalive, bind_address); - if(rc) return rc; - - pthread_mutex_lock(&mosq->state_mutex); - mosq->state = mosq_cs_new; - pthread_mutex_unlock(&mosq->state_mutex); - - return mosquitto__reconnect(mosq, true); -} - -int mosquitto_connect_async(struct mosquitto *mosq, const char *host, int port, int keepalive) -{ - return mosquitto_connect_bind_async(mosq, host, port, keepalive, NULL); -} - -int mosquitto_connect_bind_async(struct mosquitto *mosq, const char *host, int port, int keepalive, const char *bind_address) -{ - int rc = mosquitto__connect_init(mosq, host, port, keepalive, bind_address); - if(rc) return rc; - - pthread_mutex_lock(&mosq->state_mutex); - mosq->state = mosq_cs_connect_async; - pthread_mutex_unlock(&mosq->state_mutex); - - return mosquitto__reconnect(mosq, false); -} - -int mosquitto_reconnect_async(struct mosquitto *mosq) -{ - return mosquitto__reconnect(mosq, false); -} - -int mosquitto_reconnect(struct mosquitto *mosq) -{ - return mosquitto__reconnect(mosq, true); -} - -static int mosquitto__reconnect(struct mosquitto *mosq, bool blocking) -{ - int rc; - struct mosquitto__packet *packet; - if(!mosq) return MOSQ_ERR_INVAL; - if(!mosq->host || mosq->port <= 0) return MOSQ_ERR_INVAL; - - pthread_mutex_lock(&mosq->state_mutex); -#ifdef WITH_SOCKS - if(mosq->socks5_host){ - mosq->state = mosq_cs_socks5_new; - }else -#endif - { - mosq->state = mosq_cs_new; - } - pthread_mutex_unlock(&mosq->state_mutex); - - pthread_mutex_lock(&mosq->msgtime_mutex); - mosq->last_msg_in = mosquitto_time(); - mosq->next_msg_out = mosq->last_msg_in + mosq->keepalive; - pthread_mutex_unlock(&mosq->msgtime_mutex); - - mosq->ping_t = 0; - - packet__cleanup(&mosq->in_packet); - - pthread_mutex_lock(&mosq->current_out_packet_mutex); - pthread_mutex_lock(&mosq->out_packet_mutex); - - if(mosq->out_packet && !mosq->current_out_packet){ - mosq->current_out_packet = mosq->out_packet; - mosq->out_packet = mosq->out_packet->next; - } - - while(mosq->current_out_packet){ - packet = mosq->current_out_packet; - /* Free data and reset values */ - mosq->current_out_packet = mosq->out_packet; - if(mosq->out_packet){ - mosq->out_packet = mosq->out_packet->next; - } - - packet__cleanup(packet); - mosquitto__free(packet); - } - pthread_mutex_unlock(&mosq->out_packet_mutex); - pthread_mutex_unlock(&mosq->current_out_packet_mutex); - - message__reconnect_reset(mosq); - - if(mosq->sock != INVALID_SOCKET){ - net__socket_close(mosq); //close socket - } - -#ifdef WITH_SOCKS - if(mosq->socks5_host){ - rc = net__socket_connect(mosq, mosq->socks5_host, mosq->socks5_port, mosq->bind_address, blocking); - }else -#endif - { - rc = net__socket_connect(mosq, mosq->host, mosq->port, mosq->bind_address, blocking); - } - if(rc>0){ - return rc; - } - -#ifdef WITH_SOCKS - if(mosq->socks5_host){ - return socks5__send(mosq); - }else -#endif - { - return send__connect(mosq, mosq->keepalive, mosq->clean_session); - } -} - -int mosquitto_disconnect(struct mosquitto *mosq) -{ - if(!mosq) return MOSQ_ERR_INVAL; - - pthread_mutex_lock(&mosq->state_mutex); - mosq->state = mosq_cs_disconnecting; - pthread_mutex_unlock(&mosq->state_mutex); - - if(mosq->sock == INVALID_SOCKET) return MOSQ_ERR_NO_CONN; - return send__disconnect(mosq); -} - -int mosquitto_publish(struct mosquitto *mosq, int *mid, const char *topic, int payloadlen, const void *payload, int qos, bool retain) -{ - struct mosquitto_message_all *message; - uint16_t local_mid; - int queue_status; - - if(!mosq || !topic || qos<0 || qos>2) return MOSQ_ERR_INVAL; - if(STREMPTY(topic)) return MOSQ_ERR_INVAL; - if(mosquitto_validate_utf8(topic, strlen(topic))) return MOSQ_ERR_MALFORMED_UTF8; - if(payloadlen < 0 || payloadlen > MQTT_MAX_PAYLOAD) return MOSQ_ERR_PAYLOAD_SIZE; - - if(mosquitto_pub_topic_check(topic) != MOSQ_ERR_SUCCESS){ - return MOSQ_ERR_INVAL; - } - - local_mid = mosquitto__mid_generate(mosq); - if(mid){ - *mid = local_mid; - } - - if(qos == 0){ - return send__publish(mosq, local_mid, topic, payloadlen, payload, qos, retain, false); - }else{ - message = mosquitto__calloc(1, sizeof(struct mosquitto_message_all)); - if(!message) return MOSQ_ERR_NOMEM; - - message->next = NULL; - message->timestamp = mosquitto_time(); - message->msg.mid = local_mid; - message->msg.topic = mosquitto__strdup(topic); - if(!message->msg.topic){ - message__cleanup(&message); - return MOSQ_ERR_NOMEM; - } - if(payloadlen){ - message->msg.payloadlen = payloadlen; - message->msg.payload = mosquitto__malloc(payloadlen*sizeof(uint8_t)); - if(!message->msg.payload){ - message__cleanup(&message); - return MOSQ_ERR_NOMEM; - } - memcpy(message->msg.payload, payload, payloadlen*sizeof(uint8_t)); - }else{ - message->msg.payloadlen = 0; - message->msg.payload = NULL; - } - message->msg.qos = qos; - message->msg.retain = retain; - message->dup = false; - - pthread_mutex_lock(&mosq->out_message_mutex); - queue_status = message__queue(mosq, message, mosq_md_out); - if(queue_status == 0){ - if(qos == 1){ - message->state = mosq_ms_wait_for_puback; - }else if(qos == 2){ - message->state = mosq_ms_wait_for_pubrec; - } - pthread_mutex_unlock(&mosq->out_message_mutex); - return send__publish(mosq, message->msg.mid, message->msg.topic, message->msg.payloadlen, message->msg.payload, message->msg.qos, message->msg.retain, message->dup); - }else{ - message->state = mosq_ms_invalid; - pthread_mutex_unlock(&mosq->out_message_mutex); - return MOSQ_ERR_SUCCESS; - } - } -} - -int mosquitto_subscribe(struct mosquitto *mosq, int *mid, const char *sub, int qos) -{ - if(!mosq) return MOSQ_ERR_INVAL; - if(mosq->sock == INVALID_SOCKET) return MOSQ_ERR_NO_CONN; - - if(mosquitto_sub_topic_check(sub)) return MOSQ_ERR_INVAL; - if(mosquitto_validate_utf8(sub, strlen(sub))) return MOSQ_ERR_MALFORMED_UTF8; - - return send__subscribe(mosq, mid, sub, qos); -} - -int mosquitto_unsubscribe(struct mosquitto *mosq, int *mid, const char *sub) -{ - if(!mosq) return MOSQ_ERR_INVAL; - if(mosq->sock == INVALID_SOCKET) return MOSQ_ERR_NO_CONN; - - if(mosquitto_sub_topic_check(sub)) return MOSQ_ERR_INVAL; - if(mosquitto_validate_utf8(sub, strlen(sub))) return MOSQ_ERR_MALFORMED_UTF8; - - return send__unsubscribe(mosq, mid, sub); -} - -int mosquitto_tls_set(struct mosquitto *mosq, const char *cafile, const char *capath, const char *certfile, const char *keyfile, int (*pw_callback)(char *buf, int size, int rwflag, void *userdata)) -{ -#ifdef WITH_TLS - FILE *fptr; - - if(!mosq || (!cafile && !capath) || (certfile && !keyfile) || (!certfile && keyfile)) return MOSQ_ERR_INVAL; - - mosquitto__free(mosq->tls_cafile); - mosq->tls_cafile = NULL; - if(cafile){ - fptr = mosquitto__fopen(cafile, "rt", false); - if(fptr){ - fclose(fptr); - }else{ - return MOSQ_ERR_INVAL; - } - mosq->tls_cafile = mosquitto__strdup(cafile); - - if(!mosq->tls_cafile){ - return MOSQ_ERR_NOMEM; - } - } - - mosquitto__free(mosq->tls_capath); - mosq->tls_capath = NULL; - if(capath){ - mosq->tls_capath = mosquitto__strdup(capath); - if(!mosq->tls_capath){ - return MOSQ_ERR_NOMEM; - } - } - - mosquitto__free(mosq->tls_certfile); - mosq->tls_certfile = NULL; - if(certfile){ - fptr = mosquitto__fopen(certfile, "rt", false); - if(fptr){ - fclose(fptr); - }else{ - mosquitto__free(mosq->tls_cafile); - mosq->tls_cafile = NULL; - - mosquitto__free(mosq->tls_capath); - mosq->tls_capath = NULL; - return MOSQ_ERR_INVAL; - } - mosq->tls_certfile = mosquitto__strdup(certfile); - if(!mosq->tls_certfile){ - return MOSQ_ERR_NOMEM; - } - } - - mosquitto__free(mosq->tls_keyfile); - mosq->tls_keyfile = NULL; - if(keyfile){ - fptr = mosquitto__fopen(keyfile, "rt", false); - if(fptr){ - fclose(fptr); - }else{ - mosquitto__free(mosq->tls_cafile); - mosq->tls_cafile = NULL; - - mosquitto__free(mosq->tls_capath); - mosq->tls_capath = NULL; - - mosquitto__free(mosq->tls_certfile); - mosq->tls_certfile = NULL; - return MOSQ_ERR_INVAL; - } - mosq->tls_keyfile = mosquitto__strdup(keyfile); - if(!mosq->tls_keyfile){ - return MOSQ_ERR_NOMEM; - } - } - - mosq->tls_pw_callback = pw_callback; - - - return MOSQ_ERR_SUCCESS; -#else - return MOSQ_ERR_NOT_SUPPORTED; - -#endif -} - -int mosquitto_tls_opts_set(struct mosquitto *mosq, int cert_reqs, const char *tls_version, const char *ciphers) -{ -#ifdef WITH_TLS - if(!mosq) return MOSQ_ERR_INVAL; - - mosq->tls_cert_reqs = cert_reqs; - if(tls_version){ -#if OPENSSL_VERSION_NUMBER >= 0x10001000L - if(!strcasecmp(tls_version, "tlsv1.2") - || !strcasecmp(tls_version, "tlsv1.1") - || !strcasecmp(tls_version, "tlsv1")){ - - mosq->tls_version = mosquitto__strdup(tls_version); - if(!mosq->tls_version) return MOSQ_ERR_NOMEM; - }else{ - return MOSQ_ERR_INVAL; - } -#else - if(!strcasecmp(tls_version, "tlsv1")){ - mosq->tls_version = mosquitto__strdup(tls_version); - if(!mosq->tls_version) return MOSQ_ERR_NOMEM; - }else{ - return MOSQ_ERR_INVAL; - } -#endif - }else{ -#if OPENSSL_VERSION_NUMBER >= 0x10001000L - mosq->tls_version = mosquitto__strdup("tlsv1.2"); -#else - mosq->tls_version = mosquitto__strdup("tlsv1"); -#endif - if(!mosq->tls_version) return MOSQ_ERR_NOMEM; - } - if(ciphers){ - mosq->tls_ciphers = mosquitto__strdup(ciphers); - if(!mosq->tls_ciphers) return MOSQ_ERR_NOMEM; - }else{ - mosq->tls_ciphers = NULL; - } - - - return MOSQ_ERR_SUCCESS; -#else - return MOSQ_ERR_NOT_SUPPORTED; - -#endif -} - - -int mosquitto_tls_insecure_set(struct mosquitto *mosq, bool value) -{ -#ifdef WITH_TLS - if(!mosq) return MOSQ_ERR_INVAL; - mosq->tls_insecure = value; - return MOSQ_ERR_SUCCESS; -#else - return MOSQ_ERR_NOT_SUPPORTED; -#endif -} - - -int mosquitto_tls_psk_set(struct mosquitto *mosq, const char *psk, const char *identity, const char *ciphers) -{ -#ifdef REAL_WITH_TLS_PSK - if(!mosq || !psk || !identity) return MOSQ_ERR_INVAL; - - /* Check for hex only digits */ - if(strspn(psk, "0123456789abcdefABCDEF") < strlen(psk)){ - return MOSQ_ERR_INVAL; - } - mosq->tls_psk = mosquitto__strdup(psk); - if(!mosq->tls_psk) return MOSQ_ERR_NOMEM; - - mosq->tls_psk_identity = mosquitto__strdup(identity); - if(!mosq->tls_psk_identity){ - mosquitto__free(mosq->tls_psk); - return MOSQ_ERR_NOMEM; - } - if(ciphers){ - mosq->tls_ciphers = mosquitto__strdup(ciphers); - if(!mosq->tls_ciphers) return MOSQ_ERR_NOMEM; - }else{ - mosq->tls_ciphers = NULL; - } - - return MOSQ_ERR_SUCCESS; -#else - return MOSQ_ERR_NOT_SUPPORTED; -#endif -} - - -int mosquitto_loop(struct mosquitto *mosq, int timeout, int max_packets) -{ -#ifdef HAVE_PSELECT - struct timespec local_timeout; -#else - struct timeval local_timeout; -#endif - fd_set readfds, writefds; - int fdcount; - int rc; - char pairbuf; - int maxfd = 0; - time_t now; - - if(!mosq || max_packets < 1) return MOSQ_ERR_INVAL; -#ifndef WIN32 - if(mosq->sock >= FD_SETSIZE || mosq->sockpairR >= FD_SETSIZE){ - return MOSQ_ERR_INVAL; - } -#endif - - FD_ZERO(&readfds); - FD_ZERO(&writefds); - if(mosq->sock != INVALID_SOCKET){ - maxfd = mosq->sock; - FD_SET(mosq->sock, &readfds); - pthread_mutex_lock(&mosq->current_out_packet_mutex); - pthread_mutex_lock(&mosq->out_packet_mutex); - if(mosq->out_packet || mosq->current_out_packet){ - FD_SET(mosq->sock, &writefds); - } -#ifdef WITH_TLS - if(mosq->ssl){ - if(mosq->want_write){ - FD_SET(mosq->sock, &writefds); - }else if(mosq->want_connect){ - /* Remove possible FD_SET from above, we don't want to check - * for writing if we are still connecting, unless want_write is - * definitely set. The presence of outgoing packets does not - * matter yet. */ - FD_CLR(mosq->sock, &writefds); - } - } -#endif - pthread_mutex_unlock(&mosq->out_packet_mutex); - pthread_mutex_unlock(&mosq->current_out_packet_mutex); - }else{ -#ifdef WITH_SRV - if(mosq->achan){ - pthread_mutex_lock(&mosq->state_mutex); - if(mosq->state == mosq_cs_connect_srv){ - rc = ares_fds(mosq->achan, &readfds, &writefds); - if(rc > maxfd){ - maxfd = rc; - } - }else{ - pthread_mutex_unlock(&mosq->state_mutex); - return MOSQ_ERR_NO_CONN; - } - pthread_mutex_unlock(&mosq->state_mutex); - } -#else - return MOSQ_ERR_NO_CONN; -#endif - } - if(mosq->sockpairR != INVALID_SOCKET){ - /* sockpairR is used to break out of select() before the timeout, on a - * call to publish() etc. */ - FD_SET(mosq->sockpairR, &readfds); - if(mosq->sockpairR > maxfd){ - maxfd = mosq->sockpairR; - } - } - - if(timeout < 0){ - timeout = 1000; - } - - now = mosquitto_time(); - if(mosq->next_msg_out && now + timeout/1000 > mosq->next_msg_out){ - timeout = (mosq->next_msg_out - now)*1000; - } - - if(timeout < 0){ - /* There has been a delay somewhere which means we should have already - * sent a message. */ - timeout = 0; - } - - local_timeout.tv_sec = timeout/1000; -#ifdef HAVE_PSELECT - local_timeout.tv_nsec = (timeout-local_timeout.tv_sec*1000)*1e6; -#else - local_timeout.tv_usec = (timeout-local_timeout.tv_sec*1000)*1000; -#endif - -#ifdef HAVE_PSELECT - fdcount = pselect(maxfd+1, &readfds, &writefds, NULL, &local_timeout, NULL); -#else - fdcount = select(maxfd+1, &readfds, &writefds, NULL, &local_timeout); -#endif - if(fdcount == -1){ -#ifdef WIN32 - errno = WSAGetLastError(); -#endif - if(errno == EINTR){ - return MOSQ_ERR_SUCCESS; - }else{ - return MOSQ_ERR_ERRNO; - } - }else{ - if(mosq->sock != INVALID_SOCKET){ - if(FD_ISSET(mosq->sock, &readfds)){ -#ifdef WITH_TLS - if(mosq->want_connect){ - rc = net__socket_connect_tls(mosq); - if(rc) return rc; - }else -#endif - { - do{ - rc = mosquitto_loop_read(mosq, max_packets); - if(rc || mosq->sock == INVALID_SOCKET){ - return rc; - } - }while(SSL_DATA_PENDING(mosq)); - } - } - if(mosq->sockpairR != INVALID_SOCKET && FD_ISSET(mosq->sockpairR, &readfds)){ -#ifndef WIN32 - if(read(mosq->sockpairR, &pairbuf, 1) == 0){ - } -#else - recv(mosq->sockpairR, &pairbuf, 1, 0); -#endif - /* Fake write possible, to stimulate output write even though - * we didn't ask for it, because at that point the publish or - * other command wasn't present. */ - if(mosq->sock != INVALID_SOCKET) - FD_SET(mosq->sock, &writefds); - } - if(mosq->sock != INVALID_SOCKET && FD_ISSET(mosq->sock, &writefds)){ -#ifdef WITH_TLS - if(mosq->want_connect){ - rc = net__socket_connect_tls(mosq); - if(rc) return rc; - }else -#endif - { - rc = mosquitto_loop_write(mosq, max_packets); - if(rc || mosq->sock == INVALID_SOCKET){ - return rc; - } - } - } - } -#ifdef WITH_SRV - if(mosq->achan){ - ares_process(mosq->achan, &readfds, &writefds); - } -#endif - } - return mosquitto_loop_misc(mosq); -} - -int mosquitto_loop_forever(struct mosquitto *mosq, int timeout, int max_packets) -{ - int run = 1; - int rc; - unsigned int reconnects = 0; - unsigned long reconnect_delay; - - if(!mosq) return MOSQ_ERR_INVAL; - - if(mosq->state == mosq_cs_connect_async){ - mosquitto_reconnect(mosq); - } - - while(run){ - do{ - rc = mosquitto_loop(mosq, timeout, max_packets); - if (reconnects !=0 && rc == MOSQ_ERR_SUCCESS){ - reconnects = 0; - } - }while(run && rc == MOSQ_ERR_SUCCESS); - /* Quit after fatal errors. */ - switch(rc){ - case MOSQ_ERR_NOMEM: - case MOSQ_ERR_PROTOCOL: - case MOSQ_ERR_INVAL: - case MOSQ_ERR_NOT_FOUND: - case MOSQ_ERR_TLS: - case MOSQ_ERR_PAYLOAD_SIZE: - case MOSQ_ERR_NOT_SUPPORTED: - case MOSQ_ERR_AUTH: - case MOSQ_ERR_ACL_DENIED: - case MOSQ_ERR_UNKNOWN: - case MOSQ_ERR_EAI: - case MOSQ_ERR_PROXY: - return rc; - case MOSQ_ERR_ERRNO: - break; - } - if(errno == EPROTO){ - return rc; - } - do{ - rc = MOSQ_ERR_SUCCESS; - pthread_mutex_lock(&mosq->state_mutex); - if(mosq->state == mosq_cs_disconnecting){ - run = 0; - pthread_mutex_unlock(&mosq->state_mutex); - }else{ - pthread_mutex_unlock(&mosq->state_mutex); - - if(mosq->reconnect_delay > 0 && mosq->reconnect_exponential_backoff){ - reconnect_delay = mosq->reconnect_delay*reconnects*reconnects; - }else{ - reconnect_delay = mosq->reconnect_delay; - } - - if(reconnect_delay > mosq->reconnect_delay_max){ - reconnect_delay = mosq->reconnect_delay_max; - }else{ - reconnects++; - } - -#ifdef WIN32 - Sleep(reconnect_delay*1000); -#else - sleep(reconnect_delay); -#endif - - pthread_mutex_lock(&mosq->state_mutex); - if(mosq->state == mosq_cs_disconnecting){ - run = 0; - pthread_mutex_unlock(&mosq->state_mutex); - }else{ - pthread_mutex_unlock(&mosq->state_mutex); - rc = mosquitto_reconnect(mosq); - } - } - }while(run && rc != MOSQ_ERR_SUCCESS); - } - return rc; -} - -int mosquitto_loop_misc(struct mosquitto *mosq) -{ - time_t now; - int rc; - - if(!mosq) return MOSQ_ERR_INVAL; - if(mosq->sock == INVALID_SOCKET) return MOSQ_ERR_NO_CONN; - - mosquitto__check_keepalive(mosq); - now = mosquitto_time(); - if(mosq->ping_t && now - mosq->ping_t >= mosq->keepalive){ - /* mosq->ping_t != 0 means we are waiting for a pingresp. - * This hasn't happened in the keepalive time so we should disconnect. - */ - net__socket_close(mosq); - pthread_mutex_lock(&mosq->state_mutex); - if(mosq->state == mosq_cs_disconnecting){ - rc = MOSQ_ERR_SUCCESS; - }else{ - rc = 1; - } - pthread_mutex_unlock(&mosq->state_mutex); - pthread_mutex_lock(&mosq->callback_mutex); - if(mosq->on_disconnect){ - mosq->in_callback = true; - mosq->on_disconnect(mosq, mosq->userdata, rc); - mosq->in_callback = false; - } - pthread_mutex_unlock(&mosq->callback_mutex); - return MOSQ_ERR_CONN_LOST; - } - return MOSQ_ERR_SUCCESS; -} - -static int mosquitto__loop_rc_handle(struct mosquitto *mosq, int rc) -{ - if(rc){ - net__socket_close(mosq); - pthread_mutex_lock(&mosq->state_mutex); - if(mosq->state == mosq_cs_disconnecting){ - rc = MOSQ_ERR_SUCCESS; - } - pthread_mutex_unlock(&mosq->state_mutex); - pthread_mutex_lock(&mosq->callback_mutex); - if(mosq->on_disconnect){ - mosq->in_callback = true; - mosq->on_disconnect(mosq, mosq->userdata, rc); - mosq->in_callback = false; - } - pthread_mutex_unlock(&mosq->callback_mutex); - return rc; - } - return rc; -} - -int mosquitto_loop_read(struct mosquitto *mosq, int max_packets) -{ - int rc; - int i; - if(max_packets < 1) return MOSQ_ERR_INVAL; - - pthread_mutex_lock(&mosq->out_message_mutex); - max_packets = mosq->out_queue_len; - pthread_mutex_unlock(&mosq->out_message_mutex); - - pthread_mutex_lock(&mosq->in_message_mutex); - max_packets += mosq->in_queue_len; - pthread_mutex_unlock(&mosq->in_message_mutex); - - if(max_packets < 1) max_packets = 1; - /* Queue len here tells us how many messages are awaiting processing and - * have QoS > 0. We should try to deal with that many in this loop in order - * to keep up. */ - for(i=0; isocks5_host){ - rc = socks5__read(mosq); - }else -#endif - { - rc = packet__read(mosq); - } - if(rc || errno == EAGAIN || errno == COMPAT_EWOULDBLOCK){ - return mosquitto__loop_rc_handle(mosq, rc); - } - } - return rc; -} - -int mosquitto_loop_write(struct mosquitto *mosq, int max_packets) -{ - int rc; - int i; - if(max_packets < 1) return MOSQ_ERR_INVAL; - - pthread_mutex_lock(&mosq->out_message_mutex); - max_packets = mosq->out_queue_len; - pthread_mutex_unlock(&mosq->out_message_mutex); - - pthread_mutex_lock(&mosq->in_message_mutex); - max_packets += mosq->in_queue_len; - pthread_mutex_unlock(&mosq->in_message_mutex); - - if(max_packets < 1) max_packets = 1; - /* Queue len here tells us how many messages are awaiting processing and - * have QoS > 0. We should try to deal with that many in this loop in order - * to keep up. */ - for(i=0; iprotocol = mosq_p_mqtt31; - }else if(ival == MQTT_PROTOCOL_V311){ - mosq->protocol = mosq_p_mqtt311; - }else{ - return MOSQ_ERR_INVAL; - } - break; - default: - return MOSQ_ERR_INVAL; - } - return MOSQ_ERR_SUCCESS; -} - - -void mosquitto_connect_callback_set(struct mosquitto *mosq, void (*on_connect)(struct mosquitto *, void *, int)) -{ - pthread_mutex_lock(&mosq->callback_mutex); - mosq->on_connect = on_connect; - pthread_mutex_unlock(&mosq->callback_mutex); -} - -void mosquitto_connect_with_flags_callback_set(struct mosquitto *mosq, void (*on_connect)(struct mosquitto *, void *, int, int)) -{ - pthread_mutex_lock(&mosq->callback_mutex); - mosq->on_connect_with_flags = on_connect; - pthread_mutex_unlock(&mosq->callback_mutex); -} - -void mosquitto_disconnect_callback_set(struct mosquitto *mosq, void (*on_disconnect)(struct mosquitto *, void *, int)) -{ - pthread_mutex_lock(&mosq->callback_mutex); - mosq->on_disconnect = on_disconnect; - pthread_mutex_unlock(&mosq->callback_mutex); -} - -void mosquitto_publish_callback_set(struct mosquitto *mosq, void (*on_publish)(struct mosquitto *, void *, int)) -{ - pthread_mutex_lock(&mosq->callback_mutex); - mosq->on_publish = on_publish; - pthread_mutex_unlock(&mosq->callback_mutex); -} - -void mosquitto_message_callback_set(struct mosquitto *mosq, void (*on_message)(struct mosquitto *, void *, const struct mosquitto_message *)) -{ - pthread_mutex_lock(&mosq->callback_mutex); - mosq->on_message = on_message; - pthread_mutex_unlock(&mosq->callback_mutex); -} - -void mosquitto_subscribe_callback_set(struct mosquitto *mosq, void (*on_subscribe)(struct mosquitto *, void *, int, int, const int *)) -{ - pthread_mutex_lock(&mosq->callback_mutex); - mosq->on_subscribe = on_subscribe; - pthread_mutex_unlock(&mosq->callback_mutex); -} - -void mosquitto_unsubscribe_callback_set(struct mosquitto *mosq, void (*on_unsubscribe)(struct mosquitto *, void *, int)) -{ - pthread_mutex_lock(&mosq->callback_mutex); - mosq->on_unsubscribe = on_unsubscribe; - pthread_mutex_unlock(&mosq->callback_mutex); -} - -void mosquitto_log_callback_set(struct mosquitto *mosq, void (*on_log)(struct mosquitto *, void *, int, const char *)) -{ - pthread_mutex_lock(&mosq->log_callback_mutex); - mosq->on_log = on_log; - pthread_mutex_unlock(&mosq->log_callback_mutex); -} - -void mosquitto_user_data_set(struct mosquitto *mosq, void *userdata) -{ - if(mosq){ - mosq->userdata = userdata; - } -} const char *mosquitto_strerror(int mosq_errno) { diff --git a/lib/options.c b/lib/options.c new file mode 100644 index 00000000..f6dc3165 --- /dev/null +++ b/lib/options.c @@ -0,0 +1,293 @@ +/* +Copyright (c) 2010-2018 Roger Light + +All rights reserved. This program and the accompanying materials +are made available under the terms of the Eclipse Public License v1.0 +and Eclipse Distribution License v1.0 which accompany this distribution. + +The Eclipse Public License is available at + http://www.eclipse.org/legal/epl-v10.html +and the Eclipse Distribution License is available at + http://www.eclipse.org/org/documents/edl-v10.php. + +Contributors: + Roger Light - initial implementation and documentation. +*/ + +#include "config.h" + +#ifndef WIN32 +# include +#endif + +#include + +#include "mosquitto.h" +#include "mosquitto_internal.h" +#include "memory_mosq.h" +#include "util_mosq.h" +#include "will_mosq.h" + + +int mosquitto_will_set(struct mosquitto *mosq, const char *topic, int payloadlen, const void *payload, int qos, bool retain) +{ + if(!mosq) return MOSQ_ERR_INVAL; + return will__set(mosq, topic, payloadlen, payload, qos, retain); +} + + +int mosquitto_will_clear(struct mosquitto *mosq) +{ + if(!mosq) return MOSQ_ERR_INVAL; + return will__clear(mosq); +} + + +int mosquitto_username_pw_set(struct mosquitto *mosq, const char *username, const char *password) +{ + if(!mosq) return MOSQ_ERR_INVAL; + + mosquitto__free(mosq->username); + mosq->username = NULL; + + mosquitto__free(mosq->password); + mosq->password = NULL; + + if(username){ + mosq->username = mosquitto__strdup(username); + if(!mosq->username) return MOSQ_ERR_NOMEM; + if(password){ + mosq->password = mosquitto__strdup(password); + if(!mosq->password){ + mosquitto__free(mosq->username); + mosq->username = NULL; + return MOSQ_ERR_NOMEM; + } + } + } + return MOSQ_ERR_SUCCESS; +} + + +int mosquitto_reconnect_delay_set(struct mosquitto *mosq, unsigned int reconnect_delay, unsigned int reconnect_delay_max, bool reconnect_exponential_backoff) +{ + if(!mosq) return MOSQ_ERR_INVAL; + + mosq->reconnect_delay = reconnect_delay; + mosq->reconnect_delay_max = reconnect_delay_max; + mosq->reconnect_exponential_backoff = reconnect_exponential_backoff; + + return MOSQ_ERR_SUCCESS; + +} + + +int mosquitto_tls_set(struct mosquitto *mosq, const char *cafile, const char *capath, const char *certfile, const char *keyfile, int (*pw_callback)(char *buf, int size, int rwflag, void *userdata)) +{ +#ifdef WITH_TLS + FILE *fptr; + + if(!mosq || (!cafile && !capath) || (certfile && !keyfile) || (!certfile && keyfile)) return MOSQ_ERR_INVAL; + + mosquitto__free(mosq->tls_cafile); + mosq->tls_cafile = NULL; + if(cafile){ + fptr = mosquitto__fopen(cafile, "rt", false); + if(fptr){ + fclose(fptr); + }else{ + return MOSQ_ERR_INVAL; + } + mosq->tls_cafile = mosquitto__strdup(cafile); + + if(!mosq->tls_cafile){ + return MOSQ_ERR_NOMEM; + } + } + + mosquitto__free(mosq->tls_capath); + mosq->tls_capath = NULL; + if(capath){ + mosq->tls_capath = mosquitto__strdup(capath); + if(!mosq->tls_capath){ + return MOSQ_ERR_NOMEM; + } + } + + mosquitto__free(mosq->tls_certfile); + mosq->tls_certfile = NULL; + if(certfile){ + fptr = mosquitto__fopen(certfile, "rt", false); + if(fptr){ + fclose(fptr); + }else{ + mosquitto__free(mosq->tls_cafile); + mosq->tls_cafile = NULL; + + mosquitto__free(mosq->tls_capath); + mosq->tls_capath = NULL; + return MOSQ_ERR_INVAL; + } + mosq->tls_certfile = mosquitto__strdup(certfile); + if(!mosq->tls_certfile){ + return MOSQ_ERR_NOMEM; + } + } + + mosquitto__free(mosq->tls_keyfile); + mosq->tls_keyfile = NULL; + if(keyfile){ + fptr = mosquitto__fopen(keyfile, "rt", false); + if(fptr){ + fclose(fptr); + }else{ + mosquitto__free(mosq->tls_cafile); + mosq->tls_cafile = NULL; + + mosquitto__free(mosq->tls_capath); + mosq->tls_capath = NULL; + + mosquitto__free(mosq->tls_certfile); + mosq->tls_certfile = NULL; + return MOSQ_ERR_INVAL; + } + mosq->tls_keyfile = mosquitto__strdup(keyfile); + if(!mosq->tls_keyfile){ + return MOSQ_ERR_NOMEM; + } + } + + mosq->tls_pw_callback = pw_callback; + + + return MOSQ_ERR_SUCCESS; +#else + return MOSQ_ERR_NOT_SUPPORTED; + +#endif +} + + +int mosquitto_tls_opts_set(struct mosquitto *mosq, int cert_reqs, const char *tls_version, const char *ciphers) +{ +#ifdef WITH_TLS + if(!mosq) return MOSQ_ERR_INVAL; + + mosq->tls_cert_reqs = cert_reqs; + if(tls_version){ +#if OPENSSL_VERSION_NUMBER >= 0x10001000L + if(!strcasecmp(tls_version, "tlsv1.2") + || !strcasecmp(tls_version, "tlsv1.1") + || !strcasecmp(tls_version, "tlsv1")){ + + mosq->tls_version = mosquitto__strdup(tls_version); + if(!mosq->tls_version) return MOSQ_ERR_NOMEM; + }else{ + return MOSQ_ERR_INVAL; + } +#else + if(!strcasecmp(tls_version, "tlsv1")){ + mosq->tls_version = mosquitto__strdup(tls_version); + if(!mosq->tls_version) return MOSQ_ERR_NOMEM; + }else{ + return MOSQ_ERR_INVAL; + } +#endif + }else{ +#if OPENSSL_VERSION_NUMBER >= 0x10001000L + mosq->tls_version = mosquitto__strdup("tlsv1.2"); +#else + mosq->tls_version = mosquitto__strdup("tlsv1"); +#endif + if(!mosq->tls_version) return MOSQ_ERR_NOMEM; + } + if(ciphers){ + mosq->tls_ciphers = mosquitto__strdup(ciphers); + if(!mosq->tls_ciphers) return MOSQ_ERR_NOMEM; + }else{ + mosq->tls_ciphers = NULL; + } + + + return MOSQ_ERR_SUCCESS; +#else + return MOSQ_ERR_NOT_SUPPORTED; + +#endif +} + + +int mosquitto_tls_insecure_set(struct mosquitto *mosq, bool value) +{ +#ifdef WITH_TLS + if(!mosq) return MOSQ_ERR_INVAL; + mosq->tls_insecure = value; + return MOSQ_ERR_SUCCESS; +#else + return MOSQ_ERR_NOT_SUPPORTED; +#endif +} + + +int mosquitto_tls_psk_set(struct mosquitto *mosq, const char *psk, const char *identity, const char *ciphers) +{ +#ifdef REAL_WITH_TLS_PSK + if(!mosq || !psk || !identity) return MOSQ_ERR_INVAL; + + /* Check for hex only digits */ + if(strspn(psk, "0123456789abcdefABCDEF") < strlen(psk)){ + return MOSQ_ERR_INVAL; + } + mosq->tls_psk = mosquitto__strdup(psk); + if(!mosq->tls_psk) return MOSQ_ERR_NOMEM; + + mosq->tls_psk_identity = mosquitto__strdup(identity); + if(!mosq->tls_psk_identity){ + mosquitto__free(mosq->tls_psk); + return MOSQ_ERR_NOMEM; + } + if(ciphers){ + mosq->tls_ciphers = mosquitto__strdup(ciphers); + if(!mosq->tls_ciphers) return MOSQ_ERR_NOMEM; + }else{ + mosq->tls_ciphers = NULL; + } + + return MOSQ_ERR_SUCCESS; +#else + return MOSQ_ERR_NOT_SUPPORTED; +#endif +} + + +int mosquitto_opts_set(struct mosquitto *mosq, enum mosq_opt_t option, void *value) +{ + int ival; + + if(!mosq || !value) return MOSQ_ERR_INVAL; + + switch(option){ + case MOSQ_OPT_PROTOCOL_VERSION: + ival = *((int *)value); + if(ival == MQTT_PROTOCOL_V31){ + mosq->protocol = mosq_p_mqtt31; + }else if(ival == MQTT_PROTOCOL_V311){ + mosq->protocol = mosq_p_mqtt311; + }else{ + return MOSQ_ERR_INVAL; + } + break; + default: + return MOSQ_ERR_INVAL; + } + return MOSQ_ERR_SUCCESS; +} + + +void mosquitto_user_data_set(struct mosquitto *mosq, void *userdata) +{ + if(mosq){ + mosq->userdata = userdata; + } +} +