diff --git a/ChangeLog.txt b/ChangeLog.txt index 5732c585..16039fcd 100644 --- a/ChangeLog.txt +++ b/ChangeLog.txt @@ -208,6 +208,7 @@ - Fix `-f` and `-s` options in mosquitto_rr. - Add `--latency` option to mosquitto_rr, for printing the request/response latency. +- Add `--retain-handling` option. ## mosquitto_sub - Fix incorrect output formatting in mosquitto_sub when using field widths @@ -216,6 +217,7 @@ - mosquitto_sub payload hex output can now be split by fixed field length. - Add `--message-rate` option to mosquitto_sub, for printing the count of messages received each second. +- Add `--retain-handling` option. # Apps diff --git a/client/client_shared.c b/client/client_shared.c index c00c1a98..1b3607d3 100644 --- a/client/client_shared.c +++ b/client/client_shared.c @@ -1040,6 +1040,26 @@ int client_config_line_proc(struct mosq_config *cfg, int pub_or_sub, int argc, c goto unknown_option; } cfg->sub_opts |= MQTT_SUB_OPT_RETAIN_AS_PUBLISHED; + }else if(!strcmp(argv[i], "--retain-handling")){ + if(pub_or_sub == CLIENT_PUB){ + goto unknown_option; + } + if(i==argc-1){ + fprintf(stderr, "Error: --retain-handling argument given but no option specified.\n\n"); + return 1; + }else{ + if(!strcmp(argv[i+1],"always")){ + MQTT_SUB_OPT_SET_RETAIN_HANDLING(cfg->sub_opts, MQTT_SUB_OPT_SEND_RETAIN_ALWAYS); + }else if(!strcmp(argv[i+1],"new")){ + MQTT_SUB_OPT_SET_RETAIN_HANDLING(cfg->sub_opts, MQTT_SUB_OPT_SEND_RETAIN_NEW); + }else if(!strcmp(argv[i+1],"never")){ + MQTT_SUB_OPT_SET_RETAIN_HANDLING(cfg->sub_opts, MQTT_SUB_OPT_SEND_RETAIN_NEVER); + }else{ + fprintf(stderr, "Error: Unknown value '%s' for --retain-handling.\n\n", argv[i+1]); + return 1; + } + } + i++; }else if(!strcmp(argv[i], "--retained-only")){ if(pub_or_sub != CLIENT_SUB){ goto unknown_option; diff --git a/client/rr_client.c b/client/rr_client.c index ea9cf995..23333034 100644 --- a/client/rr_client.c +++ b/client/rr_client.c @@ -115,7 +115,7 @@ void my_connect_callback(struct mosquitto *mosq, void *obj, int result, int flag connack_result = result; if(!result){ client_state = rr_s_connected; - mosquitto_subscribe_v5(mosq, NULL, cfg.response_topic, cfg.qos, 0, cfg.subscribe_props); + mosquitto_subscribe_v5(mosq, NULL, cfg.response_topic, cfg.qos, cfg.sub_opts, cfg.subscribe_props); }else{ client_state = rr_s_disconnect; if(result){ diff --git a/include/mosquitto/mqtt_protocol.h b/include/mosquitto/mqtt_protocol.h index 48efe416..b5a6b475 100644 --- a/include/mosquitto/mqtt_protocol.h +++ b/include/mosquitto/mqtt_protocol.h @@ -286,8 +286,11 @@ enum mqtt5_sub_options { #define MQTT_SUB_OPT_GET_NO_LOCAL(opt) ((opt) & MQTT_SUB_OPT_NO_LOCAL) #define MQTT_SUB_OPT_GET_RETAIN_AS_PUBLISHED(opt) ((opt) & MQTT_SUB_OPT_RETAIN_AS_PUBLISHED) #define MQTT_SUB_OPT_GET_SEND_RETAIN(opt) ((opt) & (MQTT_SUB_OPT_SEND_RETAIN_NEW | MQTT_SUB_OPT_SEND_RETAIN_NEVER)) +#define MQTT_SUB_OPT_GET_RETAIN_HANDLING(opt) ((opt) & (MQTT_SUB_OPT_SEND_RETAIN_NEW | MQTT_SUB_OPT_SEND_RETAIN_NEVER)) #define MQTT_SUB_OPT_SET_QOS(opt, qos) ((opt) = ((opt) & 0xFC) | ((qos) & 0x03)) +#define MQTT_SUB_OPT_SET_NO_LOCAL(opt, qos) ((opt) = ((opt) & 0xFC) | ((qos) & 0x03)) +#define MQTT_SUB_OPT_SET_RETAIN_HANDLING(opt, rh) ((opt) = ((opt) & 0xCF) | ((rh) & 0x30)) #define MQTT_SUB_OPT_SET(opt, val) ((opt) |= val) #define MQTT_SUB_OPT_CLEAR(opt, val) ((opt) = (opt) & !val) diff --git a/man/common/option-retain-handling.xml b/man/common/option-retain-handling.xml new file mode 100644 index 00000000..aecebd9b --- /dev/null +++ b/man/common/option-retain-handling.xml @@ -0,0 +1,24 @@ + + always | new | never + + + Use this option to control the retain handling option when making a + subscription. This controls under what circumstances an existing + retained message is sent to the client when the subscription is + made. + + + + - always deliver retained messages + + - deliver retained messages the first time + a subscription is made, but not on subsequent subscriptions. This + is useful for the case where you have a long running client using + a non-clean session. If the connection is dropped briefly, when the + client reconnects you will not receive the retained messages again. + + - never deliver retained messages + + + + diff --git a/man/mosquitto_rr.1.xml b/man/mosquitto_rr.1.xml index 5a85feea..a417a14f 100644 --- a/man/mosquitto_rr.1.xml +++ b/man/mosquitto_rr.1.xml @@ -76,6 +76,7 @@ keepalive-time message-QoS protocol-version + always | new | never session-expiry-interval @@ -607,6 +608,7 @@ their display. + diff --git a/man/mosquitto_sub.1.xml b/man/mosquitto_sub.1.xml index 4fa6618e..0027977c 100644 --- a/man/mosquitto_sub.1.xml +++ b/man/mosquitto_sub.1.xml @@ -78,6 +78,7 @@ keepalive-time message-QoS + always | new | never protocol-version session-expiry-interval @@ -663,6 +664,7 @@ mosquitto_sub -t 'bbc/#' -T bbc/bbc1 --remove-retained clients. + diff --git a/src/handle_subscribe.c b/src/handle_subscribe.c index cec73400..8ac38451 100644 --- a/src/handle_subscribe.c +++ b/src/handle_subscribe.c @@ -144,7 +144,7 @@ int handle__subscribe(struct mosquitto *context) mosquitto_FREE(payload); return MOSQ_ERR_PROTOCOL; } - retain_handling = MQTT_SUB_OPT_GET_SEND_RETAIN(sub.options); + retain_handling = MQTT_SUB_OPT_GET_RETAIN_HANDLING(sub.options); if(retain_handling == 0x30 || (sub.options & 0xC0) != 0){ mosquitto_FREE(sub.topic_filter); mosquitto_FREE(payload); diff --git a/test/client/02-subscribe-argv-errors-without-tls.py b/test/client/02-subscribe-argv-errors-without-tls.py index 36ef1f9a..f14a2c7d 100755 --- a/test/client/02-subscribe-argv-errors-without-tls.py +++ b/test/client/02-subscribe-argv-errors-without-tls.py @@ -120,6 +120,7 @@ if __name__ == '__main__': do_test(['-x', 'A'], "Error: session-expiry-interval not a number.\n\n" + helps, 1) do_test(['-x', '-2'], "Error: session-expiry-interval out of range.\n\n" + helps, 1) do_test(['-x', '4294967296'], "Error: session-expiry-interval out of range.\n\n" + helps, 1) + do_test(['--retain-handling', 'invalid'], "Error: Unknown value 'invalid' for --retain-handling.\n\n" + helps, 1) # Unknown options do_test(['--unknown'], "Error: Unknown option '--unknown'.\n" + helps, 1) diff --git a/test/client/02-subscribe-retain-handling.py b/test/client/02-subscribe-retain-handling.py new file mode 100755 index 00000000..80b1b802 --- /dev/null +++ b/test/client/02-subscribe-retain-handling.py @@ -0,0 +1,76 @@ +#!/usr/bin/env python3 + +# + +from mosq_test_helper import * + +def do_test(): + rc = 1 + + port = mosq_test.get_port() + + env = { + 'XDG_CONFIG_HOME':'/tmp/missing' + } + env = mosq_test.env_add_ld_library_path(env) + cmd = [f'{mosq_test.get_build_root()}/client/mosquitto_sub', + '-p', str(port), + '-q', '0', + '-t', 'retain-handling', + '-V', '5', + '-C', '1' + ] + + sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) + sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) + sock.settimeout(10) + sock.bind(('', port)) + sock.listen(5) + + props = mqtt5_props.gen_uint16_prop(mqtt5_props.PROP_RECEIVE_MAXIMUM, 1) + connect_packet = mosq_test.gen_connect("", proto_ver=5, properties=props) + connack_packet = mosq_test.gen_connack(rc=0, proto_ver=5) + + subscribe_packet_always = mosq_test.gen_subscribe(mid=1, topic="retain-handling", qos=0x00, proto_ver=5) + subscribe_packet_new = mosq_test.gen_subscribe(mid=1, topic="retain-handling", qos=0x10, proto_ver=5) + subscribe_packet_never = mosq_test.gen_subscribe(mid=1, topic="retain-handling", qos=0x20, proto_ver=5) + suback_packet = mosq_test.gen_suback(mid=1, qos=0) + + publish_packet = mosq_test.gen_publish("retain-handling", qos=0, payload="m", proto_ver=5) + + client_terminate_rc = 0 + + try: + for subscribe_packet, handling in [ + (subscribe_packet_always, 'always'), + (subscribe_packet_new, 'new'), + (subscribe_packet_never, 'never')]: + + client_cmd = cmd + ['--retain-handling', handling] + client = subprocess.Popen(client_cmd, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, env=env) + + (conn, address) = sock.accept() + conn.settimeout(5) + + mosq_test.expect_packet(conn, "connect", connect_packet) + conn.send(connack_packet) + + mosq_test.expect_packet(conn, f"subscribe {handling}", subscribe_packet) + conn.send(suback_packet) + conn.send(publish_packet) + + if mosq_test.wait_for_subprocess(client): + print("client not terminated") + client_terminate_rc = 1 + sock.close() + except mosq_test.TestError: + pass + except Exception as e: + print(e) + finally: + sock.close() + client.terminate() + exit(client_terminate_rc) + + +do_test() diff --git a/test/client/03-publish-argv-errors-without-tls.py b/test/client/03-publish-argv-errors-without-tls.py index 1554a8ae..34e6bd0a 100755 --- a/test/client/03-publish-argv-errors-without-tls.py +++ b/test/client/03-publish-argv-errors-without-tls.py @@ -116,6 +116,7 @@ if __name__ == '__main__': do_test(['--random-filter'], "Error: Unknown option '--random-filter'.\n" + helps, 1) do_test(['--remove-retained'], "Error: Unknown option '--remove-retained'.\n" + helps, 1) do_test(['--retain-as-published'], "Error: Unknown option '--retain-as-published'.\n" + helps, 1) + do_test(['--retain-handling', 'invalid'], "Error: Unknown option '--retain-handling'.\n" + helps, 1) do_test(['--retained-only'], "Error: Unknown option '--retained-only'.\n" + helps, 1) do_test(['-T'], "Error: Unknown option '-T'.\n" + helps, 1) do_test(['-U'], "Error: Unknown option '-U'.\n" + helps, 1) diff --git a/test/client/04-rr-argv-errors-without-tls.py b/test/client/04-rr-argv-errors-without-tls.py index d2556c38..24663ffe 100755 --- a/test/client/04-rr-argv-errors-without-tls.py +++ b/test/client/04-rr-argv-errors-without-tls.py @@ -105,6 +105,7 @@ if __name__ == '__main__': do_test(['-x', 'A'], "Error: session-expiry-interval not a number.\n\n" + helps, 1) do_test(['-x', '-2'], "Error: session-expiry-interval out of range.\n\n" + helps, 1) do_test(['-x', '4294967296'], "Error: session-expiry-interval out of range.\n\n" + helps, 1) + do_test(['--retain-handling', 'invalid'], "Error: Unknown value 'invalid' for --retain-handling.\n\n" + helps, 1) # Mixed message types do_test(['-m', 'message', '-f', 'file'], "Error: Only one type of message can be sent at once.\n\n" + helps, 1) diff --git a/test/client/04-rr-retain-handling.py b/test/client/04-rr-retain-handling.py new file mode 100755 index 00000000..e8b81b46 --- /dev/null +++ b/test/client/04-rr-retain-handling.py @@ -0,0 +1,81 @@ +#!/usr/bin/env python3 + +# + +from mosq_test_helper import * + +def do_test(): + rc = 1 + + port = mosq_test.get_port() + + env = { + 'XDG_CONFIG_HOME':'/tmp/missing' + } + env = mosq_test.env_add_ld_library_path(env) + cmd = [f'{mosq_test.get_build_root()}/client/mosquitto_rr', + '-p', str(port), + '-q', '0', + '-e', 'retain-handling', + '-t', 'retain-handling', + '-m', 'm', + '-V', '5', + ] + + sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) + sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) + sock.settimeout(5) + sock.bind(('', port)) + sock.listen(5) + + props = mqtt5_props.gen_uint16_prop(mqtt5_props.PROP_RECEIVE_MAXIMUM, 1) + connect_packet = mosq_test.gen_connect("", proto_ver=5, properties=props) + connack_packet = mosq_test.gen_connack(rc=0, proto_ver=5) + + subscribe_packet_always = mosq_test.gen_subscribe(mid=1, topic="retain-handling", qos=0x00, proto_ver=5) + subscribe_packet_new = mosq_test.gen_subscribe(mid=1, topic="retain-handling", qos=0x10, proto_ver=5) + subscribe_packet_never = mosq_test.gen_subscribe(mid=1, topic="retain-handling", qos=0x20, proto_ver=5) + suback_packet = mosq_test.gen_suback(mid=1, qos=0, proto_ver=5) + + props = mqtt5_props.gen_string_prop(mqtt5_props.PROP_RESPONSE_TOPIC, "retain-handling") + publish_packet_in = mosq_test.gen_publish("retain-handling", qos=0, payload="m", proto_ver=5, properties=props) + publish_packet_out = mosq_test.gen_publish("retain-handling", qos=0, payload="m", proto_ver=5) + + client_terminate_rc = 0 + + try: + for subscribe_packet, handling in [ + (subscribe_packet_always, 'always'), + (subscribe_packet_new, 'new'), + (subscribe_packet_never, 'never')]: + + client_cmd = cmd + ['--retain-handling', handling] + client = subprocess.Popen(client_cmd, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, env=env) + + (conn, address) = sock.accept() + conn.settimeout(5) + + mosq_test.expect_packet(conn, "connect", connect_packet) + conn.send(connack_packet) + + mosq_test.expect_packet(conn, f"subscribe {handling}", subscribe_packet) + conn.send(suback_packet) + + mosq_test.expect_packet(conn, "publish}", publish_packet_in) + conn.send(publish_packet_out) + + if mosq_test.wait_for_subprocess(client): + print("client not terminated") + client_terminate_rc = 1 + sock.close() + except mosq_test.TestError: + pass + except Exception as e: + print(e) + finally: + sock.close() + client.terminate() + exit(client_terminate_rc) + + +do_test() diff --git a/test/client/Makefile b/test/client/Makefile index 13067b4a..d182596c 100644 --- a/test/client/Makefile +++ b/test/client/Makefile @@ -32,6 +32,7 @@ endif ifeq ($(WITH_WEBSOCKETS),yes) ./02-subscribe-qos1-ws.py endif + ./02-subscribe-retain-handling.py ./02-subscribe-format.py ./02-subscribe-null.py ./02-subscribe-verbose.py @@ -76,6 +77,7 @@ endif ifeq ($(WITH_WEBSOCKETS),yes) ./04-rr-qos1-ws.py endif + ./04-rr-retain-handling.py ptest : ./test.sh diff --git a/test/client/test.py b/test/client/test.py index 4b66142f..93302d59 100755 --- a/test/client/test.py +++ b/test/client/test.py @@ -19,6 +19,7 @@ tests = [ (1, './02-subscribe-null.py'), (1, './02-subscribe-qos1.py'), (2, './02-subscribe-qos1-ws.py'), + (1, './02-subscribe-retain-handling.py'), (1, './02-subscribe-verbose.py'), (1, './03-publish-argv-errors-tls-psk.py'), @@ -48,6 +49,7 @@ tests = [ (1, './04-rr-env.py'), (1, './04-rr-qos1.py'), (2, './04-rr-qos1-ws.py'), + (1, './04-rr-retain-handling.py'), ] if __name__ == "__main__":