Mirrors_Framework/mosquitto
1
0
Fork 0
mirror of https://github.com/eclipse-mosquitto/mosquitto.git synced 2026-09-23 08:33:48 +08:00
Code Issues Packages Projects Releases Wiki Activity

Add --retain-handling to sub and rr clients

Browse Source
This commit is contained in:
Roger A. Light
2025-08-11 12:17:17 +01:00
parent d48eaabef5
commit 32c2b64081
15 changed files with 219 additions and 2 deletions
Download Patch File Download Diff File Expand all files Collapse all files
ChangeLog.txt
+2
View File
@@ -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
client/client_shared.c
+20
View File
@@ -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;
client/rr_client.c
+1 -1
View File
@@ -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){
include/mosquitto/mqtt_protocol.h
+3
View File
@@ -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)
man/common/option-retain-handling.xml
+24
View File
@@ -0,0 +1,24 @@
<varlistentry>
<term><option>--retain-handling</option> always | new | never</term>
<listitem>
<para>
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.
</para>
<itemizedlist mark="circle">
<listitem><para><option>always</option> - always deliver retained messages</para></listitem>
<listitem><para>
<option>new</option> - 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.
</para></listitem>
<listitem><para><option>never</option> - never deliver retained messages</para></listitem>
</itemizedlist>
</listitem>
</varlistentry>
man/mosquitto_rr.1.xml
+2
View File
@@ -76,6 +76,7 @@
<arg><option>-k</option> <replaceable>keepalive-time</replaceable></arg>
<arg><option>-q</option> <replaceable>message-QoS</replaceable></arg>
<arg><option>-V</option> <replaceable>protocol-version</replaceable></arg>
<arg><option>--retain-handling</option> always | new | never</arg>
<arg><option>-x</option> <replaceable>session-expiry-interval</replaceable></arg>
<sbr/>
<arg>
@@ -607,6 +608,7 @@
their display.</para>
</listitem>
</varlistentry>
<xi:include xmlns:xi="http://www.w3.org/2001/XInclude" href="common/option-retain-handling.xml" />
<varlistentry>
<term><option>-S</option></term>
<listitem>
man/mosquitto_sub.1.xml
+2
View File
@@ -78,6 +78,7 @@
<arg><option>-k</option> <replaceable>keepalive-time</replaceable></arg>
<arg><option>-q</option> <replaceable>message-QoS</replaceable></arg>
<arg><option>--retain-as-published</option></arg>
<arg><option>--retain-handling</option> always | new | never</arg>
<arg><option>-V</option> <replaceable>protocol-version</replaceable></arg>
<arg><option>-x</option> <replaceable>session-expiry-interval</replaceable></arg>
<sbr/>
@@ -663,6 +664,7 @@ mosquitto_sub -t 'bbc/#' -T bbc/bbc1 --remove-retained</programlisting>
clients.</para>
</listitem>
</varlistentry>
<xi:include xmlns:xi="http://www.w3.org/2001/XInclude" href="common/option-retain-handling.xml" />
<varlistentry>
<term><option>-S</option></term>
<listitem>
src/handle_subscribe.c
+1 -1
View File
@@ -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);
test/client/02-subscribe-argv-errors-without-tls.py
+1
View File
@@ -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)
test/client/02-subscribe-retain-handling.py
Executable
+76
View File
@@ -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()
test/client/03-publish-argv-errors-without-tls.py
+1
View File
@@ -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)
test/client/04-rr-argv-errors-without-tls.py
+1
View File
@@ -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)
test/client/04-rr-retain-handling.py
Executable
+81
View File
@@ -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()
test/client/Makefile
+2
View File
@@ -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
test/client/test.py
+2
View File
@@ -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__":