diff --git a/src/handle_subscribe.c b/src/handle_subscribe.c index aac6ad6d..4d59da6f 100644 --- a/src/handle_subscribe.c +++ b/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; - if(retain_handling == 3 || (subscription_options & 0xC0) != 0){ + retain_handling = (subscription_options & 0x30); + 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) - || (rc2 == MOSQ_ERR_SUB_EXISTS && retain_handling == 1)){ + if((retain_handling == MQTT_SUB_OPT_SEND_RETAIN_ALWAYS) + || (rc2 == MOSQ_ERR_SUCCESS && retain_handling == MQTT_SUB_OPT_SEND_RETAIN_NEW)){ if(sub__retain_queue(db, context, sub, qos)) rc = 1; } diff --git a/src/subs.c b/src/subs.c index 0899383c..3e3bf3be 100644 --- a/src/subs.c +++ b/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; diff --git a/test/broker/02-subpub-qos0-retain-as-publish.py b/test/broker/02-subpub-qos0-retain-as-publish.py new file mode 100755 index 00000000..d6436eef --- /dev/null +++ b/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) + diff --git a/test/broker/02-subpub-qos0-send-retain.py b/test/broker/02-subpub-qos0-send-retain.py new file mode 100755 index 00000000..ed9ba735 --- /dev/null +++ b/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) + diff --git a/test/broker/02-subpub-qos1-nolocal.py b/test/broker/02-subpub-qos1-nolocal.py new file mode 100755 index 00000000..d66be515 --- /dev/null +++ b/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) + diff --git a/test/broker/Makefile b/test/broker/Makefile index 3105c62a..1d80d79d 100644 --- a/test/broker/Makefile +++ b/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 diff --git a/test/broker/mosq_test_helper.py b/test/broker/mosq_test_helper.py index 0e861dba..9b8f2113 100644 --- a/test/broker/mosq_test_helper.py +++ b/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 diff --git a/test/broker/ptest.py b/test/broker/ptest.py index 2804dfd8..68f1ed25 100755 --- a/test/broker/ptest.py +++ b/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'), diff --git a/test/mqtt5_opts.py b/test/mqtt5_opts.py new file mode 100644 index 00000000..55675a73 --- /dev/null +++ b/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