Merge branch 'fixes'

2026-09-20 14:57:34 +08:00 · 2019-11-28 17:14:53 +00:00
parent a26157643d b410568299
commit e55f7facce
55 changed files with 1224 additions and 329 deletions
@@ -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.7)
+set (VERSION 1.6.8)

 add_definitions (-DCMAKE -DVERSION=\"${VERSION}\")

@@ -111,9 +111,9 @@ install(FILES mosquitto.conf aclfile.example pskfile.example pwfile.example DEST
 # ========================================

 configure_file(libmosquitto.pc.in libmosquitto.pc @ONLY)
-install(FILES "${CMAKE_CURRENT_BINARY_DIR}/libmosquitto.pc" DESTINATION "${CMAKE_INSTALL_PREFIX}/share/pkgconfig")
+install(FILES "${CMAKE_CURRENT_BINARY_DIR}/libmosquitto.pc" DESTINATION "${CMAKE_INSTALL_LIBDIR}/pkgconfig")
 configure_file(libmosquittopp.pc.in libmosquittopp.pc @ONLY)
-install(FILES "${CMAKE_CURRENT_BINARY_DIR}/libmosquittopp.pc" DESTINATION "${CMAKE_INSTALL_PREFIX}/share/pkgconfig")
+install(FILES "${CMAKE_CURRENT_BINARY_DIR}/libmosquittopp.pc" DESTINATION "${CMAKE_INSTALL_LIBDIR}/pkgconfig")

 # ========================================
 # Testing
@@ -1,3 +1,51 @@
+1.6.8 - 20191128
+================
+
+Broker:
+- Various fixes for `allow_zero_length_clientid` config, where this option was
+  not being set correctly. Closes #1429.
+- Fix incorrect memory tracking causing problems with memory_limit option.
+  Closes #1437.
+- Fix subscription topics being limited to 200 characters instead of 200
+  hierarchy levels. Closes #1441.
+- Only a single CRL could be loaded at once. This has been fixed.
+  Closes #1442.
+- Fix problems with reloading config when `per_listener_settings` was true.
+  Closes #1459.
+- Fix retained messages with an expiry interval not being expired after being
+  restored from persistence. Closes #1464.
+- Fix messages with an expiry interval being sent without an expiry interval
+  property just before they were expired. Closes #1464.
+- Fix TLS Websockets clients not receiving messages after taking over a
+  previous connection. Closes #1489.
+- Fix MQTT 3.1.1 clients using clean session false, or MQTT 5.0 clients using
+  session-expiry-interval set to infinity never expiring, even when the global
+  `persistent_client_expiration` option was set. Closes #1494.
+
+Client library:
+- Fix publish properties not being passed to on_message_v5 callback for QoS 2
+  messages. Closes #1432.
+- Fix documentation issues in mosquitto.h. Closes #1478.
+- Document `mosquitto_connect_srv()`. Closes #1499.
+
+Clients:
+- Fix duplicate cfg definition in rr_client. Closes #1453.
+- Fix `mosquitto_pub -l` hang when stdin stream ends. Closes #1448.
+- Fix `mosquitto_pub -l` not sending the final line of stdin if it does not
+  end with a new line. Closes #1473.
+- Make documentation for `mosquitto_pub -l` match reality - blank lines are
+  sent as empty messages. Closes #1474.
+- Free memory in `mosquitto_sub` when quiting without having made a successful
+  connection. Closes #1513.
+
+Build:
+- Added `CLIENT_STATIC_LDADD` to makefile builds to allow more libraries to be
+  linked when compiling the clients with a static libmosquitto, as required
+  for e.g. openssl on some systems.
+
+Installer:
+- Fix mosquitto_rr.exe not being included in Windows installers. Closes #1463.
+
 1.6.7 - 20190925
 ================

@@ -21,13 +21,13 @@ static : static_pub static_sub static_rr
 	# libmosquitto only.

 static_pub : pub_client.o pub_shared.o client_props.o client_shared.o ../lib/libmosquitto.a
-	${CROSS_COMPILE}${CC} $^ -o mosquitto_pub ${CLIENT_LDFLAGS} ${STATIC_LIB_DEPS}
+	${CROSS_COMPILE}${CC} $^ -o mosquitto_pub ${CLIENT_LDFLAGS} ${STATIC_LIB_DEPS} ${CLIENT_STATIC_LDADD}

 static_sub : sub_client.o sub_client_output.o client_props.o client_shared.o ../lib/libmosquitto.a
-	${CROSS_COMPILE}${CC} $^ -o mosquitto_sub ${CLIENT_LDFLAGS} ${STATIC_LIB_DEPS}
+	${CROSS_COMPILE}${CC} $^ -o mosquitto_sub ${CLIENT_LDFLAGS} ${STATIC_LIB_DEPS} ${CLIENT_STATIC_LDADD}

 static_rr : rr_client.o client_props.o client_shared.o pub_shared.o sub_client_output.o ../lib/libmosquitto.a
-	${CROSS_COMPILE}${CC} $^ -o mosquitto_rr ${CLIENT_LDFLAGS} ${STATIC_LIB_DEPS}
+	${CROSS_COMPILE}${CC} $^ -o mosquitto_rr ${CLIENT_LDFLAGS} ${STATIC_LIB_DEPS} ${CLIENT_STATIC_LDADD}

 mosquitto_pub : pub_client.o pub_shared.o client_shared.o client_props.o
 	${CROSS_COMPILE}${CC} $(CLIENT_LDFLAGS) $^ -o $@ $(CLIENT_LDADD)
@@ -45,6 +45,7 @@ static int line_buf_len = 1024;
 static bool disconnect_sent = false;
 static int publish_count = 0;
 static bool ready_for_repeat = false;
+static volatile int status = STATUS_CONNECTING;

 #ifdef WIN32
 static uint64_t next_publish_tv;
@@ -219,111 +220,122 @@ int pub_shared_init(void)
 }


-int pub_shared_loop(struct mosquitto *mosq)
+int pub_stdin_line_loop(struct mosquitto *mosq)
 {
-	int read_len;
-	int pos;
-	int rc, rc2;
 	char *buf2;
-	int buf_len_actual;
-	int mode;
-	int loop_delay = 1000;
+	int buf_len_actual = 0;
+	int pos;
+	int rc = MOSQ_ERR_SUCCESS;
+	int read_len;
 	bool stdin_finished = false;

+	mosquitto_loop_start(mosq);
+	stdin_finished = false;
+	do{
+		if(status == STATUS_CONNACK_RECVD){
+			pos = 0;
+			read_len = line_buf_len;
+			while(status == STATUS_CONNACK_RECVD && fgets(&line_buf[pos], read_len, stdin)){
+				buf_len_actual = strlen(line_buf);
+				if(line_buf[buf_len_actual-1] == '\n'){
+					line_buf[buf_len_actual-1] = '\0';
+					rc = my_publish(mosq, &mid_sent, cfg.topic, buf_len_actual-1, line_buf, cfg.qos, cfg.retain);
+					pos = 0;
+					if(rc){
+						err_printf(&cfg, "Error: Publish returned %d, disconnecting.\n", rc);
+						mosquitto_disconnect_v5(mosq, MQTT_RC_DISCONNECT_WITH_WILL_MSG, cfg.disconnect_props);
+					}
+					break;
+				}else{
+					line_buf_len += 1024;
+					pos += 1023;
+					read_len = 1024;
+					buf2 = realloc(line_buf, line_buf_len);
+					if(!buf2){
+						err_printf(&cfg, "Error: Out of memory.\n");
+						return MOSQ_ERR_NOMEM;
+					}
+					line_buf = buf2;
+				}
+			}
+			if(pos != 0){
+				rc = my_publish(mosq, &mid_sent, cfg.topic, buf_len_actual, line_buf, cfg.qos, cfg.retain);
+				if(rc){
+					err_printf(&cfg, "Error: Publish returned %d, disconnecting.\n", rc);
+					mosquitto_disconnect_v5(mosq, MQTT_RC_DISCONNECT_WITH_WILL_MSG, cfg.disconnect_props);
+				}
+			}
+			if(feof(stdin)){
+				if(mid_sent == -1){
+					/* Empty file */
+					mosquitto_disconnect_v5(mosq, 0, cfg.disconnect_props);
+					disconnect_sent = true;
+					status = STATUS_DISCONNECTING;
+				}else{
+					last_mid = mid_sent;
+					status = STATUS_WAITING;
+				}
+				stdin_finished = true;
+			}else if(status == STATUS_DISCONNECTED){
+				/* Not end of stdin, so we've lost our connection and must
+				 * reconnect */
+			}
+		}
+
+		if(status == STATUS_WAITING){
+			if(last_mid_sent == last_mid && disconnect_sent == false){
+				mosquitto_disconnect_v5(mosq, 0, cfg.disconnect_props);
+				disconnect_sent = true;
+			}
+#ifdef WIN32
+			Sleep(100);
+#else
+			struct timespec ts;
+			ts.tv_sec = 0;
+			ts.tv_nsec = 100000000;
+			nanosleep(&ts, NULL);
+#endif
+		}
+	}while(stdin_finished == false);
+	mosquitto_loop_stop(mosq, false);
+
+	if(status == STATUS_DISCONNECTED){
+		return MOSQ_ERR_SUCCESS;
+	}else{
+		return rc;
+	}
+}
+
+
+int pub_other_loop(struct mosquitto *mosq)
+{
+	int rc;
+	int loop_delay = 1000;
+
 	if(cfg.repeat_count > 1 && (cfg.repeat_delay.tv_sec == 0 || cfg.repeat_delay.tv_usec != 0)){
 		loop_delay = cfg.repeat_delay.tv_usec / 2000;
 	}

-	mode = cfg.pub_mode;
-
-	if(mode == MSGMODE_STDIN_LINE){
-		mosquitto_loop_start(mosq);
-		stdin_finished = false;
-	}
-
 	do{
-		if(mode == MSGMODE_STDIN_LINE){
-			if(status == STATUS_CONNACK_RECVD){
-				pos = 0;
-				read_len = line_buf_len;
-				while(status == STATUS_CONNACK_RECVD && fgets(&line_buf[pos], read_len, stdin)){
-					buf_len_actual = strlen(line_buf);
-					if(line_buf[buf_len_actual-1] == '\n'){
-						line_buf[buf_len_actual-1] = '\0';
-						rc2 = my_publish(mosq, &mid_sent, cfg.topic, buf_len_actual-1, line_buf, cfg.qos, cfg.retain);
-						if(rc2){
-							err_printf(&cfg, "Error: Publish returned %d, disconnecting.\n", rc2);
-							mosquitto_disconnect_v5(mosq, MQTT_RC_DISCONNECT_WITH_WILL_MSG, cfg.disconnect_props);
-						}
-						break;
-					}else{
-						line_buf_len += 1024;
-						pos += 1023;
-						read_len = 1024;
-						buf2 = realloc(line_buf, line_buf_len);
-						if(!buf2){
-							err_printf(&cfg, "Error: Out of memory.\n");
-							return MOSQ_ERR_NOMEM;
-						}
-						line_buf = buf2;
-					}
-				}
-				if(feof(stdin)){
-					if(mid_sent == -1){
-						/* Empty file */
-						mosquitto_disconnect_v5(mosq, 0, cfg.disconnect_props);
-						disconnect_sent = true;
-						status = STATUS_DISCONNECTING;
-					}else{
-						last_mid = mid_sent;
-						status = STATUS_WAITING;
-					}
-					stdin_finished = true;
-				}else if(status == STATUS_DISCONNECTED){
-					/* Not end of stdin, so we've lost our connection and must
-					 * reconnect */
-				}
-			}else if(status == STATUS_WAITING){
-				if(last_mid_sent == last_mid && disconnect_sent == false){
-					mosquitto_disconnect_v5(mosq, 0, cfg.disconnect_props);
-					disconnect_sent = true;
-				}
-#ifdef WIN32
-				Sleep(100);
-#else
-				struct timespec ts;
-				ts.tv_sec = 0;
-				ts.tv_nsec = 100000000;
-				nanosleep(&ts, NULL);
-#endif
-			}
+		rc = mosquitto_loop(mosq, loop_delay, 1);
+		if(ready_for_repeat && check_repeat_time()){
 			rc = MOSQ_ERR_SUCCESS;
-		}else{
-			rc = mosquitto_loop(mosq, loop_delay, 1);
-			if(ready_for_repeat && check_repeat_time()){
-				rc = 0;
-				switch(cfg.pub_mode){
-					case MSGMODE_CMD:
-					case MSGMODE_FILE:
-					case MSGMODE_STDIN_FILE:
-						rc = my_publish(mosq, &mid_sent, cfg.topic, cfg.msglen, cfg.message, cfg.qos, cfg.retain);
-						break;
-					case MSGMODE_NULL:
-						rc = my_publish(mosq, &mid_sent, cfg.topic, 0, NULL, cfg.qos, cfg.retain);
-						break;
-					case MSGMODE_STDIN_LINE:
-						break;
-				}
-				if(rc){
-					err_printf(&cfg, "Error sending repeat publish: %s", mosquitto_strerror(rc));
-				}
+			switch(cfg.pub_mode){
+				case MSGMODE_CMD:
+				case MSGMODE_FILE:
+				case MSGMODE_STDIN_FILE:
+					rc = my_publish(mosq, &mid_sent, cfg.topic, cfg.msglen, cfg.message, cfg.qos, cfg.retain);
+					break;
+				case MSGMODE_NULL:
+					rc = my_publish(mosq, &mid_sent, cfg.topic, 0, NULL, cfg.qos, cfg.retain);
+					break;
+			}
+			if(rc){
+				err_printf(&cfg, "Error sending repeat publish: %s", mosquitto_strerror(rc));
 			}
 		}
-	}while(rc == MOSQ_ERR_SUCCESS && stdin_finished == false);
+	}while(rc == MOSQ_ERR_SUCCESS);

-	if(mode == MSGMODE_STDIN_LINE){
-		mosquitto_loop_stop(mosq, false);
-	}
 	if(status == STATUS_DISCONNECTED){
 		return MOSQ_ERR_SUCCESS;
 	}else{
@@ -332,6 +344,16 @@ int pub_shared_loop(struct mosquitto *mosq)
 }


+int pub_shared_loop(struct mosquitto *mosq)
+{
+	if(cfg.pub_mode == MSGMODE_STDIN_LINE){
+		return pub_stdin_line_loop(mosq);
+	}else{
+		return pub_other_loop(mosq);
+	}
+}
+
+
 void pub_shared_cleanup(void)
 {
 	free(line_buf);
@@ -37,7 +37,6 @@ Contributors:
 /* Global variables for use in callbacks. See sub_client.c for an example of
 * using a struct to hold variables for use in callbacks. */
 int mid_sent = -1;
-int status = STATUS_CONNECTING;
 struct mosq_config cfg;

 void my_log_callback(struct mosquitto *mosq, void *obj, int level, const char *str)
@@ -23,7 +23,6 @@ Contributors:
 #define STATUS_DISCONNECTED 4

 extern int mid_sent;
-extern int status;
 extern struct mosq_config cfg;


@@ -47,7 +47,8 @@ enum rr__state {

 static enum rr__state client_state = rr_s_new;

-struct mosq_config cfg;
+extern struct mosq_config cfg;
+
 bool process_messages = true;
 int msg_count = 0;
 struct mosquitto *mosq = NULL;
@@ -44,7 +44,7 @@ int last_mid = 0;
 #ifndef WIN32
 void my_signal_handler(int signum)
 {
-	if(signum == SIGALRM){
+	if(signum == SIGALRM || signum == SIGTERM || signum == SIGINT){
 		process_messages = false;
 		mosquitto_disconnect_v5(mosq, MQTT_RC_DISCONNECT_WITH_WILL_MSG, cfg.disconnect_props);
 	}
@@ -344,6 +344,16 @@ int main(int argc, char *argv[])
 		goto cleanup;
 	}

+	if(sigaction(SIGTERM, &sigact, NULL) == -1){
+		perror("sigaction");
+		goto cleanup;
+	}
+
+	if(sigaction(SIGINT, &sigact, NULL) == -1){
+		perror("sigaction");
+		goto cleanup;
+	}
+
 	if(cfg.timeout){
 		alarm(cfg.timeout);
 	}
@@ -364,6 +374,7 @@ int main(int argc, char *argv[])
 	return rc;

 cleanup:
+	mosquitto_destroy(mosq);
 	mosquitto_lib_cleanup();
 	client_config_cleanup(&cfg);
 	return 1;
@@ -82,6 +82,11 @@ WITH_STRIP:=no
 # Build static libraries
 WITH_STATIC_LIBRARIES:=no

+# Use this variable to add extra library dependencies when building the clients
+# with the static libmosquitto library. This may be required on some systems
+# where e.g. -lz or -latomic are needed for openssl.
+CLIENT_STATIC_LDADD:=
+
 # Build shared libraries
 WITH_SHARED_LIBRARIES:=yes

@@ -104,7 +109,7 @@ WITH_COVERAGE:=no

 # Also bump lib/mosquitto.h, CMakeLists.txt,
 # installer/mosquitto.nsi, installer/mosquitto64.nsi
-VERSION=1.6.7
+VERSION=1.6.8

 # Client library SO version. Bump if incompatible API/ABI changes are made.
 SOVERSION=1
@@ -312,3 +317,7 @@ ifeq ($(WITH_COVERAGE),yes)
 	CLIENT_CFLAGS:=$(CLIENT_CFLAGS) -coverage
 	CLIENT_LDFLAGS:=$(CLIENT_LDFLAGS) -coverage
 endif
+
+BROKER_LDADD:=${BROKER_LDADD} ${LDADD}
+CLIENT_LDADD:=${CLIENT_LDADD} ${LDADD}
+PASSWD_LDADD:=${PASSWD_LDADD} ${LDADD}
@@ -12,7 +12,8 @@ RUN apk --no-cache add \
      util-linux-dev \
      libwebsockets-dev \
      libxslt \
-      python2
+      python2 \
+      ca-certificates

 # This build procedure is based on:
 # https://github.com/alpinelinux/aports/blob/master/main/mosquitto/APKBUILD
@@ -9,7 +9,7 @@
 !define env_hklm 'HKLM "SYSTEM\CurrentControlSet\Control\Session Manager\Environment"'

 Name "Eclipse Mosquitto"
-!define VERSION 1.6.7
+!define VERSION 1.6.8
 OutFile "mosquitto-${VERSION}-install-windows-x86.exe"

 InstallDir "$PROGRAMFILES\mosquitto"
@@ -45,6 +45,7 @@ Section "Files" SecInstall
 	File "..\build\src\Release\mosquitto_passwd.exe"
 	File "..\build\client\Release\mosquitto_pub.exe"
 	File "..\build\client\Release\mosquitto_sub.exe"
+	File "..\build\client\Release\mosquitto_rr.exe"
 	File "..\build\lib\Release\mosquitto.dll"
 	File "..\build\lib\cpp\Release\mosquittopp.dll"
 	File "..\aclfile.example"
@@ -90,6 +91,7 @@ Section "Uninstall"
 	Delete "$INSTDIR\mosquitto_passwd.exe"
 	Delete "$INSTDIR\mosquitto_pub.exe"
 	Delete "$INSTDIR\mosquitto_sub.exe"
+	Delete "$INSTDIR\mosquitto_rr.exe"
 	Delete "$INSTDIR\mosquitto.dll"
 	Delete "$INSTDIR\mosquittopp.dll"
 	Delete "$INSTDIR\aclfile.example"
@@ -9,7 +9,7 @@
 !define env_hklm 'HKLM "SYSTEM\CurrentControlSet\Control\Session Manager\Environment"'

 Name "Eclipse Mosquitto"
-!define VERSION 1.6.7
+!define VERSION 1.6.8
 OutFile "mosquitto-${VERSION}-install-windows-x64.exe"

 !include "x64.nsh"
@@ -46,6 +46,7 @@ Section "Files" SecInstall
 	File "..\build64\src\Release\mosquitto_passwd.exe"
 	File "..\build64\client\Release\mosquitto_pub.exe"
 	File "..\build64\client\Release\mosquitto_sub.exe"
+	File "..\build64\client\Release\mosquitto_rr.exe"
 	File "..\build64\lib\Release\mosquitto.dll"
 	File "..\build64\lib\cpp\Release\mosquittopp.dll"
 	File "..\aclfile.example"
@@ -91,6 +92,7 @@ Section "Uninstall"
 	Delete "$INSTDIR\mosquitto_passwd.exe"
 	Delete "$INSTDIR\mosquitto_pub.exe"
 	Delete "$INSTDIR\mosquitto_sub.exe"
+	Delete "$INSTDIR\mosquitto_rr.exe"
 	Delete "$INSTDIR\mosquitto.dll"
 	Delete "$INSTDIR\mosquittopp.dll"
 	Delete "$INSTDIR\aclfile.example"
@@ -152,13 +152,13 @@ int handle__publish(struct mosquitto *mosq)
 			mosquitto_property_free_all(&properties);
 			return rc;
 		case 2:
+			message->properties = properties;
 			util__decrement_receive_quota(mosq);
 			rc = send__pubrec(mosq, message->msg.mid, 0);
 			pthread_mutex_lock(&mosq->msgs_in.mutex);
 			message->state = mosq_ms_wait_for_pubrel;
 			message__queue(mosq, message, mosq_md_in);
 			pthread_mutex_unlock(&mosq->msgs_in.mutex);
-			mosquitto_property_free_all(&properties);
 			return rc;
 		default:
 			message__cleanup(&message);
@@ -114,7 +114,7 @@ int handle__pubrel(struct mosquitto_db *db, struct mosquitto *mosq)
 		}
 		if(mosq->on_message_v5){
 			mosq->in_callback = true;
-			mosq->on_message_v5(mosq, mosq->userdata, &message->msg, properties);
+			mosq->on_message_v5(mosq, mosq->userdata, &message->msg, message->properties);
 			mosq->in_callback = false;
 		}
 		pthread_mutex_unlock(&mosq->callback_mutex);
@@ -856,7 +856,6 @@ ssize_t net__read(struct mosquitto *mosq, void *buf, size_t count)
 	errno = 0;
 #ifdef WITH_TLS
 	if(mosq->ssl){
-		ERR_clear_error();
 		ret = SSL_read(mosq->ssl, buf, count);
 		if(ret <= 0){
 			err = SSL_get_error(mosq->ssl, ret);
@@ -871,6 +870,7 @@ ssize_t net__read(struct mosquitto *mosq, void *buf, size_t count)
 				net__print_ssl_error(mosq);
 				errno = EPROTO;
 			}
+			ERR_clear_error();
 #ifdef WIN32
 			WSASetLastError(errno);
 #endif
@@ -904,7 +904,6 @@ ssize_t net__write(struct mosquitto *mosq, void *buf, size_t count)
 #ifdef WITH_TLS
 	if(mosq->ssl){
 		mosq->want_write = false;
-		ERR_clear_error();
 		ret = SSL_write(mosq->ssl, buf, count);
 		if(ret < 0){
 			err = SSL_get_error(mosq->ssl, ret);
@@ -919,6 +918,7 @@ ssize_t net__write(struct mosquitto *mosq, void *buf, size_t count)
 				net__print_ssl_error(mosq);
 				errno = EPROTO;
 			}
+			ERR_clear_error();
 #ifdef WIN32
 			WSASetLastError(errno);
 #endif
@@ -338,7 +338,7 @@
 				<term><option>-l</option></term>
 				<term><option>--stdin-line</option></term>
 				<listitem>
-					<para>Send messages read from stdin, splitting separate lines into separate messages. Note that blank lines won't be sent.</para>
+					<para>Send messages read from stdin, splitting separate lines into separate messages.</para>
 				</listitem>
 			</varlistentry>
 			<varlistentry>
@@ -2,7 +2,7 @@

 MAJOR=1
 MINOR=6
-REVISION=7
+REVISION=8

 sed -i "s/^VERSION=.*/VERSION=${MAJOR}.${MINOR}.${REVISION}/" config.mk

@@ -1,5 +1,5 @@
 name: mosquitto
-version: 1.6.7
+version: 1.6.8
 summary: Eclipse Mosquitto MQTT broker
 description: This is a message broker that supports version 3.1 and 3.1.1 of the MQTT
    protocol.
@@ -271,6 +271,7 @@ void config__init(struct mosquitto_db *db, struct mosquitto__config *config)
 	config->default_listener.max_connections = -1;
 	config->default_listener.protocol = mp_mqtt;
 	config->default_listener.security_options.allow_anonymous = -1;
+	config->default_listener.security_options.allow_zero_length_clientid = true;
 	config->default_listener.maximum_qos = 2;
 	config->default_listener.max_topic_alias = 10;
 }
@@ -471,6 +472,7 @@ int config__parse_args(struct mosquitto_db *db, struct mosquitto__config *config
 			|| config->default_listener.security_options.psk_file
 			|| config->default_listener.security_options.auth_plugin_config_count
 			|| config->default_listener.security_options.allow_anonymous != -1
+			|| config->default_listener.security_options.allow_zero_length_clientid != true
 			){

 		config->listener_count++;
@@ -529,6 +531,7 @@ int config__parse_args(struct mosquitto_db *db, struct mosquitto__config *config
 		config->listeners[config->listener_count-1].security_options.auth_plugin_configs = config->default_listener.security_options.auth_plugin_configs;
 		config->listeners[config->listener_count-1].security_options.auth_plugin_config_count = config->default_listener.security_options.auth_plugin_config_count;
 		config->listeners[config->listener_count-1].security_options.allow_anonymous = config->default_listener.security_options.allow_anonymous;
+		config->listeners[config->listener_count-1].security_options.allow_zero_length_clientid = config->default_listener.security_options.allow_zero_length_clientid;
 	}

 	/* Default to drop to mosquitto user if we are privileged and no user specified. */
@@ -638,6 +641,7 @@ int config__read(struct mosquitto_db *db, struct mosquitto__config *config, bool
 		config__init_reload(db, &config_reload);
 		config_reload.listeners = config->listeners;
 		config_reload.listener_count = config->listener_count;
+		cur_security_options = NULL;
 		rc = config__read_file(&config_reload, reload, db->config_file, &cr, 0, &lineno);
 	}else{
 		rc = config__read_file(config, reload, db->config_file, &cr, 0, &lineno);
@@ -1388,6 +1392,7 @@ int config__read_file_core(struct mosquitto__config *config, bool reload, struct
 						}

 						cur_listener->security_options.allow_anonymous = -1;
+						cur_listener->security_options.allow_zero_length_clientid = true;
 						cur_listener->protocol = mp_mqtt;
 						cur_listener->port = tmp_int;
 						cur_listener->maximum_qos = 2;
@@ -230,7 +230,7 @@ void context__disconnect(struct mosquitto_db *db, struct mosquitto *context)

 	context__send_will(db, context);
 	if(context->session_expiry_interval == 0){
-
+		/* Client session is due to be expired now */
 #ifdef WITH_BRIDGE
 		if(!context->bridge)
 #endif
@@ -160,6 +160,9 @@ int connect__on_authorised(struct mosquitto_db *db, struct mosquitto *context, v
 			}
 		}

+		if(context->clean_start == true){
+			sub__clean_session(db, found_context);
+		}
 		session_expiry__remove(found_context);
 		will_delay__remove(found_context);
 		will__clear(found_context);
@@ -138,8 +138,6 @@ int mosquitto_main_loop(struct mosquitto_db *db, mosq_sock_t *listensock, int li
 	int err;
 	socklen_t len;
 #endif
-	time_t expiration_check_time = 0;
-	char *id;


 #if defined(WITH_WEBSOCKETS) && LWS_LIBRARY_VERSION_NUMBER == 3002000
@@ -169,10 +167,6 @@ int mosquitto_main_loop(struct mosquitto_db *db, mosq_sock_t *listensock, int li
 	}
 #endif

-	if(db->config->persistent_client_expiration > 0){
-		expiration_check_time = time(NULL) + 3600;
-	}
-
 #ifdef WITH_EPOLL
 	db->epollfd = 0;
 	if ((db->epollfd = epoll_create(MAX_EVENTS)) == -1) {
@@ -471,31 +465,6 @@ int mosquitto_main_loop(struct mosquitto_db *db, mosq_sock_t *listensock, int li
 			}
 		}
 #endif
-		now = time(NULL);
-		if(db->config->persistent_client_expiration > 0 && now > expiration_check_time){
-			HASH_ITER(hh_id, db->contexts_by_id, context, ctxt_tmp){
-				if(context->sock == INVALID_SOCKET && context->session_expiry_interval > 0 && context->session_expiry_interval != UINT32_MAX){
-					/* This is a persistent client, check to see if the
-					 * last time it connected was longer than
-					 * persistent_client_expiration seconds ago. If so,
-					 * expire it and clean up.
-					 */
-					if(now > context->session_expiry_time){
-						if(context->id){
-							id = context->id;
-						}else{
-							id = "<unknown>";
-						}
-						log__printf(NULL, MOSQ_LOG_NOTICE, "Expiring persistent client %s due to timeout.", id);
-						G_CLIENTS_EXPIRED_INC();
-						context->session_expiry_interval = 0;
-						mosquitto__set_state(context, mosq_cs_expiring);
-						do_disconnect(db, context, MOSQ_ERR_SUCCESS);
-					}
-				}
-			}
-			expiration_check_time = time(NULL) + 3600;
-		}

 #ifndef WIN32
 		sigprocmask(SIG_SETMASK, &sigblock, &origsig);
@@ -426,7 +426,7 @@ int net__load_crl_file(struct mosquitto__listener *listener)
 	}
 	lookup = X509_STORE_add_lookup(store, X509_LOOKUP_file());
 	rc = X509_load_crl_file(lookup, listener->crlfile, X509_FILETYPE_PEM);
-	if(rc != 1){
+	if(rc < 1){
 		log__printf(NULL, MOSQ_LOG_ERR, "Error: Unable to load certificate revocation file \"%s\". Check crlfile.", listener->crlfile);
 		net__print_error(MOSQ_LOG_ERR, "Error: %s");
 		return 1;
@@ -114,6 +114,12 @@ static int persist__client_msg_restore(struct mosquitto_db *db, struct P_client_
 	struct mosquitto *context;
 	struct mosquitto_msg_data *msg_data;

+	HASH_FIND(hh, db->msg_store_load, &chunk->F.store_id, sizeof(dbid_t), load);
+	if(!load){
+		/* Can't find message - probably expired */
+		return MOSQ_ERR_SUCCESS;
+	}
+
 	cmsg = mosquitto__calloc(1, sizeof(struct mosquitto_client_msg));
 	if(!cmsg){
 		log__printf(NULL, MOSQ_LOG_ERR, "Error: Out of memory.");
@@ -131,12 +137,6 @@ static int persist__client_msg_restore(struct mosquitto_db *db, struct P_client_
 	cmsg->dup = chunk->F.retain_dup&0x0F;
 	cmsg->properties = chunk->properties;

-	HASH_FIND(hh, db->msg_store_load, &chunk->F.store_id, sizeof(dbid_t), load);
-	if(!load){
-		mosquitto__free(cmsg);
-		log__printf(NULL, MOSQ_LOG_ERR, "Error restoring persistent database, message store corrupt.");
-		return 1;
-	}
 	cmsg->store = load->store;
 	db__msg_store_ref_inc(cmsg->store);

@@ -273,7 +273,13 @@ static int persist__msg_store_chunk_restore(struct mosquitto_db *db, FILE *db_fp
 	if(chunk.F.expiry_time > 0){
 		message_expiry_interval64 = chunk.F.expiry_time - time(NULL);
 		if(message_expiry_interval64 < 0 || message_expiry_interval64 > UINT32_MAX){
-			message_expiry_interval = 0;
+			/* Expired message */
+			mosquitto__free(chunk.source.id);
+			mosquitto__free(chunk.source.username);
+			mosquitto__free(chunk.topic);
+			UHPA_FREE(chunk.payload, chunk.F.payloadlen);
+			mosquitto__free(load);
+			return MOSQ_ERR_SUCCESS;
 		}else{
 			message_expiry_interval = (uint32_t)message_expiry_interval64;
 		}
@@ -327,8 +333,7 @@ static int persist__retain_chunk_restore(struct mosquitto_db *db, FILE *db_fptr)
 	if(load){
 		sub__messages_queue(db, NULL, load->store->topic, load->store->qos, load->store->retain, &load->store);
 	}else{
-		log__printf(NULL, MOSQ_LOG_ERR, "Error: Corrupt database whilst restoring a retained message.");
-		return MOSQ_ERR_INVAL;
+		/* Can't find the message - probably expired */
 	}
 	return MOSQ_ERR_SUCCESS;
 }
@@ -22,6 +22,7 @@ Contributors:

 #include "mosquitto_broker_internal.h"
 #include "memory_mosq.h"
+#include "sys_tree.h"
 #include "time_mosq.h"

 static struct session_expiry_list *expiry_list = NULL;
@@ -38,17 +39,32 @@ int session_expiry__add(struct mosquitto_db *db, struct mosquitto *context)
 {
 	struct session_expiry_list *item;

+	if(db->config->persistent_client_expiration == 0){
+		if(context->session_expiry_interval == UINT32_MAX){
+			/* There isn't a global expiry set, and the client has asked to
+			 * never expire, so we don't add it to the list. */
+			return MOSQ_ERR_SUCCESS;
+		}
+	}
+
 	item = mosquitto__calloc(1, sizeof(struct session_expiry_list));
 	if(!item) return MOSQ_ERR_NOMEM;

 	item->context = context;
 	item->context->session_expiry_time = time(NULL);
-	if(db->config->persistent_client_expiration == 0 || 
-			db->config->persistent_client_expiration < item->context->session_expiry_interval){

+	if(db->config->persistent_client_expiration == 0){
+		/* No global expiry, so use the client expiration interval */
 		item->context->session_expiry_time += item->context->session_expiry_interval;
 	}else{
-		item->context->session_expiry_time += db->config->persistent_client_expiration;
+		/* We have a global expiry interval */
+		if(db->config->persistent_client_expiration < item->context->session_expiry_interval){
+			/* The client expiry is longer than the global expiry, so use the global */
+			item->context->session_expiry_time += db->config->persistent_client_expiration;
+		}else{
+			/* The global expiry is longer than the client expiry, so use the client */
+			item->context->session_expiry_time += item->context->session_expiry_interval;
+		}
 	}
 	context->expiry_list_item = item;

@@ -95,12 +111,16 @@ void session_expiry__check(struct mosquitto_db *db, time_t now)
 	last_check = now;

 	DL_FOREACH_SAFE(expiry_list, item, tmp){
-		if(item->context->session_expiry_interval != UINT32_MAX
-				&& item->context->session_expiry_time < now){
+		if(item->context->session_expiry_time < now){

 			context = item->context;
 			session_expiry__remove(context);

+			if(context->id){
+				log__printf(NULL, MOSQ_LOG_NOTICE, "Expiring client %s due to timeout.", context->id);
+			}
+			G_CLIENTS_EXPIRED_INC();
+
 			/* Session has now expired, so clear interval */
 			context->session_expiry_interval = 0;
 			/* Session has expired, so will delay should be cleared. */
@@ -243,9 +243,9 @@ static int sub__topic_tokenise(const char *subtopic, struct sub__token **topics)

 	stop = 0;
 	for(i=start; i<len+1; i++){
-		count++;
 		if(subtopic[i] == '/' || subtopic[i] == '\0'){
 			stop = i;
+			count++;

 			if(start != stop){
 				tlen = stop-start;
@@ -684,7 +684,7 @@ struct mosquitto__subhier *sub__add_hier_entry(struct mosquitto__subhier *parent
 	}
 	child->parent = parent;
 	child->topic_len = len;
-	child->topic = malloc(len+1);
+	child->topic = mosquitto__malloc(len+1);
 	if(!child->topic){
 		child->topic_len = 0;
 		mosquitto__free(child);
@@ -987,7 +987,7 @@ static int retain__process(struct mosquitto_db *db, struct mosquitto__subhier *b
 	mosquitto_property *properties = NULL;
 	struct mosquitto_msg_store *retained;

-	if(branch->retained->message_expiry_time > 0 && now > branch->retained->message_expiry_time){
+	if(branch->retained->message_expiry_time > 0 && now >= branch->retained->message_expiry_time){
 		db__msg_store_ref_dec(db, &branch->retained);
 		branch->retained = NULL;
 #ifdef WITH_SYS_TREE
@@ -22,8 +22,10 @@ try:
    sock.close()
    if len(data) == 0:
        rc = 0
-except socket.error:
-    rc = 0
+except socket.error as e:
+    if e.errno == errno.ECONNRESET:
+        # Connection has been closed by peer, this is the expected behaviour
+        rc = 0
 finally:
    broker.terminate()
    broker.wait()
@@ -1,7 +1,7 @@
 #!/usr/bin/env python3

 # Test whether a CONNECT with a zero length client id results in the correct CONNACK packet.
-
+# MQTT V3.1 only - zero length is invalid.
 from mosq_test_helper import *

 rc = 1
@@ -15,7 +15,10 @@ try:
    sock = mosq_test.do_client_connect(connect_packet, b"", port=port)
    sock.close()
    rc = 0
-
+except socket.error as e:
+    if e.errno == errno.ECONNRESET:
+        # Connection has been closed by peer, this is the expected behaviour
+        rc = 0
 finally:
    broker.terminate()
    broker.wait()
@@ -0,0 +1,125 @@
+#!/usr/bin/env python3
+
+# Test whether a CONNECT with a zero length client id results in the correct behaviour.
+
+# MQTT v3.1.1 - zero length is allowed, unless allow_zero_length_clientid is false, and unless clean_start is False.
+# MQTT v5.0 - zero length is allowed, unless allow_zero_length_clientid is false
+
+from mosq_test_helper import *
+
+def write_config(filename, port1, port2, per_listener, allow_zero):
+    with open(filename, 'w') as f:
+        f.write("per_listener_settings %s\n" % (per_listener))
+        f.write("port %d\n" % (port2))
+        if allow_zero != "":
+            f.write("allow_zero_length_clientid %s\n" % (allow_zero))
+        f.write("listener %d\n" % (port1))
+        if allow_zero != "":
+            f.write("allow_zero_length_clientid %s\n" % (allow_zero))
+
+
+def do_test(per_listener, proto_ver, clean_start, allow_zero, client_port, expect_fail):
+    conf_file = os.path.basename(__file__).replace('.py', '.conf')
+    write_config(conf_file, port1, port2, per_listener, allow_zero)
+
+    rc = 1
+    keepalive = 10
+    connect_packet = mosq_test.gen_connect("", keepalive=keepalive, proto_ver=proto_ver, clean_session=clean_start)
+    if proto_ver == 4:
+        if expect_fail == True:
+            connack_packet = mosq_test.gen_connack(rc=2, proto_ver=proto_ver)
+        else:
+            connack_packet = mosq_test.gen_connack(rc=0, proto_ver=proto_ver)
+    else:
+        if expect_fail == True:
+            connack_packet = mosq_test.gen_connack(rc=128, proto_ver=proto_ver, properties=None)
+        else:
+            props = mqtt5_props.gen_string_prop(mqtt5_props.PROP_ASSIGNED_CLIENT_IDENTIFIER, "auto-xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx")
+            connack_packet = mosq_test.gen_connack(rc=0, proto_ver=proto_ver, properties=props)
+            # Remove the "xxxx" part - this means the front part of the packet
+            # is correct (so remaining length etc. is correct), but we don't
+            # need to match against the random id.
+            connack_packet = connack_packet[:-36]
+
+    broker = mosq_test.start_broker(filename=os.path.basename(__file__), port=port1, use_conf=True)
+
+    try:
+        sock = mosq_test.do_client_connect(connect_packet, connack_packet, port=client_port)
+        sock.close()
+        rc = 0
+    finally:
+        broker.terminate()
+        broker.wait()
+        (stdo, stde) = broker.communicate()
+        os.remove(conf_file)
+        if rc:
+            print(stde.decode('utf-8'))
+            print("per_listener:%s proto_ver:%d client_port:%d clean_start:%d allow_zero:%s" % (per_listener, proto_ver, client_port, clean_start, allow_zero))
+            print("port1:%d port2:%d" % (port1, port2))
+            exit(rc)
+
+
+(port1, port2) = mosq_test.get_port(2)
+
+test_v4 = True
+test_v5 = True
+
+if test_v4 == True:
+    do_test(per_listener="false", proto_ver=4, client_port=port1, clean_start=True, allow_zero="true", expect_fail=False)
+    do_test(per_listener="false", proto_ver=4, client_port=port1, clean_start=True, allow_zero="false", expect_fail=True)
+    do_test(per_listener="false", proto_ver=4, client_port=port1, clean_start=False, allow_zero="true", expect_fail=True)
+    do_test(per_listener="false", proto_ver=4, client_port=port1, clean_start=False, allow_zero="false", expect_fail=True)
+    do_test(per_listener="true", proto_ver=4, client_port=port1, clean_start=True, allow_zero="true", expect_fail=False)
+    do_test(per_listener="true", proto_ver=4, client_port=port1, clean_start=True, allow_zero="false", expect_fail=True)
+    do_test(per_listener="true", proto_ver=4, client_port=port1, clean_start=False, allow_zero="true", expect_fail=True)
+    do_test(per_listener="true", proto_ver=4, client_port=port1, clean_start=False, allow_zero="false", expect_fail=True)
+
+    do_test(per_listener="false", proto_ver=4, client_port=port2, clean_start=True, allow_zero="true", expect_fail=False)
+    do_test(per_listener="false", proto_ver=4, client_port=port2, clean_start=True, allow_zero="false", expect_fail=True)
+    do_test(per_listener="false", proto_ver=4, client_port=port2, clean_start=False, allow_zero="true", expect_fail=True)
+    do_test(per_listener="false", proto_ver=4, client_port=port2, clean_start=False, allow_zero="false", expect_fail=True)
+    do_test(per_listener="true", proto_ver=4, client_port=port2, clean_start=True, allow_zero="true", expect_fail=False)
+    do_test(per_listener="true", proto_ver=4, client_port=port2, clean_start=True, allow_zero="false", expect_fail=True)
+    do_test(per_listener="true", proto_ver=4, client_port=port2, clean_start=False, allow_zero="true", expect_fail=True)
+    do_test(per_listener="true", proto_ver=4, client_port=port2, clean_start=False, allow_zero="false", expect_fail=True)
+
+    do_test(per_listener="false", proto_ver=4, client_port=port1, clean_start=True, allow_zero="", expect_fail=False)
+    do_test(per_listener="false", proto_ver=4, client_port=port1, clean_start=False, allow_zero="", expect_fail=True)
+    do_test(per_listener="true", proto_ver=4, client_port=port1, clean_start=True, allow_zero="", expect_fail=False)
+    do_test(per_listener="true", proto_ver=4, client_port=port1, clean_start=False, allow_zero="", expect_fail=True)
+
+    do_test(per_listener="false", proto_ver=4, client_port=port2, clean_start=True, allow_zero="", expect_fail=False)
+    do_test(per_listener="false", proto_ver=4, client_port=port2, clean_start=False, allow_zero="", expect_fail=True)
+    do_test(per_listener="true", proto_ver=4, client_port=port2, clean_start=True, allow_zero="", expect_fail=False)
+    do_test(per_listener="true", proto_ver=4, client_port=port2, clean_start=False, allow_zero="", expect_fail=True)
+
+if test_v5 == True:
+    do_test(per_listener="false", proto_ver=5, client_port=port1, clean_start=True, allow_zero="true", expect_fail=False)
+    do_test(per_listener="false", proto_ver=5, client_port=port1, clean_start=True, allow_zero="false", expect_fail=True)
+    do_test(per_listener="false", proto_ver=5, client_port=port1, clean_start=False, allow_zero="true", expect_fail=False)
+    do_test(per_listener="false", proto_ver=5, client_port=port1, clean_start=False, allow_zero="false", expect_fail=True)
+    do_test(per_listener="true", proto_ver=5, client_port=port1, clean_start=True, allow_zero="true", expect_fail=False)
+    do_test(per_listener="true", proto_ver=5, client_port=port1, clean_start=True, allow_zero="false", expect_fail=True)
+    do_test(per_listener="true", proto_ver=5, client_port=port1, clean_start=False, allow_zero="true", expect_fail=False)
+    do_test(per_listener="true", proto_ver=5, client_port=port1, clean_start=False, allow_zero="false", expect_fail=True)
+
+    do_test(per_listener="false", proto_ver=5, client_port=port2, clean_start=True, allow_zero="true", expect_fail=False)
+    do_test(per_listener="false", proto_ver=5, client_port=port2, clean_start=True, allow_zero="false", expect_fail=True)
+    do_test(per_listener="false", proto_ver=5, client_port=port2, clean_start=False, allow_zero="true", expect_fail=False)
+    do_test(per_listener="false", proto_ver=5, client_port=port2, clean_start=False, allow_zero="false", expect_fail=True)
+    do_test(per_listener="true", proto_ver=5, client_port=port2, clean_start=True, allow_zero="true", expect_fail=False)
+    do_test(per_listener="true", proto_ver=5, client_port=port2, clean_start=True, allow_zero="false", expect_fail=True)
+    do_test(per_listener="true", proto_ver=5, client_port=port2, clean_start=False, allow_zero="true", expect_fail=False)
+    do_test(per_listener="true", proto_ver=5, client_port=port2, clean_start=False, allow_zero="false", expect_fail=True)
+
+    do_test(per_listener="false", proto_ver=5, client_port=port1, clean_start=True, allow_zero="", expect_fail=False)
+    do_test(per_listener="false", proto_ver=5, client_port=port1, clean_start=False, allow_zero="", expect_fail=False)
+    do_test(per_listener="true", proto_ver=5, client_port=port1, clean_start=True, allow_zero="", expect_fail=False)
+    do_test(per_listener="true", proto_ver=5, client_port=port1, clean_start=False, allow_zero="", expect_fail=False)
+
+    do_test(per_listener="false", proto_ver=5, client_port=port2, clean_start=True, allow_zero="", expect_fail=False)
+    do_test(per_listener="false", proto_ver=5, client_port=port2, clean_start=False, allow_zero="", expect_fail=False)
+    do_test(per_listener="true", proto_ver=5, client_port=port2, clean_start=True, allow_zero="", expect_fail=False)
+    do_test(per_listener="true", proto_ver=5, client_port=port2, clean_start=False, allow_zero="", expect_fail=False)
+
+exit(0)
@@ -0,0 +1,50 @@
+#!/usr/bin/env python3
+
+# Test whether a client subscribed to a topic receives its own message sent to that topic, for long topics.
+
+from mosq_test_helper import *
+
+def do_test(topic, succeeds):
+    rc = 1
+    mid = 53
+    keepalive = 60
+    connect_packet = mosq_test.gen_connect("subpub-qos0-test", keepalive=keepalive)
+    connack_packet = mosq_test.gen_connack(rc=0)
+
+    subscribe_packet = mosq_test.gen_subscribe(mid, topic, 0)
+    suback_packet = mosq_test.gen_suback(mid, 0)
+
+    publish_packet = mosq_test.gen_publish(topic, qos=0, payload="message")
+
+    port = mosq_test.get_port()
+    broker = mosq_test.start_broker(filename=os.path.basename(__file__), port=port)
+
+    try:
+        sock = mosq_test.do_client_connect(connect_packet, connack_packet, timeout=20, port=port)
+
+        if succeeds == True:
+            mosq_test.do_send_receive(sock, subscribe_packet, suback_packet, "suback")
+            mosq_test.do_send_receive(sock, publish_packet, publish_packet, "publish")
+        else:
+            mosq_test.do_send_receive(sock, subscribe_packet, b"", "suback")
+
+        rc = 0
+
+        sock.close()
+    finally:
+        broker.terminate()
+        broker.wait()
+        (stdo, stde) = broker.communicate()
+        if rc:
+            print(stde.decode('utf-8'))
+            exit(rc)
+
+do_test("/"*200, True) # 200 max hierarchy limit
+do_test("abc/"*199+"d", True) # 200 max hierarchy limit, longer overall string than 200
+
+do_test("/"*201, False) # Exceeds 200 max hierarchy limit
+do_test("abc/"*200+"d", False) # Exceeds 200 max hierarchy limit, longer overall string than 200
+
+
+exit(0)
+
@@ -51,6 +51,10 @@ try:
            rc = 0

    sock.close()
+except socket.error as e:
+    if e.errno == errno.ECONNRESET:
+        # Connection has been closed by peer, this is the expected behaviour
+        rc = 0
 finally:
    broker.terminate()
    broker.wait()
@@ -47,6 +47,10 @@ try:
            rc = 0

    sock.close()
+except socket.error as e:
+    if e.errno == errno.ECONNRESET:
+        # Connection has been closed by peer, this is the expected behaviour
+        rc = 0
 finally:
    broker.terminate()
    broker.wait()
@@ -50,6 +50,10 @@ try:
            rc = 0

    sock.close()
+except socket.error as e:
+    if e.errno == errno.ECONNRESET:
+        # Connection has been closed by peer, this is the expected behaviour
+        rc = 0
 finally:
    broker.terminate()
    broker.wait()
@@ -53,6 +53,10 @@ try:
            rc = 0

    sock.close()
+except socket.error as e:
+    if e.errno == errno.ECONNRESET:
+        # Connection has been closed by peer, this is the expected behaviour
+        rc = 0
 finally:
    broker.terminate()
    broker.wait()
@@ -50,6 +50,10 @@ try:
            rc = 0

    sock.close()
+except socket.error as e:
+    if e.errno == errno.ECONNRESET:
+        # Connection has been closed by peer, this is the expected behaviour
+        rc = 0
 finally:
    broker.terminate()
    broker.wait()
@@ -38,6 +38,7 @@ try:

    if mosq_test.expect_packet(sock, "publish2", publish_packet2):
        mosq_test.do_send_receive(sock, pubrec_packet2, pubrel_packet2, "pubrel2")
+        sock.send(pubcomp_packet2)
        # Broker side of flow complete so can quit here.
        rc = 0

@@ -16,6 +16,10 @@ try:
    sock = mosq_test.do_client_connect(connect_packet, b"", timeout=30, port=port)
    rc = 0
    sock.close()
+except socket.error as e:
+    if e.errno == errno.ECONNRESET:
+        # Connection has been closed by peer, this is the expected behaviour
+        rc = 0
 finally:
    broker.terminate()
    broker.wait()
@@ -3,7 +3,6 @@
 # Test whether a client can connect without an SSL certificate if one is required.

 from mosq_test_helper import *
-import errno

 if sys.version < '2.7':
    print("WARNING: SSL not supported on Python 2.6")
@@ -44,11 +44,13 @@ ifeq ($(WITH_TLS),yes)
 else
 	./01-connect-uname-password-success-no-tls.py
 endif
+	./01-connect-zero-length-id.py


 02 :
 	./02-shared-qos0-v5.py
 	./02-subhier-crash.py
+	./02-subpub-qos0-long-topic.py
 	./02-subpub-qos0-retain-as-publish.py
 	./02-subpub-qos0-send-retain.py
 	./02-subpub-qos0-subscription-id.py
@@ -15,3 +15,4 @@ import ssl
 import struct
 import subprocess
 import time
+import errno
@@ -25,9 +25,11 @@ tests = [
    (1, './01-connect-uname-password-denied.py'),
    (1, './01-connect-uname-password-success.py'),
    (1, './01-connect-uname-pwd-no-flag.py'),
+    (2, './01-connect-zero-length-id.py'),

    (1, './02-shared-qos0-v5.py'),
    (1, './02-subhier-crash.py'),
+    (1, './02-subpub-qos0-long-topic.py'),
    (1, './02-subpub-qos0-retain-as-publish.py'),
    (1, './02-subpub-qos0-send-retain.py'),
    (1, './02-subpub-qos0-subscription-id.py'),
@@ -0,0 +1,58 @@
+#!/usr/bin/env python3
+
+# Check whether the v5 message callback gets the properties
+
+from mosq_test_helper import *
+
+port = mosq_test.get_lib_port()
+
+rc = 1
+keepalive = 60
+connect_packet = mosq_test.gen_connect("prop-test", keepalive=keepalive, proto_ver=5)
+connack_packet = mosq_test.gen_connack(rc=0, proto_ver=5)
+
+props = mqtt5_props.gen_string_prop(mqtt5_props.PROP_CONTENT_TYPE, "plain/text")
+props += mqtt5_props.gen_string_prop(mqtt5_props.PROP_RESPONSE_TOPIC, "msg/123")
+publish_packet = mosq_test.gen_publish("prop/test", qos=0, payload="message", proto_ver=5, properties=props)
+
+ok_packet = mosq_test.gen_publish("ok", qos=0, payload="ok", proto_ver=5)
+
+disconnect_packet = mosq_test.gen_disconnect(proto_ver=5)
+
+sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
+sock.settimeout(10)
+sock.bind(('', port))
+sock.listen(5)
+
+client_args = sys.argv[1:]
+env = dict(os.environ)
+env['LD_LIBRARY_PATH'] = '../../lib:../../lib/cpp'
+try:
+    pp = env['PYTHONPATH']
+except KeyError:
+    pp = ''
+env['PYTHONPATH'] = '../../lib/python:'+pp
+client = mosq_test.start_client(filename=sys.argv[1].replace('/', '-'), cmd=client_args, env=env, port=port)
+
+try:
+    (conn, address) = sock.accept()
+    conn.settimeout(10)
+
+    if mosq_test.expect_packet(conn, "connect", connect_packet):
+        conn.send(connack_packet)
+
+        conn.send(publish_packet)
+        if mosq_test.expect_packet(conn, "ok", ok_packet):
+            rc = 0
+
+    conn.close()
+finally:
+    client.terminate()
+    client.wait()
+    if rc:
+        (stdo, stde) = client.communicate()
+        print(stde)
+    sock.close()
+
+exit(rc)
@@ -0,0 +1,62 @@
+#!/usr/bin/env python3
+
+# Check whether the v5 message callback gets the properties
+
+from mosq_test_helper import *
+
+port = mosq_test.get_lib_port()
+
+rc = 1
+keepalive = 60
+connect_packet = mosq_test.gen_connect("prop-test", keepalive=keepalive, proto_ver=5)
+connack_packet = mosq_test.gen_connack(rc=0, proto_ver=5)
+
+
+mid = 1
+props = mqtt5_props.gen_string_prop(mqtt5_props.PROP_CONTENT_TYPE, "plain/text")
+props += mqtt5_props.gen_string_prop(mqtt5_props.PROP_RESPONSE_TOPIC, "msg/123")
+publish_packet = mosq_test.gen_publish("prop/test", mid=mid, qos=1, payload="message", proto_ver=5, properties=props)
+puback_packet = mosq_test.gen_puback(mid=mid, proto_ver=5)
+
+ok_packet = mosq_test.gen_publish("ok", qos=0, payload="ok", proto_ver=5)
+
+disconnect_packet = mosq_test.gen_disconnect(proto_ver=5)
+
+sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
+sock.settimeout(10)
+sock.bind(('', port))
+sock.listen(5)
+
+client_args = sys.argv[1:]
+env = dict(os.environ)
+env['LD_LIBRARY_PATH'] = '../../lib:../../lib/cpp'
+try:
+    pp = env['PYTHONPATH']
+except KeyError:
+    pp = ''
+env['PYTHONPATH'] = '../../lib/python:'+pp
+client = mosq_test.start_client(filename=sys.argv[1].replace('/', '-'), cmd=client_args, env=env, port=port)
+
+try:
+    (conn, address) = sock.accept()
+    conn.settimeout(10)
+
+    if mosq_test.expect_packet(conn, "connect", connect_packet):
+        conn.send(connack_packet)
+
+        conn.send(publish_packet)
+        if mosq_test.expect_packet(conn, "puback", puback_packet):
+            if mosq_test.expect_packet(conn, "ok", ok_packet):
+                rc = 0
+
+    conn.close()
+finally:
+    client.terminate()
+    client.wait()
+    if rc:
+        (stdo, stde) = client.communicate()
+        print(stde)
+    sock.close()
+
+exit(rc)
@@ -0,0 +1,66 @@
+#!/usr/bin/env python3
+
+# Check whether the v5 message callback gets the properties
+
+from mosq_test_helper import *
+
+port = mosq_test.get_lib_port()
+
+rc = 1
+keepalive = 60
+connect_packet = mosq_test.gen_connect("prop-test", keepalive=keepalive, proto_ver=5)
+connack_packet = mosq_test.gen_connack(rc=0, proto_ver=5)
+
+
+mid = 1
+props = mqtt5_props.gen_string_prop(mqtt5_props.PROP_CONTENT_TYPE, "plain/text")
+props += mqtt5_props.gen_string_prop(mqtt5_props.PROP_RESPONSE_TOPIC, "msg/123")
+publish_packet = mosq_test.gen_publish("prop/test", mid=mid, qos=2, payload="message", proto_ver=5, properties=props)
+pubrec_packet = mosq_test.gen_pubrec(mid=mid, proto_ver=5)
+pubrel_packet = mosq_test.gen_pubrel(mid=mid, proto_ver=5)
+pubcomp_packet = mosq_test.gen_pubcomp(mid=mid, proto_ver=5)
+
+ok_packet = mosq_test.gen_publish("ok", qos=0, payload="ok", proto_ver=5)
+
+disconnect_packet = mosq_test.gen_disconnect(proto_ver=5)
+
+sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
+sock.settimeout(10)
+sock.bind(('', port))
+sock.listen(5)
+
+client_args = sys.argv[1:]
+env = dict(os.environ)
+env['LD_LIBRARY_PATH'] = '../../lib:../../lib/cpp'
+try:
+    pp = env['PYTHONPATH']
+except KeyError:
+    pp = ''
+env['PYTHONPATH'] = '../../lib/python:'+pp
+client = mosq_test.start_client(filename=sys.argv[1].replace('/', '-'), cmd=client_args, env=env, port=port)
+
+try:
+    (conn, address) = sock.accept()
+    conn.settimeout(10)
+
+    if mosq_test.expect_packet(conn, "connect", connect_packet):
+        conn.send(connack_packet)
+
+        conn.send(publish_packet)
+        if mosq_test.expect_packet(conn, "pubrec", pubrec_packet):
+            conn.send(pubrel_packet)
+            if mosq_test.expect_packet(conn, "pubcomp", pubcomp_packet):
+                if mosq_test.expect_packet(conn, "ok", ok_packet):
+                    rc = 0
+
+    conn.close()
+finally:
+    client.terminate()
+    client.wait()
+    if rc:
+        (stdo, stde) = client.communicate()
+        print(stde)
+    sock.close()
+
+exit(rc)
@@ -68,6 +68,9 @@ endif
 	./11-prop-oversize-packet.py $@/11-prop-oversize-packet.test
 	./11-prop-send-content-type.py $@/11-prop-send-content-type.test
 	./11-prop-send-payload-format.py $@/11-prop-send-payload-format.test
+	./11-prop-recv-qos0.py $@/11-prop-recv-qos0.test
+	./11-prop-recv-qos1.py $@/11-prop-recv-qos1.test
+	./11-prop-recv-qos2.py $@/11-prop-recv-qos2.test

 clean : 
 	$(MAKE) -C c clean
@@ -42,33 +42,24 @@ int main(int argc, char *argv[])
 	mosquitto_connect_callback_set(mosq, on_connect);
 	mosquitto_disconnect_callback_set(mosq, on_disconnect);
 	mosquitto_subscribe_callback_set(mosq, on_subscribe);
-	printf("ok, about to call connect_async\n");

-	// this only works if loop_start is first.  with loop_start second,
-	// it fails on both 1.6.4 _and_ 1.6.5
-	// in this order, 1.6.4 works and 1.6.5 fails.
 	rc = mosquitto_loop_start(mosq);
-	printf("loop_start returned rc: %d\n", rc);
-	if (rc) {
-		printf("which is: %s\n", mosquitto_strerror(rc));
+	if(rc){
+		printf("loop_start failed: %s\n", mosquitto_strerror(rc));
+		return rc;
 	}
 	
-	// not sure which rc you want to be returned....
 	rc = mosquitto_connect_async(mosq, "localhost", port, 60);
-	printf("connect async returned rc: %d\n", rc);
-	if (rc) {
-		printf("which is: %s\n", mosquitto_strerror(rc));
+	if(rc){
+		printf("connect_async failed: %s\n", mosquitto_strerror(rc));
+		return rc;
 	}

-	printf("ok, so we can start just waiting now, loop_start will run in it's thread\n");
-	/* 10 millis to be system polite */
-	//struct timespec tv = { 0, 10e6 };
-	struct timespec tv = { 1, 0 };
+	/* 50 millis to be system polite */
+	struct timespec tv = { 0, 50e6 };
 	while(should_run){
 		nanosleep(&tv, NULL);
-		printf("...waiting...\n");
 	}
-	printf("Already exited should_run....\n");

 	mosquitto_disconnect(mosq);
 	mosquitto_loop_stop(mosq, false);
@@ -42,29 +42,22 @@ int main(int argc, char *argv[])
 	mosquitto_connect_callback_set(mosq, on_connect);
 	mosquitto_disconnect_callback_set(mosq, on_disconnect);
 	mosquitto_subscribe_callback_set(mosq, on_subscribe);
-	printf("ok, about to call connect_async\n");

 	rc = mosquitto_connect_async(mosq, "localhost", port, 60);
-	printf("connect async returned rc: %d\n", rc);
-	if (rc) {
-		printf("which is: %s\n", mosquitto_strerror(rc));
+	if(rc){
+		printf("connect_async failed: %s\n", mosquitto_strerror(rc));
 	}

 	rc = mosquitto_loop_start(mosq);
-	printf("loop_start returned rc: %d\n", rc);
-	if (rc) {
-		printf("which is: %s\n", mosquitto_strerror(rc));
+	if(rc){
+		printf("loop_start failed: %s\n", mosquitto_strerror(rc));
 	}
 	
-	printf("ok, so we can start just waiting now, loop_start will run in it's thread\n");
-	/* 10 millis to be system polite */
-	//struct timespec tv = { 0, 10e6 };
-	struct timespec tv = { 1, 0 };
+	/* 50 millis to be system polite */
+	struct timespec tv = { 0, 50e6 };
 	while(should_run){
 		nanosleep(&tv, NULL);
-		printf("...waiting...\n");
 	}
-	printf("Already exited should_run....\n");

 	mosquitto_disconnect(mosq);
 	mosquitto_loop_stop(mosq, false);
@@ -0,0 +1,81 @@
+#include <stdbool.h>
+#include <stdio.h>
+#include <stdlib.h>
+#include <string.h>
+#include <mosquitto.h>
+#include <mqtt_protocol.h>
+
+static int run = -1;
+static int sent_mid = -1;
+
+void on_connect(struct mosquitto *mosq, void *obj, int rc)
+{
+	int rc2;
+	mosquitto_property *proplist = NULL;
+
+	if(rc){
+		exit(1);
+	}
+}
+
+
+void on_message_v5(struct mosquitto *mosq, void *obj, const struct mosquitto_message *msg, const mosquitto_property *properties)
+{
+	int rc;
+	char *str;
+
+	if(properties){
+		if(mosquitto_property_read_string(properties, MQTT_PROP_CONTENT_TYPE, &str, false)){
+			rc = strcmp(str, "plain/text");
+			free(str);
+
+			if(rc == 0){
+				if(mosquitto_property_read_string(properties, MQTT_PROP_RESPONSE_TOPIC, &str, false)){
+					rc = strcmp(str, "msg/123");
+					free(str);
+
+					if(rc == 0){
+						if(msg->qos == 0){
+							mosquitto_publish(mosq, NULL, "ok", 2, "ok", 0, 0);
+							return;
+						}
+					}
+				}
+			}
+		}
+	}
+
+	/* No matching message, so quit with an error */
+	exit(1);
+}
+
+
+void on_publish(struct mosquitto *mosq, void *obj, int mid)
+{
+	run = 0;
+}
+
+int main(int argc, char *argv[])
+{
+	int rc;
+	int tmp;
+	struct mosquitto *mosq;
+
+	int port = atoi(argv[1]);
+
+	mosquitto_lib_init();
+
+	mosq = mosquitto_new("prop-test", true, NULL);
+	mosquitto_connect_callback_set(mosq, on_connect);
+	mosquitto_message_v5_callback_set(mosq, on_message_v5);
+	mosquitto_int_option(mosq, MOSQ_OPT_PROTOCOL_VERSION, MQTT_PROTOCOL_V5);
+
+	rc = mosquitto_connect(mosq, "localhost", port, 60);
+
+	while(run == -1){
+		rc = mosquitto_loop(mosq, -1, 1);
+	}
+
+	mosquitto_lib_cleanup();
+	return run;
+}
--- a/Show More
+++ b/Show More