Add mosquitto_plugin_publish*()

These allow plugins to publish messages on the broker that can
be consumed by other plugins in the message-in event

Author: ralight

ChangeLog.txt

@@ -149,6 +149,9 @@
 - Plugins can now use the `next_s` and `next_ms` members of the tick event data
   struct to set a minimum interval that the broker will wait before calling the
   tick callback again.
+- Add `mosquitto_plugin_publish()` and `mosquitto_plugin_publish_copy()`.
+  These allow plugins to publish messages on the broker that will also be
+  consumed by other plugins in their message-in events.
 
 # Plugins
 - Add acl-file plugin.

include/mosquitto/broker.h

@@ -804,9 +804,9 @@ mosq_EXPORT int mosquitto_apply_on_all_clients(int (*FUNC_client_functor)(const
  *
  * This function allows a plugin to publish a message. Messages published in
  * this way are treated as coming from the broker and so will not be passed to
- * `mosquitto_auth_acl_check(, MOSQ_ACL_WRITE, , )` for checking. Read access
- * will be enforced as normal for individual clients when they are due to
- * receive the message.
+ * the MOSQ_EVT_ACL_CHECK nor MOSQ_EVT_MESSAGE_IN events. Read access will be
+ * enforced as normal for individual clients when they are due to receive the
+ * message.
  *
  * It can be used to send messages to all clients that have a matching
  * subscription, or to a single client whether or not it has a matching
@@ -883,6 +883,91 @@ mosq_EXPORT int mosquitto_broker_publish_copy(
 		bool retain,
 		mosquitto_property *properties);
 
+
+/* Function: mosquitto_plugin_publish
+ *
+ * Publish a message from within a plugin.
+ *
+ * This function allows a plugin to publish a message. Messages published in
+ * this way are treated as coming from the broker and so will not be passed to
+ * the MOSQ_EVT_ACL_CHECK event. Read access will be enforced as normal for
+ * individual clients when they are due to receive the message.
+ *
+ * The important difference compared to mosquitto_broker_publish* is that
+ * messages from this function follow the same flow as messages from clients
+ * after they have been accepted. This means that messages do go through the
+ * MOSQ_EVT_MESSAGE_IN event. If your plugin generates new messages in its
+ * MOSQ_EVT_MESSAGE_IN event, it is important to ensure you do not create an
+ * infinite loop.
+ *
+ * Parameters:
+ *  topic -      message topic
+ *  payloadlen - payload length in bytes. Can be 0 for an empty payload.
+ *  payload -    payload bytes. If payloadlen > 0 this must not be NULL. Must
+ *               be allocated on the heap. Will be freed by mosquitto after use if the
+ *               function returns success.
+ *  qos -        message QoS to use.
+ *  retain -     should retain be set on the message. This does not apply if
+ *               clientid is non-NULL.
+ *  properties - MQTT v5 properties to attach to the message. If the function
+ *               returns success, then properties is owned by the broker and
+ *               will be freed at a later point.
+ *
+ * Returns:
+ *   MOSQ_ERR_SUCCESS - on success
+ *   MOSQ_ERR_INVAL - if topic is NULL, if payloadlen < 0, if payloadlen > 0
+ *                    and payload is NULL, if qos is not 0, 1, or 2.
+ *   MOSQ_ERR_NOMEM - on out of memory
+ */
+mosq_EXPORT int mosquitto_plugin_publish(
+		const char *topic,
+		int payloadlen,
+		void *payload,
+		int qos,
+		bool retain,
+		mosquitto_property *properties);
+
+
+/* Function: mosquitto_plugin_publish_copy
+ *
+ * Publish a message from within a plugin.
+ *
+ * This function is identical to mosquitto_plugin_publish, except that a copy
+ * of `payload` is taken.
+ *
+ * Parameters:
+ *  topic -      message topic
+ *  payloadlen - payload length in bytes. Can be 0 for an empty payload.
+ *  payload -    payload bytes. If payloadlen > 0 this must not be NULL.
+ *	             Memory remains the property of the calling function.
+ *  qos -        message QoS to use.
+ *  retain -     should retain be set on the message. This does not apply if
+ *               clientid is non-NULL.
+ *  properties - MQTT v5 properties to attach to the message. If the function
+ *               returns success, then properties is owned by the broker and
+ *               will be freed at a later point.
+ *
+ * Returns:
+ *   MOSQ_ERR_SUCCESS - on success
+ *   MOSQ_ERR_INVAL - if topic is NULL, if payloadlen < 0, if payloadlen > 0
+ *                    and payload is NULL, if qos is not 0, 1, or 2.
+ *   MOSQ_ERR_NOMEM - on out of memory
+ */
+mosq_EXPORT int mosquitto_plugin_publish_copy(
+		const char *topic,
+		int payloadlen,
+		const void *payload,
+		int qos,
+		bool retain,
+		mosquitto_property *properties);
+
+
+/* =========================================================================
+ *
+ * Section: Authentication functions
+ *
+ * ========================================================================= */
+
 /* Function: mosquitto_complete_basic_auth
  *
  * Complete a delayed authentication request.

src/database.c

@@ -40,6 +40,10 @@ bool db__ready_for_flight(struct mosquitto *context, enum mosquitto_msg_directio
 	bool valid_bytes;
 	bool valid_count;
 
+	if(!context){
+		return true;
+	}
+
 	if(dir == mosq_md_out){
 		msgs = &context->msgs_out;
 	}else{
@@ -1514,7 +1518,7 @@ int db__message_write_queued_in(struct mosquitto *context)
 	struct mosquitto__client_msg *client_msg, *tmp;
 	int rc;
 
-	if(context->state != mosq_cs_active){
+	if(!context || context->state != mosq_cs_active){
 		return MOSQ_ERR_SUCCESS;
 	}
 

src/handle_publish.c

@@ -64,32 +64,18 @@ static int process_bad_message(struct mosquitto *context, struct mosquitto__base
 }
 
 
-int handle__accepted_publish(struct mosquitto *context, struct mosquitto__base_msg *base_msg, uint16_t mid, int dup, uint32_t *message_expiry_interval)
+static int handle__accepted_publish_plugin(struct mosquitto__base_msg *base_msg)
 {
-	int rc;
-	int rc2;
-	struct mosquitto__base_msg *stored = NULL;
-	struct mosquitto__client_msg *cmsg_stored = NULL;
-
-	{
-		rc = plugin__handle_message_in(context, &base_msg->data);
-		if(rc == MOSQ_ERR_ACL_DENIED){
-			log__printf(NULL, MOSQ_LOG_DEBUG,
-					"Denied PUBLISH from %s (d%d, q%d, r%d, m%d, '%s', ... (%ld bytes))",
-					context->id, dup, base_msg->data.qos, base_msg->data.retain, base_msg->data.source_mid, base_msg->data.topic,
-					(long)base_msg->data.payloadlen);
+	return sub__messages_queue("plugin", base_msg->data.topic, base_msg->data.qos, base_msg->data.retain, &base_msg);
+}
 
-			return process_bad_message(context, base_msg, MQTT_RC_NOT_AUTHORIZED);
-		}else if(rc == MOSQ_ERR_QUOTA_EXCEEDED){
-			log__printf(NULL, MOSQ_LOG_DEBUG,
-					"Rejected PUBLISH from %s, quota exceeded.", context->id);
 
-			return process_bad_message(context, base_msg, MQTT_RC_QUOTA_EXCEEDED);
-		}else if(rc != MOSQ_ERR_SUCCESS){
-			db__msg_store_free(base_msg);
-			return rc;
-		}
-	}
+static int handle__accepted_publish_client(struct mosquitto *context, struct mosquitto__base_msg *base_msg, uint16_t mid, int dup, uint32_t *message_expiry_interval)
+{
+	struct mosquitto__client_msg *cmsg_stored = NULL;
+	struct mosquitto__base_msg *stored = NULL;
+	int rc;
+	int rc2;
 
 	if(base_msg->data.qos > 0){
 		db__message_store_find(context, base_msg->data.source_mid, &cmsg_stored);
@@ -107,7 +93,7 @@ int handle__accepted_publish(struct mosquitto *context, struct mosquitto__base_m
 	}
 
 	if(!cmsg_stored){
-		if(base_msg->data.qos > 0 && context->msgs_in.inflight_quota == 0){
+		if(base_msg->data.qos > 0 && context && context->msgs_in.inflight_quota == 0){
 			/* Client isn't allowed any more incoming messages, so fail early */
 			db__msg_store_free(base_msg);
 			return MOSQ_ERR_RECEIVE_MAXIMUM_EXCEEDED;
@@ -195,6 +181,41 @@ int handle__accepted_publish(struct mosquitto *context, struct mosquitto__base_m
 }
 
 
+int handle__accepted_publish(struct mosquitto *context, struct mosquitto__base_msg *base_msg, uint16_t mid, int dup, uint32_t *message_expiry_interval)
+{
+	int rc;
+	const char *clientid = context?context->id:"plugin";
+
+	/* context == NULL happens if mosquitto_plugin_publish*() is used message */
+
+	{
+		rc = plugin__handle_message_in(context, &base_msg->data);
+		if(rc == MOSQ_ERR_ACL_DENIED){
+			log__printf(NULL, MOSQ_LOG_DEBUG,
+					"Denied PUBLISH from %s (d%d, q%d, r%d, m%d, '%s', ... (%ld bytes))",
+					clientid, dup, base_msg->data.qos, base_msg->data.retain, base_msg->data.source_mid, base_msg->data.topic,
+					(long)base_msg->data.payloadlen);
+
+			return process_bad_message(context, base_msg, MQTT_RC_NOT_AUTHORIZED);
+		}else if(rc == MOSQ_ERR_QUOTA_EXCEEDED){
+			log__printf(NULL, MOSQ_LOG_DEBUG,
+					"Rejected PUBLISH from %s, quota exceeded.", clientid);
+
+			return process_bad_message(context, base_msg, MQTT_RC_QUOTA_EXCEEDED);
+		}else if(rc != MOSQ_ERR_SUCCESS){
+			db__msg_store_free(base_msg);
+			return rc;
+		}
+	}
+
+	if(context){
+		return handle__accepted_publish_client(context, base_msg, mid, dup, message_expiry_interval);
+	}else{
+		return handle__accepted_publish_plugin(base_msg);
+	}
+}
+
+
 int handle__publish(struct mosquitto *context)
 {
 	uint8_t dup;

src/linker-aix.syms

@@ -39,6 +39,8 @@ mosquitto_persist_client_update
 mosquitto_persist_retain_msg_delete
 mosquitto_persist_retain_msg_set
 mosquitto_persistence_location
+mosquitto_plugin_publish
+mosquitto_plugin_publish_copy
 mosquitto_plugin_set_info
 mosquitto_property_add_binary
 mosquitto_property_add_byte

src/linker-macosx.syms

@@ -40,6 +40,8 @@ _mosquitto_persist_client_update
 _mosquitto_persist_retain_msg_delete
 _mosquitto_persist_retain_msg_set
 _mosquitto_persistence_location
+_mosquitto_plugin_publish
+_mosquitto_plugin_publish_copy
 _mosquitto_plugin_set_info
 _mosquitto_property_add_binary
 _mosquitto_property_add_byte

src/linker.syms

@@ -41,6 +41,8 @@
 	mosquitto_persist_retain_msg_delete;
 	mosquitto_persist_retain_msg_set;
 	mosquitto_persistence_location;
+	mosquitto_plugin_publish;
+	mosquitto_plugin_publish_copy;
 	mosquitto_plugin_set_info;
 	mosquitto_property_add_binary;
 	mosquitto_property_add_byte;

src/loop.c

@@ -115,32 +115,36 @@ static void read_message_expiry_interval(mosquitto_property **proplist, uint32_t
 static void queue_plugin_msgs(void)
 {
 	struct mosquitto__base_msg *base_msg, *base_tmp;
-	struct mosquitto *context;
 	uint32_t message_expiry;
 
 	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(&base_msg->data.properties, &message_expiry);
 
-		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, base_msg, message_expiry);
+		if(base_msg->origin == mosq_mo_plugin){
+			handle__accepted_publish(NULL, base_msg, 0, 0, &message_expiry);
+		}else{
+			if(base_msg->data.source_id){
+				struct mosquitto *context;
+				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, base_msg, message_expiry);
+				}else{
+					db__msg_store_free(base_msg);
+				}
 			}else{
+				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);
 			}
-		}else{
-			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);
 		}
 	}
 }

src/mosquitto_broker_internal.h

@@ -102,6 +102,7 @@ typedef int (*FUNC_auth_plugin_psk_key_get_v2)(void *, const char *, const char
 enum mosquitto_msg_origin {
 	mosq_mo_client = 0,
 	mosq_mo_broker = 1,
+	mosq_mo_plugin = 2,
 };
 
 struct mosquitto__plugin_lib {
@@ -693,6 +694,7 @@ int handle__publish(struct mosquitto *context);
 int handle__subscribe(struct mosquitto *context);
 int handle__unsubscribe(struct mosquitto *context);
 int handle__auth(struct mosquitto *context);
+int handle__accepted_publish(struct mosquitto *context, struct mosquitto__base_msg *base_msg, uint16_t mid, int dup, uint32_t *message_expiry_interval);
 
 /* ============================================================
  * Database handling