Refactor plugin publishing to use base_msg instead of mosquitto_message_v5

Author: ralight

src/CMakeLists.txt

@@ -3,6 +3,7 @@ add_executable(mosquitto
 	../plugins/acl-file/acl_check.c
 	../plugins/acl-file/acl_parse.c
 	../lib/alias_mosq.c ../lib/alias_mosq.h
+	base_msg.c
 	bridge.c bridge_topic.c
 	broker_control.c
 	conf.c

src/Makefile

@@ -46,6 +46,7 @@ endif
 
 OBJS=	mosquitto.o \
 		acl_file.o \
+		base_msg.o \
 		bridge.o \
 		bridge_topic.o \
 		broker_control.o \

src/base_msg.c

@@ -0,0 +1,56 @@
+/*
+Copyright (c) 2009-2021 Roger Light <roger@atchoo.org>
+
+All rights reserved. This program and the accompanying materials
+are made available under the terms of the Eclipse Public License 2.0
+and Eclipse Distribution License v1.0 which accompany this distribution.
+
+The Eclipse Public License is available at
+   https://www.eclipse.org/legal/epl-2.0/
+and the Eclipse Distribution License is available at
+  http://www.eclipse.org/org/documents/edl-v10.php.
+
+SPDX-License-Identifier: EPL-2.0 OR BSD-3-Clause
+
+Contributors:
+   Roger Light - initial implementation and documentation.
+*/
+
+#include "mosquitto_broker_internal.h"
+
+/* This is using the uthash internal hh.prev/hh.next pointers to make our own
+ * double linked list. This function must not be used on a base_msg that is
+ * stored in a hash table. */
+void base_msg__dl_append(struct mosquitto__base_msg **head, struct mosquitto__base_msg *add_msg)
+{
+	if(*head){
+		add_msg->hh.prev = db.plugin_msgs->hh.prev;
+		((struct mosquitto__base_msg *)(*head)->hh.prev)->hh.next = add_msg;
+		(*head)->hh.prev = add_msg;
+		add_msg->hh.next = NULL;
+	} else {
+		(*head) = add_msg;
+		(*head)->hh.prev = (*head);
+		(*head)->hh.next = NULL;
+	}
+}
+
+/* This is using the uthash internal hh.prev/hh.next pointers to make our own
+ * double linked list. This function must not be used on a base_msg that is
+ * stored in a hash table. */
+void base_msg__dl_delete(struct mosquitto__base_msg **head, struct mosquitto__base_msg *del_msg)
+{
+	if(del_msg->hh.prev == del_msg){
+		*head = NULL;
+	}else if(del_msg == *head){
+		((struct mosquitto__base_msg *)del_msg->hh.next)->hh.prev = del_msg->hh.prev;
+		*head = del_msg->hh.next;
+	}else{
+		((struct mosquitto__base_msg *)del_msg->hh.prev)->hh.next = del_msg->hh.next;
+		if(del_msg->hh.next){
+			((struct mosquitto__base_msg *)del_msg->hh.next)->hh.prev = del_msg->hh.prev;
+		}else{
+			(*head)->hh.prev = del_msg->hh.prev;
+		}
+	}
+}

src/loop.c

@@ -66,44 +66,20 @@ static struct lws_sorted_usec_list sul;
 #endif
 
 
-static int single_publish(struct mosquitto *context, struct mosquitto__message_v5 *pub_msg, uint32_t message_expiry)
+static int single_publish(struct mosquitto *context, struct mosquitto__base_msg *base_msg, uint32_t message_expiry)
 {
-	struct mosquitto__base_msg *base_msg;
 	uint16_t mid;
 
-	base_msg = mosquitto_calloc(1, sizeof(struct mosquitto__base_msg));
-	if(base_msg == NULL){
-		return MOSQ_ERR_NOMEM;
-	}
-
-	base_msg->data.topic = pub_msg->topic;
-	pub_msg->topic = NULL;
-	base_msg->data.retain = 0;
-	base_msg->data.payloadlen = (uint32_t)pub_msg->payloadlen;
-	base_msg->data.payload = mosquitto_malloc(base_msg->data.payloadlen+1);
-	if(base_msg->data.payload == NULL){
-		db__msg_store_free(base_msg);
-		return MOSQ_ERR_NOMEM;
-	}
-	/* Ensure payload is always zero terminated, this is the reason for the extra byte above */
-	((uint8_t *)base_msg->data.payload)[base_msg->data.payloadlen] = 0;
-	memcpy(base_msg->data.payload, pub_msg->payload, base_msg->data.payloadlen);
-
-	if(pub_msg->properties){
-		base_msg->data.properties = pub_msg->properties;
-		pub_msg->properties = NULL;
-	}
-
 	if(db__message_store(context, base_msg, &message_expiry, mosq_mo_broker)){
 		return 1;
 	}
 
-	if(pub_msg->qos){
+	if(base_msg->data.qos){
 		mid = mosquitto__mid_generate(context);
 	}else{
 		mid = 0;
 	}
-	return db__message_insert_outgoing(context, 0, mid, (uint8_t)pub_msg->qos, 0, base_msg, 0, true, true);
+	return db__message_insert_outgoing(context, 0, mid, base_msg->data.qos, 0, base_msg, 0, true, true);
 }
 
 
@@ -138,28 +114,34 @@ static void read_message_expiry_interval(mosquitto_property **proplist, uint32_t
 
 static void queue_plugin_msgs(void)
 {
-	struct mosquitto__message_v5 *msg, *tmp;
+	struct mosquitto__base_msg *base_msg, *base_tmp;
 	struct mosquitto *context;
 	uint32_t message_expiry;
 
-	DL_FOREACH_SAFE(db.plugin_msgs, msg, tmp){
-		DL_DELETE(db.plugin_msgs, msg);
+	for(base_msg = db.plugin_msgs; base_msg && (base_tmp = base_msg->hh.next, 1); base_msg = base_tmp){
+		base_msg__dl_delete(&db.plugin_msgs, base_msg);
 
-		read_message_expiry_interval(&msg->properties, &message_expiry);
+		read_message_expiry_interval(&base_msg->data.properties, &message_expiry);
 
-		if(msg->clientid){
-			HASH_FIND(hh_id, db.contexts_by_id, msg->clientid, strlen(msg->clientid), context);
+		if(base_msg->data.source_id){
+			HASH_FIND(hh_id, db.contexts_by_id, base_msg->data.source_id, strlen(base_msg->data.source_id), context);
+			mosquitto_FREE(base_msg->data.source_id);
 			if(context){
-				single_publish(context, msg, message_expiry);
+				single_publish(context, base_msg, message_expiry);
+			}else{
+				db__msg_store_free(base_msg);
 			}
 		}else{
-			db__messages_easy_queue(NULL, msg->topic, (uint8_t)msg->qos, (uint32_t)msg->payloadlen, msg->payload, msg->retain, message_expiry, &msg->properties);
+			db__messages_easy_queue(NULL,
+					base_msg->data.topic,
+					base_msg->data.qos,
+					base_msg->data.payloadlen,
+					base_msg->data.payload,
+					base_msg->data.retain,
+					message_expiry,
+					&base_msg->data.properties);
+			db__msg_store_free(base_msg);
 		}
-		mosquitto_FREE(msg->topic);
-		mosquitto_FREE(msg->payload);
-		mosquitto_property_free_all(&msg->properties);
-		mosquitto_FREE(msg->clientid);
-		mosquitto_FREE(msg);
 	}
 }
 
@@ -197,7 +179,6 @@ int mosquitto_main_loop(struct mosquitto__listener_sock *listensock, int listens
 
 	while(g_run){
 		retain__expiry_check();
-		queue_plugin_msgs();
 		context__free_disused();
 
 		db.next_event_ms = 86400000;
@@ -262,6 +243,7 @@ int mosquitto_main_loop(struct mosquitto__listener_sock *listensock, int listens
 			}
 		}
 #endif
+		queue_plugin_msgs();
 	}
 
 	return MOSQ_ERR_SUCCESS;

src/mosquitto_broker_internal.h

@@ -502,7 +502,7 @@ struct mosquitto_db {
 #ifdef WITH_KQUEUE
 	int kqueuefd;
 #endif
-	struct mosquitto__message_v5 *plugin_msgs;
+	struct mosquitto__base_msg *plugin_msgs;
 #ifdef WITH_TLS
 	/* tls_keylog can't be in the config struct because it is used
 	   before the config is allocated. Config probably
@@ -739,6 +739,8 @@ void db__msg_add_to_queued_stats(struct mosquitto_msg_data *msg_data, struct mos
 uint64_t db__new_msg_id(void);
 void db__expire_all_messages(struct mosquitto *context);
 void db__check_acl_of_all_messages(struct mosquitto *context);
+void base_msg__dl_append(struct mosquitto__base_msg **head, struct mosquitto__base_msg *add_msg);
+void base_msg__dl_delete(struct mosquitto__base_msg **head, struct mosquitto__base_msg *del_msg);
 
 /* ============================================================
  * Subscription functions

src/plugin_public.c

@@ -219,7 +219,7 @@ BROKER_EXPORT int mosquitto_broker_publish(
 		bool retain,
 		mosquitto_property *properties)
 {
-	struct mosquitto__message_v5 *msg;
+	struct mosquitto__base_msg *base_msg;
 
 	if(topic == NULL
 			|| payloadlen < 0
@@ -229,35 +229,31 @@ BROKER_EXPORT int mosquitto_broker_publish(
 		return MOSQ_ERR_INVAL;
 	}
 
-	msg = mosquitto_malloc(sizeof(struct mosquitto__message_v5));
-	if(msg == NULL){
+	base_msg = mosquitto_calloc(1, sizeof(struct mosquitto__base_msg));
+	if(base_msg == NULL){
 		return MOSQ_ERR_NOMEM;
 	}
 
-	msg->next = NULL;
-	msg->prev = NULL;
 	if(clientid){
-		msg->clientid = mosquitto_strdup(clientid);
-		if(msg->clientid == NULL){
-			mosquitto_FREE(msg);
+		base_msg->data.source_id = mosquitto_strdup(clientid);
+		if(base_msg->data.source_id == NULL){
+			mosquitto_FREE(base_msg);
 			return MOSQ_ERR_NOMEM;
 		}
-	}else{
-		msg->clientid = NULL;
 	}
-	msg->topic = mosquitto_strdup(topic);
-	if(msg->topic == NULL){
-		mosquitto_FREE(msg->clientid);
-		mosquitto_FREE(msg);
+	base_msg->data.topic = mosquitto_strdup(topic);
+	if(base_msg->data.topic == NULL){
+		mosquitto_FREE(base_msg->data.source_id);
+		mosquitto_FREE(base_msg);
 		return MOSQ_ERR_NOMEM;
 	}
-	msg->payloadlen = payloadlen;
-	msg->payload = payload;
-	msg->qos = qos;
-	msg->retain = retain;
-	msg->properties = properties;
+	base_msg->data.payloadlen = (uint32_t)payloadlen;
+	base_msg->data.payload = payload;
+	base_msg->data.qos = (uint8_t)qos;
+	base_msg->data.retain = retain;
+	base_msg->data.properties = properties;
 
-	DL_APPEND(db.plugin_msgs, msg);
+	base_msg__dl_append(&db.plugin_msgs, base_msg);
 
 	loop__update_next_event(1);
 	return MOSQ_ERR_SUCCESS;

test/broker/02-subpub-duplicate-sub.py

@@ -0,0 +1,71 @@
+#!/usr/bin/env python3
+
+# Test whether a client subscribed to a topic receives its own message sent to that topic.
+
+from mosq_test_helper import *
+
+def write_config(filename, port):
+    with open(filename, 'w') as f:
+        f.write(f"listener {port}\n")
+        f.write("allow_anonymous true\n")
+        f.write("allow_duplicate_messages false\n")
+
+def do_test(start_broker, proto_ver):
+    rc = 1
+    connect_packet = mosq_test.gen_connect("subpub-qos1-test", proto_ver=proto_ver)
+    connack_packet = mosq_test.gen_connack(rc=0, proto_ver=proto_ver)
+
+    mid = 1
+    subscribe1_packet = mosq_test.gen_subscribe(mid, "subpub/+/topic", 0, proto_ver=proto_ver)
+    suback1_packet = mosq_test.gen_suback(mid, 0, proto_ver=proto_ver)
+
+    mid = 2
+    subscribe2_packet = mosq_test.gen_subscribe(mid, "subpub/topic/+", 0, proto_ver=proto_ver)
+    suback2_packet = mosq_test.gen_suback(mid, 0, proto_ver=proto_ver)
+
+    publish_packet = mosq_test.gen_publish("subpub/topic/topic", qos=0, payload="message", proto_ver=proto_ver)
+
+    port = mosq_test.get_port()
+    conf_file = os.path.basename(__file__).replace('.py', '.conf')
+    write_config(conf_file, port)
+
+    if start_broker:
+        broker = mosq_test.start_broker(filename=os.path.basename(__file__), use_conf=True, port=port)
+
+    try:
+        sock = mosq_test.do_client_connect(connect_packet, connack_packet, port=port)
+
+        mosq_test.do_send_receive(sock, subscribe1_packet, suback1_packet, "suback 1")
+        mosq_test.do_send_receive(sock, subscribe2_packet, suback2_packet, "suback 2")
+
+        sock.send(publish_packet)
+        mosq_test.expect_packet(sock, "publish 1", publish_packet)
+        mosq_test.do_ping(sock)
+        rc = 0
+
+        sock.close()
+    except mosq_test.TestError:
+        pass
+    finally:
+        if start_broker:
+            broker.terminate()
+            if mosq_test.wait_for_subprocess(broker):
+                print("broker not terminated")
+                if rc == 0: rc=1
+            os.remove(conf_file)
+            if rc:
+                (stdo, stde) = broker.communicate()
+                print(stde.decode('utf-8'))
+                print("proto_ver=%d" % (proto_ver))
+                exit(rc)
+        else:
+            return rc
+
+
+def all_tests(start_broker=False):
+    rc = do_test(start_broker, proto_ver=4)
+    if rc:
+        return rc;
+
+if __name__ == '__main__':
+    all_tests(True)

test/broker/Makefile

@@ -51,6 +51,7 @@ msg_sequence_test:
 	./02-shared-qos0-v5.py
 	./02-subhier-crash.py
 	./02-subpub-b2c-topic-alias.py
+	./02-subpub-duplicate-sub.py
 	./02-subpub-qos0-long-topic.py
 	./02-subpub-qos0-oversize-payload.py
 	./02-subpub-qos0-queued-bytes.py

test/broker/test.py

@@ -32,6 +32,7 @@
     (1, './02-shared-qos0-v5.py'),
     (1, './02-subhier-crash.py'),
     (1, './02-subpub-b2c-topic-alias.py'),
+	(1, './02-subpub-duplicate-sub.py'),
     (1, './02-subpub-qos0-long-topic.py'),
     (1, './02-subpub-qos0-oversize-payload.py'),
     (1, './02-subpub-qos0-queued-bytes.py'),