diff --git a/src/persist.h b/src/persist.h index 9fedc3ea..47a5731e 100644 --- a/src/persist.h +++ b/src/persist.h @@ -100,6 +100,7 @@ struct P_msg_store{ mosquitto__payload_uhpa payload; struct mosquitto source; char *topic; + mosquitto_property *properties; }; @@ -140,7 +141,7 @@ int persist__chunk_header_read_v5(FILE *db_fptr, int *chunk, int *length); int persist__chunk_cfg_read_v5(FILE *db_fptr, struct PF_cfg *chunk); int persist__chunk_client_read_v5(FILE *db_fptr, struct P_client *chunk); int persist__chunk_client_msg_read_v5(FILE *db_fptr, struct P_client_msg *chunk); -int persist__chunk_msg_store_read_v5(FILE *db_fptr, struct P_msg_store *chunk); +int persist__chunk_msg_store_read_v5(FILE *db_fptr, struct P_msg_store *chunk, uint32_t length); int persist__chunk_retain_read_v5(FILE *db_fptr, struct P_retain *chunk); int persist__chunk_sub_read_v5(FILE *db_fptr, struct P_sub *chunk); diff --git a/src/persist_read.c b/src/persist_read.c index a8514691..e397954a 100644 --- a/src/persist_read.c +++ b/src/persist_read.c @@ -221,7 +221,7 @@ static int persist__client_msg_chunk_restore(struct mosquitto_db *db, FILE *db_f } -static int persist__msg_store_chunk_restore(struct mosquitto_db *db, FILE *db_fptr) +static int persist__msg_store_chunk_restore(struct mosquitto_db *db, FILE *db_fptr, uint32_t length) { struct P_msg_store chunk; struct mosquitto_msg_store *stored = NULL; @@ -232,7 +232,7 @@ static int persist__msg_store_chunk_restore(struct mosquitto_db *db, FILE *db_fp memset(&chunk, 0, sizeof(struct P_msg_store)); if(db_version == 5){ - rc = persist__chunk_msg_store_read_v5(db_fptr, &chunk); + rc = persist__chunk_msg_store_read_v5(db_fptr, &chunk, length); }else{ rc = persist__chunk_msg_store_read_v234(db_fptr, &chunk, db_version); } @@ -262,7 +262,7 @@ static int persist__msg_store_chunk_restore(struct mosquitto_db *db, FILE *db_fp rc = db__message_store(db, &chunk.source, chunk.F.source_mid, chunk.topic, chunk.F.qos, chunk.F.payloadlen, - &chunk.payload, chunk.F.retain, &stored, 0, NULL, chunk.F.store_id); + &chunk.payload, chunk.F.retain, &stored, 0, chunk.properties, chunk.F.store_id); mosquitto__free(chunk.source.id); mosquitto__free(chunk.source.username); @@ -421,7 +421,7 @@ int persist__restore(struct mosquitto_db *db) break; case DB_CHUNK_MSG_STORE: - if(persist__msg_store_chunk_restore(db, fptr)) return 1; + if(persist__msg_store_chunk_restore(db, fptr, length)) return 1; break; case DB_CHUNK_CLIENT_MSG: diff --git a/src/persist_read_v5.c b/src/persist_read_v5.c index 32441915..4e1d144c 100644 --- a/src/persist_read_v5.c +++ b/src/persist_read_v5.c @@ -31,7 +31,9 @@ Contributors: #include "mosquitto_broker_internal.h" #include "memory_mosq.h" +#include "mqtt_protocol.h" #include "persist.h" +#include "property_mosq.h" #include "time_mosq.h" #include "util_mosq.h" @@ -96,9 +98,11 @@ error: } -int persist__chunk_msg_store_read_v5(FILE *db_fptr, struct P_msg_store *chunk) +int persist__chunk_msg_store_read_v5(FILE *db_fptr, struct P_msg_store *chunk, uint32_t length) { int rc = 0; + mosquitto_property *properties = NULL; + struct mosquitto__packet prop_packet; read_e(db_fptr, &chunk->F, sizeof(struct PF_msg_store)); chunk->F.payloadlen = ntohl(chunk->F.payloadlen); @@ -108,6 +112,8 @@ int persist__chunk_msg_store_read_v5(FILE *db_fptr, struct P_msg_store *chunk) chunk->F.topic_len = ntohs(chunk->F.topic_len); chunk->F.source_port = ntohs(chunk->F.source_port); + length -= (sizeof(struct PF_msg_store) + chunk->F.payloadlen + chunk->F.source_id_len + chunk->F.source_username_len + chunk->F.topic_len); + if(chunk->F.source_id_len){ rc = persist__read_string_len(db_fptr, &chunk->source.id, chunk->F.source_id_len); if(rc){ @@ -145,12 +151,35 @@ int persist__chunk_msg_store_read_v5(FILE *db_fptr, struct P_msg_store *chunk) read_e(db_fptr, UHPA_ACCESS(chunk->payload, chunk->F.payloadlen), chunk->F.payloadlen); } + if(length > 0){ + memset(&prop_packet, 0, sizeof(struct mosquitto__packet)); + prop_packet.remaining_length = length; + prop_packet.payload = mosquitto__malloc(length); + if(!prop_packet.payload){ + mosquitto__free(chunk->source.id); + mosquitto__free(chunk->source.username); + mosquitto__free(chunk->topic); + return 1; + } + read_e(db_fptr, prop_packet.payload, length); + rc = property__read_all(CMD_PUBLISH, &prop_packet, &properties); + mosquitto__free(prop_packet.payload); + if(rc){ + mosquitto__free(chunk->source.id); + mosquitto__free(chunk->source.username); + mosquitto__free(chunk->topic); + return rc; + } + } + chunk->properties = properties; + return MOSQ_ERR_SUCCESS; error: log__printf(NULL, MOSQ_LOG_ERR, "Error: %s.", strerror(errno)); mosquitto__free(chunk->source.id); mosquitto__free(chunk->source.username); mosquitto__free(chunk->topic); + mosquitto__free(prop_packet.payload); return 1; } diff --git a/src/persist_write.c b/src/persist_write.c index 8e83ed51..f79b60ff 100644 --- a/src/persist_write.c +++ b/src/persist_write.c @@ -144,6 +144,7 @@ static int persist__message_store_save(struct mosquitto_db *db, FILE *db_fptr) } chunk.F.qos = stored->qos; chunk.payload = stored->payload; + chunk.properties = stored->properties; rc = persist__chunk_message_store_write_v5(db_fptr, &chunk); if(rc){ diff --git a/src/persist_write_v5.c b/src/persist_write_v5.c index 90fc3a1f..2dd6439c 100644 --- a/src/persist_write_v5.c +++ b/src/persist_write_v5.c @@ -32,6 +32,8 @@ Contributors: #include "mosquitto_broker_internal.h" #include "memory_mosq.h" #include "persist.h" +#include "packet_mosq.h" +#include "property_mosq.h" #include "time_mosq.h" #include "util_mosq.h" @@ -104,6 +106,15 @@ int persist__chunk_message_store_write_v5(FILE *db_fptr, struct P_msg_store *chu int source_id_len = chunk->F.source_id_len; int source_username_len = chunk->F.source_username_len; int topic_len = chunk->F.topic_len; + uint32_t proplen = 0; + struct mosquitto__packet prop_packet; + int rc; + + memset(&prop_packet, 0, sizeof(struct mosquitto__packet)); + if(chunk->properties){ + proplen = property__get_length_all(chunk->properties); + proplen += packet__varint_bytes(proplen); + } chunk->F.payloadlen = htonl(chunk->F.payloadlen); chunk->F.source_mid = htons(chunk->F.source_mid); @@ -115,7 +126,7 @@ int persist__chunk_message_store_write_v5(FILE *db_fptr, struct P_msg_store *chu header.chunk = htonl(DB_CHUNK_MSG_STORE); header.length = htonl(sizeof(struct PF_msg_store) + topic_len + payloadlen + - source_id_len + source_username_len); + source_id_len + source_username_len + proplen); write_e(db_fptr, &header, sizeof(struct PF_header)); write_e(db_fptr, &chunk->F, sizeof(struct PF_msg_store)); @@ -129,10 +140,26 @@ int persist__chunk_message_store_write_v5(FILE *db_fptr, struct P_msg_store *chu if(payloadlen){ write_e(db_fptr, UHPA_ACCESS(chunk->payload, payloadlen), (unsigned int)payloadlen); } + if(chunk->properties){ + if(proplen > 0){ + prop_packet.remaining_length = proplen; + prop_packet.packet_length = proplen; + prop_packet.payload = mosquitto__malloc(proplen); + if(!prop_packet.payload){ + return MOSQ_ERR_NOMEM; + } + rc = property__write_all(&prop_packet, chunk->properties, true); + if(rc) return rc; + + write_e(db_fptr, prop_packet.payload, proplen); + mosquitto__free(prop_packet.payload); + } + } return MOSQ_ERR_SUCCESS; error: log__printf(NULL, MOSQ_LOG_ERR, "Error: %s.", strerror(errno)); + mosquitto__free(prop_packet.payload); return 1; } diff --git a/test/broker/11-pub-props.py b/test/broker/11-pub-props.py new file mode 100755 index 00000000..845ed2c4 --- /dev/null +++ b/test/broker/11-pub-props.py @@ -0,0 +1,77 @@ +#!/usr/bin/env python + +# Does a persisted PUBLISH keep its properties? + +from mosq_test_helper import * + +def write_config(filename, port): + with open(filename, 'w') as f: + f.write("port %d\n" % (port)) + f.write("persistence true\n") + f.write("persistence_file mosquitto-%d.db\n" % (port)) + +port = mosq_test.get_port() +conf_file = os.path.basename(__file__).replace('.py', '.conf') +write_config(conf_file, port) + +rc = 1 +keepalive = 60 +connect_packet = mosq_test.gen_connect( + "persistent-props-test", keepalive=keepalive, clean_session=True, proto_ver=5 +) +connack_packet = mosq_test.gen_connack(rc=0, proto_ver=5) + +mid = 1 +props = mqtt5_props.gen_byte_prop(mqtt5_props.PROP_PAYLOAD_FORMAT_INDICATOR, 1) +props += mqtt5_props.gen_string_prop(mqtt5_props.PROP_CONTENT_TYPE, "plain/text") +props += mqtt5_props.gen_string_prop(mqtt5_props.PROP_RESPONSE_TOPIC, "/dev/null") +props += mqtt5_props.gen_string_prop(mqtt5_props.PROP_CORRELATION_DATA, "2357289375902345") +props += mqtt5_props.gen_string_pair_prop(mqtt5_props.PROP_USER_PROPERTY, "name", "value") +publish_packet = mosq_test.gen_publish("subpub/qos1", qos=1, mid=mid, payload="message", proto_ver=5, properties=props, retain=True) +puback_packet = mosq_test.gen_puback(mid, reason_code=mqtt5_rc.MQTT_RC_NO_MATCHING_SUBSCRIBERS, proto_ver=5) + +publish2_packet = mosq_test.gen_publish("subpub/qos1", qos=0, payload="message", proto_ver=5, properties=props, retain=True) + +mid = 1 +subscribe_packet = mosq_test.gen_subscribe(mid, "subpub/qos1", 0, proto_ver=5) +suback_packet = mosq_test.gen_suback(mid, 0, proto_ver=5) + +if os.path.exists('mosquitto-%d.db' % (port)): + os.unlink('mosquitto-%d.db' % (port)) + +broker = mosq_test.start_broker(filename=os.path.basename(__file__), use_conf=True, port=port) + +(stdo1, stde1) = ("", "") +try: + sock = mosq_test.do_client_connect(connect_packet, connack_packet, timeout=20, port=port) + mosq_test.do_send_receive(sock, publish_packet, puback_packet, "puback") + mosq_test.do_send_receive(sock, subscribe_packet, suback_packet, "suback") + + if mosq_test.expect_packet(sock, "publish2", publish2_packet): + + broker.terminate() + broker.wait() + (stdo1, stde1) = broker.communicate() + sock.close() + broker = mosq_test.start_broker(filename=os.path.basename(__file__), use_conf=True, port=port) + + 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") + + if mosq_test.expect_packet(sock, "publish2", publish2_packet): + rc = 0 + + sock.close() +finally: + os.remove(conf_file) + broker.terminate() + broker.wait() + (stdo, stde) = broker.communicate() + if rc: + print(stde1 + stde) + if os.path.exists('mosquitto-%d.db' % (port)): + os.unlink('mosquitto-%d.db' % (port)) + + +exit(rc) + diff --git a/test/broker/Makefile b/test/broker/Makefile index 5d9a7a5d..5c312386 100644 --- a/test/broker/Makefile +++ b/test/broker/Makefile @@ -181,6 +181,7 @@ endif ./11-persistent-subscription.py ./11-persistent-subscription-v5.py ./11-persistent-subscription-no-local.py + ./11-pub-props.py 12 : ./12-prop-assigned-client-identifier.py diff --git a/test/broker/test.py b/test/broker/test.py index 3f0f5664..bb7628c9 100755 --- a/test/broker/test.py +++ b/test/broker/test.py @@ -149,6 +149,7 @@ tests = [ (1, './11-persistent-subscription.py'), (1, './11-persistent-subscription-v5.py'), (1, './11-persistent-subscription-no-local.py'), + (1, './11-pub-props.py'), (1, './12-prop-assigned-client-identifier.py'), (1, './12-prop-maximum-packet-size-broker.py'), diff --git a/test/unit/Makefile b/test/unit/Makefile index d7ca6632..26934323 100644 --- a/test/unit/Makefile +++ b/test/unit/Makefile @@ -28,9 +28,12 @@ PERSIST_READ_TEST_OBJS = \ PERSIST_READ_OBJS = \ memory_mosq.o \ + packet_datatypes.o \ persist_read.o \ persist_read_v234.o \ persist_read_v5.o \ + property_mosq.o \ + utf8_mosq.o \ util_mosq.o PERSIST_WRITE_TEST_OBJS = \ @@ -40,12 +43,15 @@ PERSIST_WRITE_TEST_OBJS = \ PERSIST_WRITE_OBJS = \ database.o \ memory_mosq.o \ + packet_datatypes.o \ persist_read.o \ persist_read_v234.o \ persist_read_v5.o \ persist_write.o \ persist_write_v5.o \ + property_mosq.o \ subs.o \ + utf8_mosq.o \ util_mosq.o all : test diff --git a/test/unit/files/persist_read/v5-message-store.test-db b/test/unit/files/persist_read/v5-message-store.test-db index 684c24de..dc9b7f77 100644 Binary files a/test/unit/files/persist_read/v5-message-store.test-db and b/test/unit/files/persist_read/v5-message-store.test-db differ diff --git a/test/unit/files/persist_read/v5-retain.test-db b/test/unit/files/persist_read/v5-retain.test-db index 0265060f..fa839288 100644 Binary files a/test/unit/files/persist_read/v5-retain.test-db and b/test/unit/files/persist_read/v5-retain.test-db differ diff --git a/test/unit/persist_write_stubs.c b/test/unit/persist_write_stubs.c index dfc3a06f..97bb8e05 100644 --- a/test/unit/persist_write_stubs.c +++ b/test/unit/persist_write_stubs.c @@ -38,15 +38,6 @@ int send__pingreq(struct mosquitto *mosq) return MOSQ_ERR_SUCCESS; } -int mosquitto_property_add_varint(mosquitto_property **proplist, int identifier, uint32_t value) -{ - return MOSQ_ERR_SUCCESS; -} - -void mosquitto_property_free_all(mosquitto_property **properties) -{ -} - int mosquitto_acl_check(struct mosquitto_db *db, struct mosquitto *context, const char *topic, long payloadlen, void* payload, int qos, bool retain, int access) { return MOSQ_ERR_SUCCESS; diff --git a/test/unit/persist_write_test.c b/test/unit/persist_write_test.c index a4e41592..1b4645e1 100644 --- a/test/unit/persist_write_test.c +++ b/test/unit/persist_write_test.c @@ -238,7 +238,7 @@ static void TEST_v5_sub(void) CU_ASSERT_EQUAL(rc, MOSQ_ERR_SUCCESS); CU_ASSERT_EQUAL(0, file_diff("files/persist_read/v5-sub.test-db", "v5-sub.db")); - //unlink("v5-sub.db"); + unlink("v5-sub.db"); }