Add MOSQ_EVT_PERSIST_CLIENT_MSG_CLEAR plugin event.

include/mosquitto_broker.h

@@ -89,7 +89,8 @@ enum mosquitto_plugin_event {
 	MOSQ_EVT_PERSIST_CLIENT_MSG_ADD = 24,
 	MOSQ_EVT_PERSIST_CLIENT_MSG_REMOVE = 25,
 	MOSQ_EVT_PERSIST_CLIENT_MSG_UPDATE = 26,
-	MOSQ_EVT_PERSIST_CLIENT_MSG_LOAD = 27,
+	MOSQ_EVT_PERSIST_CLIENT_MSG_CLEAR = 27,
+	MOSQ_EVT_PERSIST_CLIENT_MSG_LOAD = 28,
 };
 
 /* Data for the MOSQ_EVT_RELOAD event */

src/database.c

@@ -719,6 +719,7 @@ int db__messages_delete(struct mosquitto *context, bool force_free)
 		context->msgs_in.queued_bytes12 = 0;
 		context->msgs_in.queued_count = 0;
 		context->msgs_in.queued_count12 = 0;
+		plugin_persist__handle_client_msg_clear(context, mosq_md_in);
 	}
 
 	if(force_free || (context->bridge && context->bridge->clean_start_local)
@@ -734,6 +735,7 @@ int db__messages_delete(struct mosquitto *context, bool force_free)
 		context->msgs_out.queued_bytes12 = 0;
 		context->msgs_out.queued_count = 0;
 		context->msgs_out.queued_count12 = 0;
+		plugin_persist__handle_client_msg_clear(context, mosq_md_out);
 	}
 
 	return MOSQ_ERR_SUCCESS;

src/mosquitto_broker_internal.h

@@ -173,6 +173,7 @@ struct plugin__callbacks{
 	struct mosquitto__callback *persist_client_msg_add;
 	struct mosquitto__callback *persist_client_msg_remove;
 	struct mosquitto__callback *persist_client_msg_update;
+	struct mosquitto__callback *persist_client_msg_clear;
 	struct mosquitto__callback *persist_msg_add;
 	struct mosquitto__callback *persist_msg_remove;
 	struct mosquitto__callback *persist_msg_load;
@@ -865,6 +866,7 @@ void plugin_persist__handle_subscription_remove(struct mosquitto *context, const
 void plugin_persist__handle_client_msg_add(struct mosquitto *context, const struct mosquitto_client_msg *cmsg);
 void plugin_persist__handle_client_msg_remove(struct mosquitto *context, const struct mosquitto_client_msg *cmsg);
 void plugin_persist__handle_client_msg_update(struct mosquitto *context, const struct mosquitto_client_msg *cmsg);
+void plugin_persist__handle_client_msg_clear(struct mosquitto *context, uint8_t direction);
 void plugin_persist__handle_msg_add(struct mosquitto_msg_store *msg);
 void plugin_persist__handle_msg_remove(struct mosquitto_msg_store *msg);
 void plugin_persist__handle_retain_add(struct mosquitto_msg_store *msg);

src/plugin_callbacks.c

@@ -107,6 +107,8 @@ static struct mosquitto__callback **plugin__get_callback_base(struct mosquitto__
 			return &security_options->plugin_callbacks.persist_client_msg_remove;
 		case MOSQ_EVT_PERSIST_CLIENT_MSG_UPDATE:
 			return &security_options->plugin_callbacks.persist_client_msg_update;
+		case MOSQ_EVT_PERSIST_CLIENT_MSG_CLEAR:
+			return &security_options->plugin_callbacks.persist_client_msg_clear;
 		case MOSQ_EVT_PERSIST_MSG_ADD:
 			return &security_options->plugin_callbacks.persist_msg_add;
 		case MOSQ_EVT_PERSIST_MSG_REMOVE:

src/plugin_persist.c

@@ -246,6 +246,26 @@ void plugin_persist__handle_client_msg_update(struct mosquitto *context, const s
 }
 
 
+void plugin_persist__handle_client_msg_clear(struct mosquitto *context, uint8_t direction)
+{
+	struct mosquitto_evt_persist_client_msg event_data;
+	struct mosquitto__callback *cb_base;
+	struct mosquitto__security_options *opts;
+
+	if(context->session_expiry_interval == 0 || db.shutdown) return;
+
+	opts = &db.config->security_options;
+	memset(&event_data, 0, sizeof(event_data));
+
+	event_data.client_id = context->id;
+	event_data.direction = direction;
+
+	DL_FOREACH(opts->plugin_callbacks.persist_client_msg_clear, cb_base){
+		cb_base->cb(MOSQ_EVT_PERSIST_CLIENT_MSG_CLEAR, &event_data, cb_base->userdata);
+	}
+}
+
+
 void plugin_persist__handle_msg_add(struct mosquitto_msg_store *msg)
 {
 	struct mosquitto_evt_persist_msg event_data;