Tests and fixes for subscription options.

src/handle_subscribe.c

@@ -100,8 +100,8 @@ int handle__subscribe(struct mosquitto_db *db, struct mosquitto *context)
 				qos = subscription_options & 0x03;
 				subscription_options &= 0xFC;
 
-				retain_handling = (subscription_options & 0x30) >> 4;
+				retain_handling = (subscription_options & 0x30);
-				if(retain_handling == 3 || (subscription_options & 0xC0) != 0){
+				if(retain_handling == 0x30 || (subscription_options & 0xC0) != 0){
 					return MOSQ_ERR_PROTOCOL;
 				}
 			}
@@ -148,13 +148,17 @@ int handle__subscribe(struct mosquitto_db *db, struct mosquitto *context)
 
 			if(qos != 0x80){
 				rc2 = sub__add(db, context, sub, qos, subscription_options, &db->subs);
+				if(rc2 > 0){
+					mosquitto__free(sub);
+					return rc2;
+				}
 				if(context->protocol == mosq_p_mqtt311 || context->protocol == mosq_p_mqtt31){
 					if(rc2 == MOSQ_ERR_SUCCESS || rc2 == MOSQ_ERR_SUB_EXISTS){
 						if(sub__retain_queue(db, context, sub, qos)) rc = 1;
 					}
 				}else{
-					if((rc2 == MOSQ_ERR_SUCCESS && retain_handling == 0)
+					if((retain_handling == MQTT_SUB_OPT_SEND_RETAIN_ALWAYS)
-							|| (rc2 == MOSQ_ERR_SUB_EXISTS && retain_handling == 1)){
+							|| (rc2 == MOSQ_ERR_SUCCESS && retain_handling == MQTT_SUB_OPT_SEND_RETAIN_NEW)){
 
 						if(sub__retain_queue(db, context, sub, qos)) rc = 1;
 					}

src/subs.c

@@ -258,12 +258,12 @@ static int sub__add_recurse(struct mosquitto_db *db, struct mosquitto *context,
 					 * need to update QoS. Return MOSQ_ERR_SUB_EXISTS to
 					 * indicate this to the calling function. */
 					leaf->qos = qos;
-					if(context->protocol == mosq_p_mqtt31){
+					if(context->protocol == mosq_p_mqtt31 || context->protocol == mosq_p_mqtt5){
 						return MOSQ_ERR_SUB_EXISTS;
 					}else{
 						/* mqttv311/mqttv5 requires retained messages are resent on
 						 * resubscribe. */
-						return 0;
+						return MOSQ_ERR_SUCCESS;
 					}
 				}
 				last_leaf = leaf;

test/broker/02-subpub-qos0-retain-as-publish.py

@@ -0,0 +1,53 @@
+#!/usr/bin/env python
+
+# Test whether a client subscribed to a topic with retain-as-published set works as expected.
+# MQTT v5
+
+from mosq_test_helper import *
+
+rc = 1
+keepalive = 60
+connect_packet = mosq_test.gen_connect("subpub-qos1-test", keepalive=keepalive, proto_ver=5)
+connack_packet = mosq_test.gen_connack(rc=0, proto_ver=5)
+
+mid = 530
+subscribe1_packet = mosq_test.gen_subscribe(mid, "subpub/normal", 0, proto_ver=5)
+suback1_packet = mosq_test.gen_suback(mid, 0, proto_ver=5)
+
+mid = 531
+subscribe2_packet = mosq_test.gen_subscribe(mid, "subpub/rap", 0 | mqtt5_opts.MQTT_SUB_OPT_RETAIN_AS_PUBLISHED, proto_ver=5)
+suback2_packet = mosq_test.gen_suback(mid, 0, proto_ver=5)
+
+publish1_packet = mosq_test.gen_publish("subpub/normal", qos=0, retain=True, payload="message", proto_ver=5)
+publish2_packet = mosq_test.gen_publish("subpub/rap", qos=0, retain=True, payload="message", proto_ver=5)
+
+publish1r_packet = mosq_test.gen_publish("subpub/normal", qos=0, retain=False, payload="message", proto_ver=5)
+publish2r_packet = mosq_test.gen_publish("subpub/rap", qos=0, retain=True, payload="message", proto_ver=5)
+
+mid = 1
+publish3_packet = mosq_test.gen_publish("subpub/receive", qos=1, mid=mid, payload="success", proto_ver=5)
+
+
+port = mosq_test.get_port()
+broker = mosq_test.start_broker(filename=os.path.basename(__file__), port=port)
+
+try:
+    sock = mosq_test.do_client_connect(connect_packet, connack_packet, timeout=20, port=port)
+
+    mosq_test.do_send_receive(sock, subscribe1_packet, suback1_packet, "suback1")
+    mosq_test.do_send_receive(sock, subscribe2_packet, suback2_packet, "suback2")
+
+    mosq_test.do_send_receive(sock, publish1_packet, publish1r_packet, "publish1")
+    mosq_test.do_send_receive(sock, publish2_packet, publish2r_packet, "publish2")
+    rc = 0
+
+    sock.close()
+finally:
+    broker.terminate()
+    broker.wait()
+    (stdo, stde) = broker.communicate()
+    if rc:
+        print(stde)
+
+exit(rc)
+

test/broker/02-subpub-qos0-send-retain.py

@@ -0,0 +1,76 @@
+#!/usr/bin/env python
+
+# Test whether "send retain" subscribe options work
+# MQTT v5
+
+from mosq_test_helper import *
+
+rc = 1
+keepalive = 60
+connect_packet = mosq_test.gen_connect("subpub-test", keepalive=keepalive, proto_ver=5)
+connack_packet = mosq_test.gen_connack(rc=0, proto_ver=5)
+
+mid = 530
+subscribe1_packet = mosq_test.gen_subscribe(mid, "subpub/always", 0 | mqtt5_opts.MQTT_SUB_OPT_SEND_RETAIN_ALWAYS, proto_ver=5)
+suback1_packet = mosq_test.gen_suback(mid, 0, proto_ver=5)
+
+mid = 531
+subscribe2_packet = mosq_test.gen_subscribe(mid, "subpub/new", 0 | mqtt5_opts.MQTT_SUB_OPT_SEND_RETAIN_NEW, proto_ver=5)
+suback2_packet = mosq_test.gen_suback(mid, 0, proto_ver=5)
+
+mid = 532
+subscribe3_packet = mosq_test.gen_subscribe(mid, "subpub/never", 0 | mqtt5_opts.MQTT_SUB_OPT_SEND_RETAIN_NEVER, proto_ver=5)
+suback3_packet = mosq_test.gen_suback(mid, 0, proto_ver=5)
+
+
+publish1_packet = mosq_test.gen_publish("subpub/always", qos=0, retain=True, payload="message", proto_ver=5)
+publish2_packet = mosq_test.gen_publish("subpub/new", qos=0, retain=True, payload="message", proto_ver=5)
+publish3_packet = mosq_test.gen_publish("subpub/never", qos=0, retain=True, payload="message", proto_ver=5)
+
+publish1r1_packet = mosq_test.gen_publish("subpub/always", qos=0, retain=True, payload="message", proto_ver=5)
+publish1r2_packet = mosq_test.gen_publish("subpub/always", qos=0, retain=True, payload="message", proto_ver=5)
+publish2r1_packet = mosq_test.gen_publish("subpub/new", qos=0, retain=True, payload="message", proto_ver=5)
+publish2r2_packet = mosq_test.gen_publish("subpub/new", qos=0, retain=False, payload="message", proto_ver=5)
+publish3r1_packet = mosq_test.gen_publish("subpub/never", qos=0, retain=False, payload="message", proto_ver=5)
+publish3r2_packet = mosq_test.gen_publish("subpub/never", qos=0, retain=False, payload="message", proto_ver=5)
+
+
+port = mosq_test.get_port()
+broker = mosq_test.start_broker(filename=os.path.basename(__file__), port=port)
+
+try:
+    sock = mosq_test.do_client_connect(connect_packet, connack_packet, timeout=5, port=port)
+
+    sock.send(publish1_packet)
+    sock.send(publish2_packet)
+    sock.send(publish3_packet)
+
+    # Don't expect a message after this
+    mosq_test.do_send_receive(sock, subscribe3_packet, suback3_packet, "suback3")
+    # Don't expect a message after this
+    mosq_test.do_send_receive(sock, subscribe3_packet, suback3_packet, "suback3")
+
+    # Expect a message after this, because it is the first subscribe
+    mosq_test.do_send_receive(sock, subscribe2_packet, suback2_packet, "suback2")
+    if mosq_test.expect_packet(sock, "publish2r1", publish2r1_packet):
+        # Don't expect a message after this, it is the second subscribe
+        mosq_test.do_send_receive(sock, subscribe2_packet, suback2_packet, "suback2")
+
+        # Always expect a message after this
+        mosq_test.do_send_receive(sock, subscribe1_packet, suback1_packet, "suback1")
+        if mosq_test.expect_packet(sock, "publish1r1", publish1r1_packet):
+            # Always expect a message after this
+            mosq_test.do_send_receive(sock, subscribe1_packet, suback1_packet, "suback1")
+            if mosq_test.expect_packet(sock, "publish1r1", publish1r2_packet):
+                rc = 0
+
+    sock.close()
+finally:
+    broker.terminate()
+    broker.wait()
+    (stdo, stde) = broker.communicate()
+    if rc:
+        print(stde)
+
+exit(rc)
+

test/broker/02-subpub-qos1-nolocal.py

@@ -0,0 +1,58 @@
+#!/usr/bin/env python
+
+# Test whether a client subscribed to a topic does not receive its own message
+# sent to that topic if no local is set.
+# MQTT v5
+
+from mosq_test_helper import *
+
+rc = 1
+keepalive = 60
+connect_packet = mosq_test.gen_connect("subpub-qos1-test", keepalive=keepalive, proto_ver=5)
+connack_packet = mosq_test.gen_connack(rc=0, proto_ver=5)
+
+mid = 530
+subscribe_packet = mosq_test.gen_subscribe(mid, "subpub/qos1", 1 | mqtt5_opts.MQTT_SUB_OPT_NO_LOCAL, proto_ver=5)
+suback_packet = mosq_test.gen_suback(mid, 1, proto_ver=5)
+
+mid = 531
+subscribe2_packet = mosq_test.gen_subscribe(mid, "subpub/receive", 1, proto_ver=5)
+suback2_packet = mosq_test.gen_suback(mid, 1, proto_ver=5)
+
+mid = 300
+publish_packet = mosq_test.gen_publish("subpub/qos1", qos=1, mid=mid, payload="message", proto_ver=5)
+puback_packet = mosq_test.gen_puback(mid, proto_ver=5)
+
+mid = 301
+publish2_packet = mosq_test.gen_publish("subpub/receive", qos=1, mid=mid, payload="success", proto_ver=5)
+puback2_packet = mosq_test.gen_puback(mid, proto_ver=5)
+
+mid = 1
+publish3_packet = mosq_test.gen_publish("subpub/receive", qos=1, mid=mid, payload="success", proto_ver=5)
+
+
+port = mosq_test.get_port()
+broker = mosq_test.start_broker(filename=os.path.basename(__file__), port=port)
+
+try:
+    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")
+    mosq_test.do_send_receive(sock, subscribe2_packet, suback2_packet, "suback2")
+
+    mosq_test.do_send_receive(sock, publish_packet, puback_packet, "puback")
+    mosq_test.do_send_receive(sock, publish2_packet, puback2_packet, "puback2")
+
+    if mosq_test.expect_packet(sock, "publish3", publish3_packet):
+        rc = 0
+
+    sock.close()
+finally:
+    broker.terminate()
+    broker.wait()
+    (stdo, stde) = broker.communicate()
+    if rc:
+        print(stde)
+
+exit(rc)
+

test/broker/Makefile

@@ -49,6 +49,9 @@ endif
 	./02-subpub-qos0-v5.py
 	./02-subpub-qos1-v5.py
 	./02-subpub-qos2-v5.py
+	./02-subpub-qos1-nolocal.py
+	./02-subpub-qos0-retain-as-publish.py
+	./02-subpub-qos0-send-retain.py
 	./02-unsubscribe-qos0.py
 	./02-unsubscribe-qos1.py
 	./02-unsubscribe-qos2.py

test/broker/mosq_test_helper.py

@@ -6,6 +6,7 @@ if cmd_subfolder not in sys.path:
     sys.path.insert(0, cmd_subfolder)
 
 import mosq_test
+import mqtt5_opts
 import mqtt5_props
 
 import socket

test/broker/ptest.py

@@ -32,6 +32,9 @@ tests = [
     (1, './02-subpub-qos0-v5.py'),
     (1, './02-subpub-qos1-v5.py'),
     (1, './02-subpub-qos2-v5.py'),
+    (1, './02-subpub-qos1-nolocal.py'),
+    (1, './02-subpub-qos0-retain-as-publish.py'),
+    (1, './02-subpub-qos0-send-retain.py'),
     (1, './02-unsubscribe-qos0.py'),
     (1, './02-unsubscribe-qos1.py'),
     (1, './02-unsubscribe-qos2.py'),

test/mqtt5_opts.py

@@ -0,0 +1,5 @@
+MQTT_SUB_OPT_NO_LOCAL = 0x04
+MQTT_SUB_OPT_RETAIN_AS_PUBLISHED = 0x08
+MQTT_SUB_OPT_SEND_RETAIN_ALWAYS = 0x00
+MQTT_SUB_OPT_SEND_RETAIN_NEW = 0x10
+MQTT_SUB_OPT_SEND_RETAIN_NEVER = 0x20