add mosquitto_delay_puback option

Signed-off-by: Linus Basig <[email protected]>
Signed-off-by: Fabrizio Lazzaretti <[email protected]>
Signed-off-by: Linus Basig <[email protected]>

Author: linuxbasic

.gitignore

@@ -85,3 +85,4 @@ test/unit/out/
 
 www/cache/
 __pycache__
+.DS_Store

ChangeLog.txt

@@ -1,3 +1,7 @@
+Client library:
+- Add `mosquitto_delay_puback()` for delaying the PUBACK message
+  until the on_message callback returns.
+
 2.0.0 - 2020-12-03
 ==================
 

include/mosquitto.h

@@ -1665,6 +1665,18 @@ libmosq_EXPORT void mosquitto_user_data_set(struct mosquitto *mosq, void *obj);
  */
 libmosq_EXPORT void *mosquitto_userdata(struct mosquitto *mosq);
 
+/*
+ * Function: mosquitto_delay_puback
+ *
+ * Per Default, the library will acknowledge a QoS 1 message before it calls the 
+ * on_message callback. After <mosquitto_delay_puback> is called, the library
+ * delays the acknowledgment untill after the on_message callback has returned.
+ *
+ * Parameters:
+ *  mosq - a valid mosquitto instance.
+ */
+libmosq_EXPORT int mosquitto_delay_puback(struct mosquitto *mosq);
+
 
 /* ======================================================================
  *

lib/cpp/mosquittopp.cpp

@@ -352,6 +352,11 @@ void mosquittopp::user_data_set(void *userdata)
 	mosquitto_user_data_set(m_mosq, userdata);
 }
 
+int mosquittopp::delay_puback()
+{
+	mosquitto_delay_puback(m_mosq);
+}
+
 int mosquittopp::socks5_set(const char *host, int port, const char *username, const char *password)
 {
 	return mosquitto_socks5_set(m_mosq, host, port, username, password);

lib/cpp/mosquittopp.h

@@ -107,6 +107,7 @@ class mosqpp_EXPORT mosquittopp {
 		int max_inflight_messages_set(unsigned int max_inflight_messages);
 		void message_retry_set(unsigned int message_retry);
 		void user_data_set(void *userdata);
+		int delay_puback();
 		int tls_set(const char *cafile, const char *capath=NULL, const char *certfile=NULL, const char *keyfile=NULL, int (*pw_callback)(char *buf, int size, int rwflag, void *userdata)=NULL);
 		int tls_opts_set(int cert_reqs, const char *tls_version=NULL, const char *ciphers=NULL);
 		int tls_insecure_set(bool value);

lib/handle_publish.c

@@ -135,7 +135,9 @@ int handle__publish(struct mosquitto *mosq)
 			return MOSQ_ERR_SUCCESS;
 		case 1:
 			util__decrement_receive_quota(mosq);
-			rc = send__puback(mosq, mid, 0, NULL);
+			if(mosq->delayed_puback){
+				rc = send__puback(mosq, mid, 0, NULL);
+			}
 			pthread_mutex_lock(&mosq->callback_mutex);
 			if(mosq->on_message){
 				mosq->in_callback = true;
@@ -148,6 +150,9 @@ int handle__publish(struct mosquitto *mosq)
 				mosq->in_callback = false;
 			}
 			pthread_mutex_unlock(&mosq->callback_mutex);
+			if(!mosq->delayed_puback){
+				rc = send__puback(mosq, mid, 0, NULL);
+			}
 			message__cleanup(&message);
 			mosquitto_property_free_all(&properties);
 			return rc;

lib/linker.version

@@ -141,3 +141,8 @@ MOSQ_1.7 {
 		mosquitto_property_next;
 		mosquitto_ssl_get;
 } MOSQ_1.6;
+
+MOSQ_2.1 {
+	global:
+		mosquitto_delay_puback;
+} MOSQ_2.0;

lib/mosquitto.c

@@ -193,6 +193,7 @@ int mosquitto_reinitialise(struct mosquitto *mosq, const char *id, bool clean_st
 	mosq->reconnect_delay_max = 1;
 	mosq->reconnect_exponential_backoff = false;
 	mosq->threaded = mosq_ts_none;
+	mosq->delayed_puback = false;
 #ifdef WITH_TLS
 	mosq->ssl = NULL;
 	mosq->ssl_ctx = NULL;

lib/mosquitto_internal.h

@@ -351,6 +351,7 @@ struct mosquitto {
 #ifdef WITH_EPOLL
 	uint32_t events;
 #endif
+	bool delayed_puback;
 };
 
 #define STREMPTY(str) (str[0] == '\0')

lib/options.c

@@ -530,3 +530,12 @@ void *mosquitto_userdata(struct mosquitto *mosq)
 {
 	return mosq->userdata;
 }
+
+int mosquitto_delay_puback(struct mosquitto *mosq)
+{
+	if(!mosq) return MOSQ_ERR_INVAL;
+
+	mosq->delayed_puback = true;
+	
+	return MOSQ_ERR_SUCCESS;
+}