diff --git a/ChangeLog.txt b/ChangeLog.txt
index 66de676b..03641151 100644
--- a/ChangeLog.txt
+++ b/ChangeLog.txt
@@ -8,6 +8,8 @@ Broker:
an IPv6 link-local address and no other IPv6 addresses. Closes #2696.
- Fix mismatched wrapped/unwrapped memory alloc/free in properties. Closes #3192.
- Fix `allow_anonymous false` not being applied in local only mode. Closes #3198.
+- Add `retain_expiry_interval` option to fix expired retained message not
+ being removed from memory if they are not subscribed to. Closes #3221.
Client library:
- Fix potential deadlock in mosquitto_sub if `-W` is used. Closes #3175.
diff --git a/man/mosquitto.conf.5.xml b/man/mosquitto.conf.5.xml
index 898d48be..16198949 100644
--- a/man/mosquitto.conf.5.xml
+++ b/man/mosquitto.conf.5.xml
@@ -987,6 +987,34 @@ log_timestamp_format %Y-%m-%dT%H:%M:%S
Reloaded on reload signal.
+
+ minutes
+
+
+ The default behaviour of mosquitto is to remove retained
+ messages that have reached their message-expiry-interval
+ property the next time that that message is accessed -
+ either by being replaced by a new message, or on the
+ next subscription that matches the message. If you have
+ a pattern of publishing many retained messages with a
+ message-expiry-interval, but that are not subscribed to,
+ then the expired retained messages will remain in
+ memory. This option configures the broker to
+ periodically check the retained tree for expired
+ messages.
+
+
+
+ Defaults to off. Setting to a value greater than zero
+ means the broker will make a check at an interval of
+ that number of minutes.
+
+
+ This option applies globally.
+
+ Reloaded on reload signal.
+
+
[ true | false ]
diff --git a/src/conf.c b/src/conf.c
index f9433ea1..a62c4ca1 100644
--- a/src/conf.c
+++ b/src/conf.c
@@ -201,6 +201,7 @@ static void config__init_reload(struct mosquitto__config *config)
config->persistent_client_expiration = 0;
config->queue_qos0_messages = false;
config->retain_available = true;
+ config->retain_expiry_interval = 0;
config->set_tcp_nodelay = false;
config->sys_interval = 10;
config->upgrade_outgoing_qos = false;
@@ -1917,6 +1918,9 @@ static int config__read_file_core(struct mosquitto__config *config, bool reload,
#endif
}else if(!strcmp(token, "retain_available")){
if(conf__parse_bool(&token, token, &config->retain_available, saveptr)) return MOSQ_ERR_INVAL;
+ }else if(!strcmp(token, "retain_expiry_interval")){
+ if(conf__parse_int(&token, token, &config->retain_expiry_interval, saveptr)) return MOSQ_ERR_INVAL;
+ config->retain_expiry_interval *= 60;
}else if(!strcmp(token, "retry_interval")){
log__printf(NULL, MOSQ_LOG_WARNING, "Warning: The retry_interval option is no longer available.");
}else if(!strcmp(token, "round_robin")){
diff --git a/src/loop.c b/src/loop.c
index ec953398..0940bb77 100644
--- a/src/loop.c
+++ b/src/loop.c
@@ -187,6 +187,7 @@ int mosquitto_main_loop(struct mosquitto__listener_sock *listensock, int listens
#endif
while(run){
+ retain__expire();
queue_plugin_msgs();
context__free_disused();
#ifdef WITH_SYS_TREE
diff --git a/src/mosquitto_broker_internal.h b/src/mosquitto_broker_internal.h
index aff6e948..4458434f 100644
--- a/src/mosquitto_broker_internal.h
+++ b/src/mosquitto_broker_internal.h
@@ -291,6 +291,7 @@ struct mosquitto__config {
bool queue_qos0_messages;
bool per_listener_settings;
bool retain_available;
+ int retain_expiry_interval;
bool set_tcp_nodelay;
int sys_interval;
bool upgrade_outgoing_qos;
@@ -791,6 +792,7 @@ int retain__init(void);
void retain__clean(struct mosquitto__retainhier **retainhier);
int retain__queue(struct mosquitto *context, const char *sub, uint8_t sub_qos, uint32_t subscription_identifier);
int retain__store(const char *topic, struct mosquitto_msg_store *stored, char **split_topics);
+void retain__expire(void);
/* ============================================================
* Security related functions
diff --git a/src/retain.c b/src/retain.c
index 6c303c96..45765e80 100644
--- a/src/retain.c
+++ b/src/retain.c
@@ -29,6 +29,8 @@ Contributors:
#include "utlist.h"
+static time_t next_expire_check = 0;
+
static struct mosquitto__retainhier *retain__add_hier_entry(struct mosquitto__retainhier *parent, struct mosquitto__retainhier **sibling, const char *topic, uint16_t len)
{
struct mosquitto__retainhier *child;
@@ -320,3 +322,32 @@ void retain__clean(struct mosquitto__retainhier **retainhier)
}
}
+static void retain__expire_search(struct mosquitto__retainhier *retainhier)
+{
+ struct mosquitto__retainhier *branch, *branch_tmp;
+
+ HASH_ITER(hh, retainhier->children, branch, branch_tmp){
+ if(branch->children){
+ retain__expire_search(branch);
+ }
+ if(branch->retained){
+ if(branch->retained->message_expiry_time > 0 && db.now_real_s >= branch->retained->message_expiry_time){
+ db__msg_store_ref_dec(&branch->retained);
+ branch->retained = NULL;
+#ifdef WITH_SYS_TREE
+ db.retained_count--;
+#endif
+ retain__clean_empty_hierarchy(retainhier);
+ }
+ }
+ }
+}
+
+
+void retain__expire(void)
+{
+ if(db.config->retain_expiry_interval > 0 && db.now_s > next_expire_check){
+ retain__expire_search(db.retains);
+ next_expire_check = db.now_s + db.config->retain_expiry_interval;
+ }
+}