Make behaviour of mosquitto_connect[_async]() consistent.

`mosquitto_connect_async()` is now consistent with `mosquitto_connect()`
when connecting to a non-existent server.

Closes #1345. Thanks to Mohammad Reza.

ChangeLog.txt

@@ -10,6 +10,9 @@ Broker:
 Client library:
 - Fix MQTT v5 subscription options being incorrectly set for MQTT v3
  subscriptions. Closes #1353.
+- Make behaviour of `mosquitto_connect_async()` consistent with
+ `mosquitto_connect()` when connecting to a non-existent server.
+ Closes #1345.
 
 Clients:
 - mosquitto_pub: fix error codes not being returned when mosquitto_pub exits.

lib/connect.c

@@ -163,7 +163,6 @@ static int mosquitto__reconnect(struct mosquitto *mosq, bool blocking, const mos
  const mosquitto_property *outgoing_properties = NULL;
  mosquitto_property local_property;
  int rc;
- struct mosquitto__packet *packet;
  if(!mosq) return MOSQ_ERR_INVAL;
  if(!mosq->host || mosq->port <= 0) return MOSQ_ERR_INVAL;
  if(mosq->protocol != mosq_p_mqtt5 && properties) return MOSQ_ERR_NOT_SUPPORTED;
@@ -201,27 +200,7 @@ static int mosquitto__reconnect(struct mosquitto *mosq, bool blocking, const mos
 
  packet__cleanup(&mosq->in_packet);
 
- pthread_mutex_lock(&mosq->current_out_packet_mutex);
- pthread_mutex_lock(&mosq->out_packet_mutex);
-
- if(mosq->out_packet && !mosq->current_out_packet){
- mosq->current_out_packet = mosq->out_packet;
- mosq->out_packet = mosq->out_packet->next;
- }
-
- while(mosq->current_out_packet){
- packet = mosq->current_out_packet;
- /* Free data and reset values */
- mosq->current_out_packet = mosq->out_packet;
- if(mosq->out_packet){
- mosq->out_packet = mosq->out_packet->next;
- }
-
- packet__cleanup(packet);
- mosquitto__free(packet);
- }
- pthread_mutex_unlock(&mosq->out_packet_mutex);
- pthread_mutex_unlock(&mosq->current_out_packet_mutex);
+ packet__cleanup_all(mosq);
 
  message__reconnect_reset(mosq);
 
@@ -250,7 +229,15 @@ static int mosquitto__reconnect(struct mosquitto *mosq, bool blocking, const mos
  }else
 #endif
  {
- return send__connect(mosq, mosq->keepalive, mosq->clean_start, outgoing_properties);
+ rc = send__connect(mosq, mosq->keepalive, mosq->clean_start, outgoing_properties);
+ if(rc){
+ packet__cleanup_all(mosq);
+ net__socket_close(mosq);
+ pthread_mutex_lock(&mosq->state_mutex);
+ mosq->state = mosq_cs_new;
+ pthread_mutex_unlock(&mosq->state_mutex);
+ }
+ return rc;
  }
 }
 

lib/packet_mosq.c

@@ -98,6 +98,38 @@ void packet__cleanup(struct mosquitto__packet *packet)
  packet->pos = 0;
 }
 
+
+void packet__cleanup_all(struct mosquitto *mosq)
+{
+ struct mosquitto__packet *packet;
+
+ pthread_mutex_lock(&mosq->current_out_packet_mutex);
+ pthread_mutex_lock(&mosq->out_packet_mutex);
+
+ /* Out packet cleanup */
+ if(mosq->out_packet && !mosq->current_out_packet){
+ mosq->current_out_packet = mosq->out_packet;
+ mosq->out_packet = mosq->out_packet->next;
+ }
+ while(mosq->current_out_packet){
+ packet = mosq->current_out_packet;
+ /* Free data and reset values */
+ mosq->current_out_packet = mosq->out_packet;
+ if(mosq->out_packet){
+ mosq->out_packet = mosq->out_packet->next;
+ }
+
+ packet__cleanup(packet);
+ mosquitto__free(packet);
+ }
+
+ packet__cleanup(&mosq->in_packet);
+
+ pthread_mutex_unlock(&mosq->out_packet_mutex);
+ pthread_mutex_unlock(&mosq->current_out_packet_mutex);
+}
+
+
 int packet__queue(struct mosquitto *mosq, struct mosquitto__packet *packet)
 {
 #ifndef WITH_BROKER

lib/packet_mosq.h

@@ -25,6 +25,7 @@ struct mosquitto_db;
 
 int packet__alloc(struct mosquitto__packet *packet);
 void packet__cleanup(struct mosquitto__packet *packet);
+void packet__cleanup_all(struct mosquitto *mosq);
 int packet__queue(struct mosquitto *mosq, struct mosquitto__packet *packet);
 
 int packet__check_oversize(struct mosquitto *mosq, uint32_t remaining_length);