[456899] Broker: Add bridge_attempt_unsubscribe option.

The bridge_attempt_unsubscribe option has been added, to allow the sending
of UNSUBSCRIBE requests to be disabled for topics with "out" direction.

Bug: https://bugs.eclipse.org/bugs/show_bug.cgi?id=456899

ChangeLog.txt

@@ -48,6 +48,9 @@ Broker:
   and has been fixed so the broker no longer asks.
 - When using syslog logging on non-Windows OSs, it is now possible to specify
   the logging facility to one of local0-7 instead of the default "daemon".
+- The bridge_attempt_unsubscribe option has been added, to allow the sending
+  of UNSUBSCRIBE requests to be disabled for topics with "out" direction.
+  Closes bug #456899.
 
 Clients:
 - Both clients can now load default configuration options from a file.

man/mosquitto.conf.5.xml

@@ -1220,6 +1220,21 @@ topic clients/total in 0 test/mosquitto/org $SYS/broker/
 			<para>The following options are available for all bridges to
 				configure SSL/TLS support.</para>
 			<variablelist>
+				<varlistentry>
+					<term><option>bridge_attempt_unsubscribe</option> [ true | false ]</term>
+					<listitem>
+						<para>If a bridge has topics that have "out" direction,
+							the default behaviour is to send an unsubscribe
+							request to the remote broker on that topic. This
+							means that changing a topic direction from "in" to
+							"out" will not keep receiving incoming messages.
+							Sending these unsubscribe requests is not always
+							desirable, setting
+							<option>bridge_attempt_unsubscribe</option> to
+							<replaceable>false</replaceable> will disable
+							sending the unsubscribe request.</para>
+					</listitem>
+				</varlistentry>
 				<varlistentry>
 					<term><option>bridge_cafile</option> <replaceable>file path</replaceable></term>
 					<listitem>

mosquitto.conf

@@ -631,6 +631,14 @@
 #address <host>[:<port>] [<host>[:<port>]]
 #topic <topic> [[[out | in | both] qos-level] local-prefix remote-prefix]
 
+# If a bridge has topics that have "out" direction, the default behaviour is to
+# send an unsubscribe request to the remote broker on that topic. This means
+# that changing a topic direction from "in" to "out" will not keep receiving
+# incoming messages. Sending these unsubscribe requests is not always
+# desirable, setting bridge_attempt_unsubscribe to false will disable sending
+# the unsubscribe request.
+#bridge_attempt_unsubscribe true
+
 # If the bridge has more than one address given in the address/addresses
 # configuration, the round_robin option defines the behaviour of the bridge on
 # a failure of the bridge connection. If round_robin is false, the default

src/conf.c

@@ -681,6 +681,17 @@ int _config_read_file_core(struct mqtt3_config *config, bool reload, const char
 					if(_conf_attempt_resolve(config->default_listener.host, "bind_address", MOSQ_LOG_ERR, "Error")){
 						return MOSQ_ERR_INVAL;
 					}
+				}else if(!strcmp(token, "bridge_attempt_unsubscribe")){
+#ifdef WITH_BRIDGE
+					if(reload) continue; // FIXME
+					if(!cur_bridge){
+						_mosquitto_log_printf(NULL, MOSQ_LOG_ERR, "Error: Invalid bridge configuration.");
+						return MOSQ_ERR_INVAL;
+					}
+					if(_conf_parse_bool(&token, "bridge_attempt_unsubscribe", &cur_bridge->attempt_unsubscribe, saveptr)) return MOSQ_ERR_INVAL;
+#else
+					_mosquitto_log_printf(NULL, MOSQ_LOG_WARNING, "Warning: Bridge support not available.");
+#endif
 				}else if(!strcmp(token, "bridge_cafile")){
 #if defined(WITH_BRIDGE) && defined(WITH_TLS)
 					if(reload) continue; // FIXME
@@ -1007,6 +1018,7 @@ int _config_read_file_core(struct mqtt3_config *config, bool reload, const char
 						cur_bridge->restart_timeout = 30;
 						cur_bridge->threshold = 10;
 						cur_bridge->try_private = true;
+						cur_bridge->attempt_unsubscribe = true;
 					}else{
 						_mosquitto_log_printf(NULL, MOSQ_LOG_ERR, "Error: Empty connection value in configuration.");
 						return MOSQ_ERR_INVAL;

src/mosquitto_broker.h

@@ -300,6 +300,7 @@ struct _mqtt3_bridge{
 	int restart_timeout;
 	int threshold;
 	bool lazy_reconnect;
+	bool attempt_unsubscribe;
 #ifdef WITH_TLS
 	char *tls_cafile;
 	char *tls_capath;

src/read_handle_client.c

@@ -74,6 +74,7 @@ int mqtt3_handle_connack(struct mosquitto_db *db, struct mosquitto *context)
 							return 1;
 						}
 					}else{
+						if(context->bridge->attempt_unsubscribe){
 							if(_mosquitto_send_unsubscribe(context, NULL, context->bridge->topics[i].remote_topic)){
 								/* direction = inwards only. This means we should not be subscribed
 								* to the topic. It is possible that we used to be subscribed to
@@ -83,6 +84,7 @@ int mqtt3_handle_connack(struct mosquitto_db *db, struct mosquitto *context)
 						}
 					}
 				}
+			}
 			context->state = mosq_cs_connected;
 			return MOSQ_ERR_SUCCESS;
 		case CONNACK_REFUSED_PROTOCOL_VERSION: