Mirrors_Framework/mosquitto
1
0
Fork 0
mirror of https://github.com/eclipse-mosquitto/mosquitto.git synced 2026-09-24 09:14:02 +08:00
Code Issues Packages Projects Releases Wiki Activity

Reorganise functions into separate files as appropriate.

Browse Source
This commit is contained in:
Roger A. Light
2016-03-12 21:49:48 +00:00
parent bd61d18e92
commit db2e66c543
28 changed files with 1623 additions and 1015 deletions
Download Patch File Download Diff File Expand all files Collapse all files
lib/CMakeLists.txt
+14 -3
View File
@@ -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
lib/Makefile
+47 -6
View File
@@ -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
lib/read_handle_client.c → lib/handle_connack.c
+2 -2
View File
@@ -4,12 +4,12 @@ Copyright (c) 2009-2015 Roger Light <roger@atchoo.org>
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.
*/
lib/handle_ping.c
+57
View File
@@ -0,0 +1,57 @@
/*
Copyright (c) 2009-2015 Roger Light <roger@atchoo.org>
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 <assert.h>
#include <stdio.h>
#include <string.h>
#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;
}
lib/handle_pubackcomp.c
+73
View File
@@ -0,0 +1,73 @@
/*
Copyright (c) 2009-2015 Roger Light <roger@atchoo.org>
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 <assert.h>
#include <stdio.h>
#include <string.h>
#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;
}
lib/handle_publish.c
+120
View File
@@ -0,0 +1,120 @@
/*
Copyright (c) 2009-2015 Roger Light <roger@atchoo.org>
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 <assert.h>
#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;
}
}
lib/handle_pubrec.c
+59
View File
@@ -0,0 +1,59 @@
/*
Copyright (c) 2009-2015 Roger Light <roger@atchoo.org>
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 <assert.h>
#include <stdio.h>
#include <string.h>
#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;
}
lib/handle_pubrel.c
+81
View File
@@ -0,0 +1,81 @@
/*
Copyright (c) 2009-2015 Roger Light <roger@atchoo.org>
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 <assert.h>
#include <stdio.h>
#include <string.h>
#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;
}
lib/handle_suback.c
+73
View File
@@ -0,0 +1,73 @@
/*
Copyright (c) 2009-2015 Roger Light <roger@atchoo.org>
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 <assert.h>
#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;
}
lib/handle_unsuback.c
+61
View File
@@ -0,0 +1,61 @@
/*
Copyright (c) 2009-2015 Roger Light <roger@atchoo.org>
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 <assert.h>
#include <stdio.h>
#include <string.h>
#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;
}
lib/read_handle.c
-92
View File
@@ -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;
}
}
lib/read_handle_shared.c
-233
View File
@@ -1,233 +0,0 @@
/*
Copyright (c) 2009-2015 Roger Light <roger@atchoo.org>
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 <assert.h>
#include <stdio.h>
#include <string.h>
#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;
}
lib/send_client_mosq.c → lib/send_connect.c
+5 -108
View File
@@ -4,30 +4,27 @@ Copyright (c) 2009-2015 Roger Light <roger@atchoo.org>
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 <assert.h>
#include <string.h>
#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);
}
lib/send_disconnect.c
+42
View File
@@ -0,0 +1,42 @@
/*
Copyright (c) 2009-2015 Roger Light <roger@atchoo.org>
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 <assert.h>
#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);
}
lib/send_mosq.c
-125
View File
@@ -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; i<mosq->bridge->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);
}
lib/send_publish.c
+166
View File
@@ -0,0 +1,166 @@
/*
Copyright (c) 2009-2015 Roger Light <roger@atchoo.org>
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 <assert.h>
//#include <stdio.h>
//#include <string.h>
#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; i<mosq->bridge->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);
}
lib/send_subscribe.c
+75
View File
@@ -0,0 +1,75 @@
/*
Copyright (c) 2009-2015 Roger Light <roger@atchoo.org>
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 <assert.h>
#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);
}
lib/send_unsubscribe.c
+74
View File
@@ -0,0 +1,74 @@
/*
Copyright (c) 2009-2015 Roger Light <roger@atchoo.org>
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 <assert.h>
//#include <string.h>
#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);
}
src/CMakeLists.txt
+21 -4
View File
@@ -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)
src/Makefile
+94 -12
View File
@@ -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
src/read_handle_client.c → src/handle_connack.c
+2 -2
View File
@@ -4,12 +4,12 @@ Copyright (c) 2009-2015 Roger Light <roger@atchoo.org>
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.
*/
src/read_handle_server.c → src/handle_connect.c
+3 -194
View File
@@ -4,12 +4,12 @@ Copyright (c) 2009-2015 Roger Light <roger@atchoo.org>
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);
}
src/handle_publish.c
+239
View File
@@ -0,0 +1,239 @@
/*
Copyright (c) 2009-2015 Roger Light <roger@atchoo.org>
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 <assert.h>
#include <stdio.h>
#include <string.h>
#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; i<context->bridge->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;
}
src/handle_subscribe.c
+181
View File
@@ -0,0 +1,181 @@
/*
Copyright (c) 2009-2016 Roger Light <roger@atchoo.org>
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 <stdio.h>
#include <string.h>
#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;
}
src/handle_unsubscribe.c
+81
View File
@@ -0,0 +1,81 @@
/*
Copyright (c) 2009-2015 Roger Light <roger@atchoo.org>
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 <stdio.h>
#include <string.h>
#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);
}
src/read_handle.c
+2 -208
View File
@@ -1,5 +1,5 @@
/*
Copyright (c) 2009-2015 Roger Light <roger@atchoo.org>
Copyright (c) 2009-2016 Roger Light <roger@atchoo.org>
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; i<context->bridge->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;
}
src/send_server.c → src/send_connack.c
+2 -26
View File
@@ -4,12 +4,12 @@ Copyright (c) 2009-2015 Roger Light <roger@atchoo.org>
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);
}
src/send_suback.c
+49
View File
@@ -0,0 +1,49 @@
/*
Copyright (c) 2009-2015 Roger Light <roger@atchoo.org>
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);
}