Merge branch 'fixes'

CMakeLists.txt

@@ -11,7 +11,7 @@ project(mosquitto)
 cmake_minimum_required(VERSION 2.8)
 # Only for version 3 and up. cmake_policy(SET CMP0042 NEW)
 
-set (VERSION 1.6.10)
+set (VERSION 1.6.11)
 
 add_definitions (-DCMAKE -DVERSION=\"${VERSION}\")
 

ChangeLog.txt

@@ -1,3 +1,45 @@
+1.6.11 - 2020-08-11
+===================
+
+Security:
+- On Windows the Mosquitto service was being installed without appropriate
+  path quoting, this has been fixed.
+
+Broker:
+- Fix usage message only mentioning v3.1.1. Closes #1713.
+- Fix broker refusing to start if only websockets listeners were defined.
+  Closes #1740.
+- Change systemd unit files to create /var/log/mosquitto before starting.
+  Closes #821.
+- Don't quit with an error if opening the log file isn't possible.
+  Closes #821.
+- Fix bridge topic remapping when using "" as the topic. Closes #1749.
+- Fix messages being queued for disconnected bridges when clean start was
+  set to true. Closes #1729.
+- Fix `autosave_interval` not being triggered by messages being delivered.
+  Closes #1726.
+- Fix websockets clients sometimes not being disconnected promptly.
+  Closes #1718.
+- Fix "slow" file based logging by switching to line based buffering.
+  Closes #1689. Closes #1741.
+- Log protocol error message where appropriate from a bad UNSUBSCRIBE, rather
+  than the generic "socket error".
+- Don't try to start DLT logging if DLT unavailable, to avoid a long delay
+  when shutting down the broker. Closes #1735.
+- Fix potential memory leaks. Closes #1773. Closes #1774.
+- Fix clients not receiving messages after a previous client with the same
+  client ID and positive will delay interval quit. Closes #1752.
+- Fix overly broad HAVE_PTHREAD_CANCEL compile guard. Closes #1547.
+
+Client library:
+- Improved documentation around connect callback return codes. Close #1730.
+- Fix `mosquitto_publish*()` no longer returning `MOSQ_ERR_NO_CONN` when not
+  connected. Closes #1725.
+- `mosquitto_loop_start()` now sets a thread name on Linux, FreeBSD, NetBSD,
+  and OpenBSD. Closes #1777.
+- Fix `mosquitto_loop_stop()` not stopping on Windows. Closes #1748. Closes #117.
+
+
 1.6.10 - 2020-05-25
 ===================
 

client/pub_shared.c

@@ -71,7 +71,7 @@ int load_stdin(void)
 		pos += rlen;
 	}
 	if(pos > MQTT_MAX_PAYLOAD){
-		err_printf(&cfg, "Error: Message length must be less that %u bytes.\n\n", MQTT_MAX_PAYLOAD);
+		err_printf(&cfg, "Error: Message length must be less than %u bytes.\n\n", MQTT_MAX_PAYLOAD);
 		free(cfg.message);
 		return 1;
 	}
@@ -100,13 +100,9 @@ int load_file(const char *filename)
 	fseek(fptr, 0, SEEK_END);
 	flen = ftell(fptr);
 	if(flen > MQTT_MAX_PAYLOAD){
-		err_printf(&cfg, "Error: Message length must be less that %u bytes.\n\n", MQTT_MAX_PAYLOAD);
-		free(cfg.message);
-		return 1;
-	}
-	if(flen > 268435455){
 		fclose(fptr);
-		err_printf(&cfg, "Error: File \"%s\" is too large (>268,435,455 bytes).\n", filename);
+		err_printf(&cfg, "Error: File must be less than %u bytes.\n\n", MQTT_MAX_PAYLOAD);
+		free(cfg.message);
 		return 1;
 	}else if(flen == 0){
 		fclose(fptr);

config.mk

@@ -109,7 +109,7 @@ WITH_COVERAGE:=no
 
 # Also bump lib/mosquitto.h, CMakeLists.txt,
 # installer/mosquitto.nsi, installer/mosquitto64.nsi
-VERSION=1.6.10
+VERSION=1.6.11
 
 # Client library SO version. Bump if incompatible API/ABI changes are made.
 SOVERSION=1

docker/1.5/docker-entrypoint.sh

@@ -1,4 +1,15 @@
 #!/bin/ash
 set -e
 
+docker_set_permissions() {
+	local user; user="$(id -u)"
+
+	if [ "$user" = '0' ]; then
+		chown -R mosquitto:mosquitto /mosquitto
+	fi
+}
+
+
+docker_set_permissions()
+
 exec "$@"

docker/1.6/docker-entrypoint.sh

@@ -1,4 +1,15 @@
 #!/bin/ash
 set -e
 
+docker_set_permissions() {
+	local user; user="$(id -u)"
+
+	if [ "$user" = '0' ]; then
+		chown -R mosquitto:mosquitto /mosquitto
+	fi
+}
+
+
+docker_set_permissions()
+
 exec "$@"

docker/generic/Dockerfile

@@ -6,7 +6,7 @@ RUN test -n "${VERSION}"
 
 RUN apk --no-cache add \
       build-base \
-      libressl-dev \
+      openssl-dev \
       c-ares-dev \
       curl \
       util-linux-dev \
@@ -48,8 +48,7 @@ LABEL maintainer="Jonathan Hanson <jonathan@jonathan-hanson.org>" \
 RUN apk --no-cache add \
       busybox \
       ca-certificates \
-      libcrypto1.0 \
-      libssl1.0 \
+      openssl \
       libuuid \
       libwebsockets \
       musl

docker/generic/docker-entrypoint.sh

@@ -1,4 +1,15 @@
 #!/bin/ash
 set -e
 
+docker_set_permissions() {
+	local user; user="$(id -u)"
+
+	if [ "$user" = '0' ]; then
+		chown -R mosquitto:mosquitto /mosquitto
+	fi
+}
+
+
+docker_set_permissions()
+
 exec "$@"

docker/local/docker-entrypoint.sh

@@ -1,4 +1,15 @@
 #!/bin/ash
 set -e
 
+docker_set_permissions() {
+	local user; user="$(id -u)"
+
+	if [ "$user" = '0' ]; then
+		chown -R mosquitto:mosquitto /mosquitto
+	fi
+}
+
+
+docker_set_permissions()
+
 exec "$@"

examples/temperature_conversion/readme.txt

@@ -2,5 +2,5 @@ This is a simple example of the C++ library mosquittopp.
 
 It is a client that subscribes to the topic temperature/celsius which should
 have temperature data in text form being published to it. It reads this data as
-a Celsius temperature, converts to Farenheit and republishes on
-temperature/farenheit.
+a Celsius temperature, converts to Fahrenheit and republishes on
+temperature/fahrenheit.

examples/temperature_conversion/temperature_conversion.cpp

@@ -28,7 +28,7 @@ void mqtt_tempconv::on_connect(int rc)
 
 void mqtt_tempconv::on_message(const struct mosquitto_message *message)
 {
-	double temp_celsius, temp_farenheit;
+	double temp_celsius, temp_fahrenheit;
 	char buf[51];
 
 	if(!strcmp(message->topic, "temperature/celsius")){
@@ -36,9 +36,9 @@ void mqtt_tempconv::on_message(const struct mosquitto_message *message)
 		/* Copy N-1 bytes to ensure always 0 terminated. */
 		memcpy(buf, message->payload, 50*sizeof(char));
 		temp_celsius = atof(buf);
-		temp_farenheit = temp_celsius*9.0/5.0 + 32.0;
-		snprintf(buf, 50, "%f", temp_farenheit);
-		publish(NULL, "temperature/farenheit", strlen(buf), buf);
+		temp_fahrenheit = temp_celsius*9.0/5.0 + 32.0;
+		snprintf(buf, 50, "%f", temp_fahrenheit);
+		publish(NULL, "temperature/fahrenheit", strlen(buf), buf);
 	}
 }
 

installer/mosquitto.nsi

@@ -9,7 +9,7 @@
 !define env_hklm 'HKLM "SYSTEM\CurrentControlSet\Control\Session Manager\Environment"'
 
 Name "Eclipse Mosquitto"
-!define VERSION 1.6.10
+!define VERSION 1.6.11
 OutFile "mosquitto-${VERSION}-install-windows-x86.exe"
 
 InstallDir "$PROGRAMFILES\mosquitto"

installer/mosquitto64.nsi

@@ -9,7 +9,7 @@
 !define env_hklm 'HKLM "SYSTEM\CurrentControlSet\Control\Session Manager\Environment"'
 
 Name "Eclipse Mosquitto"
-!define VERSION 1.6.10
+!define VERSION 1.6.11
 OutFile "mosquitto-${VERSION}-install-windows-x64.exe"
 
 !include "x64.nsh"

lib/actions.c

@@ -151,9 +151,9 @@ int mosquitto_publish_v5(struct mosquitto *mosq, int *mid, const char *topic, in
 
 		pthread_mutex_lock(&mosq->msgs_out.mutex);
 		message->state = mosq_ms_invalid;
-		message__queue(mosq, message, mosq_md_out);
+		rc = message__queue(mosq, message, mosq_md_out);
 		pthread_mutex_unlock(&mosq->msgs_out.mutex);
-		return MOSQ_ERR_SUCCESS;
+		return rc;
 	}
 }
 

lib/dummypthread.h

@@ -4,6 +4,7 @@
 #define pthread_create(A, B, C, D)
 #define pthread_join(A, B)
 #define pthread_cancel(A)
+#define pthread_testcancel()
 
 #define pthread_mutex_init(A, B)
 #define pthread_mutex_destroy(A)

lib/loop.c

@@ -256,6 +256,9 @@ int mosquitto_loop_forever(struct mosquitto *mosq, int timeout, int max_packets)
 
 	while(run){
 		do{
+#ifdef HAVE_PTHREAD_CANCEL
+			pthread_testcancel();
+#endif
 			rc = mosquitto_loop(mosq, timeout, max_packets);
 		}while(run && rc == MOSQ_ERR_SUCCESS);
 		/* Quit after fatal errors. */
@@ -280,6 +283,9 @@ int mosquitto_loop_forever(struct mosquitto *mosq, int timeout, int max_packets)
 			return rc;
 		}
 		do{
+#ifdef HAVE_PTHREAD_CANCEL
+			pthread_testcancel();
+#endif
 			rc = MOSQ_ERR_SUCCESS;
 			state = mosquitto__get_state(mosq);
 			if(state == mosq_cs_disconnecting || state == mosq_cs_disconnected){

lib/mosquitto.c

@@ -151,10 +151,6 @@ int mosquitto_reinitialise(struct mosquitto *mosq, const char *id, bool clean_st
 	}
 	mosq->protocol = mosq_p_mqtt311;
 	mosq->sock = INVALID_SOCKET;
-	if(net__socketpair(&mosq->sockpairR, &mosq->sockpairW)){
-		log__printf(mosq, MOSQ_LOG_WARNING,
-				"Warning: Unable to open socket pair, outgoing publish commands may be delayed.");
-	}
 	mosq->keepalive = 60;
 	mosq->clean_start = clean_start;
 	if(id){
@@ -213,6 +209,12 @@ int mosquitto_reinitialise(struct mosquitto *mosq, const char *id, bool clean_st
 	pthread_mutex_init(&mosq->mid_mutex, NULL);
 	mosq->thread_id = pthread_self();
 #endif
+	/* This must be after pthread_mutex_init(), otherwise the log mutex may be
+	 * used before being initialised. */
+	if(net__socketpair(&mosq->sockpairR, &mosq->sockpairW)){
+		log__printf(mosq, MOSQ_LOG_WARNING,
+				"Warning: Unable to open socket pair, outgoing publish commands may be delayed.");
+	}
 
 	return MOSQ_ERR_SUCCESS;
 }

lib/mosquitto.h

@@ -48,7 +48,7 @@ extern "C" {
 
 #define LIBMOSQUITTO_MAJOR 1
 #define LIBMOSQUITTO_MINOR 6
-#define LIBMOSQUITTO_REVISION 10
+#define LIBMOSQUITTO_REVISION 11
 /* LIBMOSQUITTO_VERSION_NUMBER looks like 1002001 for e.g. version 1.2.1. */
 #define LIBMOSQUITTO_VERSION_NUMBER (LIBMOSQUITTO_MAJOR*1000000+LIBMOSQUITTO_MINOR*1000+LIBMOSQUITTO_REVISION)
 
@@ -434,7 +434,11 @@ libmosq_EXPORT int mosquitto_username_pw_set(struct mosquitto *mosq, const char
  *
  * Returns:
  * 	MOSQ_ERR_SUCCESS - on success.
- * 	MOSQ_ERR_INVAL -   if the input parameters were invalid.
+ * 	MOSQ_ERR_INVAL -   if the input parameters were invalid, which could be any of:
+ * 	                   * mosq == NULL
+ * 	                   * host == NULL
+ * 	                   * port < 0
+ * 	                   * keepalive < 5
  * 	MOSQ_ERR_ERRNO -   if a system call returned an error. The variable errno
  *                     contains the error code, even on Windows.
  *                     Use strerror_r() where available or FormatMessage() on
@@ -500,7 +504,11 @@ libmosq_EXPORT int mosquitto_connect_bind(struct mosquitto *mosq, const char *ho
  *
  * Returns:
  * 	MOSQ_ERR_SUCCESS - on success.
- * 	MOSQ_ERR_INVAL -   if the input parameters were invalid.
+ * 	MOSQ_ERR_INVAL -   if the input parameters were invalid, which could be any of:
+ * 	                   * mosq == NULL
+ * 	                   * host == NULL
+ * 	                   * port < 0
+ * 	                   * keepalive < 5
  * 	MOSQ_ERR_ERRNO -   if a system call returned an error. The variable errno
  *                     contains the error code, even on Windows.
  *                     Use strerror_r() where available or FormatMessage() on
@@ -570,7 +578,11 @@ libmosq_EXPORT int mosquitto_connect_async(struct mosquitto *mosq, const char *h
  *
  * Returns:
  * 	MOSQ_ERR_SUCCESS - on success.
- * 	MOSQ_ERR_INVAL -   if the input parameters were invalid.
+ * 	MOSQ_ERR_INVAL -   if the input parameters were invalid, which could be any of:
+ * 	                   * mosq == NULL
+ * 	                   * host == NULL
+ * 	                   * port < 0
+ * 	                   * keepalive < 5
  * 	MOSQ_ERR_ERRNO -   if a system call returned an error. The variable errno
  *                     contains the error code, even on Windows.
  *                     Use strerror_r() where available or FormatMessage() on
@@ -604,7 +616,11 @@ libmosq_EXPORT int mosquitto_connect_bind_async(struct mosquitto *mosq, const ch
  *
  * Returns:
  * 	MOSQ_ERR_SUCCESS - on success.
- * 	MOSQ_ERR_INVAL -   if the input parameters were invalid.
+ * 	MOSQ_ERR_INVAL -   if the input parameters were invalid, which could be any of:
+ * 	                   * mosq == NULL
+ * 	                   * host == NULL
+ * 	                   * port < 0
+ * 	                   * keepalive < 5
  * 	MOSQ_ERR_ERRNO -   if a system call returned an error. The variable errno
  *                     contains the error code, even on Windows.
  *                     Use strerror_r() where available or FormatMessage() on
@@ -1763,13 +1779,10 @@ libmosq_EXPORT int mosquitto_tls_psk_set(struct mosquitto *mosq, const char *psk
  * Callback Parameters:
  *  mosq - the mosquitto instance making the callback.
  *  obj - the user data provided in <mosquitto_new>
- *  rc -  the return code of the connection response, one of:
- *
- * * 0 - success
- * * 1 - connection refused (unacceptable protocol version)
- * * 2 - connection refused (identifier rejected)
- * * 3 - connection refused (broker unavailable)
- * * 4-255 - reserved for future use
+ *  rc -  the return code of the connection response. The values are defined by
+ *        the MQTT protocol version in use.
+ *        For MQTT v5.0, look at section 3.2.2.2 Connect Reason code: https://docs.oasis-open.org/mqtt/mqtt/v5.0/os/mqtt-v5.0-os.html
+ *        For MQTT v3.1.1, look at section 3.2.2.3 Connect Return code: http://docs.oasis-open.org/mqtt/mqtt/v3.1.1/mqtt-v3.1.1.html
  */
 libmosq_EXPORT void mosquitto_connect_callback_set(struct mosquitto *mosq, void (*on_connect)(struct mosquitto *, void *, int));
 
@@ -1787,14 +1800,11 @@ libmosq_EXPORT void mosquitto_connect_callback_set(struct mosquitto *mosq, void
  * Callback Parameters:
  *  mosq - the mosquitto instance making the callback.
  *  obj - the user data provided in <mosquitto_new>
- *  rc -  the return code of the connection response, one of:
+ *  rc -  the return code of the connection response. The values are defined by
+ *        the MQTT protocol version in use.
+ *        For MQTT v5.0, look at section 3.2.2.2 Connect Reason code: https://docs.oasis-open.org/mqtt/mqtt/v5.0/os/mqtt-v5.0-os.html
+ *        For MQTT v3.1.1, look at section 3.2.2.3 Connect Return code: http://docs.oasis-open.org/mqtt/mqtt/v3.1.1/mqtt-v3.1.1.html
  *  flags - the connect flags.
- *
- * * 0 - success
- * * 1 - connection refused (unacceptable protocol version)
- * * 2 - connection refused (identifier rejected)
- * * 3 - connection refused (broker unavailable)
- * * 4-255 - reserved for future use
  */
 libmosq_EXPORT void mosquitto_connect_with_flags_callback_set(struct mosquitto *mosq, void (*on_connect)(struct mosquitto *, void *, int, int));
 
@@ -1812,12 +1822,10 @@ libmosq_EXPORT void mosquitto_connect_with_flags_callback_set(struct mosquitto *
  * Callback Parameters:
  *  mosq - the mosquitto instance making the callback.
  *  obj - the user data provided in <mosquitto_new>
- *  rc -  the return code of the connection response, one of:
- * * 0 - success
- * * 1 - connection refused (unacceptable protocol version)
- * * 2 - connection refused (identifier rejected)
- * * 3 - connection refused (broker unavailable)
- * * 4-255 - reserved for future use
+ *  rc -  the return code of the connection response. The values are defined by
+ *        the MQTT protocol version in use.
+ *        For MQTT v5.0, look at section 3.2.2.2 Connect Reason code: https://docs.oasis-open.org/mqtt/mqtt/v5.0/os/mqtt-v5.0-os.html
+ *        For MQTT v3.1.1, look at section 3.2.2.3 Connect Return code: http://docs.oasis-open.org/mqtt/mqtt/v3.1.1/mqtt-v3.1.1.html
  *  flags - the connect flags.
  *  props - list of MQTT 5 properties, or NULL
  *

lib/socks_mosq.c

@@ -27,7 +27,7 @@ Contributors:
 #else
 #  include <arpa/inet.h>
 #endif
-#ifdef __FreeBSD__
+#if defined(__FreeBSD__) || defined(__OpenBSD__)
 #  include <sys/socket.h>
 #  include <netinet/in.h>
 #endif

lib/thread_mosq.c

@@ -20,6 +20,12 @@ Contributors:
 #include <time.h>
 #endif
 
+#if defined(__linux__) || defined(__NetBSD__)
+#  include <pthread.h>
+#elif defined(__FreeBSD__) || defined(__OpenBSD__)
+#  include <pthread_np.h>
+#endif
+
 #include "mosquitto_internal.h"
 #include "net_mosq.h"
 #include "util_mosq.h"
@@ -28,11 +34,18 @@ void *mosquitto__thread_main(void *obj);
 
 int mosquitto_loop_start(struct mosquitto *mosq)
 {
-#if defined(WITH_THREADING) && defined(HAVE_PTHREAD_CANCEL)
+#if defined(WITH_THREADING)
 	if(!mosq || mosq->threaded != mosq_ts_none) return MOSQ_ERR_INVAL;
 
 	mosq->threaded = mosq_ts_self;
 	if(!pthread_create(&mosq->thread_id, NULL, mosquitto__thread_main, mosq)){
+#if defined(__linux__)
+		pthread_setname_np(mosq->thread_id, "mosquitto loop");
+#elif defined(__NetBSD__)
+		pthread_setname_np(mosq->thread_id, "%s", "mosquitto loop");
+#elif defined(__FreeBSD__) || defined(__OpenBSD__)
+		pthread_set_name_np(mosq->thread_id, "mosquitto loop");
+#endif
 		return MOSQ_ERR_SUCCESS;
 	}else{
 		return MOSQ_ERR_ERRNO;
@@ -44,7 +57,7 @@ int mosquitto_loop_start(struct mosquitto *mosq)
 
 int mosquitto_loop_stop(struct mosquitto *mosq, bool force)
 {
-#if defined(WITH_THREADING) && defined(HAVE_PTHREAD_CANCEL)
+#if defined(WITH_THREADING)
 #  ifndef WITH_BROKER
 	char sockpair_data = 0;
 #  endif
@@ -63,9 +76,11 @@ int mosquitto_loop_stop(struct mosquitto *mosq, bool force)
 #endif
 	}
 	
+#ifdef HAVE_PTHREAD_CANCEL
 	if(force){
 		pthread_cancel(mosq->thread_id);
 	}
+#endif
 	pthread_join(mosq->thread_id, NULL);
 	mosq->thread_id = pthread_self();
 	mosq->threaded = mosq_ts_none;

lib/will_mosq.c

@@ -120,6 +120,7 @@ int will__clear(struct mosquitto *mosq)
 
 	mosquitto__free(mosq->will);
 	mosq->will = NULL;
+	mosq->will_delay_interval = 0;
 
 	return MOSQ_ERR_SUCCESS;
 }

service/systemd/mosquitto.service.notify

@@ -1,5 +1,5 @@
 [Unit]
-Description=Mosquitto MQTT v3.1/v3.1.1 Broker
+Description=Mosquitto MQTT Broker
 Documentation=man:mosquitto.conf(5) man:mosquitto(8)
 After=network.target
 Wants=network.target
@@ -10,6 +10,8 @@ NotifyAccess=main
 ExecStart=/usr/sbin/mosquitto -c /etc/mosquitto/mosquitto.conf
 ExecReload=/bin/kill -HUP $MAINPID
 Restart=on-failure
+ExecStartPre=/bin/mkdir -m 740 -p /var/log/mosquitto
+ExecStartPre=/bin/chown mosquitto: /var/log/mosquitto
 
 [Install]
 WantedBy=multi-user.target

service/systemd/mosquitto.service.simple

@@ -1,5 +1,5 @@
 [Unit]
-Description=Mosquitto MQTT v3.1/v3.1.1 Broker
+Description=Mosquitto MQTT Broker
 Documentation=man:mosquitto.conf(5) man:mosquitto(8)
 After=network.target
 Wants=network.target
@@ -8,6 +8,8 @@ Wants=network.target
 ExecStart=/usr/sbin/mosquitto -c /etc/mosquitto/mosquitto.conf
 ExecReload=/bin/kill -HUP $MAINPID
 Restart=on-failure
+ExecStartPre=/bin/mkdir -m 740 -p /var/log/mosquitto
+ExecStartPre=/bin/chown mosquitto: /var/log/mosquitto
 
 [Install]
 WantedBy=multi-user.target

set-version.sh

@@ -2,7 +2,7 @@
 
 MAJOR=1
 MINOR=6
-REVISION=10
+REVISION=11
 
 sed -i "s/^VERSION=.*/VERSION=${MAJOR}.${MINOR}.${REVISION}/" config.mk
 

snap/snapcraft.yaml

@@ -1,5 +1,5 @@
 name: mosquitto
-version: 1.6.10
+version: 1.6.11
 summary: Eclipse Mosquitto MQTT broker
 description: This is a message broker that supports version 5.0, 3.1.1, and 3.1 of the MQTT
     protocol.

src/CMakeLists.txt

@@ -150,6 +150,8 @@ endif (HAVE_GETADDRINFO_A AND WITH_ADNS)
 if (UNIX)
 	if (APPLE)
 		set (MOSQ_LIBS ${MOSQ_LIBS} dl m)
+    elseif (${CMAKE_SYSTEM_NAME} MATCHES "OpenBSD")
+	    set (MOSQ_LIBS ${MOSQ_LIBS} m)
     elseif(QNX)
         set(MOSQ_LIBS ${MOSQ_LIBS} m socket)
     else(APPLE)

src/conf.c

@@ -356,7 +356,7 @@ void config__cleanup(struct mosquitto__config *config)
 static void print_usage(void)
 {
 	printf("mosquitto version %s\n\n", VERSION);
-	printf("mosquitto is an MQTT v3.1.1 broker.\n\n");
+	printf("mosquitto is an MQTT v5.0/v3.1.1/v3.1 broker.\n\n");
 	printf("Usage: mosquitto [-c config_file] [-d] [-h] [-p port]\n\n");
 	printf(" -c : specify the broker config file.\n");
 	printf(" -d : put the broker into the background after starting.\n");
@@ -365,7 +365,7 @@ static void print_usage(void)
 	printf("      Not recommended in conjunction with the -c option.\n");
 	printf(" -v : verbose mode - enable all logging types. This overrides\n");
 	printf("      any logging options given in the config file.\n");
-	printf("\nSee http://mosquitto.org/ for more information.\n\n");
+	printf("\nSee https://mosquitto.org/ for more information.\n\n");
 }
 
 int config__parse_args(struct mosquitto_db *db, struct mosquitto__config *config, int argc, char *argv[])
@@ -653,7 +653,7 @@ int config__read(struct mosquitto_db *db, struct mosquitto__config *config, bool
 				}
 
 				/* Check plugins loaded to see if they have username/password checks enabled */
-				for(j=0; j<config->listeners[i].security_options.auth_plugin_config_count; j++){ 
+				for(j=0; j<config->listeners[i].security_options.auth_plugin_config_count; j++){
 					plugin = &config->listeners[i].security_options.auth_plugin_configs[j].plugin;
 
 					if(plugin->version == 3 || plugin->version == 2){
@@ -685,7 +685,7 @@ int config__read(struct mosquitto_db *db, struct mosquitto__config *config, bool
 			}
 
 			/* Check plugins loaded to see if they have username/password checks enabled */
-			for(j=0; j<config->security_options.auth_plugin_config_count; j++){ 
+			for(j=0; j<config->security_options.auth_plugin_config_count; j++){
 				plugin = &config->security_options.auth_plugin_configs[j].plugin;
 
 				if(plugin->version == 3 || plugin->version == 2){
@@ -759,6 +759,76 @@ int config__read(struct mosquitto_db *db, struct mosquitto__config *config, bool
 	return MOSQ_ERR_SUCCESS;
 }
 
+
+static int config__create_bridge_remap_topic(const char *prefix, const char *topic, char **remap_topic)
+{
+	int len;
+
+	if(prefix){
+		if(topic){
+			len = strlen(topic) + strlen(prefix)+1;
+			*remap_topic = mosquitto__malloc(len+1);
+			if(!(*remap_topic)){
+				log__printf(NULL, MOSQ_LOG_ERR, "Error: Out of memory.");
+				return MOSQ_ERR_NOMEM;
+			}
+			snprintf(*remap_topic, len+1, "%s%s", prefix, topic);
+			(*remap_topic)[len] = '\0';
+		}else{
+			*remap_topic = mosquitto__strdup(prefix);
+			if(!(*remap_topic)){
+				log__printf(NULL, MOSQ_LOG_ERR, "Error: Out of memory.");
+				return MOSQ_ERR_NOMEM;
+			}
+		}
+	}else{
+		*remap_topic = mosquitto__strdup(topic);
+		if(!(*remap_topic)){
+			log__printf(NULL, MOSQ_LOG_ERR, "Error: Out of memory.");
+			return MOSQ_ERR_NOMEM;
+		}
+	}
+	return MOSQ_ERR_SUCCESS;
+}
+
+
+static int config__create_bridge_prefix(char **prefix, const char *topic, const char *token, const char *direction)
+{
+	int len;
+
+	if(topic){
+		len = strlen(topic) + strlen(token) + 1;
+	}else{
+		len = strlen(token) + 1;
+	}
+	*prefix = malloc(len);
+	if(*prefix == NULL){
+		log__printf(NULL, MOSQ_LOG_ERR, "Error: Out of memory.");
+		return MOSQ_ERR_NOMEM;
+	}
+
+	if(topic){
+		/* Print prefix+pattern to check for validity */
+		snprintf(*prefix, len, "%s%s", token, topic);
+	}else{
+		snprintf(*prefix, len, "%s", token);
+	}
+
+	if(mosquitto_sub_topic_check(*prefix) != MOSQ_ERR_SUCCESS){
+		log__printf(NULL, MOSQ_LOG_ERR,
+				"Error: Invalid bridge topic %s prefix and pattern combination '%s'.",
+				direction, *prefix);
+
+		return MOSQ_ERR_INVAL;
+	}
+
+	/* Print just the prefix for storage */
+	snprintf(*prefix, len, "%s", token);
+
+	return MOSQ_ERR_SUCCESS;
+}
+
+
 int config__read_file_core(struct mosquitto__config *config, bool reload, struct config_recurse *cr, int level, int *lineno, FILE *fptr, char **buf, int *buflen)
 {
 	int rc;
@@ -769,7 +839,6 @@ int config__read_file_core(struct mosquitto__config *config, bool reload, struct
 	char *tmp_char;
 	struct mosquitto__bridge *cur_bridge = NULL;
 	struct mosquitto__bridge_topic *cur_topic;
-	int len;
 #endif
 	struct mosquitto__auth_plugin_config *cur_auth_plugin_config = NULL;
 
@@ -2021,22 +2090,12 @@ int config__read_file_core(struct mosquitto__config *config, bool reload, struct
 										log__printf(NULL, MOSQ_LOG_ERR, "Error: Invalid bridge topic local prefix '%s'.", token);
 										return MOSQ_ERR_INVAL;
 									}
-									cur_topic->local_prefix = malloc(strlen(cur_topic->topic) + strlen(token) + 1);
-									if(cur_topic->local_prefix == NULL){
-										log__printf(NULL, MOSQ_LOG_ERR, "Error: Out of memory.");
-										return MOSQ_ERR_NOMEM;
-									}
-									/* Print prefix+pattern to check for validity */
-									snprintf(cur_topic->local_prefix, strlen(cur_topic->topic) + strlen(token)+1,
-											"%s%s", token, cur_topic->topic);
-									if(mosquitto_sub_topic_check(cur_topic->local_prefix) != MOSQ_ERR_SUCCESS){
-										log__printf(NULL, MOSQ_LOG_ERR, "Error: Invalid bridge topic local prefix and pattern combination '%s'.", cur_topic->local_prefix);
+
+									if(config__create_bridge_prefix(&cur_topic->local_prefix,
+												cur_topic->topic, token, "local")){
+
 										return MOSQ_ERR_INVAL;
 									}
-									
-									/* Print just the prefix for storage */
-									snprintf(cur_topic->local_prefix, strlen(cur_topic->topic) + strlen(token)+1,
-											"%s", token);
 								}
 
 								token = strtok_r(NULL, " ", &saveptr);
@@ -2048,24 +2107,11 @@ int config__read_file_core(struct mosquitto__config *config, bool reload, struct
 											log__printf(NULL, MOSQ_LOG_ERR, "Error: Invalid bridge topic remote prefix '%s'.", token);
 											return MOSQ_ERR_INVAL;
 										}
-										cur_topic->remote_prefix = malloc(strlen(cur_topic->topic) + strlen(token) + 1);
-										if(cur_topic == NULL){
-											log__printf(NULL, MOSQ_LOG_ERR, "Error: Out of memory.");
-											return MOSQ_ERR_NOMEM;
-										}
-										/* Print prefix+pattern to check for validity */
-										snprintf(cur_topic->remote_prefix, strlen(cur_topic->topic) + strlen(token)+1,
-												"%s%s", token, cur_topic->topic);
-										if(mosquitto_sub_topic_check(cur_topic->remote_prefix) != MOSQ_ERR_SUCCESS){
-											log__printf(NULL, MOSQ_LOG_ERR,
-													"Error: Invalid bridge topic remote prefix and pattern combination '%s'.",
-													cur_topic->remote_prefix);
+										if(config__create_bridge_prefix(&cur_topic->remote_prefix,
+													cur_topic->topic, token, "remote")){
+
 											return MOSQ_ERR_INVAL;
 										}
-									
-										/* Print just the prefix for storage */
-										snprintf(cur_topic->remote_prefix, strlen(cur_topic->topic) + strlen(token)+1,
-												"%s", token);
 									}
 								}
 							}
@@ -2077,54 +2123,16 @@ int config__read_file_core(struct mosquitto__config *config, bool reload, struct
 						log__printf(NULL, MOSQ_LOG_ERR, "Error: Invalid bridge remapping.");
 						return MOSQ_ERR_INVAL;
 					}
-					if(cur_topic->local_prefix){
-						if(cur_topic->topic){
-							len = strlen(cur_topic->topic) + strlen(cur_topic->local_prefix)+1;
-							cur_topic->local_topic = mosquitto__malloc(len+1);
-							if(!cur_topic->local_topic){
-								log__printf(NULL, MOSQ_LOG_ERR, "Error: Out of memory.");
-								return MOSQ_ERR_NOMEM;
-							}
-							snprintf(cur_topic->local_topic, len+1, "%s%s", cur_topic->local_prefix, cur_topic->topic);
-							cur_topic->local_topic[len] = '\0';
-						}else{
-							cur_topic->local_topic = mosquitto__strdup(cur_topic->local_prefix);
-							if(!cur_topic->local_topic){
-								log__printf(NULL, MOSQ_LOG_ERR, "Error: Out of memory.");
-								return MOSQ_ERR_NOMEM;
-							}
-						}
-					}else{
-						cur_topic->local_topic = mosquitto__strdup(cur_topic->topic);
-						if(!cur_topic->local_topic){
-							log__printf(NULL, MOSQ_LOG_ERR, "Error: Out of memory.");
-							return MOSQ_ERR_NOMEM;
-						}
+					if(config__create_bridge_remap_topic(cur_topic->local_prefix,
+								cur_topic->topic, &cur_topic->local_topic)){
+
+						return MOSQ_ERR_INVAL;
 					}
 
-					if(cur_topic->remote_prefix){
-						if(cur_topic->topic){
-							len = strlen(cur_topic->topic) + strlen(cur_topic->remote_prefix)+1;
-							cur_topic->remote_topic = mosquitto__malloc(len+1);
-							if(!cur_topic->remote_topic){
-								log__printf(NULL, MOSQ_LOG_ERR, "Error: Out of memory.");
-								return MOSQ_ERR_NOMEM;
-							}
-							snprintf(cur_topic->remote_topic, len, "%s%s", cur_topic->remote_prefix, cur_topic->topic);
-							cur_topic->remote_topic[len] = '\0';
-						}else{
-							cur_topic->remote_topic = mosquitto__strdup(cur_topic->remote_prefix);
-							if(!cur_topic->remote_topic){
-								log__printf(NULL, MOSQ_LOG_ERR, "Error: Out of memory.");
-								return MOSQ_ERR_NOMEM;
-							}
-						}
-					}else{
-						cur_topic->remote_topic = mosquitto__strdup(cur_topic->topic);
-						if(!cur_topic->remote_topic){
-							log__printf(NULL, MOSQ_LOG_ERR, "Error: Out of memory.");
-							return MOSQ_ERR_NOMEM;
-						}
+					if(config__create_bridge_remap_topic(cur_topic->remote_prefix,
+								cur_topic->topic, &cur_topic->remote_topic)){
+
+						return MOSQ_ERR_INVAL;
 					}
 #else
 					log__printf(NULL, MOSQ_LOG_WARNING, "Warning: Bridge support not available.");

src/database.c

@@ -339,6 +339,9 @@ int db__message_delete_outgoing(struct mosquitto_db *db, struct mosquitto *conte
 		}
 		db__message_dequeue_first(context, &context->msgs_out);
 	}
+#ifdef WITH_PERSISTENCE
+	db->persistence_changes++;
+#endif
 
 	return MOSQ_ERR_SUCCESS;
 }
@@ -394,6 +397,10 @@ int db__message_insert(struct mosquitto_db *db, struct mosquitto *context, uint1
 				}
 			}
 		}
+		if(context->bridge && context->bridge->clean_start == true){
+			mosquitto_property_free_all(&properties);
+			return 2;
+		}
 	}
 
 	if(context->sock != INVALID_SOCKET){

src/db_dump/db_dump.c

@@ -110,8 +110,8 @@ static int dump__cfg_chunk_process(struct mosquitto_db *db, FILE *db_fd, uint32_
 
 	memset(&chunk, 0, sizeof(struct PF_cfg));
 
-	if(db_version == 5){
-		rc = persist__chunk_cfg_read_v5(db_fd, &chunk);
+	if(db_version == 6 || db_version == 5){
+		rc = persist__chunk_cfg_read_v56(db_fd, &chunk);
 	}else{
 		rc = persist__chunk_cfg_read_v234(db_fd, &chunk);
 	}
@@ -147,8 +147,8 @@ static int dump__client_chunk_process(struct mosquitto_db *db, FILE *db_fd, uint
 
 	memset(&chunk, 0, sizeof(struct P_client));
 
-	if(db_version == 5){
-		rc = persist__chunk_client_read_v5(db_fd, &chunk);
+	if(db_version == 6 || db_version == 5){
+		rc = persist__chunk_client_read_v56(db_fd, &chunk, db_version);
 	}else{
 		rc = persist__chunk_client_read_v234(db_fd, &chunk, db_version);
 	}
@@ -189,8 +189,8 @@ static int dump__client_msg_chunk_process(struct mosquitto_db *db, FILE *db_fd,
 	client_msg_count++;
 
 	memset(&chunk, 0, sizeof(struct P_client_msg));
-	if(db_version == 5){
-		rc = persist__chunk_client_msg_read_v5(db_fd, &chunk, length);
+	if(db_version == 6 || db_version == 5){
+		rc = persist__chunk_client_msg_read_v56(db_fd, &chunk, length);
 	}else{
 		rc = persist__chunk_client_msg_read_v234(db_fd, &chunk);
 	}
@@ -234,8 +234,8 @@ static int dump__msg_store_chunk_process(struct mosquitto_db *db, FILE *db_fptr,
 	msg_store_count++;
 
 	memset(&chunk, 0, sizeof(struct P_msg_store));
-	if(db_version == 5){
-		rc = persist__chunk_msg_store_read_v5(db_fptr, &chunk, length);
+	if(db_version == 6 || db_version == 5){
+		rc = persist__chunk_msg_store_read_v56(db_fptr, &chunk, length);
 	}else{
 		rc = persist__chunk_msg_store_read_v234(db_fptr, &chunk, db_version);
 	}
@@ -321,8 +321,8 @@ static int dump__retain_chunk_process(struct mosquitto_db *db, FILE *db_fd, uint
 	if(do_print) printf("DB_CHUNK_RETAIN:\n");
 	if(do_print) printf("\tLength: %d\n", length);
 
-	if(db_version == 5){
-		rc = persist__chunk_retain_read_v5(db_fd, &chunk);
+	if(db_version == 6 || db_version == 5){
+		rc = persist__chunk_retain_read_v56(db_fd, &chunk);
 	}else{
 		rc = persist__chunk_retain_read_v234(db_fd, &chunk);
 	}
@@ -345,8 +345,8 @@ static int dump__sub_chunk_process(struct mosquitto_db *db, FILE *db_fd, uint32_
 	sub_count++;
 
 	memset(&chunk, 0, sizeof(struct P_sub));
-	if(db_version == 5){
-		rc = persist__chunk_sub_read_v5(db_fd, &chunk);
+	if(db_version == 6 || db_version == 5){
+		rc = persist__chunk_sub_read_v56(db_fd, &chunk);
 	}else{
 		rc = persist__chunk_sub_read_v234(db_fd, &chunk);
 	}

src/db_dump/print.c

@@ -145,6 +145,12 @@ void print__client(struct P_client *chunk, int length)
 	printf("DB_CHUNK_CLIENT:\n");
 	printf("\tLength: %d\n", length);
 	printf("\tClient ID: %s\n", chunk->client_id);
+	if(chunk->username){
+		printf("\tUsername: %s\n", chunk->username);
+	}
+	if(chunk->F.listener_port > 0){
+		printf("\tListener port: %u\n", chunk->F.listener_port);
+	}
 	printf("\tLast MID: %d\n", chunk->F.last_mid);
 	printf("\tSession expiry time: %" PRIu64 "\n", chunk->F.session_expiry_time);
 	printf("\tSession expiry interval: %u\n", chunk->F.session_expiry_interval);

src/handle_publish.c

@@ -109,7 +109,10 @@ int handle__publish(struct mosquitto_db *db, struct mosquitto *context)
 	/* Handle properties */
 	if(context->protocol == mosq_p_mqtt5){
 		rc = property__read_all(CMD_PUBLISH, &context->in_packet, &properties);
-		if(rc) return rc;
+		if(rc){
+			mosquitto__free(topic);
+			return rc;
+		}
 
 		p = properties;
 		p_prev = NULL;

src/handle_subscribe.c

@@ -119,6 +119,8 @@ int handle__subscribe(struct mosquitto_db *db, struct mosquitto *context)
 
 				retain_handling = (subscription_options & 0x30);
 				if(retain_handling == 0x30 || (subscription_options & 0xC0) != 0){
+					mosquitto__free(sub);
+					mosquitto__free(payload);
 					return MOSQ_ERR_PROTOCOL;
 				}
 			}

src/handle_unsubscribe.c

@@ -49,7 +49,7 @@ int handle__unsubscribe(struct mosquitto_db *db, struct mosquitto *context)
 			return MOSQ_ERR_PROTOCOL;
 		}
 	}
-	if(packet__read_uint16(&context->in_packet, &mid)) return 1;
+	if(packet__read_uint16(&context->in_packet, &mid)) return MOSQ_ERR_PROTOCOL;
 	if(mid == 0) return MOSQ_ERR_PROTOCOL;
 
 	if(context->protocol == mosq_p_mqtt5){
@@ -76,7 +76,7 @@ int handle__unsubscribe(struct mosquitto_db *db, struct mosquitto *context)
 		sub = NULL;
 		if(packet__read_string(&context->in_packet, &sub, &slen)){
 			mosquitto__free(reason_codes);
-			return 1;
+			return MOSQ_ERR_PROTOCOL;
 		}
 
 		if(!slen){
@@ -85,7 +85,7 @@ int handle__unsubscribe(struct mosquitto_db *db, struct mosquitto *context)
 					context->id);
 			mosquitto__free(sub);
 			mosquitto__free(reason_codes);
-			return 1;
+			return MOSQ_ERR_PROTOCOL;
 		}
 		if(mosquitto_sub_topic_check(sub)){
 			log__printf(NULL, MOSQ_LOG_INFO,
@@ -93,7 +93,7 @@ int handle__unsubscribe(struct mosquitto_db *db, struct mosquitto *context)
 					context->id);
 			mosquitto__free(sub);
 			mosquitto__free(reason_codes);
-			return 1;
+			return MOSQ_ERR_PROTOCOL;
 		}
 
 		log__printf(NULL, MOSQ_LOG_DEBUG, "\t%s", sub);

src/logging.c

@@ -24,6 +24,7 @@ Contributors:
 #include <time.h>
 
 #ifdef WITH_DLT
+#include <sys/stat.h>
 #include <dlt/dlt.h>
 #endif
 
@@ -55,6 +56,29 @@ static int log_priorities = MOSQ_LOG_ERR | MOSQ_LOG_WARNING | MOSQ_LOG_NOTICE |
 
 #ifdef WITH_DLT
 static DltContext dltContext;
+static bool dlt_allowed = false;
+
+void dlt_fifo_check(void)
+{
+	struct stat statbuf;
+	int fd;
+
+	/* If we start DLT but the /tmp/dlt fifo doesn't exist, or isn't available
+	 * for writing then there is a big delay when we try and close the log
+	 * later, so check for it first. This has the side effect of not letting
+	 * people using DLT create the fifo after Mosquitto has started, but at the
+	 * benefit of not having a massive delay for everybody else. */
+	memset(&statbuf, 0, sizeof(statbuf));
+	if(stat("/tmp/dlt", &statbuf) == 0){
+		if(S_ISFIFO(statbuf.st_mode)){
+			fd = open("/tmp/dlt", O_NONBLOCK | O_WRONLY);
+			if(fd != -1){
+				dlt_allowed = true;
+				close(fd);
+			}
+		}
+	}
+}
 #endif
 
 static int get_time(struct tm **ti)
@@ -110,17 +134,21 @@ int log__init(struct mosquitto__config *config)
 			return 1;
 		}
 		config->log_fptr = mosquitto__fopen(config->log_file, "at", true);
-		if(!config->log_fptr){
+		if(config->log_fptr){
+			setvbuf(config->log_fptr, NULL, _IOLBF, 0);
+		}else{
 			log_destinations = MQTT3_LOG_STDERR;
 			log_priorities = MOSQ_LOG_ERR;
 			log__printf(NULL, MOSQ_LOG_ERR, "Error: Unable to open log file %s for writing.", config->log_file);
-			return MOSQ_ERR_INVAL;
 		}
 		restore_privileges();
 	}
 #ifdef WITH_DLT
-	DLT_REGISTER_APP("MQTT","mosquitto log");
-	dlt_register_context(&dltContext, "MQTT", "mosquitto DLT context");
+	dlt_fifo_check();
+	if(dlt_allowed){
+		DLT_REGISTER_APP("MQTT","mosquitto log");
+		dlt_register_context(&dltContext, "MQTT", "mosquitto DLT context");
+	}
 #endif
 	return rc;
 }
@@ -142,8 +170,10 @@ int log__close(struct mosquitto__config *config)
 	}
 
 #ifdef WITH_DLT
-	dlt_unregister_context(&dltContext);
-	DLT_UNREGISTER_APP();
+	if(dlt_allowed){
+		dlt_unregister_context(&dltContext);
+		DLT_UNREGISTER_APP();
+	}
 #endif
 	/* FIXME - do something for all destinations! */
 	return MOSQ_ERR_SUCCESS;
@@ -182,7 +212,6 @@ int log__vprintf(int priority, const char *fmt, va_list va)
 	const char *topic;
 	int syslog_priority;
 	time_t now = time(NULL);
-	static time_t last_flush = 0;
 	char time_buf[50];
 	bool log_timestamp = true;
 	char *log_timestamp_format = NULL;
@@ -294,7 +323,6 @@ int log__vprintf(int priority, const char *fmt, va_list va)
 			}else{
 				fprintf(stdout, "%s\n", s);
 			}
-			fflush(stdout);
 		}
 		if(log_destinations & MQTT3_LOG_STDERR){
 			if(log_timestamp){
@@ -306,7 +334,6 @@ int log__vprintf(int priority, const char *fmt, va_list va)
 			}else{
 				fprintf(stderr, "%s\n", s);
 			}
-			fflush(stderr);
 		}
 		if(log_destinations & MQTT3_LOG_FILE && log_fptr){
 			if(log_timestamp){
@@ -318,10 +345,6 @@ int log__vprintf(int priority, const char *fmt, va_list va)
 			}else{
 				fprintf(log_fptr, "%s\n", s);
 			}
-			if(now - last_flush > 1){
-				fflush(log_fptr);
-				last_flush = now;
-			}
 		}
 		if(log_destinations & MQTT3_LOG_SYSLOG){
 #ifndef WIN32

src/mosquitto.c

@@ -289,6 +289,18 @@ int main(int argc, char *argv[])
 	rc = mosquitto_security_init(&int_db, false);
 	if(rc) return rc;
 
+	/* After loading persisted clients and ACLs, try to associate them,
+	 * so persisted subscriptions can start storing messages */
+	HASH_ITER(hh_id, int_db.contexts_by_id, ctxt, ctxt_tmp){
+		if(ctxt && !ctxt->clean_start && ctxt->username){
+			rc = acl__find_acls(&int_db, ctxt);
+			if(rc){
+				log__printf(NULL, MOSQ_LOG_WARNING, "Failed to associate persisted user %s with ACLs, "
+					"likely due to changed ports while using a per_listener_settings configuration.", ctxt->username);
+			}
+		}
+	}
+
 #ifdef WITH_SYS_TREE
 	sys_tree__init(&int_db);
 #endif
@@ -333,10 +345,6 @@ int main(int argc, char *argv[])
 #endif
 		}
 	}
-	if(listensock == NULL){
-		log__printf(NULL, MOSQ_LOG_ERR, "Error: Unable to start any listening sockets, exiting.");
-		return 1;
-	}
 
 	rc = drop_privileges(&config, false);
 	if(rc != MOSQ_ERR_SUCCESS) return rc;
@@ -364,6 +372,7 @@ int main(int argc, char *argv[])
 	}
 #endif
 
+	log__printf(NULL, MOSQ_LOG_INFO, "mosquitto version %s running", VERSION);
 #ifdef WITH_SYSTEMD
 	sd_notify(0, "READY=1");
 #endif

src/net.c

@@ -86,7 +86,7 @@ static void net__print_error(int log, const char *format_str)
 
 #ifdef WIN32
 	FormatMessage(FORMAT_MESSAGE_ALLOCATE_BUFFER | FORMAT_MESSAGE_FROM_SYSTEM,
-			NULL, WSAGetLastError(), LANG_NEUTRAL, &buf, 0, NULL);
+			NULL, WSAGetLastError(), LANG_NEUTRAL, (LPTSTR)&buf, 0, NULL);
 
 	log__printf(NULL, log, format_str, buf);
 	LocalFree(buf);
@@ -165,7 +165,11 @@ int net__socket_accept(struct mosquitto_db *db, mosq_sock_t listensock)
 
 	if(db->config->set_tcp_nodelay){
 		int flag = 1;
+#ifdef WIN32
+			if (setsockopt(new_sock, IPPROTO_TCP, TCP_NODELAY, (char *)&flag, sizeof(int)) != 0) {
+#else
 		if(setsockopt(new_sock, IPPROTO_TCP, TCP_NODELAY, &flag, sizeof(int)) != 0){
+#endif
 			log__printf(NULL, MOSQ_LOG_WARNING, "Warning: Unable to set TCP_NODELAY.");
 		}
 	}
@@ -588,11 +592,7 @@ int net__socket_listen(struct mosquitto__listener *listener)
 	struct addrinfo *ainfo, *rp;
 	char service[10];
 	int rc;
-#ifndef WIN32
 	int ss_opt = 1;
-#else
-	char ss_opt = 1;
-#endif
 #ifdef SO_BINDTODEVICE
 	struct ifreq ifr;
 #endif

src/persist.h

@@ -17,7 +17,7 @@ Contributors:
 #ifndef PERSIST_H
 #define PERSIST_H
 
-#define MOSQ_DB_VERSION 5
+#define MOSQ_DB_VERSION 6
 
 /* DB read/write */
 extern const unsigned char magic[15];
@@ -57,15 +57,28 @@ struct PF_cfg{
 	uint8_t dbid_size;
 };
 
-struct PF_client{
+struct PF_client_v5{
 	int64_t session_expiry_time;
 	uint32_t session_expiry_interval;
 	uint16_t last_mid;
 	uint16_t id_len;
 };
+struct PF_client{
+	/* struct PF_client_v5; */
+	int64_t session_expiry_time;
+	uint32_t session_expiry_interval;
+	uint16_t last_mid;
+	uint16_t id_len;
+
+	uint16_t listener_port;
+	uint16_t username_len;
+	/* tail: 4 byte padding, because 64bit member 
+	 * forces multiple of 8 for struct size */
+};
 struct P_client{
 	struct PF_client F;
 	char *client_id;
+	char *username;
 };
 
 
@@ -141,19 +154,19 @@ int persist__chunk_msg_store_read_v234(FILE *db_fptr, struct P_msg_store *chunk,
 int persist__chunk_retain_read_v234(FILE *db_fptr, struct P_retain *chunk);
 int persist__chunk_sub_read_v234(FILE *db_fptr, struct P_sub *chunk);
 
-int persist__chunk_header_read_v5(FILE *db_fptr, int *chunk, int *length);
-int persist__chunk_cfg_read_v5(FILE *db_fptr, struct PF_cfg *chunk);
-int persist__chunk_client_read_v5(FILE *db_fptr, struct P_client *chunk);
-int persist__chunk_client_msg_read_v5(FILE *db_fptr, struct P_client_msg *chunk, uint32_t length);
-int persist__chunk_msg_store_read_v5(FILE *db_fptr, struct P_msg_store *chunk, uint32_t length);
-int persist__chunk_retain_read_v5(FILE *db_fptr, struct P_retain *chunk);
-int persist__chunk_sub_read_v5(FILE *db_fptr, struct P_sub *chunk);
+int persist__chunk_header_read_v56(FILE *db_fptr, int *chunk, int *length);
+int persist__chunk_cfg_read_v56(FILE *db_fptr, struct PF_cfg *chunk);
+int persist__chunk_client_read_v56(FILE *db_fptr, struct P_client *chunk, int db_version);
+int persist__chunk_client_msg_read_v56(FILE *db_fptr, struct P_client_msg *chunk, uint32_t length);
+int persist__chunk_msg_store_read_v56(FILE *db_fptr, struct P_msg_store *chunk, uint32_t length);
+int persist__chunk_retain_read_v56(FILE *db_fptr, struct P_retain *chunk);
+int persist__chunk_sub_read_v56(FILE *db_fptr, struct P_sub *chunk);
 
-int persist__chunk_cfg_write_v5(FILE *db_fptr, struct PF_cfg *chunk);
-int persist__chunk_client_write_v5(FILE *db_fptr, struct P_client *chunk);
-int persist__chunk_client_msg_write_v5(FILE *db_fptr, struct P_client_msg *chunk);
-int persist__chunk_message_store_write_v5(FILE *db_fptr, struct P_msg_store *chunk);
-int persist__chunk_retain_write_v5(FILE *db_fptr, struct P_retain *chunk);
-int persist__chunk_sub_write_v5(FILE *db_fptr, struct P_sub *chunk);
+int persist__chunk_cfg_write_v6(FILE *db_fptr, struct PF_cfg *chunk);
+int persist__chunk_client_write_v6(FILE *db_fptr, struct P_client *chunk);
+int persist__chunk_client_msg_write_v6(FILE *db_fptr, struct P_client_msg *chunk);
+int persist__chunk_message_store_write_v6(FILE *db_fptr, struct P_msg_store *chunk);
+int persist__chunk_retain_write_v6(FILE *db_fptr, struct P_retain *chunk);
+int persist__chunk_sub_write_v6(FILE *db_fptr, struct P_sub *chunk);
 
 #endif

src/persist_read.c

@@ -175,14 +175,14 @@ static int persist__client_msg_restore(struct mosquitto_db *db, struct P_client_
 
 static int persist__client_chunk_restore(struct mosquitto_db *db, FILE *db_fptr)
 {
-	int rc = 0;
+	int i, rc = 0;
 	struct mosquitto *context;
 	struct P_client chunk;
 
 	memset(&chunk, 0, sizeof(struct P_client));
 
-	if(db_version == 5){
-		rc = persist__chunk_client_read_v5(db_fptr, &chunk);
+	if(db_version == 6 || db_version == 5){
+		rc = persist__chunk_client_read_v56(db_fptr, &chunk, db_version);
 	}else{
 		rc = persist__chunk_client_read_v234(db_fptr, &chunk, db_version);
 	}
@@ -195,13 +195,29 @@ static int persist__client_chunk_restore(struct mosquitto_db *db, FILE *db_fptr)
 	if(context){
 		context->session_expiry_time = chunk.F.session_expiry_time;
 		context->session_expiry_interval = chunk.F.session_expiry_interval;
+		if(chunk.username && !context->username){
+			/* username is not freed here, it is now owned by context */
+			context->username = chunk.username;
+			chunk.username = NULL;
+			/* in per_listener_settings mode, try to find the listener by persisted port */
+			if(db->config->per_listener_settings && !context->listener && chunk.F.listener_port > 0){
+				for(i=0; i < db->config->listener_count; i++){
+					if(db->config->listeners[i].port == chunk.F.listener_port){
+						context->listener = &db->config->listeners[i];
+						break;
+					}
+				}
+			}
+		}
 		/* FIXME - we should expire clients here if they have exceeded their time */
 	}else{
 		rc = 1;
 	}
 
 	mosquitto__free(chunk.client_id);
-
+	if(chunk.username){
+		mosquitto__free(chunk.username);
+	}
 	return rc;
 }
 
@@ -213,8 +229,8 @@ static int persist__client_msg_chunk_restore(struct mosquitto_db *db, FILE *db_f
 
 	memset(&chunk, 0, sizeof(struct P_client_msg));
 
-	if(db_version == 5){
-		rc = persist__chunk_client_msg_read_v5(db_fptr, &chunk, length);
+	if(db_version == 6 || db_version == 5){
+		rc = persist__chunk_client_msg_read_v56(db_fptr, &chunk, length);
 	}else{
 		rc = persist__chunk_client_msg_read_v234(db_fptr, &chunk);
 	}
@@ -242,8 +258,8 @@ static int persist__msg_store_chunk_restore(struct mosquitto_db *db, FILE *db_fp
 
 	memset(&chunk, 0, sizeof(struct P_msg_store));
 
-	if(db_version == 5){
-		rc = persist__chunk_msg_store_read_v5(db_fptr, &chunk, length);
+	if(db_version == 6 || db_version == 5){
+		rc = persist__chunk_msg_store_read_v56(db_fptr, &chunk, length);
 	}else{
 		rc = persist__chunk_msg_store_read_v234(db_fptr, &chunk, db_version);
 	}
@@ -320,8 +336,8 @@ static int persist__retain_chunk_restore(struct mosquitto_db *db, FILE *db_fptr)
 
 	memset(&chunk, 0, sizeof(struct P_retain));
 
-	if(db_version == 5){
-		rc = persist__chunk_retain_read_v5(db_fptr, &chunk);
+	if(db_version == 6 || db_version == 5){
+		rc = persist__chunk_retain_read_v56(db_fptr, &chunk);
 	}else{
 		rc = persist__chunk_retain_read_v234(db_fptr, &chunk);
 	}
@@ -346,8 +362,8 @@ static int persist__sub_chunk_restore(struct mosquitto_db *db, FILE *db_fptr)
 
 	memset(&chunk, 0, sizeof(struct P_sub));
 
-	if(db_version == 5){
-		rc = persist__chunk_sub_read_v5(db_fptr, &chunk);
+	if(db_version == 6 || db_version == 5){
+		rc = persist__chunk_sub_read_v56(db_fptr, &chunk);
 	}else{
 		rc = persist__chunk_sub_read_v234(db_fptr, &chunk);
 	}
@@ -367,8 +383,8 @@ static int persist__sub_chunk_restore(struct mosquitto_db *db, FILE *db_fptr)
 
 int persist__chunk_header_read(FILE *db_fptr, int *chunk, int *length)
 {
-	if(db_version == 5){
-		return persist__chunk_header_read_v5(db_fptr, chunk, length);
+	if(db_version == 6 || db_version == 5){
+		return persist__chunk_header_read_v56(db_fptr, chunk, length);
 	}else{
 		return persist__chunk_header_read_v234(db_fptr, chunk, length);
 	}
@@ -416,7 +432,9 @@ int persist__restore(struct mosquitto_db *db)
 		 * Is your DB change still compatible with previous versions?
 		 */
 		if(db_version != MOSQ_DB_VERSION){
-			if(db_version == 4){
+			if(db_version == 5){
+				/* Addition of username and listener_port to client chunk in v6 */
+			}else if(db_version == 4){
 			}else if(db_version == 3){
 				/* Addition of source_username and source_port to msg_store chunk in v4, v1.5.6 */
 			}else if(db_version == 2){
@@ -431,8 +449,8 @@ int persist__restore(struct mosquitto_db *db)
 		while(persist__chunk_header_read(fptr, &chunk, &length) == MOSQ_ERR_SUCCESS){
 			switch(chunk){
 				case DB_CHUNK_CFG:
-					if(db_version == 5){
-						if(persist__chunk_cfg_read_v5(fptr, &cfg_chunk)){
+					if(db_version == 6 || db_version == 5){
+						if(persist__chunk_cfg_read_v56(fptr, &cfg_chunk)){
 							fclose(fptr);
 							return 1;
 						}

src/persist_read_v5.c

@@ -38,7 +38,7 @@ Contributors:
 #include "util_mosq.h"
 
 
-int persist__chunk_header_read_v5(FILE *db_fptr, int *chunk, int *length)
+int persist__chunk_header_read_v56(FILE *db_fptr, int *chunk, int *length)
 {
 	size_t rlen;
 	struct PF_header header;
@@ -53,7 +53,7 @@ int persist__chunk_header_read_v5(FILE *db_fptr, int *chunk, int *length)
 }
 
 
-int persist__chunk_cfg_read_v5(FILE *db_fptr, struct PF_cfg *chunk)
+int persist__chunk_cfg_read_v56(FILE *db_fptr, struct PF_cfg *chunk)
 {
 	if(fread(chunk, sizeof(struct PF_cfg), 1, db_fptr) != 1){
 		log__printf(NULL, MOSQ_LOG_ERR, "Error: %s.", strerror(errno));
@@ -64,28 +64,46 @@ int persist__chunk_cfg_read_v5(FILE *db_fptr, struct PF_cfg *chunk)
 }
 
 
-int persist__chunk_client_read_v5(FILE *db_fptr, struct P_client *chunk)
+int persist__chunk_client_read_v56(FILE *db_fptr, struct P_client *chunk, int db_version)
 {
 	int rc;
 
-	read_e(db_fptr, &chunk->F, sizeof(struct PF_client));
+	if(db_version == 6){
+		read_e(db_fptr, &chunk->F, sizeof(struct PF_client));
+		chunk->F.username_len = ntohs(chunk->F.username_len);
+		chunk->F.listener_port = ntohs(chunk->F.listener_port);
+	}else if(db_version == 5){
+		read_e(db_fptr, &chunk->F, sizeof(struct PF_client_v5));
+	}else{
+		return 1;
+	}
+	
 	chunk->F.session_expiry_interval = ntohl(chunk->F.session_expiry_interval);
 	chunk->F.last_mid = ntohs(chunk->F.last_mid);
 	chunk->F.id_len = ntohs(chunk->F.id_len);
 
+
 	rc = persist__read_string_len(db_fptr, &chunk->client_id, chunk->F.id_len);
 	if(rc || !chunk->client_id){
 		return 1;
-	}else{
-		return MOSQ_ERR_SUCCESS;
 	}
+
+	if(chunk->F.username_len > 0){
+		rc = persist__read_string_len(db_fptr, &chunk->username, chunk->F.username_len);
+		if(rc || !chunk->username){
+			mosquitto__free(chunk->client_id);
+			return 1;
+		}
+	}
+
+	return MOSQ_ERR_SUCCESS;
 error:
 	log__printf(NULL, MOSQ_LOG_ERR, "Error: %s.", strerror(errno));
 	return 1;
 }
 
 
-int persist__chunk_client_msg_read_v5(FILE *db_fptr, struct P_client_msg *chunk, uint32_t length)
+int persist__chunk_client_msg_read_v56(FILE *db_fptr, struct P_client_msg *chunk, uint32_t length)
 {
 	mosquitto_property *properties = NULL;
 	struct mosquitto__packet prop_packet;
@@ -125,7 +143,7 @@ error:
 }
 
 
-int persist__chunk_msg_store_read_v5(FILE *db_fptr, struct P_msg_store *chunk, uint32_t length)
+int persist__chunk_msg_store_read_v56(FILE *db_fptr, struct P_msg_store *chunk, uint32_t length)
 {
 	int rc = 0;
 	mosquitto_property *properties = NULL;
@@ -215,7 +233,7 @@ error:
 }
 
 
-int persist__chunk_retain_read_v5(FILE *db_fptr, struct P_retain *chunk)
+int persist__chunk_retain_read_v56(FILE *db_fptr, struct P_retain *chunk)
 {
 	if(fread(&chunk->F, sizeof(struct P_retain), 1, db_fptr) != 1){
 		log__printf(NULL, MOSQ_LOG_ERR, "Error: %s.", strerror(errno));
@@ -225,7 +243,7 @@ int persist__chunk_retain_read_v5(FILE *db_fptr, struct P_retain *chunk)
 }
 
 
-int persist__chunk_sub_read_v5(FILE *db_fptr, struct P_sub *chunk)
+int persist__chunk_sub_read_v56(FILE *db_fptr, struct P_sub *chunk)
 {
 	int rc;
 

src/persist_write.c

@@ -70,7 +70,7 @@ static int persist__client_messages_save(struct mosquitto_db *db, FILE *db_fptr,
 		chunk.client_id = context->id;
 		chunk.properties = cmsg->properties;
 
-		rc = persist__chunk_client_msg_write_v5(db_fptr, &chunk);
+		rc = persist__chunk_client_msg_write_v6(db_fptr, &chunk);
 		if(rc){
 			return rc;
 		}
@@ -146,7 +146,7 @@ static int persist__message_store_save(struct mosquitto_db *db, FILE *db_fptr)
 		chunk.payload = stored->payload;
 		chunk.properties = stored->properties;
 
-		rc = persist__chunk_message_store_write_v5(db_fptr, &chunk);
+		rc = persist__chunk_message_store_write_v6(db_fptr, &chunk);
 		if(rc){
 			return rc;
 		}
@@ -174,8 +174,15 @@ static int persist__client_save(struct mosquitto_db *db, FILE *db_fptr)
 			chunk.F.last_mid = context->last_mid;
 			chunk.F.id_len = strlen(context->id);
 			chunk.client_id = context->id;
+			if(context->username){
+				chunk.F.username_len = strlen(context->username);
+				chunk.username = context->username;
+				if(context->listener){
+					chunk.F.listener_port = context->listener->port;
+				}
+			}
 
-			rc = persist__chunk_client_write_v5(db_fptr, &chunk);
+			rc = persist__chunk_client_write_v6(db_fptr, &chunk);
 			if(rc){
 				return rc;
 			}
@@ -224,7 +231,7 @@ static int persist__subs_retain_save(struct mosquitto_db *db, FILE *db_fptr, str
 			sub_chunk.client_id = sub->context->id;
 			sub_chunk.topic = thistopic;
 
-			rc = persist__chunk_sub_write_v5(db_fptr, &sub_chunk);
+			rc = persist__chunk_sub_write_v6(db_fptr, &sub_chunk);
 			if(rc){
 				mosquitto__free(thistopic);
 				return rc;
@@ -236,7 +243,7 @@ static int persist__subs_retain_save(struct mosquitto_db *db, FILE *db_fptr, str
 		if(strncmp(node->retained->topic, "$SYS", 4)){
 			/* Don't save $SYS messages. */
 			retain_chunk.F.store_id = node->retained->db_id;
-			rc = persist__chunk_retain_write_v5(db_fptr, &retain_chunk);
+			rc = persist__chunk_retain_write_v6(db_fptr, &retain_chunk);
 			if(rc){
 				mosquitto__free(thistopic);
 				return rc;
@@ -334,7 +341,7 @@ int persist__backup(struct mosquitto_db *db, bool shutdown)
 	cfg_chunk.last_db_id = db->last_db_id;
 	cfg_chunk.shutdown = shutdown;
 	cfg_chunk.dbid_size = sizeof(dbid_t);
-	if(persist__chunk_cfg_write_v5(db_fptr, &cfg_chunk)){
+	if(persist__chunk_cfg_write_v6(db_fptr, &cfg_chunk)){
 		goto error;
 	}
 

src/persist_write_v5.c

@@ -37,7 +37,7 @@ Contributors:
 #include "time_mosq.h"
 #include "util_mosq.h"
 
-int persist__chunk_cfg_write_v5(FILE *db_fptr, struct PF_cfg *chunk)
+int persist__chunk_cfg_write_v6(FILE *db_fptr, struct PF_cfg *chunk)
 {
 	struct PF_header header;
 
@@ -53,22 +53,28 @@ error:
 }
 
 
-int persist__chunk_client_write_v5(FILE *db_fptr, struct P_client *chunk)
+int persist__chunk_client_write_v6(FILE *db_fptr, struct P_client *chunk)
 {
 	struct PF_header header;
 	uint16_t id_len = chunk->F.id_len;
+	uint16_t username_len = chunk->F.username_len;
 
 	chunk->F.session_expiry_interval = htonl(chunk->F.session_expiry_interval);
 	chunk->F.last_mid = htons(chunk->F.last_mid);
 	chunk->F.id_len = htons(chunk->F.id_len);
+	chunk->F.username_len = htons(chunk->F.username_len);
+	chunk->F.listener_port = htons(chunk->F.listener_port);
 
 	header.chunk = htonl(DB_CHUNK_CLIENT);
-	header.length = htonl(sizeof(struct PF_client)+id_len);
+	header.length = htonl(sizeof(struct PF_client)+id_len+username_len);
 
 	write_e(db_fptr, &header, sizeof(struct PF_header));
 	write_e(db_fptr, &chunk->F, sizeof(struct PF_client));
 
 	write_e(db_fptr, chunk->client_id, id_len);
+	if(username_len > 0){
+		write_e(db_fptr, chunk->username, username_len);
+	}
 
 	return MOSQ_ERR_SUCCESS;
 error:
@@ -77,7 +83,7 @@ error:
 }
 
 
-int persist__chunk_client_msg_write_v5(FILE *db_fptr, struct P_client_msg *chunk)
+int persist__chunk_client_msg_write_v6(FILE *db_fptr, struct P_client_msg *chunk)
 {
 	struct PF_header header;
 	struct mosquitto__packet prop_packet;
@@ -123,7 +129,7 @@ error:
 }
 
 
-int persist__chunk_message_store_write_v5(FILE *db_fptr, struct P_msg_store *chunk)
+int persist__chunk_message_store_write_v6(FILE *db_fptr, struct P_msg_store *chunk)
 {
 	struct PF_header header;
 	uint32_t payloadlen = chunk->F.payloadlen;
@@ -188,7 +194,7 @@ error:
 }
 
 
-int persist__chunk_retain_write_v5(FILE *db_fptr, struct P_retain *chunk)
+int persist__chunk_retain_write_v6(FILE *db_fptr, struct P_retain *chunk)
 {
 	struct PF_header header;
 
@@ -205,7 +211,7 @@ error:
 }
 
 
-int persist__chunk_sub_write_v5(FILE *db_fptr, struct P_sub *chunk)
+int persist__chunk_sub_write_v6(FILE *db_fptr, struct P_sub *chunk)
 {
 	struct PF_header header;
 	uint16_t id_len = chunk->F.id_len;

src/security.c

@@ -33,7 +33,7 @@ void LIB_ERROR(void)
 #ifdef WIN32
 	char *buf;
 	FormatMessage(FORMAT_MESSAGE_ALLOCATE_BUFFER | FORMAT_MESSAGE_FROM_STRING,
-			NULL, GetLastError(), LANG_NEUTRAL, &buf, 0, NULL);
+			NULL, GetLastError(), LANG_NEUTRAL, (LPTSTR)&buf, 0, NULL);
 	log__printf(NULL, MOSQ_LOG_ERR, "Load error: %s", buf);
 	LocalFree(buf);
 #else

src/service.c

@@ -32,7 +32,7 @@ static void print_error(void)
 	char *buf;
 
 	FormatMessage(FORMAT_MESSAGE_ALLOCATE_BUFFER | FORMAT_MESSAGE_FROM_SYSTEM,
-		NULL, GetLastError(), LANG_NEUTRAL, &buf, 0, NULL);
+		NULL, GetLastError(), LANG_NEUTRAL, (LPTSTR)&buf, 0, NULL);
 
 	fprintf(stderr, "Error: %s\n", buf);
 	LocalFree(buf);
@@ -70,7 +70,7 @@ void __stdcall service_main(DWORD dwArgc, LPTSTR *lpszArgv)
 
 	service_handle = RegisterServiceCtrlHandler("mosquitto", service_handler);
 	if(service_handle){
-		memset(conf_path, 0, MAX_PATH + 20);
+		memset(conf_path, 0, sizeof(conf_path));
 		rc = GetEnvironmentVariable("MOSQUITTO_DIR", conf_path, MAX_PATH);
 		if(!rc || rc == MAX_PATH){
 			service_status.dwCurrentState = SERVICE_STOPPED;
@@ -103,25 +103,26 @@ void __stdcall service_main(DWORD dwArgc, LPTSTR *lpszArgv)
 void service_install(void)
 {
 	SC_HANDLE sc_manager, svc_handle;
-	char exe_path[MAX_PATH + 5];
+	char service_string[MAX_PATH + 20];
+	char exe_path[MAX_PATH + 1];
 	SERVICE_DESCRIPTION svc_desc;
 
-	memset(exe_path, 0, MAX_PATH+5);
+	memset(exe_path, 0, sizeof(exe_path));
 	if(GetModuleFileName(NULL, exe_path, MAX_PATH) == MAX_PATH){
 		fprintf(stderr, "Error: Path too long.\n");
 		return;
 	}
-	strcat(exe_path, " run");
+	snprintf(service_string, sizeof(service_string), "\"%s\" run", exe_path);
 
 	sc_manager = OpenSCManager(NULL, NULL, SC_MANAGER_CREATE_SERVICE);
 	if(sc_manager){
 		svc_handle = CreateService(sc_manager, "mosquitto", "Mosquitto Broker", 
 				SERVICE_START | SERVICE_STOP | SERVICE_CHANGE_CONFIG,
 				SERVICE_WIN32_OWN_PROCESS, SERVICE_AUTO_START, SERVICE_ERROR_NORMAL,
-				exe_path, NULL, NULL, NULL, NULL, NULL);
+				service_string, NULL, NULL, NULL, NULL, NULL);
 
 		if(svc_handle){
-			svc_desc.lpDescription = "MQTT v3.1.1 broker";
+			svc_desc.lpDescription = "Eclipse Mosquitto MQTT v5/v3.1.1 broker";
 			ChangeServiceConfig2(svc_handle, SERVICE_CONFIG_DESCRIPTION, &svc_desc);
 			CloseServiceHandle(svc_handle);
 		}else{

src/sys_tree.h

@@ -54,10 +54,10 @@ extern unsigned int g_connection_count;
 #define G_MSGS_SENT_INC(A)
 #define G_PUB_MSGS_RECEIVED_INC(A)
 #define G_PUB_MSGS_SENT_INC(A)
-#define G_MSGS_DROPPED_INC(A)
-#define G_CLIENTS_EXPIRED_INC(A)
-#define G_SOCKET_CONNECTIONS_INC(A)
-#define G_CONNECTION_COUNT_INC(A)
+#define G_MSGS_DROPPED_INC()
+#define G_CLIENTS_EXPIRED_INC()
+#define G_SOCKET_CONNECTIONS_INC()
+#define G_CONNECTION_COUNT_INC()
 
 #endif
 

src/websockets.c

@@ -302,7 +302,10 @@ static int callback_mqtt(struct libwebsocket_context *context,
 				}
 				count = libwebsocket_write(wsi, &packet->payload[packet->pos], txlen, LWS_WRITE_BINARY);
 				if(count < 0){
-					if (mosq->state == mosq_cs_disconnect_ws || mosq->state == mosq_cs_disconnecting){
+					if (mosq->state == mosq_cs_disconnect_ws
+							|| mosq->state == mosq_cs_disconnecting
+							|| mosq->state == mosq_cs_disused){
+
 						return -1;
 					}
 					return 0;
@@ -313,7 +316,10 @@ static int callback_mqtt(struct libwebsocket_context *context,
 				packet->to_process -= count;
 				packet->pos += count;
 				if(packet->to_process > 0){
-					if (mosq->state == mosq_cs_disconnect_ws || mosq->state == mosq_cs_disconnecting){
+					if (mosq->state == mosq_cs_disconnect_ws
+							|| mosq->state == mosq_cs_disconnecting
+							|| mosq->state == mosq_cs_disused){
+
 						return -1;
 					}
 					break;
@@ -340,7 +346,10 @@ static int callback_mqtt(struct libwebsocket_context *context,
 
 				mosq->next_msg_out = mosquitto_time() + mosq->keepalive;
 			}
-			if (mosq->state == mosq_cs_disconnect_ws || mosq->state == mosq_cs_disconnecting){
+			if (mosq->state == mosq_cs_disconnect_ws
+					|| mosq->state == mosq_cs_disconnecting
+					|| mosq->state == mosq_cs_disused){
+
 				return -1;
 			}
 			if(mosq->current_out_packet){