Fix failure to close thread in some situations.

Closes #2545. Thanks to p-luke.

ChangeLog.txt

@@ -36,6 +36,7 @@ Client library:
 - Fix documentation omission around mosquitto_reinitialise. Closes #2489.
 - Fix use of MOSQ_OPT_SSL_CTX when used in conjunction with
   MOSQ_OPT_SSL_CTX_DEFAULTS. Closes #2463.
+- Fix failure to close thread in some situations. Closes #2545.
 
 Clients:
 - Fix mosquitto_pub incorrectly reusing topic aliases when reconnecting.

lib/connect.c

@@ -76,6 +76,7 @@ static int mosquitto__connect_init(struct mosquitto *mosq, const char *host, int
 	mosq->msgs_in.inflight_quota = mosq->msgs_in.inflight_maximum;
 	mosq->msgs_out.inflight_quota = mosq->msgs_out.inflight_maximum;
 	mosq->retain_available = 1;
+	mosquitto__set_request_disconnect(mosq, false);
 
 	return MOSQ_ERR_SUCCESS;
 }
@@ -255,6 +256,7 @@ int mosquitto_disconnect_v5(struct mosquitto *mosq, int reason_code, const mosqu
 	}
 
 	mosquitto__set_state(mosq, mosq_cs_disconnected);
+	mosquitto__set_request_disconnect(mosq, true);
 	if(mosq->sock == INVALID_SOCKET){
 		return MOSQ_ERR_NO_CONN;
 	}else{

lib/loop.c

@@ -242,7 +242,6 @@ int mosquitto_loop_forever(struct mosquitto *mosq, int timeout, int max_packets)
 	int run = 1;
 	int rc = MOSQ_ERR_SUCCESS;
 	unsigned long reconnect_delay;
-	enum mosquitto_client_state state;
 
 	if(!mosq) return MOSQ_ERR_INVAL;
 
@@ -281,8 +280,7 @@ int mosquitto_loop_forever(struct mosquitto *mosq, int timeout, int max_packets)
 			pthread_testcancel();
 #endif
 			rc = MOSQ_ERR_SUCCESS;
-			state = mosquitto__get_state(mosq);
-			if(state == mosq_cs_disconnecting || state == mosq_cs_disconnected){
+			if(mosquitto__get_request_disconnect(mosq)){
 				run = 0;
 			}else{
 				if(mosq->reconnect_delay_max > mosq->reconnect_delay){
@@ -304,8 +302,7 @@ int mosquitto_loop_forever(struct mosquitto *mosq, int timeout, int max_packets)
 				rc = interruptible_sleep(mosq, (time_t)reconnect_delay);
 				if(rc) return rc;
 
-				state = mosquitto__get_state(mosq);
-				if(state == mosq_cs_disconnecting || state == mosq_cs_disconnected){
+				if(mosquitto__get_request_disconnect(mosq)){
 					run = 0;
 				}else{
 					rc = mosquitto_reconnect(mosq);

lib/mosquitto_internal.h

@@ -339,6 +339,7 @@ struct mosquitto {
 	unsigned int reconnect_delay;
 	unsigned int reconnect_delay_max;
 	bool reconnect_exponential_backoff;
+	bool request_disconnect;
 	char threaded;
 	struct mosquitto__packet *out_packet_last;
 	mosquitto_property *connect_properties;

lib/util_mosq.c

@@ -302,3 +302,23 @@ enum mosquitto_client_state mosquitto__get_state(struct mosquitto *mosq)
 
 	return state;
 }
+
+#ifndef WITH_BROKER
+void mosquitto__set_request_disconnect(struct mosquitto *mosq, bool request_disconnect)
+{
+	pthread_mutex_lock(&mosq->state_mutex);
+	mosq->request_disconnect = request_disconnect;
+	pthread_mutex_unlock(&mosq->state_mutex);
+}
+
+bool mosquitto__get_request_disconnect(struct mosquitto *mosq)
+{
+	bool request_disconnect;
+
+	pthread_mutex_lock(&mosq->state_mutex);
+	request_disconnect = mosq->request_disconnect;
+	pthread_mutex_unlock(&mosq->state_mutex);
+
+	return request_disconnect;
+}
+#endif

lib/util_mosq.h

@@ -32,6 +32,10 @@ uint16_t mosquitto__mid_generate(struct mosquitto *mosq);
 
 int mosquitto__set_state(struct mosquitto *mosq, enum mosquitto_client_state state);
 enum mosquitto_client_state mosquitto__get_state(struct mosquitto *mosq);
+#ifndef WITH_BROKER
+void mosquitto__set_request_disconnect(struct mosquitto *mosq, bool request_disconnect);
+bool mosquitto__get_request_disconnect(struct mosquitto *mosq);
+#endif
 
 #ifdef WITH_TLS
 int mosquitto__hex2bin_sha1(const char *hex, unsigned char **bin);