Message out event should be able to deny publishes.

lib/send_publish.c

@@ -70,7 +70,7 @@ int send__publish(struct mosquitto *mosq, uint16_t mid, const char *topic, uint3
 		tmp_msg.retain = retain;
 		tmp_msg.properties = (mosquitto_property *) store_props;
 
-		plugin__handle_message_out(mosq, &tmp_msg);
+		rc = plugin__handle_message_out(mosq, &tmp_msg);
 
 		if(tmp_msg.payload != payload) payload_changed = true;
 		if(tmp_msg.topic != topic) topic_changed = true;
@@ -82,6 +82,23 @@ int send__publish(struct mosquitto *mosq, uint16_t mid, const char *topic, uint3
 		qos = tmp_msg.qos;
 		retain = tmp_msg.retain;
 		store_props = tmp_msg.properties;
+
+		if(rc != MOSQ_ERR_SUCCESS){
+			if(payload_changed) mosquitto__free((void *) payload);
+			if(topic_changed) mosquitto__free((char *) topic);
+			if(properties_changed) mosquitto_property_free_all((mosquitto_property **) &store_props);
+			if(rc == MOSQ_ERR_ACL_DENIED){
+				log__printf(NULL, MOSQ_LOG_DEBUG,
+						"Denied PUBLISH to %s (q%d, r%d, '%s', ... (%ld bytes))",
+						mosq->id, qos, retain, topic, (long)payloadlen);
+
+			}else if(rc == MOSQ_ERR_QUOTA_EXCEEDED){
+				log__printf(NULL, MOSQ_LOG_DEBUG,
+						"Rejected PUBLISH to %s, quota exceeded.", mosq->id);
+			}
+
+			return MOSQ_ERR_SUCCESS;
+		}
 	}
 #endif
 

test/broker/09-plugin-evt-message-out.py

@@ -15,9 +15,11 @@ def do_test():
     connack_packet = mosq_test.gen_connack(rc=0, proto_ver=5)
 
     mid = 1
-    subscribe_packet = mosq_test.gen_subscribe(mid, "out-topic", 2, proto_ver=5)
+    subscribe_packet = mosq_test.gen_subscribe(mid, "#", 2, proto_ver=5)
     suback_packet = mosq_test.gen_suback(mid, 2, proto_ver=5)
 
+    publish_packet_deny = mosq_test.gen_publish("deny", qos=0, payload="message1", proto_ver=5)
+
     props = mqtt5_props.gen_string_prop(mqtt5_props.PROP_RESPONSE_TOPIC, "response/topic")
     publish_packet1 = mosq_test.gen_publish("out-topic", qos=0, payload="message1", proto_ver=5, properties=props)
 
@@ -33,8 +35,13 @@ def do_test():
         sock = mosq_test.do_client_connect(connect_packet, connack_packet, timeout=20, port=port)
 
         mosq_test.do_send_receive(sock, subscribe_packet, suback_packet, "suback")
+
+        sock.send(publish_packet_deny)
+        mosq_test.do_ping(sock)
+
         sock.send(publish_packet1)
         mosq_test.expect_packet(sock, "publish2", publish_packet2)
+        mosq_test.do_ping(sock)
 
         rc = 0
 

test/broker/c/plugin_evt_message_out.c

@@ -18,6 +18,9 @@ int callback_message_out(int event, void *event_data, void *user_data)
 	if(event != MOSQ_EVT_MESSAGE_OUT){
 		abort();
 	}
+	if(!strcmp(ed->topic, "deny")){
+		return MOSQ_ERR_ACL_DENIED;
+	}
 	ed->topic = mosquitto_strdup("new-topic");
 	ed->payload = mosquitto_strdup("new-message");
 	ed->payloadlen = strlen(ed->payload);