diff --git a/lib/CMakeLists.txt b/lib/CMakeLists.txt index aabb27ad..fc4c8da7 100644 --- a/lib/CMakeLists.txt +++ b/lib/CMakeLists.txt @@ -25,6 +25,14 @@ include_directories(${mosquitto_SOURCE_DIR} ${mosquitto_SOURCE_DIR}/lib link_directories(${mosquitto_SOURCE_DIR}/lib) add_library(libmosquitto SHARED + handle_connack.c + handle_ping.c + handle_pubackcomp.c + handle_publish.c + handle_pubrec.c + handle_pubrel.c + handle_suback.c + handle_unsuback.c helpers.c logging_mosq.c logging_mosq.h memory_mosq.c memory_mosq.h @@ -35,9 +43,12 @@ add_library(libmosquitto SHARED net_mosq.c net_mosq.h packet_mosq.c packet_mosq.h read_handle.c read_handle.h - read_handle_client.c - read_handle_shared.c - send_client_mosq.c + send_connect.c + send_disconnect.c + send_mosq.c + send_publish.c + send_subscribe.c + send_unsubscribe.c send_mosq.c send_mosq.h socks_mosq.c srv_mosq.c diff --git a/lib/Makefile b/lib/Makefile index cd599fdd..598d1ddf 100644 --- a/lib/Makefile +++ b/lib/Makefile @@ -3,6 +3,14 @@ include ../config.mk .PHONY : really clean install MOSQ_OBJS=mosquitto.o \ + handle_connack.o \ + handle_ping.o \ + handle_pubackcomp.o \ + handle_publish.o \ + handle_pubrec.o \ + handle_pubrel.o \ + handle_suback.o \ + handle_unsuback.o \ helpers.o \ logging_mosq.o \ memory_mosq.o \ @@ -10,10 +18,12 @@ MOSQ_OBJS=mosquitto.o \ net_mosq.o \ packet_mosq.o \ read_handle.o \ - read_handle_client.o \ - read_handle_shared.o \ + send_connect.o \ + send_disconnect.o \ send_mosq.o \ - send_client_mosq.o \ + send_publish.o \ + send_subscribe.o \ + send_unsubscribe.o \ socks_mosq.o \ srv_mosq.o \ thread_mosq.o \ @@ -49,9 +59,34 @@ libmosquitto.so.${SOVERSION} : ${MOSQ_OBJS} libmosquitto.a : ${MOSQ_OBJS} ${CROSS_COMPILE}$(AR) cr $@ $^ + mosquitto.o : mosquitto.c mosquitto.h ${CROSS_COMPILE}$(CC) $(LIB_CFLAGS) -c $< -o $@ +handle_connack.o : handle_connack.c read_handle.h + ${CROSS_COMPILE}$(CC) $(LIB_CFLAGS) -c $< -o $@ + +handle_publish.o : handle_publish.c read_handle.h + ${CROSS_COMPILE}$(CC) $(LIB_CFLAGS) -c $< -o $@ + +handle_ping.o : handle_ping.c read_handle.h + ${CROSS_COMPILE}$(CC) $(LIB_CFLAGS) -c $< -o $@ + +handle_pubackcomp.o : handle_pubackcomp.c read_handle.h + ${CROSS_COMPILE}$(CC) $(LIB_CFLAGS) -c $< -o $@ + +handle_pubrec.o : handle_pubrec.c read_handle.h + ${CROSS_COMPILE}$(CC) $(LIB_CFLAGS) -c $< -o $@ + +handle_pubrel.o : handle_pubrel.c read_handle.h + ${CROSS_COMPILE}$(CC) $(LIB_CFLAGS) -c $< -o $@ + +handle_suback.o : handle_suback.c read_handle.h + ${CROSS_COMPILE}$(CC) $(LIB_CFLAGS) -c $< -o $@ + +handle_unsuback.o : handle_unsuback.c read_handle.h + ${CROSS_COMPILE}$(CC) $(LIB_CFLAGS) -c $< -o $@ + helpers.o : helpers.c ${CROSS_COMPILE}$(CC) $(LIB_CFLAGS) -c $< -o $@ @@ -73,16 +108,22 @@ packet_mosq.o : packet_mosq.c packet_mosq.h read_handle.o : read_handle.c read_handle.h ${CROSS_COMPILE}$(CC) $(LIB_CFLAGS) -c $< -o $@ -read_handle_client.o : read_handle_client.c read_handle.h +send_connect.o : send_connect.c send_mosq.h ${CROSS_COMPILE}$(CC) $(LIB_CFLAGS) -c $< -o $@ -read_handle_shared.o : read_handle_shared.c read_handle.h +send_disconnect.o : send_disconnect.c send_mosq.h ${CROSS_COMPILE}$(CC) $(LIB_CFLAGS) -c $< -o $@ send_mosq.o : send_mosq.c send_mosq.h ${CROSS_COMPILE}$(CC) $(LIB_CFLAGS) -c $< -o $@ -send_client_mosq.o : send_client_mosq.c send_mosq.h +send_publish.o : send_publish.c send_mosq.h + ${CROSS_COMPILE}$(CC) $(LIB_CFLAGS) -c $< -o $@ + +send_subscribe.o : send_subscribe.c send_mosq.h + ${CROSS_COMPILE}$(CC) $(LIB_CFLAGS) -c $< -o $@ + +send_unsubscribe.o : send_unsubscribe.c send_mosq.h ${CROSS_COMPILE}$(CC) $(LIB_CFLAGS) -c $< -o $@ socks_mosq.o : socks_mosq.c diff --git a/lib/read_handle_client.c b/lib/handle_connack.c similarity index 99% rename from lib/read_handle_client.c rename to lib/handle_connack.c index 7ec782ce..0fda21c4 100644 --- a/lib/read_handle_client.c +++ b/lib/handle_connack.c @@ -4,12 +4,12 @@ Copyright (c) 2009-2015 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. */ diff --git a/lib/handle_ping.c b/lib/handle_ping.c new file mode 100644 index 00000000..c682542f --- /dev/null +++ b/lib/handle_ping.c @@ -0,0 +1,57 @@ +/* +Copyright (c) 2009-2015 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 +#include +#include + +#include "mosquitto.h" +#include "logging_mosq.h" +#include "memory_mosq.h" +#include "messages_mosq.h" +#include "mqtt3_protocol.h" +#include "net_mosq.h" +#include "packet_mosq.h" +#include "read_handle.h" +#include "send_mosq.h" +#include "util_mosq.h" +#ifdef WITH_BROKER +#include "mosquitto_broker.h" +#endif + +int handle__pingreq(struct mosquitto *mosq) +{ + assert(mosq); +#ifdef WITH_BROKER + log__printf(NULL, MOSQ_LOG_DEBUG, "Received PINGREQ from %s", mosq->id); +#else + log__printf(mosq, MOSQ_LOG_DEBUG, "Client %s received PINGREQ", mosq->id); +#endif + return send__pingresp(mosq); +} + +int handle__pingresp(struct mosquitto *mosq) +{ + assert(mosq); + mosq->ping_t = 0; /* No longer waiting for a PINGRESP. */ +#ifdef WITH_BROKER + log__printf(NULL, MOSQ_LOG_DEBUG, "Received PINGRESP from %s", mosq->id); +#else + log__printf(mosq, MOSQ_LOG_DEBUG, "Client %s received PINGRESP", mosq->id); +#endif + return MOSQ_ERR_SUCCESS; +} + diff --git a/lib/handle_pubackcomp.c b/lib/handle_pubackcomp.c new file mode 100644 index 00000000..a79d52a0 --- /dev/null +++ b/lib/handle_pubackcomp.c @@ -0,0 +1,73 @@ +/* +Copyright (c) 2009-2015 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 +#include +#include + +#include "mosquitto.h" +#include "logging_mosq.h" +#include "memory_mosq.h" +#include "messages_mosq.h" +#include "mqtt3_protocol.h" +#include "net_mosq.h" +#include "packet_mosq.h" +#include "read_handle.h" +#include "send_mosq.h" +#include "util_mosq.h" + +#ifdef WITH_BROKER +# include "mosquitto_broker.h" +#endif + + +#ifdef WITH_BROKER +int handle__pubackcomp(struct mosquitto_db *db, struct mosquitto *mosq, const char *type) +#else +int handle__pubackcomp(struct mosquitto *mosq, const char *type) +#endif +{ + uint16_t mid; + int rc; + + assert(mosq); + rc = packet__read_uint16(&mosq->in_packet, &mid); + if(rc) return rc; +#ifdef WITH_BROKER + log__printf(NULL, MOSQ_LOG_DEBUG, "Received %s from %s (Mid: %d)", type, mosq->id, mid); + + if(mid){ + rc = db__message_delete(db, mosq, mid, mosq_md_out); + if(rc) return rc; + } +#else + log__printf(mosq, MOSQ_LOG_DEBUG, "Client %s received %s (Mid: %d)", mosq->id, type, mid); + + if(!message__delete(mosq, mid, mosq_md_out)){ + /* Only inform the client the message has been sent once. */ + pthread_mutex_lock(&mosq->callback_mutex); + if(mosq->on_publish){ + mosq->in_callback = true; + mosq->on_publish(mosq, mosq->userdata, mid); + mosq->in_callback = false; + } + pthread_mutex_unlock(&mosq->callback_mutex); + } +#endif + + return MOSQ_ERR_SUCCESS; +} + diff --git a/lib/handle_publish.c b/lib/handle_publish.c new file mode 100644 index 00000000..8efbae09 --- /dev/null +++ b/lib/handle_publish.c @@ -0,0 +1,120 @@ +/* +Copyright (c) 2009-2015 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 + +#include "mosquitto.h" +#include "mosquitto_internal.h" +#include "logging_mosq.h" +#include "memory_mosq.h" +#include "messages_mosq.h" +#include "packet_mosq.h" +#include "send_mosq.h" +#include "time_mosq.h" + + +int handle__publish(struct mosquitto *mosq) +{ + uint8_t header; + struct mosquitto_message_all *message; + int rc = 0; + uint16_t mid; + + assert(mosq); + + message = mosquitto__calloc(1, sizeof(struct mosquitto_message_all)); + if(!message) return MOSQ_ERR_NOMEM; + + header = mosq->in_packet.command; + + message->dup = (header & 0x08)>>3; + message->msg.qos = (header & 0x06)>>1; + message->msg.retain = (header & 0x01); + + rc = packet__read_string(&mosq->in_packet, &message->msg.topic); + if(rc){ + message__cleanup(&message); + return rc; + } + if(!strlen(message->msg.topic)){ + message__cleanup(&message); + return MOSQ_ERR_PROTOCOL; + } + + if(message->msg.qos > 0){ + rc = packet__read_uint16(&mosq->in_packet, &mid); + if(rc){ + message__cleanup(&message); + return rc; + } + message->msg.mid = (int)mid; + } + + message->msg.payloadlen = mosq->in_packet.remaining_length - mosq->in_packet.pos; + if(message->msg.payloadlen){ + message->msg.payload = mosquitto__calloc(message->msg.payloadlen+1, sizeof(uint8_t)); + if(!message->msg.payload){ + message__cleanup(&message); + return MOSQ_ERR_NOMEM; + } + rc = packet__read_bytes(&mosq->in_packet, message->msg.payload, message->msg.payloadlen); + if(rc){ + message__cleanup(&message); + return rc; + } + } + log__printf(mosq, MOSQ_LOG_DEBUG, + "Client %s received PUBLISH (d%d, q%d, r%d, m%d, '%s', ... (%ld bytes))", + mosq->id, message->dup, message->msg.qos, message->msg.retain, + message->msg.mid, message->msg.topic, + (long)message->msg.payloadlen); + + message->timestamp = mosquitto_time(); + switch(message->msg.qos){ + case 0: + pthread_mutex_lock(&mosq->callback_mutex); + if(mosq->on_message){ + mosq->in_callback = true; + mosq->on_message(mosq, mosq->userdata, &message->msg); + mosq->in_callback = false; + } + pthread_mutex_unlock(&mosq->callback_mutex); + message__cleanup(&message); + return MOSQ_ERR_SUCCESS; + case 1: + rc = send__puback(mosq, message->msg.mid); + pthread_mutex_lock(&mosq->callback_mutex); + if(mosq->on_message){ + mosq->in_callback = true; + mosq->on_message(mosq, mosq->userdata, &message->msg); + mosq->in_callback = false; + } + pthread_mutex_unlock(&mosq->callback_mutex); + message__cleanup(&message); + return rc; + case 2: + rc = send__pubrec(mosq, message->msg.mid); + pthread_mutex_lock(&mosq->in_message_mutex); + message->state = mosq_ms_wait_for_pubrel; + message__queue(mosq, message, mosq_md_in); + pthread_mutex_unlock(&mosq->in_message_mutex); + return rc; + default: + message__cleanup(&message); + return MOSQ_ERR_PROTOCOL; + } +} + diff --git a/lib/handle_pubrec.c b/lib/handle_pubrec.c new file mode 100644 index 00000000..f044e02d --- /dev/null +++ b/lib/handle_pubrec.c @@ -0,0 +1,59 @@ +/* +Copyright (c) 2009-2015 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 +#include +#include + +#include "mosquitto.h" +#include "logging_mosq.h" +#include "memory_mosq.h" +#include "messages_mosq.h" +#include "mqtt3_protocol.h" +#include "net_mosq.h" +#include "packet_mosq.h" +#include "read_handle.h" +#include "send_mosq.h" +#include "util_mosq.h" +#ifdef WITH_BROKER +#include "mosquitto_broker.h" +#endif + + +int handle__pubrec(struct mosquitto *mosq) +{ + uint16_t mid; + int rc; + + assert(mosq); + rc = packet__read_uint16(&mosq->in_packet, &mid); + if(rc) return rc; +#ifdef WITH_BROKER + log__printf(NULL, MOSQ_LOG_DEBUG, "Received PUBREC from %s (Mid: %d)", mosq->id, mid); + + rc = db__message_update(mosq, mid, mosq_md_out, mosq_ms_wait_for_pubcomp); +#else + log__printf(mosq, MOSQ_LOG_DEBUG, "Client %s received PUBREC (Mid: %d)", mosq->id, mid); + + rc = message__out_update(mosq, mid, mosq_ms_wait_for_pubcomp); +#endif + if(rc) return rc; + rc = send__pubrel(mosq, mid); + if(rc) return rc; + + return MOSQ_ERR_SUCCESS; +} + diff --git a/lib/handle_pubrel.c b/lib/handle_pubrel.c new file mode 100644 index 00000000..c70b105f --- /dev/null +++ b/lib/handle_pubrel.c @@ -0,0 +1,81 @@ +/* +Copyright (c) 2009-2015 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 +#include +#include + +#include "mosquitto.h" +#include "logging_mosq.h" +#include "memory_mosq.h" +#include "messages_mosq.h" +#include "mqtt3_protocol.h" +#include "net_mosq.h" +#include "packet_mosq.h" +#include "read_handle.h" +#include "send_mosq.h" +#include "util_mosq.h" + +#ifdef WITH_BROKER +# include "mosquitto_broker.h" +#endif + + +int handle__pubrel(struct mosquitto_db *db, struct mosquitto *mosq) +{ + uint16_t mid; +#ifndef WITH_BROKER + struct mosquitto_message_all *message = NULL; +#endif + int rc; + + assert(mosq); + if(mosq->protocol == mosq_p_mqtt311){ + if((mosq->in_packet.command&0x0F) != 0x02){ + return MOSQ_ERR_PROTOCOL; + } + } + rc = packet__read_uint16(&mosq->in_packet, &mid); + if(rc) return rc; +#ifdef WITH_BROKER + log__printf(NULL, MOSQ_LOG_DEBUG, "Received PUBREL from %s (Mid: %d)", mosq->id, mid); + + if(db__message_release(db, mosq, mid, mosq_md_in)){ + /* Message not found. Still send a PUBCOMP anyway because this could be + * due to a repeated PUBREL after a client has reconnected. */ + } +#else + log__printf(mosq, MOSQ_LOG_DEBUG, "Client %s received PUBREL (Mid: %d)", mosq->id, mid); + + if(!message__remove(mosq, mid, mosq_md_in, &message)){ + /* Only pass the message on if we have removed it from the queue - this + * prevents multiple callbacks for the same message. */ + pthread_mutex_lock(&mosq->callback_mutex); + if(mosq->on_message){ + mosq->in_callback = true; + mosq->on_message(mosq, mosq->userdata, &message->msg); + mosq->in_callback = false; + } + pthread_mutex_unlock(&mosq->callback_mutex); + message__cleanup(&message); + } +#endif + rc = send__pubcomp(mosq, mid); + if(rc) return rc; + + return MOSQ_ERR_SUCCESS; +} + diff --git a/lib/handle_suback.c b/lib/handle_suback.c new file mode 100644 index 00000000..a183e5a4 --- /dev/null +++ b/lib/handle_suback.c @@ -0,0 +1,73 @@ +/* +Copyright (c) 2009-2015 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 + +#include "mosquitto.h" +#include "mosquitto_internal.h" +#include "logging_mosq.h" +#include "memory_mosq.h" +#include "packet_mosq.h" + +#ifdef WITH_BROKER +# include "mosquitto_broker.h" +#endif + + +int handle__suback(struct mosquitto *mosq) +{ + uint16_t mid; + uint8_t qos; + int *granted_qos; + int qos_count; + int i = 0; + int rc; + + assert(mosq); +#ifdef WITH_BROKER + log__printf(NULL, MOSQ_LOG_DEBUG, "Received SUBACK from %s", mosq->id); +#else + log__printf(mosq, MOSQ_LOG_DEBUG, "Client %s received SUBACK", mosq->id); +#endif + rc = packet__read_uint16(&mosq->in_packet, &mid); + if(rc) return rc; + + qos_count = mosq->in_packet.remaining_length - mosq->in_packet.pos; + granted_qos = mosquitto__malloc(qos_count*sizeof(int)); + if(!granted_qos) return MOSQ_ERR_NOMEM; + while(mosq->in_packet.pos < mosq->in_packet.remaining_length){ + rc = packet__read_byte(&mosq->in_packet, &qos); + if(rc){ + mosquitto__free(granted_qos); + return rc; + } + granted_qos[i] = (int)qos; + i++; + } +#ifndef WITH_BROKER + pthread_mutex_lock(&mosq->callback_mutex); + if(mosq->on_subscribe){ + mosq->in_callback = true; + mosq->on_subscribe(mosq, mosq->userdata, mid, qos_count, granted_qos); + mosq->in_callback = false; + } + pthread_mutex_unlock(&mosq->callback_mutex); +#endif + mosquitto__free(granted_qos); + + return MOSQ_ERR_SUCCESS; +} + diff --git a/lib/handle_unsuback.c b/lib/handle_unsuback.c new file mode 100644 index 00000000..50f65d17 --- /dev/null +++ b/lib/handle_unsuback.c @@ -0,0 +1,61 @@ +/* +Copyright (c) 2009-2015 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 +#include +#include + +#include "mosquitto.h" +#include "logging_mosq.h" +#include "memory_mosq.h" +#include "messages_mosq.h" +#include "mqtt3_protocol.h" +#include "net_mosq.h" +#include "packet_mosq.h" +#include "read_handle.h" +#include "send_mosq.h" +#include "util_mosq.h" +#ifdef WITH_BROKER +#include "mosquitto_broker.h" +#endif + + +int handle__unsuback(struct mosquitto *mosq) +{ + uint16_t mid; + int rc; + + assert(mosq); +#ifdef WITH_BROKER + log__printf(NULL, MOSQ_LOG_DEBUG, "Received UNSUBACK from %s", mosq->id); +#else + log__printf(mosq, MOSQ_LOG_DEBUG, "Client %s received UNSUBACK", mosq->id); +#endif + rc = packet__read_uint16(&mosq->in_packet, &mid); + if(rc) return rc; +#ifndef WITH_BROKER + pthread_mutex_lock(&mosq->callback_mutex); + if(mosq->on_unsubscribe){ + mosq->in_callback = true; + mosq->on_unsubscribe(mosq, mosq->userdata, mid); + mosq->in_callback = false; + } + pthread_mutex_unlock(&mosq->callback_mutex); +#endif + + return MOSQ_ERR_SUCCESS; +} + diff --git a/lib/read_handle.c b/lib/read_handle.c index 30d5d455..ac8fb473 100644 --- a/lib/read_handle.c +++ b/lib/read_handle.c @@ -62,95 +62,3 @@ int handle__packet(struct mosquitto *mosq) } } -int handle__publish(struct mosquitto *mosq) -{ - uint8_t header; - struct mosquitto_message_all *message; - int rc = 0; - uint16_t mid; - - assert(mosq); - - message = mosquitto__calloc(1, sizeof(struct mosquitto_message_all)); - if(!message) return MOSQ_ERR_NOMEM; - - header = mosq->in_packet.command; - - message->dup = (header & 0x08)>>3; - message->msg.qos = (header & 0x06)>>1; - message->msg.retain = (header & 0x01); - - rc = packet__read_string(&mosq->in_packet, &message->msg.topic); - if(rc){ - message__cleanup(&message); - return rc; - } - if(!strlen(message->msg.topic)){ - message__cleanup(&message); - return MOSQ_ERR_PROTOCOL; - } - - if(message->msg.qos > 0){ - rc = packet__read_uint16(&mosq->in_packet, &mid); - if(rc){ - message__cleanup(&message); - return rc; - } - message->msg.mid = (int)mid; - } - - message->msg.payloadlen = mosq->in_packet.remaining_length - mosq->in_packet.pos; - if(message->msg.payloadlen){ - message->msg.payload = mosquitto__calloc(message->msg.payloadlen+1, sizeof(uint8_t)); - if(!message->msg.payload){ - message__cleanup(&message); - return MOSQ_ERR_NOMEM; - } - rc = packet__read_bytes(&mosq->in_packet, message->msg.payload, message->msg.payloadlen); - if(rc){ - message__cleanup(&message); - return rc; - } - } - log__printf(mosq, MOSQ_LOG_DEBUG, - "Client %s received PUBLISH (d%d, q%d, r%d, m%d, '%s', ... (%ld bytes))", - mosq->id, message->dup, message->msg.qos, message->msg.retain, - message->msg.mid, message->msg.topic, - (long)message->msg.payloadlen); - - message->timestamp = mosquitto_time(); - switch(message->msg.qos){ - case 0: - pthread_mutex_lock(&mosq->callback_mutex); - if(mosq->on_message){ - mosq->in_callback = true; - mosq->on_message(mosq, mosq->userdata, &message->msg); - mosq->in_callback = false; - } - pthread_mutex_unlock(&mosq->callback_mutex); - message__cleanup(&message); - return MOSQ_ERR_SUCCESS; - case 1: - rc = send__puback(mosq, message->msg.mid); - pthread_mutex_lock(&mosq->callback_mutex); - if(mosq->on_message){ - mosq->in_callback = true; - mosq->on_message(mosq, mosq->userdata, &message->msg); - mosq->in_callback = false; - } - pthread_mutex_unlock(&mosq->callback_mutex); - message__cleanup(&message); - return rc; - case 2: - rc = send__pubrec(mosq, message->msg.mid); - pthread_mutex_lock(&mosq->in_message_mutex); - message->state = mosq_ms_wait_for_pubrel; - message__queue(mosq, message, mosq_md_in); - pthread_mutex_unlock(&mosq->in_message_mutex); - return rc; - default: - message__cleanup(&message); - return MOSQ_ERR_PROTOCOL; - } -} - diff --git a/lib/read_handle_shared.c b/lib/read_handle_shared.c deleted file mode 100644 index ecd991f8..00000000 --- a/lib/read_handle_shared.c +++ /dev/null @@ -1,233 +0,0 @@ -/* -Copyright (c) 2009-2015 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 -#include -#include - -#include "mosquitto.h" -#include "logging_mosq.h" -#include "memory_mosq.h" -#include "messages_mosq.h" -#include "mqtt3_protocol.h" -#include "net_mosq.h" -#include "packet_mosq.h" -#include "read_handle.h" -#include "send_mosq.h" -#include "util_mosq.h" -#ifdef WITH_BROKER -#include "mosquitto_broker.h" -#endif - -int handle__pingreq(struct mosquitto *mosq) -{ - assert(mosq); -#ifdef WITH_BROKER - log__printf(NULL, MOSQ_LOG_DEBUG, "Received PINGREQ from %s", mosq->id); -#else - log__printf(mosq, MOSQ_LOG_DEBUG, "Client %s received PINGREQ", mosq->id); -#endif - return send__pingresp(mosq); -} - -int handle__pingresp(struct mosquitto *mosq) -{ - assert(mosq); - mosq->ping_t = 0; /* No longer waiting for a PINGRESP. */ -#ifdef WITH_BROKER - log__printf(NULL, MOSQ_LOG_DEBUG, "Received PINGRESP from %s", mosq->id); -#else - log__printf(mosq, MOSQ_LOG_DEBUG, "Client %s received PINGRESP", mosq->id); -#endif - return MOSQ_ERR_SUCCESS; -} - -#ifdef WITH_BROKER -int handle__pubackcomp(struct mosquitto_db *db, struct mosquitto *mosq, const char *type) -#else -int handle__pubackcomp(struct mosquitto *mosq, const char *type) -#endif -{ - uint16_t mid; - int rc; - - assert(mosq); - rc = packet__read_uint16(&mosq->in_packet, &mid); - if(rc) return rc; -#ifdef WITH_BROKER - log__printf(NULL, MOSQ_LOG_DEBUG, "Received %s from %s (Mid: %d)", type, mosq->id, mid); - - if(mid){ - rc = db__message_delete(db, mosq, mid, mosq_md_out); - if(rc) return rc; - } -#else - log__printf(mosq, MOSQ_LOG_DEBUG, "Client %s received %s (Mid: %d)", mosq->id, type, mid); - - if(!message__delete(mosq, mid, mosq_md_out)){ - /* Only inform the client the message has been sent once. */ - pthread_mutex_lock(&mosq->callback_mutex); - if(mosq->on_publish){ - mosq->in_callback = true; - mosq->on_publish(mosq, mosq->userdata, mid); - mosq->in_callback = false; - } - pthread_mutex_unlock(&mosq->callback_mutex); - } -#endif - - return MOSQ_ERR_SUCCESS; -} - -int handle__pubrec(struct mosquitto *mosq) -{ - uint16_t mid; - int rc; - - assert(mosq); - rc = packet__read_uint16(&mosq->in_packet, &mid); - if(rc) return rc; -#ifdef WITH_BROKER - log__printf(NULL, MOSQ_LOG_DEBUG, "Received PUBREC from %s (Mid: %d)", mosq->id, mid); - - rc = db__message_update(mosq, mid, mosq_md_out, mosq_ms_wait_for_pubcomp); -#else - log__printf(mosq, MOSQ_LOG_DEBUG, "Client %s received PUBREC (Mid: %d)", mosq->id, mid); - - rc = message__out_update(mosq, mid, mosq_ms_wait_for_pubcomp); -#endif - if(rc) return rc; - rc = send__pubrel(mosq, mid); - if(rc) return rc; - - return MOSQ_ERR_SUCCESS; -} - -int handle__pubrel(struct mosquitto_db *db, struct mosquitto *mosq) -{ - uint16_t mid; -#ifndef WITH_BROKER - struct mosquitto_message_all *message = NULL; -#endif - int rc; - - assert(mosq); - if(mosq->protocol == mosq_p_mqtt311){ - if((mosq->in_packet.command&0x0F) != 0x02){ - return MOSQ_ERR_PROTOCOL; - } - } - rc = packet__read_uint16(&mosq->in_packet, &mid); - if(rc) return rc; -#ifdef WITH_BROKER - log__printf(NULL, MOSQ_LOG_DEBUG, "Received PUBREL from %s (Mid: %d)", mosq->id, mid); - - if(db__message_release(db, mosq, mid, mosq_md_in)){ - /* Message not found. Still send a PUBCOMP anyway because this could be - * due to a repeated PUBREL after a client has reconnected. */ - } -#else - log__printf(mosq, MOSQ_LOG_DEBUG, "Client %s received PUBREL (Mid: %d)", mosq->id, mid); - - if(!message__remove(mosq, mid, mosq_md_in, &message)){ - /* Only pass the message on if we have removed it from the queue - this - * prevents multiple callbacks for the same message. */ - pthread_mutex_lock(&mosq->callback_mutex); - if(mosq->on_message){ - mosq->in_callback = true; - mosq->on_message(mosq, mosq->userdata, &message->msg); - mosq->in_callback = false; - } - pthread_mutex_unlock(&mosq->callback_mutex); - message__cleanup(&message); - } -#endif - rc = send__pubcomp(mosq, mid); - if(rc) return rc; - - return MOSQ_ERR_SUCCESS; -} - -int handle__suback(struct mosquitto *mosq) -{ - uint16_t mid; - uint8_t qos; - int *granted_qos; - int qos_count; - int i = 0; - int rc; - - assert(mosq); -#ifdef WITH_BROKER - log__printf(NULL, MOSQ_LOG_DEBUG, "Received SUBACK from %s", mosq->id); -#else - log__printf(mosq, MOSQ_LOG_DEBUG, "Client %s received SUBACK", mosq->id); -#endif - rc = packet__read_uint16(&mosq->in_packet, &mid); - if(rc) return rc; - - qos_count = mosq->in_packet.remaining_length - mosq->in_packet.pos; - granted_qos = mosquitto__malloc(qos_count*sizeof(int)); - if(!granted_qos) return MOSQ_ERR_NOMEM; - while(mosq->in_packet.pos < mosq->in_packet.remaining_length){ - rc = packet__read_byte(&mosq->in_packet, &qos); - if(rc){ - mosquitto__free(granted_qos); - return rc; - } - granted_qos[i] = (int)qos; - i++; - } -#ifndef WITH_BROKER - pthread_mutex_lock(&mosq->callback_mutex); - if(mosq->on_subscribe){ - mosq->in_callback = true; - mosq->on_subscribe(mosq, mosq->userdata, mid, qos_count, granted_qos); - mosq->in_callback = false; - } - pthread_mutex_unlock(&mosq->callback_mutex); -#endif - mosquitto__free(granted_qos); - - return MOSQ_ERR_SUCCESS; -} - -int handle__unsuback(struct mosquitto *mosq) -{ - uint16_t mid; - int rc; - - assert(mosq); -#ifdef WITH_BROKER - log__printf(NULL, MOSQ_LOG_DEBUG, "Received UNSUBACK from %s", mosq->id); -#else - log__printf(mosq, MOSQ_LOG_DEBUG, "Client %s received UNSUBACK", mosq->id); -#endif - rc = packet__read_uint16(&mosq->in_packet, &mid); - if(rc) return rc; -#ifndef WITH_BROKER - pthread_mutex_lock(&mosq->callback_mutex); - if(mosq->on_unsubscribe){ - mosq->in_callback = true; - mosq->on_unsubscribe(mosq, mosq->userdata, mid); - mosq->in_callback = false; - } - pthread_mutex_unlock(&mosq->callback_mutex); -#endif - - return MOSQ_ERR_SUCCESS; -} - diff --git a/lib/send_client_mosq.c b/lib/send_connect.c similarity index 57% rename from lib/send_client_mosq.c rename to lib/send_connect.c index 1695ba85..d1ebf027 100644 --- a/lib/send_client_mosq.c +++ b/lib/send_connect.c @@ -4,30 +4,27 @@ Copyright (c) 2009-2015 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 -#include -#include "mosquitto.h" #include "logging_mosq.h" #include "memory_mosq.h" +#include "mosquitto.h" +#include "mosquitto_internal.h" #include "mqtt3_protocol.h" -#include "net_mosq.h" #include "packet_mosq.h" -#include "send_mosq.h" -#include "util_mosq.h" #ifdef WITH_BROKER -#include "mosquitto_broker.h" +# include "mosquitto_broker.h" #endif int send__connect(struct mosquitto *mosq, uint16_t keepalive, bool clean_session) @@ -145,103 +142,3 @@ int send__connect(struct mosquitto *mosq, uint16_t keepalive, bool clean_session return packet__queue(mosq, packet); } -int send__disconnect(struct mosquitto *mosq) -{ - assert(mosq); -#ifdef WITH_BROKER -# ifdef WITH_BRIDGE - log__printf(mosq, MOSQ_LOG_DEBUG, "Bridge %s sending DISCONNECT", mosq->id); -# endif -#else - log__printf(mosq, MOSQ_LOG_DEBUG, "Client %s sending DISCONNECT", mosq->id); -#endif - return send__simple_command(mosq, DISCONNECT); -} - -int send__subscribe(struct mosquitto *mosq, int *mid, const char *topic, uint8_t topic_qos) -{ - /* FIXME - only deals with a single topic */ - struct mosquitto__packet *packet = NULL; - uint32_t packetlen; - uint16_t local_mid; - int rc; - - assert(mosq); - assert(topic); - - packet = mosquitto__calloc(1, sizeof(struct mosquitto__packet)); - if(!packet) return MOSQ_ERR_NOMEM; - - packetlen = 2 + 2+strlen(topic) + 1; - - packet->command = SUBSCRIBE | (1<<1); - packet->remaining_length = packetlen; - rc = packet__alloc(packet); - if(rc){ - mosquitto__free(packet); - return rc; - } - - /* Variable header */ - local_mid = mosquitto__mid_generate(mosq); - if(mid) *mid = (int)local_mid; - packet__write_uint16(packet, local_mid); - - /* Payload */ - packet__write_string(packet, topic, strlen(topic)); - packet__write_byte(packet, topic_qos); - -#ifdef WITH_BROKER -# ifdef WITH_BRIDGE - log__printf(mosq, MOSQ_LOG_DEBUG, "Bridge %s sending SUBSCRIBE (Mid: %d, Topic: %s, QoS: %d)", mosq->id, local_mid, topic, topic_qos); -# endif -#else - log__printf(mosq, MOSQ_LOG_DEBUG, "Client %s sending SUBSCRIBE (Mid: %d, Topic: %s, QoS: %d)", mosq->id, local_mid, topic, topic_qos); -#endif - - return packet__queue(mosq, packet); -} - - -int send__unsubscribe(struct mosquitto *mosq, int *mid, const char *topic) -{ - /* FIXME - only deals with a single topic */ - struct mosquitto__packet *packet = NULL; - uint32_t packetlen; - uint16_t local_mid; - int rc; - - assert(mosq); - assert(topic); - - packet = mosquitto__calloc(1, sizeof(struct mosquitto__packet)); - if(!packet) return MOSQ_ERR_NOMEM; - - packetlen = 2 + 2+strlen(topic); - - packet->command = UNSUBSCRIBE | (1<<1); - packet->remaining_length = packetlen; - rc = packet__alloc(packet); - if(rc){ - mosquitto__free(packet); - return rc; - } - - /* Variable header */ - local_mid = mosquitto__mid_generate(mosq); - if(mid) *mid = (int)local_mid; - packet__write_uint16(packet, local_mid); - - /* Payload */ - packet__write_string(packet, topic, strlen(topic)); - -#ifdef WITH_BROKER -# ifdef WITH_BRIDGE - log__printf(mosq, MOSQ_LOG_DEBUG, "Bridge %s sending UNSUBSCRIBE (Mid: %d, Topic: %s)", mosq->id, local_mid, topic); -# endif -#else - log__printf(mosq, MOSQ_LOG_DEBUG, "Client %s sending UNSUBSCRIBE (Mid: %d, Topic: %s)", mosq->id, local_mid, topic); -#endif - return packet__queue(mosq, packet); -} - diff --git a/lib/send_disconnect.c b/lib/send_disconnect.c new file mode 100644 index 00000000..19937cfb --- /dev/null +++ b/lib/send_disconnect.c @@ -0,0 +1,42 @@ +/* +Copyright (c) 2009-2015 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 + +#include "mosquitto.h" +#include "mosquitto_internal.h" +#include "logging_mosq.h" +#include "mqtt3_protocol.h" +#include "send_mosq.h" + +#ifdef WITH_BROKER +# include "mosquitto_broker.h" +#endif + + +int send__disconnect(struct mosquitto *mosq) +{ + assert(mosq); +#ifdef WITH_BROKER +# ifdef WITH_BRIDGE + log__printf(mosq, MOSQ_LOG_DEBUG, "Bridge %s sending DISCONNECT", mosq->id); +# endif +#else + log__printf(mosq, MOSQ_LOG_DEBUG, "Client %s sending DISCONNECT", mosq->id); +#endif + return send__simple_command(mosq, DISCONNECT); +} + diff --git a/lib/send_mosq.c b/lib/send_mosq.c index 5244eda1..f796ed0b 100644 --- a/lib/send_mosq.c +++ b/lib/send_mosq.c @@ -82,96 +82,6 @@ int send__pubcomp(struct mosquitto *mosq, uint16_t mid) return send__command_with_mid(mosq, PUBCOMP, mid, false); } -int send__publish(struct mosquitto *mosq, uint16_t mid, const char *topic, uint32_t payloadlen, const void *payload, int qos, bool retain, bool dup) -{ -#ifdef WITH_BROKER - size_t len; -#ifdef WITH_BRIDGE - int i; - struct mosquitto__bridge_topic *cur_topic; - bool match; - int rc; - char *mapped_topic = NULL; - char *topic_temp = NULL; -#endif -#endif - assert(mosq); - assert(topic); - -#if defined(WITH_BROKER) && defined(WITH_WEBSOCKETS) - if(mosq->sock == INVALID_SOCKET && !mosq->wsi) return MOSQ_ERR_NO_CONN; -#else - if(mosq->sock == INVALID_SOCKET) return MOSQ_ERR_NO_CONN; -#endif - -#ifdef WITH_BROKER - if(mosq->listener && mosq->listener->mount_point){ - len = strlen(mosq->listener->mount_point); - if(len < strlen(topic)){ - topic += len; - }else{ - /* Invalid topic string. Should never happen, but silently swallow the message anyway. */ - return MOSQ_ERR_SUCCESS; - } - } -#ifdef WITH_BRIDGE - if(mosq->bridge && mosq->bridge->topics && mosq->bridge->topic_remapping){ - for(i=0; ibridge->topic_count; i++){ - cur_topic = &mosq->bridge->topics[i]; - if((cur_topic->direction == bd_both || cur_topic->direction == bd_out) - && (cur_topic->remote_prefix || cur_topic->local_prefix)){ - /* Topic mapping required on this topic if the message matches */ - - rc = mosquitto_topic_matches_sub(cur_topic->local_topic, topic, &match); - if(rc){ - return rc; - } - if(match){ - mapped_topic = mosquitto__strdup(topic); - if(!mapped_topic) return MOSQ_ERR_NOMEM; - if(cur_topic->local_prefix){ - /* This prefix needs removing. */ - if(!strncmp(cur_topic->local_prefix, mapped_topic, strlen(cur_topic->local_prefix))){ - topic_temp = mosquitto__strdup(mapped_topic+strlen(cur_topic->local_prefix)); - mosquitto__free(mapped_topic); - if(!topic_temp){ - return MOSQ_ERR_NOMEM; - } - mapped_topic = topic_temp; - } - } - - if(cur_topic->remote_prefix){ - /* This prefix needs adding. */ - len = strlen(mapped_topic) + strlen(cur_topic->remote_prefix)+1; - topic_temp = mosquitto__malloc(len+1); - if(!topic_temp){ - mosquitto__free(mapped_topic); - return MOSQ_ERR_NOMEM; - } - snprintf(topic_temp, len, "%s%s", cur_topic->remote_prefix, mapped_topic); - topic_temp[len] = '\0'; - mosquitto__free(mapped_topic); - mapped_topic = topic_temp; - } - log__printf(NULL, MOSQ_LOG_DEBUG, "Sending PUBLISH to %s (d%d, q%d, r%d, m%d, '%s', ... (%ld bytes))", mosq->id, dup, qos, retain, mid, mapped_topic, (long)payloadlen); - G_PUB_BYTES_SENT_INC(payloadlen); - rc = send__real_publish(mosq, mid, mapped_topic, payloadlen, payload, qos, retain, dup); - mosquitto__free(mapped_topic); - return rc; - } - } - } - } -#endif - log__printf(NULL, MOSQ_LOG_DEBUG, "Sending PUBLISH to %s (d%d, q%d, r%d, m%d, '%s', ... (%ld bytes))", mosq->id, dup, qos, retain, mid, topic, (long)payloadlen); - G_PUB_BYTES_SENT_INC(payloadlen); -#else - log__printf(mosq, MOSQ_LOG_DEBUG, "Client %s sending PUBLISH (d%d, q%d, r%d, m%d, '%s', ... (%ld bytes))", mosq->id, dup, qos, retain, mid, topic, (long)payloadlen); -#endif - - return send__real_publish(mosq, mid, topic, payloadlen, payload, qos, retain, dup); -} int send__pubrec(struct mosquitto *mosq, uint16_t mid) { @@ -242,38 +152,3 @@ int send__simple_command(struct mosquitto *mosq, uint8_t command) return packet__queue(mosq, packet); } -int send__real_publish(struct mosquitto *mosq, uint16_t mid, const char *topic, uint32_t payloadlen, const void *payload, int qos, bool retain, bool dup) -{ - struct mosquitto__packet *packet = NULL; - int packetlen; - int rc; - - assert(mosq); - assert(topic); - - packetlen = 2+strlen(topic) + payloadlen; - if(qos > 0) packetlen += 2; /* For message id */ - packet = mosquitto__calloc(1, sizeof(struct mosquitto__packet)); - if(!packet) return MOSQ_ERR_NOMEM; - - packet->mid = mid; - packet->command = PUBLISH | ((dup&0x1)<<3) | (qos<<1) | retain; - packet->remaining_length = packetlen; - rc = packet__alloc(packet); - if(rc){ - mosquitto__free(packet); - return rc; - } - /* Variable header (topic string) */ - packet__write_string(packet, topic, strlen(topic)); - if(qos > 0){ - packet__write_uint16(packet, mid); - } - - /* Payload */ - if(payloadlen){ - packet__write_bytes(packet, payload, payloadlen); - } - - return packet__queue(mosq, packet); -} diff --git a/lib/send_publish.c b/lib/send_publish.c new file mode 100644 index 00000000..7294f8fb --- /dev/null +++ b/lib/send_publish.c @@ -0,0 +1,166 @@ +/* +Copyright (c) 2009-2015 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 +//#include +//#include + +#include "mosquitto.h" +#include "mosquitto_internal.h" +#include "logging_mosq.h" +#include "mqtt3_protocol.h" +#include "memory_mosq.h" +#include "net_mosq.h" +#include "packet_mosq.h" +#include "send_mosq.h" +//#include "time_mosq.h" +//#include "util_mosq.h" + +#ifdef WITH_BROKER +# include "mosquitto_broker.h" +# include "sys_tree.h" +#else +# define G_PUB_BYTES_SENT_INC(A) +#endif + + +int send__publish(struct mosquitto *mosq, uint16_t mid, const char *topic, uint32_t payloadlen, const void *payload, int qos, bool retain, bool dup) +{ +#ifdef WITH_BROKER + size_t len; +#ifdef WITH_BRIDGE + int i; + struct mosquitto__bridge_topic *cur_topic; + bool match; + int rc; + char *mapped_topic = NULL; + char *topic_temp = NULL; +#endif +#endif + assert(mosq); + assert(topic); + +#if defined(WITH_BROKER) && defined(WITH_WEBSOCKETS) + if(mosq->sock == INVALID_SOCKET && !mosq->wsi) return MOSQ_ERR_NO_CONN; +#else + if(mosq->sock == INVALID_SOCKET) return MOSQ_ERR_NO_CONN; +#endif + +#ifdef WITH_BROKER + if(mosq->listener && mosq->listener->mount_point){ + len = strlen(mosq->listener->mount_point); + if(len < strlen(topic)){ + topic += len; + }else{ + /* Invalid topic string. Should never happen, but silently swallow the message anyway. */ + return MOSQ_ERR_SUCCESS; + } + } +#ifdef WITH_BRIDGE + if(mosq->bridge && mosq->bridge->topics && mosq->bridge->topic_remapping){ + for(i=0; ibridge->topic_count; i++){ + cur_topic = &mosq->bridge->topics[i]; + if((cur_topic->direction == bd_both || cur_topic->direction == bd_out) + && (cur_topic->remote_prefix || cur_topic->local_prefix)){ + /* Topic mapping required on this topic if the message matches */ + + rc = mosquitto_topic_matches_sub(cur_topic->local_topic, topic, &match); + if(rc){ + return rc; + } + if(match){ + mapped_topic = mosquitto__strdup(topic); + if(!mapped_topic) return MOSQ_ERR_NOMEM; + if(cur_topic->local_prefix){ + /* This prefix needs removing. */ + if(!strncmp(cur_topic->local_prefix, mapped_topic, strlen(cur_topic->local_prefix))){ + topic_temp = mosquitto__strdup(mapped_topic+strlen(cur_topic->local_prefix)); + mosquitto__free(mapped_topic); + if(!topic_temp){ + return MOSQ_ERR_NOMEM; + } + mapped_topic = topic_temp; + } + } + + if(cur_topic->remote_prefix){ + /* This prefix needs adding. */ + len = strlen(mapped_topic) + strlen(cur_topic->remote_prefix)+1; + topic_temp = mosquitto__malloc(len+1); + if(!topic_temp){ + mosquitto__free(mapped_topic); + return MOSQ_ERR_NOMEM; + } + snprintf(topic_temp, len, "%s%s", cur_topic->remote_prefix, mapped_topic); + topic_temp[len] = '\0'; + mosquitto__free(mapped_topic); + mapped_topic = topic_temp; + } + log__printf(NULL, MOSQ_LOG_DEBUG, "Sending PUBLISH to %s (d%d, q%d, r%d, m%d, '%s', ... (%ld bytes))", mosq->id, dup, qos, retain, mid, mapped_topic, (long)payloadlen); + G_PUB_BYTES_SENT_INC(payloadlen); + rc = send__real_publish(mosq, mid, mapped_topic, payloadlen, payload, qos, retain, dup); + mosquitto__free(mapped_topic); + return rc; + } + } + } + } +#endif + log__printf(NULL, MOSQ_LOG_DEBUG, "Sending PUBLISH to %s (d%d, q%d, r%d, m%d, '%s', ... (%ld bytes))", mosq->id, dup, qos, retain, mid, topic, (long)payloadlen); + G_PUB_BYTES_SENT_INC(payloadlen); +#else + log__printf(mosq, MOSQ_LOG_DEBUG, "Client %s sending PUBLISH (d%d, q%d, r%d, m%d, '%s', ... (%ld bytes))", mosq->id, dup, qos, retain, mid, topic, (long)payloadlen); +#endif + + return send__real_publish(mosq, mid, topic, payloadlen, payload, qos, retain, dup); +} + + +int send__real_publish(struct mosquitto *mosq, uint16_t mid, const char *topic, uint32_t payloadlen, const void *payload, int qos, bool retain, bool dup) +{ + struct mosquitto__packet *packet = NULL; + int packetlen; + int rc; + + assert(mosq); + assert(topic); + + packetlen = 2+strlen(topic) + payloadlen; + if(qos > 0) packetlen += 2; /* For message id */ + packet = mosquitto__calloc(1, sizeof(struct mosquitto__packet)); + if(!packet) return MOSQ_ERR_NOMEM; + + packet->mid = mid; + packet->command = PUBLISH | ((dup&0x1)<<3) | (qos<<1) | retain; + packet->remaining_length = packetlen; + rc = packet__alloc(packet); + if(rc){ + mosquitto__free(packet); + return rc; + } + /* Variable header (topic string) */ + packet__write_string(packet, topic, strlen(topic)); + if(qos > 0){ + packet__write_uint16(packet, mid); + } + + /* Payload */ + if(payloadlen){ + packet__write_bytes(packet, payload, payloadlen); + } + + return packet__queue(mosq, packet); +} diff --git a/lib/send_subscribe.c b/lib/send_subscribe.c new file mode 100644 index 00000000..5c55adfd --- /dev/null +++ b/lib/send_subscribe.c @@ -0,0 +1,75 @@ +/* +Copyright (c) 2009-2015 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 + +#include "mosquitto.h" +#include "mosquitto_internal.h" +#include "logging_mosq.h" +#include "memory_mosq.h" +#include "mqtt3_protocol.h" +#include "packet_mosq.h" +#include "util_mosq.h" + +#ifdef WITH_BROKER +# include "mosquitto_broker.h" +#endif + + +int send__subscribe(struct mosquitto *mosq, int *mid, const char *topic, uint8_t topic_qos) +{ + /* FIXME - only deals with a single topic */ + struct mosquitto__packet *packet = NULL; + uint32_t packetlen; + uint16_t local_mid; + int rc; + + assert(mosq); + assert(topic); + + packet = mosquitto__calloc(1, sizeof(struct mosquitto__packet)); + if(!packet) return MOSQ_ERR_NOMEM; + + packetlen = 2 + 2+strlen(topic) + 1; + + packet->command = SUBSCRIBE | (1<<1); + packet->remaining_length = packetlen; + rc = packet__alloc(packet); + if(rc){ + mosquitto__free(packet); + return rc; + } + + /* Variable header */ + local_mid = mosquitto__mid_generate(mosq); + if(mid) *mid = (int)local_mid; + packet__write_uint16(packet, local_mid); + + /* Payload */ + packet__write_string(packet, topic, strlen(topic)); + packet__write_byte(packet, topic_qos); + +#ifdef WITH_BROKER +# ifdef WITH_BRIDGE + log__printf(mosq, MOSQ_LOG_DEBUG, "Bridge %s sending SUBSCRIBE (Mid: %d, Topic: %s, QoS: %d)", mosq->id, local_mid, topic, topic_qos); +# endif +#else + log__printf(mosq, MOSQ_LOG_DEBUG, "Client %s sending SUBSCRIBE (Mid: %d, Topic: %s, QoS: %d)", mosq->id, local_mid, topic, topic_qos); +#endif + + return packet__queue(mosq, packet); +} + diff --git a/lib/send_unsubscribe.c b/lib/send_unsubscribe.c new file mode 100644 index 00000000..92307685 --- /dev/null +++ b/lib/send_unsubscribe.c @@ -0,0 +1,74 @@ +/* +Copyright (c) 2009-2015 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 +//#include + +#include "mosquitto.h" +#include "logging_mosq.h" +#include "memory_mosq.h" +#include "mqtt3_protocol.h" +//#include "net_mosq.h" +#include "packet_mosq.h" +//#include "send_mosq.h" +#include "util_mosq.h" + +#ifdef WITH_BROKER +# include "mosquitto_broker.h" +#endif + +int send__unsubscribe(struct mosquitto *mosq, int *mid, const char *topic) +{ + /* FIXME - only deals with a single topic */ + struct mosquitto__packet *packet = NULL; + uint32_t packetlen; + uint16_t local_mid; + int rc; + + assert(mosq); + assert(topic); + + packet = mosquitto__calloc(1, sizeof(struct mosquitto__packet)); + if(!packet) return MOSQ_ERR_NOMEM; + + packetlen = 2 + 2+strlen(topic); + + packet->command = UNSUBSCRIBE | (1<<1); + packet->remaining_length = packetlen; + rc = packet__alloc(packet); + if(rc){ + mosquitto__free(packet); + return rc; + } + + /* Variable header */ + local_mid = mosquitto__mid_generate(mosq); + if(mid) *mid = (int)local_mid; + packet__write_uint16(packet, local_mid); + + /* Payload */ + packet__write_string(packet, topic, strlen(topic)); + +#ifdef WITH_BROKER +# ifdef WITH_BRIDGE + log__printf(mosq, MOSQ_LOG_DEBUG, "Bridge %s sending UNSUBSCRIBE (Mid: %d, Topic: %s)", mosq->id, local_mid, topic); +# endif +#else + log__printf(mosq, MOSQ_LOG_DEBUG, "Client %s sending UNSUBSCRIBE (Mid: %d, Topic: %s)", mosq->id, local_mid, topic); +#endif + return packet__queue(mosq, packet); +} + diff --git a/src/CMakeLists.txt b/src/CMakeLists.txt index f7bc8c21..09ae1f3f 100644 --- a/src/CMakeLists.txt +++ b/src/CMakeLists.txt @@ -6,6 +6,17 @@ set (MOSQ_SRCS conf.c context.c database.c + handle_connack.c + handle_connect.c + ../lib/handle_ping.c + ../lib/handle_pubackcomp.c + handle_publish.c + ../lib/handle_pubrec.c + ../lib/handle_pubrel.c + ../lib/handle_suback.c + handle_subscribe.c + ../lib/handle_unsuback.c + handle_unsubscribe.c lib_load.h logging.c loop.c @@ -16,13 +27,18 @@ set (MOSQ_SRCS ../lib/net_mosq.c ../lib/net_mosq.h ../lib/packet_mosq.c ../lib/packet_mosq.h persist.c persist.h - read_handle.c read_handle_client.c read_handle_server.c - ../lib/read_handle_shared.c ../lib/read_handle.h + read_handle.c + ../lib/read_handle.h subs.c security.c security_default.c - ../lib/send_client_mosq.c ../lib/send_mosq.h ../lib/send_mosq.c ../lib/send_mosq.h - send_server.c + send_connack.c + ../lib/send_connect.c + ../lib/send_disconnect.c + ../lib/send_publish.c + send_suback.c + ../lib/send_subscribe.c + ../lib/send_unsubscribe.c sys_tree.c sys_tree.h ../lib/time_mosq.c ../lib/tls_mosq.c @@ -30,6 +46,7 @@ set (MOSQ_SRCS websockets.c ../lib/will_mosq.c ../lib/will_mosq.h) + option(INC_BRIDGE_SUPPORT "Include bridge support for connecting to other brokers?" ON) if (${INC_BRIDGE_SUPPORT} STREQUAL ON) diff --git a/src/Makefile b/src/Makefile index 4879ec89..198ff29c 100644 --- a/src/Makefile +++ b/src/Makefile @@ -8,7 +8,50 @@ else all : mosquitto endif -mosquitto : mosquitto.o bridge.o conf.o context.o database.o logging.o loop.o memory_mosq.o persist.o net.o net_mosq.o packet_mosq.o read_handle.o read_handle_client.o read_handle_server.o read_handle_shared.o security.o security_default.o send_client_mosq.o send_mosq.o send_server.o service.o subs.o sys_tree.o time_mosq.o tls_mosq.o util_mosq.o websockets.o will_mosq.o +OBJS= mosquitto.o \ + bridge.o \ + conf.o \ + context.o \ + database.o \ + handle_connack.o \ + handle_connect.o \ + handle_ping.o \ + handle_pubackcomp.o \ + handle_publish.o \ + handle_pubrec.o \ + handle_pubrel.o \ + handle_suback.o \ + handle_subscribe.o \ + handle_unsuback.o \ + handle_unsubscribe.o \ + logging.o \ + loop.o \ + memory_mosq.o \ + persist.o \ + net.o \ + net_mosq.o \ + packet_mosq.o \ + read_handle.o \ + security.o \ + security_default.o \ + send_connack.o \ + send_connect.o \ + send_disconnect.o \ + send_mosq.o \ + send_publish.o \ + send_suback.o \ + send_subscribe.o \ + send_unsubscribe.o \ + service.o \ + subs.o \ + sys_tree.o \ + time_mosq.o \ + tls_mosq.o \ + util_mosq.o \ + websockets.o \ + will_mosq.o + +mosquitto : ${OBJS} ${CROSS_COMPILE}${CC} $^ -o $@ ${LDFLAGS} $(BROKER_LIBS) mosquitto.o : mosquitto.c mosquitto_broker.h @@ -26,6 +69,39 @@ context.o : context.c mosquitto_broker.h database.o : database.c mosquitto_broker.h ${CROSS_COMPILE}${CC} $(BROKER_CFLAGS) -c $< -o $@ +handle_connack.o : handle_connack.c mosquitto_broker.h + ${CROSS_COMPILE}${CC} $(BROKER_CFLAGS) -c $< -o $@ + +handle_connect.o : handle_connect.c mosquitto_broker.h + ${CROSS_COMPILE}${CC} $(BROKER_CFLAGS) -c $< -o $@ + +handle_ping.o : ../lib/handle_ping.c ../lib/read_handle.h + ${CROSS_COMPILE}${CC} $(BROKER_CFLAGS) -c $< -o $@ + +handle_pubackcomp.o : ../lib/handle_pubackcomp.c ../lib/read_handle.h + ${CROSS_COMPILE}${CC} $(BROKER_CFLAGS) -c $< -o $@ + +handle_publish.o : handle_publish.c mosquitto_broker.h + ${CROSS_COMPILE}${CC} $(BROKER_CFLAGS) -c $< -o $@ + +handle_pubrec.o : ../lib/handle_pubrec.c ../lib/read_handle.h + ${CROSS_COMPILE}${CC} $(BROKER_CFLAGS) -c $< -o $@ + +handle_pubrel.o : ../lib/handle_pubrel.c ../lib/read_handle.h + ${CROSS_COMPILE}${CC} $(BROKER_CFLAGS) -c $< -o $@ + +handle_suback.o : ../lib/handle_suback.c ../lib/read_handle.h + ${CROSS_COMPILE}${CC} $(BROKER_CFLAGS) -c $< -o $@ + +handle_subscribe.o : handle_subscribe.c mosquitto_broker.h + ${CROSS_COMPILE}${CC} $(BROKER_CFLAGS) -c $< -o $@ + +handle_unsuback.o : ../lib/handle_unsuback.c ../lib/read_handle.h + ${CROSS_COMPILE}${CC} $(BROKER_CFLAGS) -c $< -o $@ + +handle_unsubscribe.o : handle_unsubscribe.c mosquitto_broker.h + ${CROSS_COMPILE}${CC} $(BROKER_CFLAGS) -c $< -o $@ + logging.o : logging.c mosquitto_broker.h ${CROSS_COMPILE}${CC} $(BROKER_CFLAGS) -c $< -o $@ @@ -50,28 +126,34 @@ packet_mosq.o : ../lib/packet_mosq.c ../lib/packet_mosq.h read_handle.o : read_handle.c mosquitto_broker.h ${CROSS_COMPILE}${CC} $(BROKER_CFLAGS) -c $< -o $@ -read_handle_client.o : read_handle_client.c mosquitto_broker.h - ${CROSS_COMPILE}${CC} $(BROKER_CFLAGS) -c $< -o $@ - -read_handle_server.o : read_handle_server.c mosquitto_broker.h - ${CROSS_COMPILE}${CC} $(BROKER_CFLAGS) -c $< -o $@ - -read_handle_shared.o : ../lib/read_handle_shared.c ../lib/read_handle.h - ${CROSS_COMPILE}${CC} $(BROKER_CFLAGS) -c $< -o $@ - security.o : security.c mosquitto_broker.h ${CROSS_COMPILE}${CC} $(BROKER_CFLAGS) -c $< -o $@ security_default.o : security_default.c mosquitto_broker.h ${CROSS_COMPILE}${CC} $(BROKER_CFLAGS) -c $< -o $@ -send_client_mosq.o : ../lib/send_client_mosq.c ../lib/send_mosq.h +send_connect.o : ../lib/send_connect.c ../lib/send_mosq.h + ${CROSS_COMPILE}${CC} $(BROKER_CFLAGS) -c $< -o $@ + +send_disconnect.o : ../lib/send_disconnect.c ../lib/send_mosq.h + ${CROSS_COMPILE}${CC} $(BROKER_CFLAGS) -c $< -o $@ + +send_connack.o : send_connack.c mosquitto_broker.h ${CROSS_COMPILE}${CC} $(BROKER_CFLAGS) -c $< -o $@ send_mosq.o : ../lib/send_mosq.c ../lib/send_mosq.h ${CROSS_COMPILE}${CC} $(BROKER_CFLAGS) -c $< -o $@ -send_server.o : send_server.c mosquitto_broker.h +send_publish.o : ../lib/send_publish.c ../lib/send_mosq.h + ${CROSS_COMPILE}${CC} $(BROKER_CFLAGS) -c $< -o $@ + +send_suback.o : send_suback.c mosquitto_broker.h + ${CROSS_COMPILE}${CC} $(BROKER_CFLAGS) -c $< -o $@ + +send_subscribe.o : ../lib/send_subscribe.c ../lib/send_mosq.h + ${CROSS_COMPILE}${CC} $(BROKER_CFLAGS) -c $< -o $@ + +send_unsubscribe.o : ../lib/send_unsubscribe.c ../lib/send_mosq.h ${CROSS_COMPILE}${CC} $(BROKER_CFLAGS) -c $< -o $@ service.o : service.c mosquitto_broker.h diff --git a/src/read_handle_client.c b/src/handle_connack.c similarity index 99% rename from src/read_handle_client.c rename to src/handle_connack.c index 45c33b06..be015974 100644 --- a/src/read_handle_client.c +++ b/src/handle_connack.c @@ -4,12 +4,12 @@ Copyright (c) 2009-2015 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. */ diff --git a/src/read_handle_server.c b/src/handle_connect.c similarity index 76% rename from src/read_handle_server.c rename to src/handle_connect.c index d137f2aa..c0b1b83a 100644 --- a/src/read_handle_server.c +++ b/src/handle_connect.c @@ -4,12 +4,12 @@ Copyright (c) 2009-2015 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. */ @@ -145,7 +145,7 @@ int handle__connect(struct mosquitto_db *db, struct mosquitto *context) goto handle_connect_error; } if((context->in_packet.command&0x0F) != 0x00){ - /* Reserved flags not set to 0, must disconnect. */ + /* Reserved flags not set to 0, must disconnect. */ mosquitto__free(protocol_name); rc = MOSQ_ERR_PROTOCOL; goto handle_connect_error; @@ -605,194 +605,3 @@ int handle__disconnect(struct mosquitto_db *db, struct mosquitto *context) } -int handle__subscribe(struct mosquitto_db *db, struct mosquitto *context) -{ - int rc = 0; - int rc2; - uint16_t mid; - char *sub; - uint8_t qos; - uint8_t *payload = NULL, *tmp_payload; - uint32_t payloadlen = 0; - int len; - char *sub_mount; - - if(!context) return MOSQ_ERR_INVAL; - log__printf(NULL, MOSQ_LOG_DEBUG, "Received SUBSCRIBE from %s", context->id); - /* FIXME - plenty of potential for memory leaks here */ - - if(context->protocol == mosq_p_mqtt311){ - if((context->in_packet.command&0x0F) != 0x02){ - return MOSQ_ERR_PROTOCOL; - } - } - if(packet__read_uint16(&context->in_packet, &mid)) return 1; - - while(context->in_packet.pos < context->in_packet.remaining_length){ - sub = NULL; - if(packet__read_string(&context->in_packet, &sub)){ - mosquitto__free(payload); - return 1; - } - - if(sub){ - if(STREMPTY(sub)){ - log__printf(NULL, MOSQ_LOG_INFO, "Empty subscription string from %s, disconnecting.", - context->address); - mosquitto__free(sub); - mosquitto__free(payload); - return 1; - } - if(mosquitto_sub_topic_check(sub)){ - log__printf(NULL, MOSQ_LOG_INFO, "Invalid subscription string from %s, disconnecting.", - context->address); - mosquitto__free(sub); - mosquitto__free(payload); - return 1; - } - - if(packet__read_byte(&context->in_packet, &qos)){ - mosquitto__free(sub); - mosquitto__free(payload); - return 1; - } - if(qos > 2){ - log__printf(NULL, MOSQ_LOG_INFO, "Invalid QoS in subscription command from %s, disconnecting.", - context->address); - mosquitto__free(sub); - mosquitto__free(payload); - return 1; - } - if(context->listener && context->listener->mount_point){ - len = strlen(context->listener->mount_point) + strlen(sub) + 1; - sub_mount = mosquitto__malloc(len+1); - if(!sub_mount){ - mosquitto__free(sub); - mosquitto__free(payload); - return MOSQ_ERR_NOMEM; - } - snprintf(sub_mount, len, "%s%s", context->listener->mount_point, sub); - sub_mount[len] = '\0'; - - mosquitto__free(sub); - sub = sub_mount; - - } - log__printf(NULL, MOSQ_LOG_DEBUG, "\t%s (QoS %d)", sub, qos); - -#if 0 - /* FIXME - * This section has been disabled temporarily. mosquitto_acl_check - * calls mosquitto_topic_matches_sub, which can't cope with - * checking subscriptions that have wildcards against ACLs that - * have wildcards. Bug #1374291 is related. - * - * It's a very difficult problem when an ACL looks like foo/+/bar - * and a subscription request to foo/# is made. - * - * This should be changed to using MOSQ_ACL_SUBSCRIPTION in the - * future anyway. - */ - if(context->protocol == mosq_p_mqtt311){ - rc = mosquitto_acl_check(db, context, sub, MOSQ_ACL_READ); - switch(rc){ - case MOSQ_ERR_SUCCESS: - break; - case MOSQ_ERR_ACL_DENIED: - qos = 0x80; - break; - default: - mosquitto__free(sub); - return rc; - } - } -#endif - - if(qos != 0x80){ - rc2 = sub__add(db, context, sub, qos, &db->subs); - if(rc2 == MOSQ_ERR_SUCCESS){ - if(sub__retain_queue(db, context, sub, qos)) rc = 1; - }else if(rc2 != -1){ - rc = rc2; - } - log__printf(NULL, MOSQ_LOG_SUBSCRIBE, "%s %d %s", context->id, qos, sub); - } - mosquitto__free(sub); - - tmp_payload = mosquitto__realloc(payload, payloadlen + 1); - if(tmp_payload){ - payload = tmp_payload; - payload[payloadlen] = qos; - payloadlen++; - }else{ - mosquitto__free(payload); - - return MOSQ_ERR_NOMEM; - } - } - } - - if(context->protocol == mosq_p_mqtt311){ - if(payloadlen == 0){ - /* No subscriptions specified, protocol error. */ - return MOSQ_ERR_PROTOCOL; - } - } - if(send__suback(context, mid, payloadlen, payload)) rc = 1; - mosquitto__free(payload); - -#ifdef WITH_PERSISTENCE - db->persistence_changes++; -#endif - - return rc; -} - -int handle__unsubscribe(struct mosquitto_db *db, struct mosquitto *context) -{ - uint16_t mid; - char *sub; - - if(!context) return MOSQ_ERR_INVAL; - log__printf(NULL, MOSQ_LOG_DEBUG, "Received UNSUBSCRIBE from %s", context->id); - - if(context->protocol == mosq_p_mqtt311){ - if((context->in_packet.command&0x0F) != 0x02){ - return MOSQ_ERR_PROTOCOL; - } - } - if(packet__read_uint16(&context->in_packet, &mid)) return 1; - - while(context->in_packet.pos < context->in_packet.remaining_length){ - sub = NULL; - if(packet__read_string(&context->in_packet, &sub)){ - return 1; - } - - if(sub){ - if(STREMPTY(sub)){ - log__printf(NULL, MOSQ_LOG_INFO, "Empty unsubscription string from %s, disconnecting.", - context->id); - mosquitto__free(sub); - return 1; - } - if(mosquitto_sub_topic_check(sub)){ - log__printf(NULL, MOSQ_LOG_INFO, "Invalid unsubscription string from %s, disconnecting.", - context->id); - mosquitto__free(sub); - return 1; - } - - log__printf(NULL, MOSQ_LOG_DEBUG, "\t%s", sub); - sub__remove(db, context, sub, &db->subs); - log__printf(NULL, MOSQ_LOG_UNSUBSCRIBE, "%s %s", context->id, sub); - mosquitto__free(sub); - } - } -#ifdef WITH_PERSISTENCE - db->persistence_changes++; -#endif - - return send__command_with_mid(context, UNSUBACK, mid, false); -} - diff --git a/src/handle_publish.c b/src/handle_publish.c new file mode 100644 index 00000000..7f5c0cdb --- /dev/null +++ b/src/handle_publish.c @@ -0,0 +1,239 @@ +/* +Copyright (c) 2009-2015 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 +#include +#include + +#include "config.h" + +#include "mosquitto_broker.h" +#include "mqtt3_protocol.h" +#include "memory_mosq.h" +#include "packet_mosq.h" +#include "read_handle.h" +#include "send_mosq.h" +#include "sys_tree.h" +#include "util_mosq.h" + + +int handle__publish(struct mosquitto_db *db, struct mosquitto *context) +{ + char *topic; + mosquitto__payload_uhpa payload; + uint32_t payloadlen; + uint8_t dup, qos, retain; + uint16_t mid = 0; + int rc = 0; + uint8_t header = context->in_packet.command; + int res = 0; + struct mosquitto_msg_store *stored = NULL; + int len; + char *topic_mount; +#ifdef WITH_BRIDGE + char *topic_temp; + int i; + struct mosquitto__bridge_topic *cur_topic; + bool match; +#endif + + payload.ptr = NULL; + + dup = (header & 0x08)>>3; + qos = (header & 0x06)>>1; + if(qos == 3){ + log__printf(NULL, MOSQ_LOG_INFO, + "Invalid QoS in PUBLISH from %s, disconnecting.", context->id); + return 1; + } + retain = (header & 0x01); + + if(packet__read_string(&context->in_packet, &topic)) return 1; + if(STREMPTY(topic)){ + /* Invalid publish topic, disconnect client. */ + mosquitto__free(topic); + return 1; + } +#ifdef WITH_BRIDGE + if(context->bridge && context->bridge->topics && context->bridge->topic_remapping){ + for(i=0; ibridge->topic_count; i++){ + cur_topic = &context->bridge->topics[i]; + if((cur_topic->direction == bd_both || cur_topic->direction == bd_in) + && (cur_topic->remote_prefix || cur_topic->local_prefix)){ + + /* Topic mapping required on this topic if the message matches */ + + rc = mosquitto_topic_matches_sub(cur_topic->remote_topic, topic, &match); + if(rc){ + mosquitto__free(topic); + return rc; + } + if(match){ + if(cur_topic->remote_prefix){ + /* This prefix needs removing. */ + if(!strncmp(cur_topic->remote_prefix, topic, strlen(cur_topic->remote_prefix))){ + topic_temp = mosquitto__strdup(topic+strlen(cur_topic->remote_prefix)); + if(!topic_temp){ + mosquitto__free(topic); + return MOSQ_ERR_NOMEM; + } + mosquitto__free(topic); + topic = topic_temp; + } + } + + if(cur_topic->local_prefix){ + /* This prefix needs adding. */ + len = strlen(topic) + strlen(cur_topic->local_prefix)+1; + topic_temp = mosquitto__malloc(len+1); + if(!topic_temp){ + mosquitto__free(topic); + return MOSQ_ERR_NOMEM; + } + snprintf(topic_temp, len, "%s%s", cur_topic->local_prefix, topic); + topic_temp[len] = '\0'; + + mosquitto__free(topic); + topic = topic_temp; + } + break; + } + } + } + } +#endif + if(mosquitto_pub_topic_check(topic) != MOSQ_ERR_SUCCESS){ + /* Invalid publish topic, just swallow it. */ + mosquitto__free(topic); + return 1; + } + + if(qos > 0){ + if(packet__read_uint16(&context->in_packet, &mid)){ + mosquitto__free(topic); + return 1; + } + } + + payloadlen = context->in_packet.remaining_length - context->in_packet.pos; + G_PUB_BYTES_RECEIVED_INC(payloadlen); + if(context->listener && context->listener->mount_point){ + len = strlen(context->listener->mount_point) + strlen(topic) + 1; + topic_mount = mosquitto__malloc(len+1); + if(!topic_mount){ + mosquitto__free(topic); + return MOSQ_ERR_NOMEM; + } + snprintf(topic_mount, len, "%s%s", context->listener->mount_point, topic); + topic_mount[len] = '\0'; + + mosquitto__free(topic); + topic = topic_mount; + } + + if(payloadlen){ + if(db->config->message_size_limit && payloadlen > db->config->message_size_limit){ + log__printf(NULL, MOSQ_LOG_DEBUG, "Dropped too large PUBLISH from %s (d%d, q%d, r%d, m%d, '%s', ... (%ld bytes))", context->id, dup, qos, retain, mid, topic, (long)payloadlen); + goto process_bad_message; + } + if(UHPA_ALLOC(payload, payloadlen+1) == 0){ + mosquitto__free(topic); + return MOSQ_ERR_NOMEM; + } + if(packet__read_bytes(&context->in_packet, UHPA_ACCESS(payload, payloadlen), payloadlen)){ + mosquitto__free(topic); + UHPA_FREE(payload, payloadlen); + return 1; + } + } + + /* Check for topic access */ + rc = mosquitto_acl_check(db, context, topic, MOSQ_ACL_WRITE); + if(rc == MOSQ_ERR_ACL_DENIED){ + log__printf(NULL, MOSQ_LOG_DEBUG, "Denied PUBLISH from %s (d%d, q%d, r%d, m%d, '%s', ... (%ld bytes))", context->id, dup, qos, retain, mid, topic, (long)payloadlen); + goto process_bad_message; + }else if(rc != MOSQ_ERR_SUCCESS){ + mosquitto__free(topic); + UHPA_FREE(payload, payloadlen); + return rc; + } + + log__printf(NULL, MOSQ_LOG_DEBUG, "Received PUBLISH from %s (d%d, q%d, r%d, m%d, '%s', ... (%ld bytes))", context->id, dup, qos, retain, mid, topic, (long)payloadlen); + if(qos > 0){ + db__message_store_find(context, mid, &stored); + } + if(!stored){ + dup = 0; + if(db__message_store(db, context->id, mid, topic, qos, payloadlen, &payload, retain, &stored, 0)){ + return 1; + } + }else{ + mosquitto__free(topic); + topic = stored->topic; + dup = 1; + } + + switch(qos){ + case 0: + if(sub__messages_queue(db, context->id, topic, qos, retain, &stored)) rc = 1; + break; + case 1: + if(sub__messages_queue(db, context->id, topic, qos, retain, &stored)) rc = 1; + if(send__puback(context, mid)) rc = 1; + break; + case 2: + if(!dup){ + res = db__message_insert(db, context, mid, mosq_md_in, qos, retain, stored); + }else{ + res = 0; + } + /* db__message_insert() returns 2 to indicate dropped message + * due to queue. This isn't an error so don't disconnect them. */ + if(!res){ + if(send__pubrec(context, mid)) rc = 1; + }else if(res == 1){ + rc = 1; + } + break; + } + + return rc; +process_bad_message: + mosquitto__free(topic); + UHPA_FREE(payload, payloadlen); + switch(qos){ + case 0: + return MOSQ_ERR_SUCCESS; + case 1: + return send__puback(context, mid); + case 2: + db__message_store_find(context, mid, &stored); + if(!stored){ + if(db__message_store(db, context->id, mid, NULL, qos, 0, NULL, false, &stored, 0)){ + return 1; + } + res = db__message_insert(db, context, mid, mosq_md_in, qos, false, stored); + }else{ + res = 0; + } + if(!res){ + res = send__pubrec(context, mid); + } + return res; + } + return 1; +} + diff --git a/src/handle_subscribe.c b/src/handle_subscribe.c new file mode 100644 index 00000000..847553ce --- /dev/null +++ b/src/handle_subscribe.c @@ -0,0 +1,181 @@ +/* +Copyright (c) 2009-2016 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 +#include + +#include "config.h" + +#include "mosquitto_broker.h" +#include "memory_mosq.h" +#include "packet_mosq.h" +/* + +#include "mosquitto_broker.h" +#include "mqtt3_protocol.h" +#include "send_mosq.h" +#include "sys_tree.h" +#include "time_mosq.h" +#include "tls_mosq.h" +#include "util_mosq.h" +*/ + + + +int handle__subscribe(struct mosquitto_db *db, struct mosquitto *context) +{ + int rc = 0; + int rc2; + uint16_t mid; + char *sub; + uint8_t qos; + uint8_t *payload = NULL, *tmp_payload; + uint32_t payloadlen = 0; + int len; + char *sub_mount; + + if(!context) return MOSQ_ERR_INVAL; + log__printf(NULL, MOSQ_LOG_DEBUG, "Received SUBSCRIBE from %s", context->id); + /* FIXME - plenty of potential for memory leaks here */ + + if(context->protocol == mosq_p_mqtt311){ + if((context->in_packet.command&0x0F) != 0x02){ + return MOSQ_ERR_PROTOCOL; + } + } + if(packet__read_uint16(&context->in_packet, &mid)) return 1; + + while(context->in_packet.pos < context->in_packet.remaining_length){ + sub = NULL; + if(packet__read_string(&context->in_packet, &sub)){ + mosquitto__free(payload); + return 1; + } + + if(sub){ + if(STREMPTY(sub)){ + log__printf(NULL, MOSQ_LOG_INFO, "Empty subscription string from %s, disconnecting.", + context->address); + mosquitto__free(sub); + mosquitto__free(payload); + return 1; + } + if(mosquitto_sub_topic_check(sub)){ + log__printf(NULL, MOSQ_LOG_INFO, "Invalid subscription string from %s, disconnecting.", + context->address); + mosquitto__free(sub); + mosquitto__free(payload); + return 1; + } + + if(packet__read_byte(&context->in_packet, &qos)){ + mosquitto__free(sub); + mosquitto__free(payload); + return 1; + } + if(qos > 2){ + log__printf(NULL, MOSQ_LOG_INFO, "Invalid QoS in subscription command from %s, disconnecting.", + context->address); + mosquitto__free(sub); + mosquitto__free(payload); + return 1; + } + if(context->listener && context->listener->mount_point){ + len = strlen(context->listener->mount_point) + strlen(sub) + 1; + sub_mount = mosquitto__malloc(len+1); + if(!sub_mount){ + mosquitto__free(sub); + mosquitto__free(payload); + return MOSQ_ERR_NOMEM; + } + snprintf(sub_mount, len, "%s%s", context->listener->mount_point, sub); + sub_mount[len] = '\0'; + + mosquitto__free(sub); + sub = sub_mount; + + } + log__printf(NULL, MOSQ_LOG_DEBUG, "\t%s (QoS %d)", sub, qos); + +#if 0 + /* FIXME + * This section has been disabled temporarily. mosquitto_acl_check + * calls mosquitto_topic_matches_sub, which can't cope with + * checking subscriptions that have wildcards against ACLs that + * have wildcards. Bug #1374291 is related. + * + * It's a very difficult problem when an ACL looks like foo/+/bar + * and a subscription request to foo/# is made. + * + * This should be changed to using MOSQ_ACL_SUBSCRIPTION in the + * future anyway. + */ + if(context->protocol == mosq_p_mqtt311){ + rc = mosquitto_acl_check(db, context, sub, MOSQ_ACL_READ); + switch(rc){ + case MOSQ_ERR_SUCCESS: + break; + case MOSQ_ERR_ACL_DENIED: + qos = 0x80; + break; + default: + mosquitto__free(sub); + return rc; + } + } +#endif + + if(qos != 0x80){ + rc2 = sub__add(db, context, sub, qos, &db->subs); + if(rc2 == MOSQ_ERR_SUCCESS){ + if(sub__retain_queue(db, context, sub, qos)) rc = 1; + }else if(rc2 != -1){ + rc = rc2; + } + log__printf(NULL, MOSQ_LOG_SUBSCRIBE, "%s %d %s", context->id, qos, sub); + } + mosquitto__free(sub); + + tmp_payload = mosquitto__realloc(payload, payloadlen + 1); + if(tmp_payload){ + payload = tmp_payload; + payload[payloadlen] = qos; + payloadlen++; + }else{ + mosquitto__free(payload); + + return MOSQ_ERR_NOMEM; + } + } + } + + if(context->protocol == mosq_p_mqtt311){ + if(payloadlen == 0){ + /* No subscriptions specified, protocol error. */ + return MOSQ_ERR_PROTOCOL; + } + } + if(send__suback(context, mid, payloadlen, payload)) rc = 1; + mosquitto__free(payload); + +#ifdef WITH_PERSISTENCE + db->persistence_changes++; +#endif + + return rc; +} + + diff --git a/src/handle_unsubscribe.c b/src/handle_unsubscribe.c new file mode 100644 index 00000000..af7a9587 --- /dev/null +++ b/src/handle_unsubscribe.c @@ -0,0 +1,81 @@ +/* +Copyright (c) 2009-2015 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 +#include + +#include "config.h" + +#include "mosquitto_broker.h" +#include "memory_mosq.h" +#include "mqtt3_protocol.h" +#include "packet_mosq.h" +#include "send_mosq.h" +/* +#include "sys_tree.h" +#include "time_mosq.h" +#include "tls_mosq.h" +#include "util_mosq.h" +*/ + +int handle__unsubscribe(struct mosquitto_db *db, struct mosquitto *context) +{ + uint16_t mid; + char *sub; + + if(!context) return MOSQ_ERR_INVAL; + log__printf(NULL, MOSQ_LOG_DEBUG, "Received UNSUBSCRIBE from %s", context->id); + + if(context->protocol == mosq_p_mqtt311){ + if((context->in_packet.command&0x0F) != 0x02){ + return MOSQ_ERR_PROTOCOL; + } + } + if(packet__read_uint16(&context->in_packet, &mid)) return 1; + + while(context->in_packet.pos < context->in_packet.remaining_length){ + sub = NULL; + if(packet__read_string(&context->in_packet, &sub)){ + return 1; + } + + if(sub){ + if(STREMPTY(sub)){ + log__printf(NULL, MOSQ_LOG_INFO, "Empty unsubscription string from %s, disconnecting.", + context->id); + mosquitto__free(sub); + return 1; + } + if(mosquitto_sub_topic_check(sub)){ + log__printf(NULL, MOSQ_LOG_INFO, "Invalid unsubscription string from %s, disconnecting.", + context->id); + mosquitto__free(sub); + return 1; + } + + log__printf(NULL, MOSQ_LOG_DEBUG, "\t%s", sub); + sub__remove(db, context, sub, &db->subs); + log__printf(NULL, MOSQ_LOG_UNSUBSCRIBE, "%s %s", context->id, sub); + mosquitto__free(sub); + } + } +#ifdef WITH_PERSISTENCE + db->persistence_changes++; +#endif + + return send__command_with_mid(context, UNSUBACK, mid, false); +} + diff --git a/src/read_handle.c b/src/read_handle.c index 0e002616..193a5e2e 100644 --- a/src/read_handle.c +++ b/src/read_handle.c @@ -1,5 +1,5 @@ /* -Copyright (c) 2009-2015 Roger Light +Copyright (c) 2009-2016 Roger Light All rights reserved. This program and the accompanying materials are made available under the terms of the Eclipse Public License v1.0 @@ -29,6 +29,7 @@ Contributors: #include "sys_tree.h" #include "util_mosq.h" + int handle__packet(struct mosquitto_db *db, struct mosquitto *context) { if(!context) return MOSQ_ERR_INVAL; @@ -70,210 +71,3 @@ int handle__packet(struct mosquitto_db *db, struct mosquitto *context) } } -int handle__publish(struct mosquitto_db *db, struct mosquitto *context) -{ - char *topic; - mosquitto__payload_uhpa payload; - uint32_t payloadlen; - uint8_t dup, qos, retain; - uint16_t mid = 0; - int rc = 0; - uint8_t header = context->in_packet.command; - int res = 0; - struct mosquitto_msg_store *stored = NULL; - int len; - char *topic_mount; -#ifdef WITH_BRIDGE - char *topic_temp; - int i; - struct mosquitto__bridge_topic *cur_topic; - bool match; -#endif - - payload.ptr = NULL; - - dup = (header & 0x08)>>3; - qos = (header & 0x06)>>1; - if(qos == 3){ - log__printf(NULL, MOSQ_LOG_INFO, - "Invalid QoS in PUBLISH from %s, disconnecting.", context->id); - return 1; - } - retain = (header & 0x01); - - if(packet__read_string(&context->in_packet, &topic)) return 1; - if(STREMPTY(topic)){ - /* Invalid publish topic, disconnect client. */ - mosquitto__free(topic); - return 1; - } -#ifdef WITH_BRIDGE - if(context->bridge && context->bridge->topics && context->bridge->topic_remapping){ - for(i=0; ibridge->topic_count; i++){ - cur_topic = &context->bridge->topics[i]; - if((cur_topic->direction == bd_both || cur_topic->direction == bd_in) - && (cur_topic->remote_prefix || cur_topic->local_prefix)){ - - /* Topic mapping required on this topic if the message matches */ - - rc = mosquitto_topic_matches_sub(cur_topic->remote_topic, topic, &match); - if(rc){ - mosquitto__free(topic); - return rc; - } - if(match){ - if(cur_topic->remote_prefix){ - /* This prefix needs removing. */ - if(!strncmp(cur_topic->remote_prefix, topic, strlen(cur_topic->remote_prefix))){ - topic_temp = mosquitto__strdup(topic+strlen(cur_topic->remote_prefix)); - if(!topic_temp){ - mosquitto__free(topic); - return MOSQ_ERR_NOMEM; - } - mosquitto__free(topic); - topic = topic_temp; - } - } - - if(cur_topic->local_prefix){ - /* This prefix needs adding. */ - len = strlen(topic) + strlen(cur_topic->local_prefix)+1; - topic_temp = mosquitto__malloc(len+1); - if(!topic_temp){ - mosquitto__free(topic); - return MOSQ_ERR_NOMEM; - } - snprintf(topic_temp, len, "%s%s", cur_topic->local_prefix, topic); - topic_temp[len] = '\0'; - - mosquitto__free(topic); - topic = topic_temp; - } - break; - } - } - } - } -#endif - if(mosquitto_pub_topic_check(topic) != MOSQ_ERR_SUCCESS){ - /* Invalid publish topic, just swallow it. */ - mosquitto__free(topic); - return 1; - } - - if(qos > 0){ - if(packet__read_uint16(&context->in_packet, &mid)){ - mosquitto__free(topic); - return 1; - } - } - - payloadlen = context->in_packet.remaining_length - context->in_packet.pos; - G_PUB_BYTES_RECEIVED_INC(payloadlen); - if(context->listener && context->listener->mount_point){ - len = strlen(context->listener->mount_point) + strlen(topic) + 1; - topic_mount = mosquitto__malloc(len+1); - if(!topic_mount){ - mosquitto__free(topic); - return MOSQ_ERR_NOMEM; - } - snprintf(topic_mount, len, "%s%s", context->listener->mount_point, topic); - topic_mount[len] = '\0'; - - mosquitto__free(topic); - topic = topic_mount; - } - - if(payloadlen){ - if(db->config->message_size_limit && payloadlen > db->config->message_size_limit){ - log__printf(NULL, MOSQ_LOG_DEBUG, "Dropped too large PUBLISH from %s (d%d, q%d, r%d, m%d, '%s', ... (%ld bytes))", context->id, dup, qos, retain, mid, topic, (long)payloadlen); - goto process_bad_message; - } - if(UHPA_ALLOC(payload, payloadlen+1) == 0){ - mosquitto__free(topic); - return MOSQ_ERR_NOMEM; - } - if(packet__read_bytes(&context->in_packet, UHPA_ACCESS(payload, payloadlen), payloadlen)){ - mosquitto__free(topic); - UHPA_FREE(payload, payloadlen); - return 1; - } - } - - /* Check for topic access */ - rc = mosquitto_acl_check(db, context, topic, MOSQ_ACL_WRITE); - if(rc == MOSQ_ERR_ACL_DENIED){ - log__printf(NULL, MOSQ_LOG_DEBUG, "Denied PUBLISH from %s (d%d, q%d, r%d, m%d, '%s', ... (%ld bytes))", context->id, dup, qos, retain, mid, topic, (long)payloadlen); - goto process_bad_message; - }else if(rc != MOSQ_ERR_SUCCESS){ - mosquitto__free(topic); - UHPA_FREE(payload, payloadlen); - return rc; - } - - log__printf(NULL, MOSQ_LOG_DEBUG, "Received PUBLISH from %s (d%d, q%d, r%d, m%d, '%s', ... (%ld bytes))", context->id, dup, qos, retain, mid, topic, (long)payloadlen); - if(qos > 0){ - db__message_store_find(context, mid, &stored); - } - if(!stored){ - dup = 0; - if(db__message_store(db, context->id, mid, topic, qos, payloadlen, &payload, retain, &stored, 0)){ - return 1; - } - }else{ - mosquitto__free(topic); - topic = stored->topic; - dup = 1; - } - - switch(qos){ - case 0: - if(sub__messages_queue(db, context->id, topic, qos, retain, &stored)) rc = 1; - break; - case 1: - if(sub__messages_queue(db, context->id, topic, qos, retain, &stored)) rc = 1; - if(send__puback(context, mid)) rc = 1; - break; - case 2: - if(!dup){ - res = db__message_insert(db, context, mid, mosq_md_in, qos, retain, stored); - }else{ - res = 0; - } - /* db__message_insert() returns 2 to indicate dropped message - * due to queue. This isn't an error so don't disconnect them. */ - if(!res){ - if(send__pubrec(context, mid)) rc = 1; - }else if(res == 1){ - rc = 1; - } - break; - } - - return rc; -process_bad_message: - mosquitto__free(topic); - UHPA_FREE(payload, payloadlen); - switch(qos){ - case 0: - return MOSQ_ERR_SUCCESS; - case 1: - return send__puback(context, mid); - case 2: - db__message_store_find(context, mid, &stored); - if(!stored){ - if(db__message_store(db, context->id, mid, NULL, qos, 0, NULL, false, &stored, 0)){ - return 1; - } - res = db__message_insert(db, context, mid, mosq_md_in, qos, false, stored); - }else{ - res = 0; - } - if(!res){ - res = send__pubrec(context, mid); - } - return res; - } - return 1; -} - diff --git a/src/send_server.c b/src/send_connack.c similarity index 69% rename from src/send_server.c rename to src/send_connack.c index eef5b30f..012be034 100644 --- a/src/send_server.c +++ b/src/send_connack.c @@ -4,12 +4,12 @@ Copyright (c) 2009-2015 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. */ @@ -51,27 +51,3 @@ int send__connack(struct mosquitto *context, int ack, int result) return packet__queue(context, packet); } -int send__suback(struct mosquitto *context, uint16_t mid, uint32_t payloadlen, const void *payload) -{ - struct mosquitto__packet *packet = NULL; - int rc; - - log__printf(NULL, MOSQ_LOG_DEBUG, "Sending SUBACK to %s", context->id); - - packet = mosquitto__calloc(1, sizeof(struct mosquitto__packet)); - if(!packet) return MOSQ_ERR_NOMEM; - - packet->command = SUBACK; - packet->remaining_length = 2+payloadlen; - rc = packet__alloc(packet); - if(rc){ - mosquitto__free(packet); - return rc; - } - packet__write_uint16(packet, mid); - if(payloadlen){ - packet__write_bytes(packet, payload, payloadlen); - } - - return packet__queue(context, packet); -} diff --git a/src/send_suback.c b/src/send_suback.c new file mode 100644 index 00000000..b3609040 --- /dev/null +++ b/src/send_suback.c @@ -0,0 +1,49 @@ +/* +Copyright (c) 2009-2015 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_broker.h" +#include "mqtt3_protocol.h" +#include "memory_mosq.h" +#include "packet_mosq.h" +#include "util_mosq.h" + + +int send__suback(struct mosquitto *context, uint16_t mid, uint32_t payloadlen, const void *payload) +{ + struct mosquitto__packet *packet = NULL; + int rc; + + log__printf(NULL, MOSQ_LOG_DEBUG, "Sending SUBACK to %s", context->id); + + packet = mosquitto__calloc(1, sizeof(struct mosquitto__packet)); + if(!packet) return MOSQ_ERR_NOMEM; + + packet->command = SUBACK; + packet->remaining_length = 2+payloadlen; + rc = packet__alloc(packet); + if(rc){ + mosquitto__free(packet); + return rc; + } + packet__write_uint16(packet, mid); + if(payloadlen){ + packet__write_bytes(packet, payload, payloadlen); + } + + return packet__queue(context, packet); +}