Make bridge connections non-blocking for non-TLS connections.

lib/net_mosq.c

@@ -274,7 +274,7 @@ int _mosquitto_try_connect(const char *host, uint16_t port, int *sock, const cha
 	struct addrinfo *ainfo, *rp;
 	struct addrinfo *ainfo_bind, *rp_bind;
 	int s;
-	int rc;
+	int rc = MOSQ_ERR_SUCCESS;
 #ifdef WIN32
 	uint32_t val = 1;
 #endif
@@ -338,6 +338,10 @@ int _mosquitto_try_connect(const char *host, uint16_t port, int *sock, const cha
 		errno = WSAGetLastError();
 #endif
 		if(rc == 0 || errno == EINPROGRESS || errno == COMPAT_EWOULDBLOCK){
+			if(rc < 0 && (errno == EINPROGRESS || errno == COMPAT_EWOULDBLOCK)){
+				rc = MOSQ_ERR_CONN_PENDING;
+			}
+
 			if(blocking){
 				/* Set non-blocking */
 				if(_mosquitto_socket_nonblock(*sock)){
@@ -358,7 +362,7 @@ int _mosquitto_try_connect(const char *host, uint16_t port, int *sock, const cha
 	if(!rp){
 		return MOSQ_ERR_ERRNO;
 	}
-	return MOSQ_ERR_SUCCESS;
+	return rc;
 }
 
 /* Create a socket and connect it to 'ip' on port 'port'.
@@ -383,7 +387,7 @@ int _mosquitto_socket_connect(struct mosquitto *mosq, const char *host, uint16_t
 #endif
 
 	rc = _mosquitto_try_connect(host, port, &sock, bind_address, blocking);
-	if(rc != MOSQ_ERR_SUCCESS) return rc;
+	if(rc > 0) return rc;
 
 #ifdef WITH_TLS
 	if(mosq->tls_cafile || mosq->tls_capath || mosq->tls_psk){
@@ -531,7 +535,7 @@ int _mosquitto_socket_connect(struct mosquitto *mosq, const char *host, uint16_t
 
 	mosq->sock = sock;
 
-	return MOSQ_ERR_SUCCESS;
+	return rc;
 }
 
 int _mosquitto_read_byte(struct _mosquitto_packet *packet, uint8_t *byte)
@@ -741,6 +745,10 @@ int _mosquitto_packet_write(struct mosquitto *mosq)
 	}
 	pthread_mutex_unlock(&mosq->out_packet_mutex);
 
+	if(mosq->state == mosq_cs_connect_pending){
+		return MOSQ_ERR_SUCCESS;
+	}
+
 	while(mosq->current_out_packet){
 		packet = mosq->current_out_packet;
 
@@ -860,6 +868,10 @@ int _mosquitto_packet_read(struct mosquitto *mosq)
 
 	if(!mosq) return MOSQ_ERR_INVAL;
 	if(mosq->sock == INVALID_SOCKET) return MOSQ_ERR_NO_CONN;
+	if(mosq->state == mosq_cs_connect_pending){
+		return MOSQ_ERR_SUCCESS;
+	}
+
 	/* This gets called if pselect() indicates that there is network data
 	 * available - ie. at least one byte.  What we do depends on what data we
 	 * already have.

lib/net_mosq.h

@@ -40,10 +40,6 @@ struct mosquitto_db;
 #  define COMPAT_EWOULDBLOCK EWOULDBLOCK
 #endif
 
-#ifndef WIN32
-#else
-#endif
-
 /* For when not using winsock libraries. */
 #ifndef INVALID_SOCKET
 #define INVALID_SOCKET -1

src/bridge.c

@@ -200,8 +200,8 @@ int mqtt3_bridge_connect(struct mosquitto_db *db, struct mosquitto *context)
 	}
 
 	_mosquitto_log_printf(NULL, MOSQ_LOG_NOTICE, "Connecting bridge %s (%s:%d)", context->bridge->name, context->bridge->addresses[context->bridge->cur_address].address, context->bridge->addresses[context->bridge->cur_address].port);
-	rc = _mosquitto_socket_connect(context, context->bridge->addresses[context->bridge->cur_address].address, context->bridge->addresses[context->bridge->cur_address].port, NULL, true);
-	if(rc != MOSQ_ERR_SUCCESS){
+	rc = _mosquitto_socket_connect(context, context->bridge->addresses[context->bridge->cur_address].address, context->bridge->addresses[context->bridge->cur_address].port, NULL, false);
+	if(rc > 0 ){
 		if(rc == MOSQ_ERR_TLS){
 			return rc; /* Error already printed */
 		}else if(rc == MOSQ_ERR_ERRNO){
@@ -214,6 +214,10 @@ int mqtt3_bridge_connect(struct mosquitto_db *db, struct mosquitto *context)
 	}
 
 	HASH_ADD(hh_sock, db->contexts_by_sock, sock, sizeof(context->sock), context);
+
+	if(rc == MOSQ_ERR_CONN_PENDING){
+		context->state = mosq_cs_connect_pending;
+	}
 	rc = _mosquitto_send_connect(context, context->keepalive, context->clean_session);
 	if(rc == MOSQ_ERR_SUCCESS){
 		return MOSQ_ERR_SUCCESS;

src/database.c

@@ -730,6 +730,10 @@ int mqtt3_db_message_write(struct mosquitto *context)
 		return MOSQ_ERR_INVAL;
 	}
 
+	if(context->state != mosq_cs_connected){
+		return MOSQ_ERR_SUCCESS;
+	}
+
 	tail = context->msgs;
 	while(tail){
 		if(tail->direction == mosq_md_in){

src/loop.c

@@ -31,6 +31,7 @@ Contributors:
 #include <signal.h>
 #include <stdio.h>
 #include <string.h>
+#include <sys/socket.h>
 #include <time.h>
 
 #ifdef WITH_WEBSOCKETS
@@ -39,6 +40,7 @@ Contributors:
 
 #include <mosquitto_broker.h>
 #include <memory_mosq.h>
+#include <send_mosq.h>
 #include <time_mosq.h>
 #include <util_mosq.h>
 
@@ -145,7 +147,7 @@ int mosquitto_main_loop(struct mosquitto_db *db, int *listensock, int listensock
 							&& now > context->bridge->primary_retry){
 
 						/* FIXME - this should be non-blocking */
-						if(_mosquitto_try_connect(context->bridge->addresses[0].address, context->bridge->addresses[0].port, &bridge_sock, NULL, true) == MOSQ_ERR_SUCCESS){
+						if(_mosquitto_try_connect(context->bridge->addresses[0].address, context->bridge->addresses[0].port, &bridge_sock, NULL, false) == MOSQ_ERR_SUCCESS){
 							COMPAT_CLOSE(bridge_sock);
 							_mosquitto_socket_close(db, context);
 							context->bridge->cur_address = context->bridge->address_count-1;
@@ -163,7 +165,7 @@ int mosquitto_main_loop(struct mosquitto_db *db, int *listensock, int listensock
 						pollfds[pollfd_index].fd = context->sock;
 						pollfds[pollfd_index].events = POLLIN;
 						pollfds[pollfd_index].revents = 0;
-						if(context->current_out_packet){
+						if(context->current_out_packet || context->state == mosq_cs_connect_pending){
 							pollfds[pollfd_index].events |= POLLOUT;
 						}
 						context->pollfd_index = pollfd_index;
@@ -398,6 +400,7 @@ static void loop_handle_errors(struct mosquitto_db *db, struct pollfd *pollfds)
 static void loop_handle_reads_writes(struct mosquitto_db *db, struct pollfd *pollfds)
 {
 	struct mosquitto *context, *ctxt_tmp;
+	int err, len;
 
 	HASH_ITER(hh_sock, db->contexts_by_sock, context, ctxt_tmp){
 		if(context->pollfd_index < 0){
@@ -412,6 +415,17 @@ static void loop_handle_reads_writes(struct mosquitto_db *db, struct pollfd *pol
 #else
 		if(pollfds[context->pollfd_index].revents & POLLOUT){
 #endif
+			if(context->state == mosq_cs_connect_pending){
+				len = sizeof(int);
+				if(!getsockopt(context->sock, SOL_SOCKET, SO_ERROR, &err, &len)){
+					if(err == 0){
+						context->state = mosq_cs_new;
+					}
+				}else{
+					do_disconnect(db, context);
+					continue;
+				}
+			}
 			if(_mosquitto_packet_write(context)){
 				do_disconnect(db, context);
 				continue;