feat: Add config option for payload on bridge notification

man/mosquitto.conf.5.xml

@@ -2363,6 +2363,24 @@ openssl dhparam -out dhparam.pem 2048</programlisting>
 						$SYS/broker/connection/&lt;remote_clientid&gt;/state.</para>
 				</listitem>
 			</varlistentry>
+			<varlistentry>
+				<term><option>notification_payload_up</option> <replaceable>value</replaceable></term>
+				<listitem>
+					<para>Choose the payload for "up" bridge notifications. This
+						payload is used for notifications sent to local and
+						remote brokers. Empty values are not valid. Defaults to
+						<replaceable>1</replaceable>.</para>
+				</listitem>
+			</varlistentry>
+			<varlistentry>
+				<term><option>notification_payload_down</option> <replaceable>value</replaceable></term>
+				<listitem>
+					<para>Choose the payload for "down" bridge notifications.
+						This payload is also used when setting the bridge Last
+						Will and Testament (LWT). Empty values are not valid.
+						Defaults to <replaceable>0</replaceable>.</para>
+				</listitem>
+			</varlistentry>
 			<varlistentry>
 				<term><option>remote_clientid</option> <replaceable>id</replaceable></term>
 				<listitem>

mosquitto.conf

@@ -876,6 +876,16 @@
 # $SYS/broker/connection/<clientid>/state
 #notification_topic
 
+# Set the payload that will be used for "up" bridge notifications.
+# This applies to notifications sent to local/remote brokers.
+# Defaults to 1.
+#notification_payload_up 1
+
+# Set the payload that will be used for "down" bridge notifications and bridge
+# LWT registration.
+# Defaults to 0.
+#notification_payload_down 0
+
 # Set the client id to use on the remote end of this bridge connection. If not
 # defined, this defaults to 'name.hostname' where name is the connection name
 # and hostname is the hostname of this computer.

src/bridge.c

@@ -253,7 +253,8 @@ static int bridge__connect_step1(struct mosquitto *context)
 	int rc;
 	char *notification_topic;
 	size_t notification_topic_len;
-	uint8_t notification_payload;
+	const char *notification_payload;
+	size_t notification_payload_len;
 	struct mosquitto__bridge_topic *cur_topic;
 	uint8_t qos;
 
@@ -302,19 +303,25 @@ static int bridge__connect_step1(struct mosquitto *context)
 	}
 
 	if(context->bridge->notifications){
+		if(context->bridge->notification_payload_down){
+			notification_payload = context->bridge->notification_payload_down;
+			notification_payload_len = strlen(notification_payload);
+		}else{
+			notification_payload = "0";
+			notification_payload_len = 1;
+		}
+
 		if(context->max_qos == 0){
 			qos = 0;
 		}else{
 			qos = 1;
 		}
 		if(context->bridge->notification_topic){
 			if(!context->bridge->initial_notification_done){
-				notification_payload = '0';
-				db__messages_easy_queue(context, context->bridge->notification_topic, qos, 1, &notification_payload, 1, MSG_EXPIRY_INFINITE, NULL);
+				db__messages_easy_queue(context, context->bridge->notification_topic, qos, (uint32_t)notification_payload_len, notification_payload, 1, MSG_EXPIRY_INFINITE, NULL);
 				context->bridge->initial_notification_done = true;
 			}
-			notification_payload = '0';
-			rc = will__set(context, context->bridge->notification_topic, 1, &notification_payload, qos, true, NULL);
+			rc = will__set(context, context->bridge->notification_topic, (int)notification_payload_len, notification_payload, qos, true, NULL);
 			if(rc != MOSQ_ERR_SUCCESS){
 				return rc;
 			}
@@ -328,13 +335,11 @@ static int bridge__connect_step1(struct mosquitto *context)
 			snprintf(notification_topic, notification_topic_len+1, "$SYS/broker/connection/%s/state", context->bridge->remote_clientid);
 
 			if(!context->bridge->initial_notification_done){
-				notification_payload = '0';
-				db__messages_easy_queue(context, notification_topic, qos, 1, &notification_payload, 1, MSG_EXPIRY_INFINITE, NULL);
+				db__messages_easy_queue(context, notification_topic, qos, (uint32_t)notification_payload_len, notification_payload, 1, MSG_EXPIRY_INFINITE, NULL);
 				context->bridge->initial_notification_done = true;
 			}
 
-			notification_payload = '0';
-			rc = will__set(context, notification_topic, 1, &notification_payload, qos, true, NULL);
+			rc = will__set(context, notification_topic, (int)notification_payload_len, notification_payload, qos, true, NULL);
 			mosquitto_FREE(notification_topic);
 			if(rc != MOSQ_ERR_SUCCESS){
 				return rc;
@@ -484,7 +489,8 @@ int bridge__connect(struct mosquitto *context)
 	int rc, rc2;
 	char *notification_topic = NULL;
 	size_t notification_topic_len;
-	uint8_t notification_payload;
+	const char *notification_payload;
+	size_t notification_payload_len;
 	struct mosquitto__bridge_topic *cur_topic;
 	uint8_t qos;
 
@@ -539,20 +545,26 @@ int bridge__connect(struct mosquitto *context)
 	}
 
 	if(context->bridge->notifications){
+		if(context->bridge->notification_payload_down){
+			notification_payload = context->bridge->notification_payload_down;
+			notification_payload_len = strlen(notification_payload);
+		}else{
+			notification_payload = "0";
+			notification_payload_len = 1;
+		}
+
 		if(context->max_qos == 0){
 			qos = 0;
 		}else{
 			qos = 1;
 		}
 		if(context->bridge->notification_topic){
 			if(!context->bridge->initial_notification_done){
-				notification_payload = '0';
-				db__messages_easy_queue(context, context->bridge->notification_topic, qos, 1, &notification_payload, 1, MSG_EXPIRY_INFINITE, NULL);
+				db__messages_easy_queue(context, context->bridge->notification_topic, qos, (uint32_t)notification_payload_len, notification_payload, 1, MSG_EXPIRY_INFINITE, NULL);
 				context->bridge->initial_notification_done = true;
 			}
 
-			notification_payload = '0';
-			rc = will__set(context, context->bridge->notification_topic, 1, &notification_payload, qos, true, NULL);
+			rc = will__set(context, context->bridge->notification_topic, (int)notification_payload_len, notification_payload, qos, true, NULL);
 			if(rc != MOSQ_ERR_SUCCESS){
 				return rc;
 			}
@@ -566,13 +578,11 @@ int bridge__connect(struct mosquitto *context)
 			snprintf(notification_topic, notification_topic_len+1, "$SYS/broker/connection/%s/state", context->bridge->remote_clientid);
 
 			if(!context->bridge->initial_notification_done){
-				notification_payload = '0';
-				db__messages_easy_queue(context, notification_topic, qos, 1, &notification_payload, 1, MSG_EXPIRY_INFINITE, NULL);
+				db__messages_easy_queue(context, notification_topic, qos, (uint32_t)notification_payload_len, notification_payload, 1, MSG_EXPIRY_INFINITE, NULL);
 				context->bridge->initial_notification_done = true;
 			}
 
-			notification_payload = '0';
-			rc = will__set(context, notification_topic, 1, &notification_payload, qos, true, NULL);
+			rc = will__set(context, notification_topic, (int)notification_payload_len, notification_payload, qos, true, NULL);
 			if(rc != MOSQ_ERR_SUCCESS){
 				mosquitto_FREE(notification_topic);
 				return rc;
@@ -666,12 +676,22 @@ int bridge__on_connect(struct mosquitto *context)
 {
 	char *notification_topic;
 	size_t notification_topic_len;
+	const char *notification_payload;
+	size_t notification_payload_len;
 	struct mosquitto__bridge_topic *cur_topic;
 	int sub_opts;
 	bool retain = true;
 	uint8_t qos;
 
 	if(context->bridge->notifications){
+		if(context->bridge->notification_payload_up){
+			notification_payload = context->bridge->notification_payload_up;
+			notification_payload_len = strlen(notification_payload);
+		}else{
+			notification_payload = "1";
+			notification_payload_len = 1;
+		}
+
 		if(context->max_qos == 0){
 			qos = 0;
 		}else{
@@ -680,16 +700,15 @@ int bridge__on_connect(struct mosquitto *context)
 		if(!context->retain_available){
 			retain = false;
 		}
-		char notification_payload = '1';
 		if(context->bridge->notification_topic){
 			if(!context->bridge->notifications_local_only){
 				if(send__real_publish(context, mosquitto__mid_generate(context),
-						context->bridge->notification_topic, 1, &notification_payload, qos, retain, 0, 0, NULL, 0)){
+						context->bridge->notification_topic, (uint32_t)notification_payload_len, notification_payload, qos, retain, 0, 0, NULL, 0)){
 
 					return 1;
 				}
 			}
-			db__messages_easy_queue(context, context->bridge->notification_topic, qos, 1, &notification_payload, 1, MSG_EXPIRY_INFINITE, NULL);
+			db__messages_easy_queue(context, context->bridge->notification_topic, qos, (uint32_t)notification_payload_len, notification_payload, 1, MSG_EXPIRY_INFINITE, NULL);
 		}else{
 			notification_topic_len = strlen(context->bridge->remote_clientid)+strlen("$SYS/broker/connection//state");
 			notification_topic = mosquitto_malloc(sizeof(char)*(notification_topic_len+1));
@@ -698,16 +717,15 @@ int bridge__on_connect(struct mosquitto *context)
 			}
 
 			snprintf(notification_topic, notification_topic_len+1, "$SYS/broker/connection/%s/state", context->bridge->remote_clientid);
-			notification_payload = '1';
 			if(!context->bridge->notifications_local_only){
 				if(send__real_publish(context, mosquitto__mid_generate(context),
-						notification_topic, 1, &notification_payload, qos, retain, 0, 0, NULL, 0)){
+						notification_topic, (uint32_t)notification_payload_len, notification_payload, qos, retain, 0, 0, NULL, 0)){
 
 					mosquitto_FREE(notification_topic);
 					return 1;
 				}
 			}
-			db__messages_easy_queue(context, notification_topic, qos, 1, &notification_payload, 1, MSG_EXPIRY_INFINITE, NULL);
+			db__messages_easy_queue(context, notification_topic, qos, (uint32_t)notification_payload_len, notification_payload, 1, MSG_EXPIRY_INFINITE, NULL);
 			mosquitto_FREE(notification_topic);
 		}
 	}

src/conf.c

@@ -516,6 +516,8 @@ void config__bridge_cleanup(struct mosquitto__bridge *bridge)
 		mosquitto_FREE(bridge->topics);
 	}
 	mosquitto_FREE(bridge->notification_topic);
+	mosquitto_FREE(bridge->notification_payload_up);
+	mosquitto_FREE(bridge->notification_payload_down);
 #ifdef WITH_TLS
 	mosquitto_FREE(bridge->tls_certfile);
 	mosquitto_FREE(bridge->tls_keyfile);
@@ -2327,6 +2329,24 @@ static int config__read_file_core(struct mosquitto__config *config, bool reload,
 					}
 #else
 					log__printf(NULL, MOSQ_LOG_WARNING, "Warning: Bridge support not available.");
+#endif
+				}else if(!strcmp(token, "notification_payload_up")){
+#ifdef WITH_BRIDGE
+					REQUIRE_BRIDGE(token);
+					if(conf__parse_string(&token, "notification_payload_up", &cur_bridge->notification_payload_up, &saveptr)){
+						return MOSQ_ERR_INVAL;
+					}
+#else
+					log__printf(NULL, MOSQ_LOG_WARNING, "Warning: Bridge support not available.");
+#endif
+				}else if(!strcmp(token, "notification_payload_down")){
+#ifdef WITH_BRIDGE
+					REQUIRE_BRIDGE(token);
+					if(conf__parse_string(&token, "notification_payload_down", &cur_bridge->notification_payload_down, &saveptr)){
+						return MOSQ_ERR_INVAL;
+					}
+#else
+					log__printf(NULL, MOSQ_LOG_WARNING, "Warning: Bridge support not available.");
 #endif
 				}else if(!strcmp(token, "password") || !strcmp(token, "remote_password")){
 #ifdef WITH_BRIDGE

src/mosquitto_broker_internal.h

@@ -578,6 +578,8 @@ struct mosquitto__bridge {
 	char *local_username;
 	char *local_password;
 	char *notification_topic;
+	char *notification_payload_up;
+	char *notification_payload_down;
 	char *bind_address;
 	bool notifications;
 	bool notifications_local_only;

test/broker/06-bridge-notification-payload.py

@@ -0,0 +1,74 @@
+#!/usr/bin/env python3
+
+# Check that bridge notification payloads can be configured, including spaces.
+
+from mosq_test_helper import *
+
+from broker_config import BrokerConfig, ListenerConfig, MQTTBridgeConfig
+from mosquitto_broker import MosquittoBroker
+
+mosq_test.require_features(["INC_BRIDGE_SUPPORT"])
+
+
+def do_test():
+    hostname = socket.gethostname()
+    client_id = hostname+".bridge_sample"
+
+    payload_up = "bridge is up"
+    payload_down = "bridge is down"
+    topic = f"$SYS/broker/connection/{client_id}/state"
+
+    connect_packet = mqtt_packets.gen_connect(
+        client_id,
+        clean_session=False,
+        proto_ver=5,
+        will_topic=topic,
+        will_payload=payload_down.encode(),
+        will_qos=1,
+        will_retain=True,
+    )
+    connack_packet = mqtt_packets.gen_connack(rc=0, proto_ver=5)
+
+    publish_packet = mqtt_packets.gen_publish(
+        topic,
+        qos=1,
+        mid=1,
+        payload=payload_up,
+        retain=True,
+        proto_ver=5,
+    )
+    puback_packet = mqtt_packets.gen_puback(1, proto_ver=5)
+
+    (port1, port2) = mosq_test.get_port(2)
+    ssock = mosq_test.listen_sock(port1)
+
+    broker_config = BrokerConfig(
+        listeners=[ListenerConfig(port=port2)],
+        bridges=[
+            MQTTBridgeConfig(
+                connection="bridge_sample",
+                address=f"localhost:{port1}",
+                bridge_protocol_version="mqttv50",
+                bridge_max_topic_alias=0,
+                notification_payload_up=payload_up,
+                notification_payload_down=payload_down,
+                topics=["\"bridge with space/#\" both 1"],
+            ),
+        ],
+        allow_anonymous=True,
+    )
+
+    broker = MosquittoBroker(config=broker_config)
+    with broker:
+        (bridge, _) = ssock.accept()
+        bridge.settimeout(20)
+        mosq_test.expect_packet(bridge, "connect", connect_packet)
+        bridge.send(connack_packet)
+
+        mosq_test.expect_packet(bridge, "publish", publish_packet)
+        bridge.send(puback_packet)
+        bridge.close()
+
+
+do_test()
+

test/broker/16-config-huge.py

@@ -97,6 +97,8 @@ def write_config(filename, ports, per_listener_settings, plugver, acl_file):
             f.write("idle_timeout 60\n")
             f.write("keepalive_interval 40\n")
             f.write("notification_topic notifications\n")
+            f.write("notification_payload_up bridge is up\n")
+            f.write("notification_payload_down bridge is down\n")
             f.write("notifications false\n")
             f.write("notifications_local_only true\n")
             f.write("remote_clientid brci\n")

test/broker/CMakeLists.txt

@@ -115,6 +115,7 @@ add_python_test(PY_TEST_NAMES ${PREFIX} 2 "06-bridge-connack-v5-error.py")
 add_python_test(PY_TEST_NAMES ${PREFIX} 2 "06-bridge-fail-persist-resend-qos1.py")
 add_python_test(PY_TEST_NAMES ${PREFIX} 2 "06-bridge-fail-persist-resend-qos2.py")
 add_python_test(PY_TEST_NAMES ${PREFIX} 1 "06-bridge-no-local.py")
+add_python_test(PY_TEST_NAMES ${PREFIX} 2 "06-bridge-notification-payload.py")
 add_python_test(PY_TEST_NAMES ${PREFIX} 2 "06-bridge-outgoing-retain.py")
 add_python_test(PY_TEST_NAMES ${PREFIX} 2 "06-bridge-per-listener-settings.py")
 add_python_test(PY_TEST_NAMES ${PREFIX} 2 "06-bridge-reconnect-local-out.py")

test/broker/Makefile

@@ -121,6 +121,7 @@ msg_sequence_test:
 	./06-bridge-fail-persist-resend-qos1.py
 	./06-bridge-fail-persist-resend-qos2.py
 	./06-bridge-no-local.py
+	./06-bridge-notification-payload.py
 	./06-bridge-outgoing-retain.py
 	./06-bridge-per-listener-settings.py
 	./06-bridge-reconnect-local-out.py

test/broker_config.py

@@ -75,6 +75,8 @@ class MQTTBridgeConfig:
     keepalive_interval: Optional[int] = None
     local_cleansession: bool = False
     notifications: bool = True
+    notification_payload_up: Optional[str] = None
+    notification_payload_down: Optional[str] = None
     remote_clientid: Optional[str] = None
     restart_timeout: Optional[int] = None
     try_private: Optional[bool] = None