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

Merge branch 'fixes'

Browse Source
This commit is contained in:
Roger A. Light
2019-06-18 12:45:08 +01:00
parent 7a581ab64f 4f32e94104
commit be73f79200
84 changed files with 881 additions and 499 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.github/lock.yml
+35
View File
@@ -0,0 +1,35 @@
# Configuration for Lock Threads - https://github.com/dessant/lock-threads
# Number of days of inactivity before a closed issue or pull request is locked
daysUntilLock: 90
# Skip issues and pull requests created before a given timestamp. Timestamp must
# follow ISO 8601 (`YYYY-MM-DD`). Set to `false` to disable
skipCreatedBefore: false
# Issues and pull requests with these labels will be ignored. Set to `[]` to disable
exemptLabels: []
# Label to add before locking, such as `outdated`. Set to `false` to disable
lockLabel: false
# Comment to post before locking. Set to `false` to disable
lockComment: false
# Assign `resolved` as the reason for locking. Set to `false` to disable
setLockReason: true
# Limit to only `issues` or `pulls`
only: issues
# Optionally, specify configuration settings just for `issues` or `pulls`
# issues:
# exemptLabels:
# - help-wanted
# lockLabel: outdated
# pulls:
# daysUntilLock: 30
# Repository to extend settings from
# _extends: repo
CMakeLists.txt
+1 -2
View File
@@ -11,7 +11,7 @@ project(mosquitto)
cmake_minimum_required(VERSION 2.8)
# Only for version 3 and up. cmake_policy(SET CMP0042 NEW)
set (VERSION 1.6.2)
set (VERSION 1.6.3)
add_definitions (-DCMAKE -DVERSION=\"${VERSION}\")
@@ -116,4 +116,3 @@ install(FILES "${CMAKE_CURRENT_BINARY_DIR}/libmosquittopp.pc" DESTINATION "${CMA
# Testing
# ========================================
enable_testing()
add_test("test" make -C ${mosquitto_SOURCE_DIR}/test test)
ChangeLog.txt
+59
View File
@@ -1,3 +1,62 @@
1.6.3 - 20190618
================
Broker:
- Fix detection of incoming v3.1/v3.1.1 bridges. Closes #1263.
- Fix default max_topic_alias listener config not being copied to the in-use
listener when compiled without TLS support.
- Fix random number generation if compiling using `WITH_TLS=no` and on Linux
with glibc >= 2.25. Without this fix, no random numbers would be generated
for e.g. on broker client id generation, and so clients connecting expecting
this feature would be unable to connect.
- Fix compilation problem related to `getrandom()` on non-glibc systems.
- Fix Will message for a persistent client incorrectly being sent when the
client reconnects after a clean disconnect. Closes #1273.
- Fix Will message for a persistent client not being sent on disconnect.
Closes #1273.
- Improve documentation around the upgrading of persistence files. Closes
#1276.
- Add 'extern "C"' on mosquitto_broker.h and mosquitto_plugin.h for C++ plugin
writing. Closes #1290.
- Fix persistent Websockets clients not receiving messages after they
reconnect, having sent DISCONNECT on a previous session. Closes #1227.
- Disable TLS renegotiation. Client initiated renegotiation is considered to
be a potential attack vector against servers. Closes #1257.
- Fix incorrect shared subscription topic '$shared'.
- Fix zero length client ids being rejected for MQTT v5 clients with clean
start set to true.
- Fix MQTT v5 overlapping subscription behaviour. Clients now receive message
from all matching subscriptions rather than the first one encountered, which
ensures the maximum QoS requirement is met.
- Fix incoming/outgoing quota problems for QoS>0.
- Remove obsolete `store_clean_interval` from documentation.
- Fix v4 authentication plugin never calling psk_key_get.
Client library:
- Fix typo causing build error on Windows when building without TLS support.
Closes #1264.
Clients:
- Fix -L url parsing when `/topic` part is missing.
- Stop some error messages being printed even when `--quiet` was used.
Closes #1284.
- Fix mosquitto_pub exiting with error code 0 when an error occurred.
Closes #1285.
- Fix mosquitto_pub not using the `-c` option. Closes #1273.
- Fix MQTT v5 clients not being able to specify a password without a username.
Closes #1274.
- Fix `mosquitto_pub -l` not handling network failures. Closes #1152.
- Fix `mosquitto_pub -l` not handling zero length input. Closes #1302.
- Fix double free on exit in mosquitto_pub. Closes #1280.
Documentation:
- Remove references to Python binding and C++ wrapper in libmosquitto man
page. Closes #1266.
Build:
- CLIENT_LDFLAGS now uses LDFLAGS. Closes #1294.
1.6.2 - 20190430
================
client/client_shared.c
+62 -55
View File
@@ -18,6 +18,7 @@ Contributors:
#include <errno.h>
#include <fcntl.h>
#include <stdarg.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
@@ -229,7 +230,7 @@ int client_config_load(struct mosq_config *cfg, int pub_or_sub, int argc, char *
len = strlen(env) + strlen("/mosquitto_pub") + 1;
loc = malloc(len);
if(!loc){
fprintf(stderr, "Error: Out of memory.\n");
err_printf(cfg, "Error: Out of memory.\n");
return 1;
}
if(pub_or_sub == CLIENT_PUB){
@@ -246,7 +247,7 @@ int client_config_load(struct mosq_config *cfg, int pub_or_sub, int argc, char *
len = strlen(env) + strlen("/.config/mosquitto_pub") + 1;
loc = malloc(len);
if(!loc){
fprintf(stderr, "Error: Out of memory.\n");
err_printf(cfg, "Error: Out of memory.\n");
return 1;
}
if(pub_or_sub == CLIENT_PUB){
@@ -257,8 +258,6 @@ int client_config_load(struct mosq_config *cfg, int pub_or_sub, int argc, char *
snprintf(loc, len, "%s/.config/mosquitto_rr", env);
}
loc[len-1] = '\0';
}else{
fprintf(stderr, "Warning: Unable to locate configuration directory, default config not loaded.\n");
}
}
@@ -268,7 +267,7 @@ int client_config_load(struct mosq_config *cfg, int pub_or_sub, int argc, char *
len = strlen(env) + strlen("\\mosquitto_pub.conf") + 1;
loc = malloc(len);
if(!loc){
fprintf(stderr, "Error: Out of memory.\n");
err_printf(cfg, "Error: Out of memory.\n");
return 1;
}
if(pub_or_sub == CLIENT_PUB){
@@ -279,8 +278,6 @@ int client_config_load(struct mosq_config *cfg, int pub_or_sub, int argc, char *
snprintf(loc, len, "%s\\mosquitto_rr.conf", env);
}
loc[len-1] = '\0';
}else{
fprintf(stderr, "Warning: Unable to locate configuration directory, default config not loaded.\n");
}
#endif
@@ -328,9 +325,6 @@ int client_config_load(struct mosq_config *cfg, int pub_or_sub, int argc, char *
fprintf(stderr, "Error: Will retain given, but no will topic given.\n");
return 1;
}
if(cfg->password && !cfg->username){
if(!cfg->quiet) fprintf(stderr, "Warning: Not using password since username not set.\n");
}
#ifdef WITH_TLS
if((cfg->certfile && !cfg->keyfile) || (cfg->keyfile && !cfg->certfile)){
fprintf(stderr, "Error: Both certfile and keyfile must be provided if one of them is set.\n");
@@ -347,23 +341,23 @@ int client_config_load(struct mosq_config *cfg, int pub_or_sub, int argc, char *
#endif
#ifdef FINAL_WITH_TLS_PSK
if((cfg->cafile || cfg->capath) && cfg->psk){
if(!cfg->quiet) fprintf(stderr, "Error: Only one of --psk or --cafile/--capath may be used at once.\n");
fprintf(stderr, "Error: Only one of --psk or --cafile/--capath may be used at once.\n");
return 1;
}
if(cfg->psk && !cfg->psk_identity){
if(!cfg->quiet) fprintf(stderr, "Error: --psk-identity required if --psk used.\n");
fprintf(stderr, "Error: --psk-identity required if --psk used.\n");
return 1;
}
#endif
if(cfg->clean_session == false && (cfg->id_prefix || !cfg->id)){
if(!cfg->quiet) fprintf(stderr, "Error: You must provide a client id if you are using the -c option.\n");
fprintf(stderr, "Error: You must provide a client id if you are using the -c option.\n");
return 1;
}
if(pub_or_sub == CLIENT_SUB){
if(cfg->topic_count == 0){
if(!cfg->quiet) fprintf(stderr, "Error: You must specify a topic to subscribe to.\n");
fprintf(stderr, "Error: You must specify a topic to subscribe to.\n");
return 1;
}
}
@@ -371,39 +365,39 @@ int client_config_load(struct mosq_config *cfg, int pub_or_sub, int argc, char *
if(!cfg->host){
cfg->host = strdup("localhost");
if(!cfg->host){
if(!cfg->quiet) fprintf(stderr, "Error: Out of memory.\n");
err_printf(cfg, "Error: Out of memory.\n");
return 1;
}
}
rc = mosquitto_property_check_all(CMD_CONNECT, cfg->connect_props);
if(rc){
if(!cfg->quiet) fprintf(stderr, "Error in CONNECT properties: %s\n", mosquitto_strerror(rc));
err_printf(cfg, "Error in CONNECT properties: %s\n", mosquitto_strerror(rc));
return 1;
}
rc = mosquitto_property_check_all(CMD_PUBLISH, cfg->publish_props);
if(rc){
if(!cfg->quiet) fprintf(stderr, "Error in PUBLISH properties: %s\n", mosquitto_strerror(rc));
err_printf(cfg, "Error in PUBLISH properties: %s\n", mosquitto_strerror(rc));
return 1;
}
rc = mosquitto_property_check_all(CMD_SUBSCRIBE, cfg->subscribe_props);
if(rc){
if(!cfg->quiet) fprintf(stderr, "Error in SUBSCRIBE properties: %s\n", mosquitto_strerror(rc));
err_printf(cfg, "Error in SUBSCRIBE properties: %s\n", mosquitto_strerror(rc));
return 1;
}
rc = mosquitto_property_check_all(CMD_UNSUBSCRIBE, cfg->unsubscribe_props);
if(rc){
if(!cfg->quiet) fprintf(stderr, "Error in UNSUBSCRIBE properties: %s\n", mosquitto_strerror(rc));
err_printf(cfg, "Error in UNSUBSCRIBE properties: %s\n", mosquitto_strerror(rc));
return 1;
}
rc = mosquitto_property_check_all(CMD_DISCONNECT, cfg->disconnect_props);
if(rc){
if(!cfg->quiet) fprintf(stderr, "Error in DISCONNECT properties: %s\n", mosquitto_strerror(rc));
err_printf(cfg, "Error in DISCONNECT properties: %s\n", mosquitto_strerror(rc));
return 1;
}
rc = mosquitto_property_check_all(CMD_WILL, cfg->will_props);
if(rc){
if(!cfg->quiet) fprintf(stderr, "Error in Will properties: %s\n", mosquitto_strerror(rc));
err_printf(cfg, "Error in Will properties: %s\n", mosquitto_strerror(rc));
return 1;
}
@@ -436,7 +430,7 @@ int cfg_add_topic(struct mosq_config *cfg, int type, char *topic, const char *ar
cfg->topic_count++;
cfg->topics = realloc(cfg->topics, cfg->topic_count*sizeof(char *));
if(!cfg->topics){
fprintf(stderr, "Error: Out of memory.\n");
err_printf(cfg, "Error: Out of memory.\n");
return 1;
}
cfg->topics[cfg->topic_count-1] = strdup(topic);
@@ -551,7 +545,7 @@ int client_config_line_proc(struct mosq_config *cfg, int pub_or_sub, int argc, c
cfg->pub_mode = MSGMODE_FILE;
cfg->file_input = strdup(argv[i+1]);
if(!cfg->file_input){
fprintf(stderr, "Error: Out of memory.\n");
err_printf(cfg, "Error: Out of memory.\n");
return 1;
}
}
@@ -662,6 +656,10 @@ int client_config_line_proc(struct mosq_config *cfg, int pub_or_sub, int argc, c
return 1;
}
topic = strchr(url, '/');
if(!topic){
fprintf(stderr, "Error: Invalid URL for -L argument specified - topic missing.\n");
return 1;
}
*topic++ = 0;
if(cfg_add_topic(cfg, pub_or_sub, topic, "-L topic"))
@@ -1077,14 +1075,14 @@ int client_opts_set(struct mosquitto *mosq, struct mosq_config *cfg)
cfg->will_payloadlen, cfg->will_payload, cfg->will_qos,
cfg->will_retain, cfg->will_props)){
if(!cfg->quiet) fprintf(stderr, "Error: Problem setting will.\n");
err_printf(cfg, "Error: Problem setting will.\n");
mosquitto_lib_cleanup();
return 1;
}
cfg->will_props = NULL;
if(cfg->username && mosquitto_username_pw_set(mosq, cfg->username, cfg->password)){
if(!cfg->quiet) fprintf(stderr, "Error: Problem setting username and password.\n");
if((cfg->username || cfg->password) && mosquitto_username_pw_set(mosq, cfg->username, cfg->password)){
err_printf(cfg, "Error: Problem setting username and/or password.\n");
mosquitto_lib_cleanup();
return 1;
}
@@ -1093,48 +1091,48 @@ int client_opts_set(struct mosquitto *mosq, struct mosq_config *cfg)
rc = mosquitto_tls_set(mosq, cfg->cafile, cfg->capath, cfg->certfile, cfg->keyfile, NULL);
if(rc){
if(rc == MOSQ_ERR_INVAL){
if(!cfg->quiet) fprintf(stderr, "Error: Problem setting TLS options: File not found.\n");
err_printf(cfg, "Error: Problem setting TLS options: File not found.\n");
}else{
if(!cfg->quiet) fprintf(stderr, "Error: Problem setting TLS options: %s.\n", mosquitto_strerror(rc));
err_printf(cfg, "Error: Problem setting TLS options: %s.\n", mosquitto_strerror(rc));
}
mosquitto_lib_cleanup();
return 1;
}
}
if(cfg->insecure && mosquitto_tls_insecure_set(mosq, true)){
if(!cfg->quiet) fprintf(stderr, "Error: Problem setting TLS insecure option.\n");
err_printf(cfg, "Error: Problem setting TLS insecure option.\n");
mosquitto_lib_cleanup();
return 1;
}
if(cfg->tls_engine && mosquitto_string_option(mosq, MOSQ_OPT_TLS_ENGINE, cfg->tls_engine)){
if(!cfg->quiet) fprintf(stderr, "Error: Problem setting TLS engine, is %s a valid engine?\n", cfg->tls_engine);
err_printf(cfg, "Error: Problem setting TLS engine, is %s a valid engine?\n", cfg->tls_engine);
mosquitto_lib_cleanup();
return 1;
}
if(cfg->keyform && mosquitto_string_option(mosq, MOSQ_OPT_TLS_KEYFORM, cfg->keyform)){
if(!cfg->quiet) fprintf(stderr, "Error: Problem setting key form, it must be one of 'pem' or 'engine'.\n");
err_printf(cfg, "Error: Problem setting key form, it must be one of 'pem' or 'engine'.\n");
mosquitto_lib_cleanup();
return 1;
}
if(cfg->tls_engine_kpass_sha1 && mosquitto_string_option(mosq, MOSQ_OPT_TLS_ENGINE_KPASS_SHA1, cfg->tls_engine_kpass_sha1)){
if(!cfg->quiet) fprintf(stderr, "Error: Problem setting TLS engine key pass sha, is it a 40 character hex string?\n");
err_printf(cfg, "Error: Problem setting TLS engine key pass sha, is it a 40 character hex string?\n");
mosquitto_lib_cleanup();
return 1;
}
if(cfg->tls_alpn && mosquitto_string_option(mosq, MOSQ_OPT_TLS_ALPN, cfg->tls_alpn)){
if(!cfg->quiet) fprintf(stderr, "Error: Problem setting TLS ALPN protocol.\n");
err_printf(cfg, "Error: Problem setting TLS ALPN protocol.\n");
mosquitto_lib_cleanup();
return 1;
}
# ifdef FINAL_WITH_TLS_PSK
if(cfg->psk && mosquitto_tls_psk_set(mosq, cfg->psk, cfg->psk_identity, NULL)){
if(!cfg->quiet) fprintf(stderr, "Error: Problem setting TLS-PSK options.\n");
err_printf(cfg, "Error: Problem setting TLS-PSK options.\n");
mosquitto_lib_cleanup();
return 1;
}
# endif
if((cfg->tls_version || cfg->ciphers) && mosquitto_tls_opts_set(mosq, 1, cfg->tls_version, cfg->ciphers)){
if(!cfg->quiet) fprintf(stderr, "Error: Problem setting TLS options, check the options are valid.\n");
err_printf(cfg, "Error: Problem setting TLS options, check the options are valid.\n");
mosquitto_lib_cleanup();
return 1;
}
@@ -1157,7 +1155,7 @@ int client_id_generate(struct mosq_config *cfg)
if(cfg->id_prefix){
cfg->id = malloc(strlen(cfg->id_prefix)+10);
if(!cfg->id){
if(!cfg->quiet) fprintf(stderr, "Error: Out of memory.\n");
err_printf(cfg, "Error: Out of memory.\n");
mosquitto_lib_cleanup();
return 1;
}
@@ -1203,17 +1201,15 @@ int client_connect(struct mosquitto *mosq, struct mosq_config *cfg)
rc = mosquitto_connect_bind_v5(mosq, cfg->host, port, cfg->keepalive, cfg->bind_address, cfg->connect_props);
#endif
if(rc>0){
if(!cfg->quiet){
if(rc == MOSQ_ERR_ERRNO){
if(rc == MOSQ_ERR_ERRNO){
#ifndef WIN32
err = strerror(errno);
err = strerror(errno);
#else
FormatMessage(FORMAT_MESSAGE_FROM_SYSTEM, NULL, errno, 0, (LPTSTR)&err, 1024, NULL);
FormatMessage(FORMAT_MESSAGE_FROM_SYSTEM, NULL, errno, 0, (LPTSTR)&err, 1024, NULL);
#endif
fprintf(stderr, "Error: %s\n", err);
}else{
fprintf(stderr, "Unable to connect (%s).\n", mosquitto_strerror(rc));
}
err_printf(cfg, "Error: %s\n", err);
}else{
err_printf(cfg, "Unable to connect (%s).\n", mosquitto_strerror(rc));
}
mosquitto_lib_cleanup();
return rc;
@@ -1280,7 +1276,7 @@ static int mosquitto__parse_socks_url(struct mosq_config *cfg, char *url)
if(!strncmp(url, "socks5h://", strlen("socks5h://"))){
str = url + strlen("socks5h://");
}else{
fprintf(stderr, "Error: Unsupported proxy protocol: %s\n", url);
err_printf(cfg, "Error: Unsupported proxy protocol: %s\n", url);
return 1;
}
@@ -1307,7 +1303,7 @@ static int mosquitto__parse_socks_url(struct mosq_config *cfg, char *url)
len = i-start;
host = malloc(len + 1);
if(!host){
fprintf(stderr, "Error: Out of memory.\n");
err_printf(cfg, "Error: Out of memory.\n");
goto cleanup;
}
memcpy(host, &(str[start]), len);
@@ -1320,7 +1316,7 @@ static int mosquitto__parse_socks_url(struct mosq_config *cfg, char *url)
len = i-start;
username_or_host = malloc(len + 1);
if(!username_or_host){
fprintf(stderr, "Error: Out of memory.\n");
err_printf(cfg, "Error: Out of memory.\n");
goto cleanup;
}
memcpy(username_or_host, &(str[start]), len);
@@ -1340,7 +1336,7 @@ static int mosquitto__parse_socks_url(struct mosq_config *cfg, char *url)
len = i-start;
password = malloc(len + 1);
if(!password){
fprintf(stderr, "Error: Out of memory.\n");
err_printf(cfg, "Error: Out of memory.\n");
goto cleanup;
}
memcpy(password, &(str[start]), len);
@@ -1356,7 +1352,7 @@ static int mosquitto__parse_socks_url(struct mosq_config *cfg, char *url)
len = i-start;
username = malloc(len + 1);
if(!username){
fprintf(stderr, "Error: Out of memory.\n");
err_printf(cfg, "Error: Out of memory.\n");
goto cleanup;
}
memcpy(username, &(str[start]), len);
@@ -1374,7 +1370,7 @@ static int mosquitto__parse_socks_url(struct mosq_config *cfg, char *url)
* socks5h://username[:password]@host:port */
port = malloc(len + 1);
if(!port){
fprintf(stderr, "Error: Out of memory.\n");
err_printf(cfg, "Error: Out of memory.\n");
goto cleanup;
}
memcpy(port, &(str[start]), len);
@@ -1386,7 +1382,7 @@ static int mosquitto__parse_socks_url(struct mosq_config *cfg, char *url)
username_or_host = NULL;
port = malloc(len + 1);
if(!port){
fprintf(stderr, "Error: Out of memory.\n");
err_printf(cfg, "Error: Out of memory.\n");
goto cleanup;
}
memcpy(port, &(str[start]), len);
@@ -1394,7 +1390,7 @@ static int mosquitto__parse_socks_url(struct mosq_config *cfg, char *url)
}else{
host = malloc(len + 1);
if(!host){
fprintf(stderr, "Error: Out of memory.\n");
err_printf(cfg, "Error: Out of memory.\n");
goto cleanup;
}
memcpy(host, &(str[start]), len);
@@ -1403,7 +1399,7 @@ static int mosquitto__parse_socks_url(struct mosq_config *cfg, char *url)
}
if(!host){
fprintf(stderr, "Error: Invalid proxy.\n");
err_printf(cfg, "Error: Invalid proxy.\n");
goto cleanup;
}
@@ -1416,7 +1412,7 @@ static int mosquitto__parse_socks_url(struct mosq_config *cfg, char *url)
if(port){
port_int = atoi(port);
if(port_int < 1 || port_int > 65535){
fprintf(stderr, "Error: Invalid proxy port %d\n", port_int);
err_printf(cfg, "Error: Invalid proxy port %d\n", port_int);
goto cleanup;
}
free(port);
@@ -1438,5 +1434,16 @@ cleanup:
if(port) free(port);
return 1;
}
#endif
void err_printf(const struct mosq_config *cfg, const char *fmt, ...)
{
va_list va;
if(cfg->quiet) return;
va_start(va, fmt);
vfprintf(stderr, fmt, va);
va_end(va);
}
client/client_shared.h
+2
View File
@@ -126,4 +126,6 @@ int client_connect(struct mosquitto *mosq, struct mosq_config *cfg);
int cfg_parse_property(struct mosq_config *cfg, int argc, char *argv[], int *idx);
void err_printf(const struct mosq_config *cfg, const char *fmt, ...);
#endif
client/pub_client.c
+49 -41
View File
@@ -42,7 +42,6 @@ static int last_mid = -1;
static int last_mid_sent = -1;
static char *line_buf = NULL;
static int line_buf_len = 1024;
static bool connected = true;
static bool disconnect_sent = false;
static int publish_count = 0;
static bool ready_for_repeat = false;
@@ -104,7 +103,7 @@ void my_disconnect_callback(struct mosquitto *mosq, void *obj, int rc, const mos
UNUSED(rc);
UNUSED(properties);
connected = false;
status = STATUS_DISCONNECTED;
}
int my_publish(struct mosquitto *mosq, int *mid, const char *topic, int payloadlen, void *payload, int qos, bool retain)
@@ -142,36 +141,34 @@ void my_connect_callback(struct mosquitto *mosq, void *obj, int result, int flag
break;
}
if(rc){
if(!cfg.quiet){
switch(rc){
case MOSQ_ERR_INVAL:
fprintf(stderr, "Error: Invalid input. Does your topic contain '+' or '#'?\n");
break;
case MOSQ_ERR_NOMEM:
fprintf(stderr, "Error: Out of memory when trying to publish message.\n");
break;
case MOSQ_ERR_NO_CONN:
fprintf(stderr, "Error: Client not connected when trying to publish.\n");
break;
case MOSQ_ERR_PROTOCOL:
fprintf(stderr, "Error: Protocol error when communicating with broker.\n");
break;
case MOSQ_ERR_PAYLOAD_SIZE:
fprintf(stderr, "Error: Message payload is too large.\n");
break;
case MOSQ_ERR_QOS_NOT_SUPPORTED:
fprintf(stderr, "Error: Message QoS not supported on broker, try a lower QoS.\n");
break;
}
switch(rc){
case MOSQ_ERR_INVAL:
err_printf(&cfg, "Error: Invalid input. Does your topic contain '+' or '#'?\n");
break;
case MOSQ_ERR_NOMEM:
err_printf(&cfg, "Error: Out of memory when trying to publish message.\n");
break;
case MOSQ_ERR_NO_CONN:
err_printf(&cfg, "Error: Client not connected when trying to publish.\n");
break;
case MOSQ_ERR_PROTOCOL:
err_printf(&cfg, "Error: Protocol error when communicating with broker.\n");
break;
case MOSQ_ERR_PAYLOAD_SIZE:
err_printf(&cfg, "Error: Message payload is too large.\n");
break;
case MOSQ_ERR_QOS_NOT_SUPPORTED:
err_printf(&cfg, "Error: Message QoS not supported on broker, try a lower QoS.\n");
break;
}
mosquitto_disconnect_v5(mosq, 0, cfg.disconnect_props);
}
}else{
if(result && !cfg.quiet){
if(result){
if(cfg.protocol_version == MQTT_PROTOCOL_V5){
fprintf(stderr, "%s\n", mosquitto_reason_string(result));
err_printf(&cfg, "%s\n", mosquitto_reason_string(result));
}else{
fprintf(stderr, "%s\n", mosquitto_connack_string(result));
err_printf(&cfg, "%s\n", mosquitto_connack_string(result));
}
}
}
@@ -185,7 +182,7 @@ void my_publish_callback(struct mosquitto *mosq, void *obj, int mid, int reason_
last_mid_sent = mid;
if(reason_code > 127){
if(!cfg.quiet) fprintf(stderr, "Warning: Publish %d failed: %s.\n", mid, mosquitto_reason_string(reason_code));
err_printf(&cfg, "Warning: Publish %d failed: %s.\n", mid, mosquitto_reason_string(reason_code));
}
publish_count++;
@@ -208,7 +205,7 @@ int pub_shared_init(void)
{
line_buf = malloc(line_buf_len);
if(!line_buf){
fprintf(stderr, "Error: Out of memory.\n");
err_printf(&cfg, "Error: Out of memory.\n");
return 1;
}
return 0;
@@ -224,6 +221,7 @@ int pub_shared_loop(struct mosquitto *mosq)
int buf_len_actual;
int mode;
int loop_delay = 1000;
bool stdin_finished = false;
if(cfg.repeat_count > 1 && (cfg.repeat_delay.tv_sec == 0 || cfg.repeat_delay.tv_usec != 0)){
loop_delay = cfg.repeat_delay.tv_usec / 2000;
@@ -233,6 +231,7 @@ int pub_shared_loop(struct mosquitto *mosq)
if(mode == MSGMODE_STDIN_LINE){
mosquitto_loop_start(mosq);
stdin_finished = false;
}
do{
@@ -240,13 +239,13 @@ int pub_shared_loop(struct mosquitto *mosq)
if(status == STATUS_CONNACK_RECVD){
pos = 0;
read_len = line_buf_len;
while(connected && fgets(&line_buf[pos], read_len, stdin)){
while(status == STATUS_CONNACK_RECVD && fgets(&line_buf[pos], read_len, stdin)){
buf_len_actual = strlen(line_buf);
if(line_buf[buf_len_actual-1] == '\n'){
line_buf[buf_len_actual-1] = '\0';
rc2 = my_publish(mosq, &mid_sent, cfg.topic, buf_len_actual-1, line_buf, cfg.qos, cfg.retain);
if(rc2){
if(!cfg.quiet) fprintf(stderr, "Error: Publish returned %d, disconnecting.\n", rc2);
err_printf(&cfg, "Error: Publish returned %d, disconnecting.\n", rc2);
mosquitto_disconnect_v5(mosq, MQTT_RC_DISCONNECT_WITH_WILL_MSG, cfg.disconnect_props);
}
break;
@@ -256,7 +255,7 @@ int pub_shared_loop(struct mosquitto *mosq)
read_len = 1024;
buf2 = realloc(line_buf, line_buf_len);
if(!buf2){
fprintf(stderr, "Error: Out of memory.\n");
err_printf(&cfg, "Error: Out of memory.\n");
return MOSQ_ERR_NOMEM;
}
line_buf = buf2;
@@ -272,6 +271,10 @@ int pub_shared_loop(struct mosquitto *mosq)
last_mid = mid_sent;
status = STATUS_WAITING;
}
stdin_finished = true;
}else if(status == STATUS_DISCONNECTED){
/* Not end of stdin, so we've lost our connection and must
* reconnect */
}
}else if(status == STATUS_WAITING){
if(last_mid_sent == last_mid && disconnect_sent == false){
@@ -305,16 +308,20 @@ int pub_shared_loop(struct mosquitto *mosq)
break;
}
if(rc){
fprintf(stderr, "Error sending repeat publish: %s", mosquitto_strerror(rc));
err_printf(&cfg, "Error sending repeat publish: %s", mosquitto_strerror(rc));
}
}
}
}while(rc == MOSQ_ERR_SUCCESS && connected);
}while(rc == MOSQ_ERR_SUCCESS && stdin_finished == false);
if(mode == MSGMODE_STDIN_LINE){
mosquitto_loop_stop(mosq, false);
}
return 0;
if(status == STATUS_DISCONNECTED){
return MOSQ_ERR_SUCCESS;
}else{
return rc;
}
}
@@ -331,7 +338,7 @@ void print_usage(void)
mosquitto_lib_version(&major, &minor, &revision);
printf("mosquitto_pub is a simple mqtt client that will publish a message on a single topic and exit.\n");
printf("mosquitto_pub version %s running on libmosquitto %d.%d.%d.\n\n", VERSION, major, minor, revision);
printf("Usage: mosquitto_pub {[-h host] [-p port] [-u username [-P password]] -t topic | -L URL}\n");
printf("Usage: mosquitto_pub {[-h host] [-p port] [-u username] [-P password] -t topic | -L URL}\n");
printf(" {-f file | -l | -n | -m message}\n");
printf(" [-c] [-k keepalive] [-q qos] [-r] [--repeat N] [--repeat-delay time]\n");
#ifdef WITH_SRV
@@ -457,12 +464,12 @@ int main(int argc, char *argv[])
if(cfg.pub_mode == MSGMODE_STDIN_FILE){
if(load_stdin()){
fprintf(stderr, "Error loading input from stdin.\n");
err_printf(&cfg, "Error loading input from stdin.\n");
goto cleanup;
}
}else if(cfg.file_input){
if(load_file(cfg.file_input)){
fprintf(stderr, "Error loading input file \"%s\".\n", cfg.file_input);
err_printf(&cfg, "Error loading input file \"%s\".\n", cfg.file_input);
goto cleanup;
}
}
@@ -478,14 +485,14 @@ int main(int argc, char *argv[])
goto cleanup;
}
mosq = mosquitto_new(cfg.id, true, NULL);
mosq = mosquitto_new(cfg.id, cfg.clean_session, NULL);
if(!mosq){
switch(errno){
case ENOMEM:
if(!cfg.quiet) fprintf(stderr, "Error: Out of memory.\n");
err_printf(&cfg, "Error: Out of memory.\n");
break;
case EINVAL:
if(!cfg.quiet) fprintf(stderr, "Error: Invalid id.\n");
err_printf(&cfg, "Error: Invalid id.\n");
break;
}
goto cleanup;
@@ -510,6 +517,7 @@ int main(int argc, char *argv[])
if(cfg.message && cfg.pub_mode == MSGMODE_FILE){
free(cfg.message);
cfg.message = NULL;
}
mosquitto_destroy(mosq);
mosquitto_lib_cleanup();
@@ -517,7 +525,7 @@ int main(int argc, char *argv[])
pub_shared_cleanup();
if(rc){
fprintf(stderr, "Error: %s\n", mosquitto_strerror(rc));
err_printf(&cfg, "Error: %s\n", mosquitto_strerror(rc));
}
return rc;
client/pub_shared.c
+8 -8
View File
@@ -36,7 +36,7 @@ Contributors:
/* Global variables for use in callbacks. See sub_client.c for an example of
* using a struct to hold variables for use in callbacks. */
int mid_sent = 0;
int mid_sent = -1;
int status = STATUS_CONNECTING;
struct mosq_config cfg;
@@ -61,7 +61,7 @@ int load_stdin(void)
rlen = fread(buf, 1, 1024, stdin);
aux_message = realloc(cfg.message, pos+rlen);
if(!aux_message){
if(!cfg.quiet) fprintf(stderr, "Error: Out of memory.\n");
err_printf(&cfg, "Error: Out of memory.\n");
free(cfg.message);
return 1;
} else
@@ -74,7 +74,7 @@ int load_stdin(void)
cfg.msglen = pos;
if(!cfg.msglen){
if(!cfg.quiet) fprintf(stderr, "Error: Zero length input.\n");
err_printf(&cfg, "Error: Zero length input.\n");
return 1;
}
@@ -88,7 +88,7 @@ int load_file(const char *filename)
fptr = fopen(filename, "rb");
if(!fptr){
if(!cfg.quiet) fprintf(stderr, "Error: Unable to open file \"%s\".\n", filename);
err_printf(&cfg, "Error: Unable to open file \"%s\".\n", filename);
return 1;
}
cfg.pub_mode = MSGMODE_FILE;
@@ -96,22 +96,22 @@ int load_file(const char *filename)
cfg.msglen = ftell(fptr);
if(cfg.msglen > 268435455){
fclose(fptr);
if(!cfg.quiet) fprintf(stderr, "Error: File \"%s\" is too large (>268,435,455 bytes).\n", filename);
err_printf(&cfg, "Error: File \"%s\" is too large (>268,435,455 bytes).\n", filename);
return 1;
}else if(cfg.msglen == 0){
fclose(fptr);
if(!cfg.quiet) fprintf(stderr, "Error: File \"%s\" is empty.\n", filename);
err_printf(&cfg, "Error: File \"%s\" is empty.\n", filename);
return 1;
}else if(cfg.msglen < 0){
fclose(fptr);
if(!cfg.quiet) fprintf(stderr, "Error: Unable to determine size of file \"%s\".\n", filename);
err_printf(&cfg, "Error: Unable to determine size of file \"%s\".\n", filename);
return 1;
}
fseek(fptr, 0, SEEK_SET);
cfg.message = malloc(cfg.msglen);
if(!cfg.message){
fclose(fptr);
if(!cfg.quiet) fprintf(stderr, "Error: Out of memory.\n");
err_printf(&cfg, "Error: Out of memory.\n");
return 1;
}
pos = 0;
client/pub_shared.h
+1
View File
@@ -20,6 +20,7 @@ Contributors:
#define STATUS_CONNACK_RECVD 1
#define STATUS_WAITING 2
#define STATUS_DISCONNECTING 3
#define STATUS_DISCONNECTED 4
extern int mid_sent;
extern int status;
client/rr_client.c
+8 -10
View File
@@ -121,8 +121,8 @@ void my_connect_callback(struct mosquitto *mosq, void *obj, int result, int flag
mosquitto_subscribe_v5(mosq, NULL, cfg.response_topic, cfg.qos, 0, cfg.subscribe_props);
}else{
client_state = rr_s_disconnect;
if(result && !cfg.quiet){
fprintf(stderr, "%s\n", mosquitto_connack_string(result));
if(result){
err_printf(&cfg, "%s\n", mosquitto_connack_string(result));
}
mosquitto_disconnect_v5(mosq, 0, cfg.disconnect_props);
}
@@ -135,10 +135,8 @@ void my_subscribe_callback(struct mosquitto *mosq, void *obj, int mid, int qos_c
client_state = rr_s_ready_to_publish;
}else{
client_state = rr_s_disconnect;
if(!cfg.quiet){
fprintf(stderr, "%s\n", mosquitto_reason_string(granted_qos[0]));
mosquitto_disconnect_v5(mosq, 0, cfg.disconnect_props);
}
err_printf(&cfg, "%s\n", mosquitto_reason_string(granted_qos[0]));
mosquitto_disconnect_v5(mosq, 0, cfg.disconnect_props);
}
}
@@ -158,7 +156,7 @@ void print_usage(void)
printf(" Defaults to MQTT v5, where the Request-Response feature will be used, but v3.1.1 can also be used\n");
printf(" with v3.1.1 brokers.\n");
printf("mosquitto_rr version %s running on libmosquitto %d.%d.%d.\n\n", VERSION, major, minor, revision);
printf("Usage: mosquitto_rr {[-h host] [-p port] [-u username [-P password]] -t topic | -L URL} -e response-topic\n");
printf("Usage: mosquitto_rr {[-h host] [-p port] [-u username] [-P password] -t topic | -L URL} -e response-topic\n");
printf(" [-c] [-k keepalive] [-q qos] [-R]\n");
printf(" [-F format]\n");
#ifndef WIN32
@@ -282,7 +280,7 @@ int main(int argc, char *argv[])
}
rc = mosquitto_property_check_all(CMD_PUBLISH, cfg.publish_props);
if(rc){
if(!cfg.quiet) fprintf(stderr, "Error in PUBLISH properties: Duplicate response topic.\n");
err_printf(&cfg, "Error in PUBLISH properties: Duplicate response topic.\n");
goto cleanup;
}
@@ -294,10 +292,10 @@ int main(int argc, char *argv[])
if(!mosq){
switch(errno){
case ENOMEM:
if(!cfg.quiet) fprintf(stderr, "Error: Out of memory.\n");
err_printf(&cfg, "Error: Out of memory.\n");
break;
case EINVAL:
if(!cfg.quiet) fprintf(stderr, "Error: Invalid id and/or clean_session.\n");
err_printf(&cfg, "Error: Invalid id and/or clean_session.\n");
break;
}
goto cleanup;
client/sub_client.c
+7 -7
View File
@@ -35,7 +35,7 @@ Contributors:
#include <mqtt_protocol.h>
#include "client_shared.h"
static struct mosq_config cfg;
struct mosq_config cfg;
bool process_messages = true;
int msg_count = 0;
struct mosquitto *mosq = NULL;
@@ -124,11 +124,11 @@ void my_connect_callback(struct mosquitto *mosq, void *obj, int result, int flag
mosquitto_unsubscribe_v5(mosq, NULL, cfg.unsub_topics[i], cfg.unsubscribe_props);
}
}else{
if(result && !cfg.quiet){
if(result){
if(cfg.protocol_version == MQTT_PROTOCOL_V5){
fprintf(stderr, "%s\n", mosquitto_reason_string(result));
err_printf(&cfg, "%s\n", mosquitto_reason_string(result));
}else{
fprintf(stderr, "%s\n", mosquitto_connack_string(result));
err_printf(&cfg, "%s\n", mosquitto_connack_string(result));
}
}
mosquitto_disconnect_v5(mosq, 0, cfg.disconnect_props);
@@ -168,7 +168,7 @@ void print_usage(void)
mosquitto_lib_version(&major, &minor, &revision);
printf("mosquitto_sub is a simple mqtt client that will subscribe to a set of topics and print all messages it receives.\n");
printf("mosquitto_sub version %s running on libmosquitto %d.%d.%d.\n\n", VERSION, major, minor, revision);
printf("Usage: mosquitto_sub {[-h host] [-p port] [-u username [-P password]] -t topic | -L URL [-t topic]}\n");
printf("Usage: mosquitto_sub {[-h host] [-p port] [-u username] [-P password] -t topic | -L URL [-t topic]}\n");
printf(" [-c] [-k keepalive] [-q qos]\n");
printf(" [-C msg_count] [-E] [-R] [--retained-only] [--remove-retained] [-T filter_out] [-U topic ...]\n");
printf(" [-F format]\n");
@@ -307,10 +307,10 @@ int main(int argc, char *argv[])
if(!mosq){
switch(errno){
case ENOMEM:
if(!cfg.quiet) fprintf(stderr, "Error: Out of memory.\n");
err_printf(&cfg, "Error: Out of memory.\n");
break;
case EINVAL:
if(!cfg.quiet) fprintf(stderr, "Error: Invalid id and/or clean_session.\n");
err_printf(&cfg, "Error: Invalid id and/or clean_session.\n");
break;
}
goto cleanup;
client/sub_client_output.c
+20 -19
View File
@@ -37,6 +37,7 @@ Contributors:
#include <mosquitto.h>
#include "client_shared.h"
extern struct mosq_config cfg;
static int get_time(struct tm **ti, long *ns)
{
@@ -60,7 +61,7 @@ static int get_time(struct tm **ti, long *ns)
*ns = tv.tv_usec*1000;
#else
if(clock_gettime(CLOCK_REALTIME, &ts) != 0){
fprintf(stderr, "Error obtaining system time.\n");
err_printf(&cfg, "Error obtaining system time.\n");
return 1;
}
s = ts.tv_sec;
@@ -69,7 +70,7 @@ static int get_time(struct tm **ti, long *ns)
*ti = localtime(&s);
if(!(*ti)){
fprintf(stderr, "Error obtaining system time.\n");
err_printf(&cfg, "Error obtaining system time.\n");
return 1;
}
@@ -130,7 +131,7 @@ static void json_print(const struct mosquitto_message *message, const struct tm
}
static void formatted_print(const struct mosq_config *cfg, const struct mosquitto_message *message)
static void formatted_print(const struct mosq_config *lcfg, const struct mosquitto_message *message)
{
int len;
int i;
@@ -139,13 +140,13 @@ static void formatted_print(const struct mosq_config *cfg, const struct mosquitt
char strf[3];
char buf[100];
len = strlen(cfg->format);
len = strlen(lcfg->format);
for(i=0; i<len; i++){
if(cfg->format[i] == '%'){
if(lcfg->format[i] == '%'){
if(i < len-1){
i++;
switch(cfg->format[i]){
switch(lcfg->format[i]){
case '%':
fputc('%', stdout);
break;
@@ -153,7 +154,7 @@ static void formatted_print(const struct mosq_config *cfg, const struct mosquitt
case 'I':
if(!ti){
if(get_time(&ti, &ns)){
fprintf(stderr, "Error obtaining system time.\n");
err_printf(lcfg, "Error obtaining system time.\n");
return;
}
}
@@ -165,7 +166,7 @@ static void formatted_print(const struct mosq_config *cfg, const struct mosquitt
case 'j':
if(!ti){
if(get_time(&ti, &ns)){
fprintf(stderr, "Error obtaining system time.\n");
err_printf(lcfg, "Error obtaining system time.\n");
return;
}
}
@@ -175,7 +176,7 @@ static void formatted_print(const struct mosq_config *cfg, const struct mosquitt
case 'J':
if(!ti){
if(get_time(&ti, &ns)){
fprintf(stderr, "Error obtaining system time.\n");
err_printf(lcfg, "Error obtaining system time.\n");
return;
}
}
@@ -213,7 +214,7 @@ static void formatted_print(const struct mosq_config *cfg, const struct mosquitt
case 'U':
if(!ti){
if(get_time(&ti, &ns)){
fprintf(stderr, "Error obtaining system time.\n");
err_printf(lcfg, "Error obtaining system time.\n");
return;
}
}
@@ -231,24 +232,24 @@ static void formatted_print(const struct mosq_config *cfg, const struct mosquitt
break;
}
}
}else if(cfg->format[i] == '@'){
}else if(lcfg->format[i] == '@'){
if(i < len-1){
i++;
if(cfg->format[i] == '@'){
if(lcfg->format[i] == '@'){
fputc('@', stdout);
}else{
if(!ti){
if(get_time(&ti, &ns)){
fprintf(stderr, "Error obtaining system time.\n");
err_printf(lcfg, "Error obtaining system time.\n");
return;
}
}
strf[0] = '%';
strf[1] = cfg->format[i];
strf[1] = lcfg->format[i];
strf[2] = 0;
if(cfg->format[i] == 'N'){
if(lcfg->format[i] == 'N'){
printf("%09ld", ns);
}else{
if(strftime(buf, 100, strf, ti) != 0){
@@ -257,10 +258,10 @@ static void formatted_print(const struct mosq_config *cfg, const struct mosquitt
}
}
}
}else if(cfg->format[i] == '\\'){
}else if(lcfg->format[i] == '\\'){
if(i < len-1){
i++;
switch(cfg->format[i]){
switch(lcfg->format[i]){
case '\\':
fputc('\\', stdout);
break;
@@ -295,10 +296,10 @@ static void formatted_print(const struct mosq_config *cfg, const struct mosquitt
}
}
}else{
fputc(cfg->format[i], stdout);
fputc(lcfg->format[i], stdout);
}
}
if(cfg->eol){
if(lcfg->eol){
fputc('\n', stdout);
}
fflush(stdout);
config.h
+6
View File
@@ -1,4 +1,5 @@
#ifndef CONFIG_H
#define CONFIG_H
/* ============================================================
* Platform options
* ============================================================ */
@@ -60,4 +61,9 @@
#define UNUSED(A) (void)(A)
/* Android Bionic libpthread implementation doesn't have pthread_cancel */
#ifndef ANDROID
# define HAVE_PTHREAD_CANCEL
#endif
#endif
config.mk
+2 -2
View File
@@ -104,7 +104,7 @@ WITH_COVERAGE:=no
# Also bump lib/mosquitto.h, CMakeLists.txt,
# installer/mosquitto.nsi, installer/mosquitto64.nsi
VERSION=1.6.2
VERSION=1.6.3
# Client library SO version. Bump if incompatible API/ABI changes are made.
SOVERSION=1
@@ -143,7 +143,7 @@ BROKER_LDADD:=
CLIENT_CPPFLAGS:=$(CPPFLAGS) -I.. -I../lib
CLIENT_CFLAGS:=${CFLAGS} -DVERSION="\"${VERSION}\""
CLIENT_LDFLAGS:=-L../lib
CLIENT_LDFLAGS:=$(LDFLAGS) -L../lib
CLIENT_LDADD:=
PASSWD_LDADD:=
installer/mosquitto.nsi
+1 -1
View File
@@ -9,7 +9,7 @@
!define env_hklm 'HKLM "SYSTEM\CurrentControlSet\Control\Session Manager\Environment"'
Name "Eclipse Mosquitto"
!define VERSION 1.6.2
!define VERSION 1.6.3
OutFile "mosquitto-${VERSION}-install-windows-x86.exe"
InstallDir "$PROGRAMFILES\mosquitto"
installer/mosquitto64.nsi
+1 -1
View File
@@ -9,7 +9,7 @@
!define env_hklm 'HKLM "SYSTEM\CurrentControlSet\Control\Session Manager\Environment"'
Name "Eclipse Mosquitto"
!define VERSION 1.6.2
!define VERSION 1.6.3
OutFile "mosquitto-${VERSION}-install-windows-x64.exe"
!include "x64.nsh"
lib/handle_pubrel.c
-1
View File
@@ -83,7 +83,6 @@ int handle__pubrel(struct mosquitto_db *db, struct mosquitto *mosq)
}else if(rc != MOSQ_ERR_SUCCESS){
/* Message not found. Still send a PUBCOMP anyway because this could be
* due to a repeated PUBREL after a client has reconnected. */
log__printf(mosq, MOSQ_LOG_WARNING, "Warning: Received PUBREL from %s for an unknown packet identifier %d.", mosq->id, mid);
}
rc = send__pubcomp(mosq, mid);
lib/mosquitto.c
+2
View File
@@ -201,11 +201,13 @@ void mosquitto__destroy(struct mosquitto *mosq)
if(!mosq) return;
#ifdef WITH_THREADING
# ifdef HAVE_PTHREAD_CANCEL
if(mosq->threaded == mosq_ts_self && !pthread_equal(mosq->thread_id, pthread_self())){
pthread_cancel(mosq->thread_id);
pthread_join(mosq->thread_id, NULL);
mosq->threaded = mosq_ts_none;
}
# endif
if(mosq->id){
/* If mosq->id is not NULL then the client has already been initialised
lib/mosquitto.h
+1 -1
View File
@@ -48,7 +48,7 @@ extern "C" {
#define LIBMOSQUITTO_MAJOR 1
#define LIBMOSQUITTO_MINOR 6
#define LIBMOSQUITTO_REVISION 2
#define LIBMOSQUITTO_REVISION 3
/* LIBMOSQUITTO_VERSION_NUMBER looks like 1002001 for e.g. version 1.2.1. */
#define LIBMOSQUITTO_VERSION_NUMBER (LIBMOSQUITTO_MAJOR*1000000+LIBMOSQUITTO_MINOR*1000+LIBMOSQUITTO_REVISION)
lib/mosquitto_internal.h
+7
View File
@@ -180,6 +180,12 @@ enum mosquitto__keyform {
};
#endif
struct will_delay_list {
struct mosquitto *context;
struct will_delay_list *prev;
struct will_delay_list *next;
};
struct mosquitto_msg_data{
#ifdef WITH_BROKER
struct mosquitto_client_msg *inflight;
@@ -224,6 +230,7 @@ struct mosquitto {
struct mosquitto__packet *out_packet;
struct mosquitto_message_all *will;
struct mosquitto__alias *aliases;
struct will_delay_list *will_delay_entry;
uint32_t maximum_packet_size;
int alias_count;
uint32_t will_delay_interval;
lib/net_mosq.c
+17
View File
@@ -50,6 +50,7 @@ Contributors:
#include <openssl/conf.h>
#include <openssl/engine.h>
#include <openssl/err.h>
#include <openssl/ui.h>
#include <tls_mosq.h>
#endif
@@ -140,7 +141,9 @@ int net__init(void)
| OPENSSL_INIT_ADD_ALL_DIGESTS \
| OPENSSL_INIT_LOAD_CONFIG, NULL);
# endif
#if !defined(OPENSSL_NO_ENGINE)
ENGINE_load_builtin_engines();
#endif
setup_ui_method();
if(tls_ex_index_mosq == -1){
tls_ex_index_mosq = SSL_get_ex_new_index(0, "client context", NULL, NULL, NULL);
@@ -598,6 +601,7 @@ static int net__init_ssl_ctx(struct mosquitto *mosq)
SSL_CTX_set_mode(mosq->ssl_ctx, SSL_MODE_RELEASE_BUFFERS);
#endif
#if !defined(OPENSSL_NO_ENGINE)
if(mosq->tls_engine){
engine = ENGINE_by_id(mosq->tls_engine);
if(!engine){
@@ -614,12 +618,15 @@ static int net__init_ssl_ctx(struct mosquitto *mosq)
ENGINE_set_default(engine, ENGINE_METHOD_ALL);
ENGINE_free(engine); /* release the structural reference from ENGINE_by_id() */
}
#endif
if(mosq->tls_ciphers){
ret = SSL_CTX_set_cipher_list(mosq->ssl_ctx, mosq->tls_ciphers);
if(ret == 0){
log__printf(mosq, MOSQ_LOG_ERR, "Error: Unable to set TLS ciphers. Check cipher list \"%s\".", mosq->tls_ciphers);
#if !defined(OPENSSL_NO_ENGINE)
ENGINE_FINISH(engine);
#endif
COMPAT_CLOSE(mosq->sock);
mosq->sock = INVALID_SOCKET;
net__print_ssl_error(mosq);
@@ -646,7 +653,9 @@ static int net__init_ssl_ctx(struct mosquitto *mosq)
log__printf(mosq, MOSQ_LOG_ERR, "Error: Unable to load CA certificates, check capath \"%s\".", mosq->tls_capath);
}
#endif
#if !defined(OPENSSL_NO_ENGINE)
ENGINE_FINISH(engine);
#endif
COMPAT_CLOSE(mosq->sock);
mosq->sock = INVALID_SOCKET;
net__print_ssl_error(mosq);
@@ -671,7 +680,9 @@ static int net__init_ssl_ctx(struct mosquitto *mosq)
#else
log__printf(mosq, MOSQ_LOG_ERR, "Error: Unable to load client certificate \"%s\".", mosq->tls_certfile);
#endif
#if !defined(OPENSSL_NO_ENGINE)
ENGINE_FINISH(engine);
#endif
COMPAT_CLOSE(mosq->sock);
mosq->sock = INVALID_SOCKET;
net__print_ssl_error(mosq);
@@ -680,6 +691,7 @@ static int net__init_ssl_ctx(struct mosquitto *mosq)
}
if(mosq->tls_keyfile){
if(mosq->tls_keyform == mosq_k_engine){
#if !defined(OPENSSL_NO_ENGINE)
UI_METHOD *ui_method = net__get_ui_method();
if(mosq->tls_engine_kpass_sha1){
if(!ENGINE_ctrl_cmd(engine, ENGINE_SECRET_MODE, ENGINE_SECRET_MODE_SHA, NULL, NULL, 0)){
@@ -713,6 +725,7 @@ static int net__init_ssl_ctx(struct mosquitto *mosq)
net__print_ssl_error(mosq);
return MOSQ_ERR_TLS;
}
#endif
}else{
ret = SSL_CTX_use_PrivateKey_file(mosq->ssl_ctx, mosq->tls_keyfile, SSL_FILETYPE_PEM);
if(ret != 1){
@@ -721,7 +734,9 @@ static int net__init_ssl_ctx(struct mosquitto *mosq)
#else
log__printf(mosq, MOSQ_LOG_ERR, "Error: Unable to load client key file \"%s\".", mosq->tls_keyfile);
#endif
#if !defined(OPENSSL_NO_ENGINE)
ENGINE_FINISH(engine);
#endif
COMPAT_CLOSE(mosq->sock);
mosq->sock = INVALID_SOCKET;
net__print_ssl_error(mosq);
@@ -731,7 +746,9 @@ static int net__init_ssl_ctx(struct mosquitto *mosq)
ret = SSL_CTX_check_private_key(mosq->ssl_ctx);
if(ret != 1){
log__printf(mosq, MOSQ_LOG_ERR, "Error: Client certificate/key are inconsistent.");
#if !defined(OPENSSL_NO_ENGINE)
ENGINE_FINISH(engine);
#endif
COMPAT_CLOSE(mosq->sock);
mosq->sock = INVALID_SOCKET;
net__print_ssl_error(mosq);
lib/options.c
+16 -7
View File
@@ -69,6 +69,12 @@ int mosquitto_username_pw_set(struct mosquitto *mosq, const char *username, cons
{
if(!mosq) return MOSQ_ERR_INVAL;
if(mosq->protocol == mosq_p_mqtt311 || mosq->protocol == mosq_p_mqtt31){
if(password != NULL && username == NULL){
return MOSQ_ERR_INVAL;
}
}
mosquitto__free(mosq->username);
mosq->username = NULL;
@@ -81,13 +87,14 @@ int mosquitto_username_pw_set(struct mosquitto *mosq, const char *username, cons
}
mosq->username = mosquitto__strdup(username);
if(!mosq->username) return MOSQ_ERR_NOMEM;
if(password){
mosq->password = mosquitto__strdup(password);
if(!mosq->password){
mosquitto__free(mosq->username);
mosq->username = NULL;
return MOSQ_ERR_NOMEM;
}
}
if(password){
mosq->password = mosquitto__strdup(password);
if(!mosq->password){
mosquitto__free(mosq->username);
mosq->username = NULL;
return MOSQ_ERR_NOMEM;
}
}
return MOSQ_ERR_SUCCESS;
@@ -255,6 +262,7 @@ int mosquitto_string_option(struct mosquitto *mosq, enum mosq_opt_t option, cons
switch(option){
case MOSQ_OPT_TLS_ENGINE:
#ifdef WITH_TLS
# if !defined(OPENSSL_NO_ENGINE)
eng = ENGINE_by_id(value);
if(!eng){
return MOSQ_ERR_INVAL;
@@ -265,6 +273,7 @@ int mosquitto_string_option(struct mosquitto *mosq, enum mosq_opt_t option, cons
return MOSQ_ERR_NOMEM;
}
return MOSQ_ERR_SUCCESS;
#endif
#else
return MOSQ_ERR_NOT_SUPPORTED;
#endif
lib/send_connect.c
+20 -9
View File
@@ -111,11 +111,21 @@ int send__connect(struct mosquitto *mosq, uint16_t keepalive, bool clean_session
payloadlen += will_proplen + varbytes;
}
}
/* After this check we can be sure that the username and password are
* always valid for the current protocol, so there is no need to check
* username before checking password. */
if(mosq->protocol == mosq_p_mqtt31 || mosq->protocol == mosq_p_mqtt311){
if(password != NULL && username == NULL){
return MOSQ_ERR_INVAL;
}
}
if(username){
payloadlen += 2+strlen(username);
if(password){
payloadlen += 2+strlen(password);
}
}
if(password){
payloadlen += 2+strlen(password);
}
packet->command = CMD_CONNECT;
@@ -145,9 +155,9 @@ int send__connect(struct mosquitto *mosq, uint16_t keepalive, bool clean_session
}
if(username){
byte = byte | 0x1<<7;
if(mosq->password){
byte = byte | 0x1<<6;
}
}
if(mosq->password){
byte = byte | 0x1<<6;
}
packet__write_byte(packet, byte);
packet__write_uint16(packet, keepalive);
@@ -173,11 +183,12 @@ int send__connect(struct mosquitto *mosq, uint16_t keepalive, bool clean_session
packet__write_string(packet, mosq->will->msg.topic, strlen(mosq->will->msg.topic));
packet__write_string(packet, (const char *)mosq->will->msg.payload, mosq->will->msg.payloadlen);
}
if(username){
packet__write_string(packet, username, strlen(username));
if(password){
packet__write_string(packet, password, strlen(password));
}
}
if(password){
packet__write_string(packet, password, strlen(password));
}
mosq->keepalive = keepalive;
lib/thread_mosq.c
+2 -2
View File
@@ -27,7 +27,7 @@ void *mosquitto__thread_main(void *obj);
int mosquitto_loop_start(struct mosquitto *mosq)
{
#ifdef WITH_THREADING
#if defined(WITH_THREADING) && defined(HAVE_PTHREAD_CANCEL)
if(!mosq || mosq->threaded != mosq_ts_none) return MOSQ_ERR_INVAL;
mosq->threaded = mosq_ts_self;
@@ -43,7 +43,7 @@ int mosquitto_loop_start(struct mosquitto *mosq)
int mosquitto_loop_stop(struct mosquitto *mosq, bool force)
{
#ifdef WITH_THREADING
#if defined(WITH_THREADING) && defined(HAVE_PTHREAD_CANCEL)
# ifndef WITH_BROKER
char sockpair_data = 0;
# endif
lib/util_mosq.c
+6 -5
View File
@@ -28,9 +28,10 @@ Contributors:
# include <sys/stat.h>
#endif
#if !defined(WITH_TLS) && defined(__linux__)
# if defined(__GLIBC__) && __GLIBC_PREREQ(2, 25)
#if !defined(WITH_TLS) && defined(__linux__) && defined(__GLIBC__)
# if __GLIBC_PREREQ(2, 25)
# include <sys/random.h>
# define HAVE_GETRANDOM 1
# endif
#endif
@@ -325,12 +326,12 @@ int util__random_bytes(void *bytes, int count)
if(RAND_bytes(bytes, count) == 1){
rc = MOSQ_ERR_SUCCESS;
}
#elif defined(__GLIBC__) && __GLIBC_PREREQ(2, 25)
if(getrandom(bytes, count, 0) == 0){
#elif defined(HAVE_GETRANDOM)
if(getrandom(bytes, count, 0) == count){
rc = MOSQ_ERR_SUCCESS;
}
#elif defined(WIN32)
HRYPTPROV provider;
HCRYPTPROV provider;
if(!CryptAcquireContext(&provider, NULL, NULL, PROV_RSA_FULL, CRYPT_VERIFYCONTEXT)){
return MOSQ_ERR_UNKNOWN;
man/libmosquitto.3.xml
+4 -5
View File
@@ -18,11 +18,10 @@
<title>Description</title>
<para>This is an overview of how to use libmosquitto to create MQTT
aware client programs. There may be separate man pages on each of the
functions described here in the future. There is also a binding for
libmosquitto for C++ and a Python implementation. They are not
documented here but operate in a similar way.</para>
<para>This is fairly incomplete, please see mosquitto.h for a better
description of the functions.</para>
functions described here in the future.</para>
<para>This man page is woefully incomplete, please see the comments
in mosquitto.h for missing functions and a description of the
functions.</para>
</refsect1>
<refsect1>
man/mosquitto.conf.5.xml
+7 -13
View File
@@ -686,6 +686,13 @@ log_timestamp_format %Y-%m-%dT%H:%M:%S
signal. If <replaceable>false</replaceable>, the data
will be stored in memory only. Defaults to
<replaceable>false</replaceable>.</para>
<para>The persistence file may change its format in a new
version. The broker can currently read all old formats,
but will only save in the latest format. It should always
be safe to upgrade, but cautious users may wish to take a
copy of the persistence file before installing a new
version so that they can roll back to an earlier version
if necessary.</para>
<para>Reloaded on reload signal.</para>
</listitem>
</varlistentry>
@@ -806,19 +813,6 @@ log_timestamp_format %Y-%m-%dT%H:%M:%S
<para>Reloaded on reload signal.</para>
</listitem>
</varlistentry>
<varlistentry>
<term><option>store_clean_interval</option> <replaceable>seconds</replaceable></term>
<listitem>
<para>The integer number of seconds between the internal
message store being cleaned of messages that are no
longer referenced. Lower values will result in lower
memory usage but more processor time, higher values
will have the opposite effect. Setting a value of 0
means the unreferenced messages will be disposed of as
quickly as possible. Defaults to 10 seconds.</para>
<para>Reloaded on reload signal.</para>
</listitem>
</varlistentry>
<varlistentry>
<term><option>sys_interval</option> <replaceable>seconds</replaceable></term>
<listitem>
man/mosquitto_pub.1.xml
+4 -6
View File
@@ -21,10 +21,8 @@
<arg choice='plain'>
<arg><option>-h</option> <replaceable>hostname</replaceable></arg>
<arg><option>-p</option> <replaceable>port-number</replaceable></arg>
<arg>
<arg><option>-u</option> <replaceable>username</replaceable></arg>
<arg><option>-P</option> <replaceable>password</replaceable></arg>
</arg>
<arg><option>-u</option> <replaceable>username</replaceable></arg>
<arg><option>-P</option> <replaceable>password</replaceable></arg>
<arg choice='plain' rep='repeat'><option>-t</option> <replaceable>message-topic</replaceable></arg>
</arg>
<arg choice='plain'><option>-L</option> <replaceable>URL</replaceable></arg>
@@ -366,8 +364,8 @@
<listitem>
<para>Provide a password to be used for authenticating with
the broker. Using this argument without also specifying
a username is invalid. See also the
<option>--username</option> option.</para>
a username is invalid when using MQTT v3.1 or v3.1.1.
See also the <option>--username</option> option.</para>
</listitem>
</varlistentry>
<varlistentry>
man/mosquitto_rr.1.xml
+4 -6
View File
@@ -21,10 +21,8 @@
<arg choice='plain'>
<arg><option>-h</option> <replaceable>hostname</replaceable></arg>
<arg><option>-p</option> <replaceable>port-number</replaceable></arg>
<arg>
<arg><option>-u</option> <replaceable>username</replaceable></arg>
<arg><option>-P</option> <replaceable>password</replaceable></arg>
</arg>
<arg><option>-u</option> <replaceable>username</replaceable></arg>
<arg><option>-P</option> <replaceable>password</replaceable></arg>
<arg choice='plain' rep='repeat'><option>-t</option> <replaceable>message-topic</replaceable></arg>
</arg>
<arg choice='plain'>
@@ -387,8 +385,8 @@
<listitem>
<para>Provide a password to be used for authenticating with
the broker. Using this argument without also specifying
a username is invalid. See also the
<option>--username</option> option.</para>
a username is invalid when using MQTT v3.1 or v3.1.1.
See also the <option>--username</option> option.</para>
</listitem>
</varlistentry>
<varlistentry>
man/mosquitto_sub.1.xml
+4 -6
View File
@@ -21,10 +21,8 @@
<arg choice='plain'>
<arg><option>-h</option> <replaceable>hostname</replaceable></arg>
<arg><option>-p</option> <replaceable>port-number</replaceable></arg>
<arg>
<arg><option>-u</option> <replaceable>username</replaceable></arg>
<arg><option>-P</option> <replaceable>password</replaceable></arg>
</arg>
<arg><option>-u</option> <replaceable>username</replaceable></arg>
<arg><option>-P</option> <replaceable>password</replaceable></arg>
<arg choice='plain' rep='repeat'><option>-t</option> <replaceable>message-topic</replaceable></arg>
</arg>
<arg choice='plain'>
@@ -396,8 +394,8 @@
<listitem>
<para>Provide a password to be used for authenticating with
the broker. Using this argument without also specifying
a username is invalid. See also the
<option>--username</option> option.</para>
a username is invalid when using MQTT v3.1 or v3.1.1.
See also the <option>--username</option> option.</para>
</listitem>
</varlistentry>
<varlistentry>
mosquitto.conf
-8
View File
@@ -178,14 +178,6 @@
# of packets being sent.
#set_tcp_nodelay false
# Time in seconds between cleaning the internal message store of
# unreferenced messages. Lower values will result in lower memory
# usage but more processor time, higher values will have the
# opposite effect.
# Setting a value of 0 means the unreferenced messages will be
# disposed of as quickly as possible.
#store_clean_interval 10
# Time in seconds between updates of the $SYS tree.
# Set to 0 to disable the publishing of the $SYS tree.
#sys_interval 10
set-version.sh
+1 -1
View File
@@ -2,7 +2,7 @@
MAJOR=1
MINOR=6
REVISION=2
REVISION=3
sed -i "s/^VERSION=.*/VERSION=${MAJOR}.${MINOR}.${REVISION}/" config.mk
snap/snapcraft.yaml
+1 -1
View File
@@ -1,5 +1,5 @@
name: mosquitto
version: 1.6.2
version: 1.6.3
summary: Eclipse Mosquitto MQTT broker
description: This is a message broker that supports version 3.1 and 3.1.1 of the MQTT
protocol.
src/conf.c
+1 -1
View File
@@ -504,12 +504,12 @@ int config__parse_args(struct mosquitto_db *db, struct mosquitto__config *config
config->listeners[config->listener_count-1].client_count = 0;
config->listeners[config->listener_count-1].use_username_as_clientid = config->default_listener.use_username_as_clientid;
config->listeners[config->listener_count-1].maximum_qos = config->default_listener.maximum_qos;
config->listeners[config->listener_count-1].max_topic_alias = config->default_listener.max_topic_alias;
#ifdef WITH_TLS
config->listeners[config->listener_count-1].tls_version = config->default_listener.tls_version;
config->listeners[config->listener_count-1].tls_engine = config->default_listener.tls_engine;
config->listeners[config->listener_count-1].tls_keyform = config->default_listener.tls_keyform;
config->listeners[config->listener_count-1].tls_engine_kpass_sha1 = config->default_listener.tls_engine_kpass_sha1;
config->listeners[config->listener_count-1].max_topic_alias = config->default_listener.max_topic_alias;
config->listeners[config->listener_count-1].cafile = config->default_listener.cafile;
config->listeners[config->listener_count-1].capath = config->default_listener.capath;
config->listeners[config->listener_count-1].certfile = config->default_listener.certfile;
src/context.c
+3 -8
View File
@@ -25,6 +25,7 @@ Contributors:
#include "packet_mosq.h"
#include "property_mosq.h"
#include "time_mosq.h"
#include "will_mosq.h"
#include "uthash.h"
@@ -218,13 +219,7 @@ void context__send_will(struct mosquitto_db *db, struct mosquitto *ctxt)
&ctxt->will->properties);
}
}
if(ctxt->will){
mosquitto_property_free_all(&ctxt->will->properties);
mosquitto__free(ctxt->will->msg.topic);
mosquitto__free(ctxt->will->msg.payload);
mosquitto__free(ctxt->will);
ctxt->will = NULL;
}
will__clear(ctxt);
}
@@ -232,8 +227,8 @@ void context__disconnect(struct mosquitto_db *db, struct mosquitto *context)
{
net__socket_close(db, context);
context__send_will(db, context);
if(context->session_expiry_interval == 0){
context__send_will(db, context);
#ifdef WITH_BRIDGE
if(!context->bridge)
src/database.c
+19 -14
View File
@@ -157,7 +157,7 @@ static void subhier_clean(struct mosquitto_db *db, struct mosquitto__subhier **s
leaf = nextleaf;
}
if(peer->retained){
db__msg_store_deref(db, &peer->retained);
db__msg_store_ref_dec(db, &peer->retained);
}
subhier_clean(db, &peer->children);
mosquitto__free(peer->topic);
@@ -232,7 +232,12 @@ void db__msg_store_clean(struct mosquitto_db *db)
}
}
void db__msg_store_deref(struct mosquitto_db *db, struct mosquitto_msg_store **store)
void db__msg_store_ref_inc(struct mosquitto_msg_store *store)
{
store->ref_count++;
}
void db__msg_store_ref_dec(struct mosquitto_db *db, struct mosquitto_msg_store **store)
{
(*store)->ref_count--;
if((*store)->ref_count == 0){
@@ -271,7 +276,7 @@ static void db__message_remove(struct mosquitto_db *db, struct mosquitto_msg_dat
msg_data->msg_count12--;
msg_data->msg_bytes12 -= item->store->payloadlen;
}
db__msg_store_deref(db, &item->store);
db__msg_store_ref_dec(db, &item->store);
}
mosquitto_property_free_all(&item->properties);
@@ -286,6 +291,9 @@ void db__message_dequeue_first(struct mosquitto *context, struct mosquitto_msg_d
msg = msg_data->queued;
DL_DELETE(msg_data->queued, msg);
DL_APPEND(msg_data->inflight, msg);
if(msg_data->inflight_quota > 0){
msg_data->inflight_quota--;
}
}
@@ -359,7 +367,8 @@ int db__message_insert(struct mosquitto_db *db, struct mosquitto *context, uint1
* multiple times for overlapping subscriptions, although this is only the
* case for SUBSCRIPTION with multiple subs in so is a minor concern.
*/
if(db->config->allow_duplicate_messages == false
if(context->protocol != mosq_p_mqtt5
&& db->config->allow_duplicate_messages == false
&& dir == mosq_md_out && retain == false && stored->dest_ids){
for(i=0; i<stored->dest_id_count; i++){
@@ -450,7 +459,7 @@ int db__message_insert(struct mosquitto_db *db, struct mosquitto *context, uint1
msg->prev = NULL;
msg->next = NULL;
msg->store = stored;
msg->store->ref_count++;
db__msg_store_ref_inc(msg->store);
msg->mid = mid;
msg->timestamp = mosquitto_time();
msg->direction = dir;
@@ -543,7 +552,7 @@ void db__messages_delete_list(struct mosquitto_db *db, struct mosquitto_client_m
DL_FOREACH_SAFE(*head, tail, tmp){
DL_DELETE(*head, tail);
db__msg_store_deref(db, &tail->store);
db__msg_store_ref_dec(db, &tail->store);
mosquitto_property_free_all(&tail->properties);
mosquitto__free(tail);
}
@@ -751,9 +760,7 @@ int db__message_reconnect_reset_outgoing(struct mosquitto_db *db, struct mosquit
if(msg->qos > 0){
context->msgs_out.msg_count12++;
context->msgs_out.msg_bytes12 += msg->store->payloadlen;
if(context->msgs_out.inflight_quota > 0){
context->msgs_out.inflight_quota--;
}
util__decrement_receive_quota(context);
}
switch(msg->qos){
@@ -822,9 +829,7 @@ int db__message_reconnect_reset_incoming(struct mosquitto_db *db, struct mosquit
if(msg->qos > 0){
context->msgs_in.msg_count12++;
context->msgs_in.msg_bytes12 += msg->store->payloadlen;
if(context->msgs_in.inflight_quota > 0){
context->msgs_in.inflight_quota--;
}
util__decrement_receive_quota(context);
}
if(msg->qos != 2){
@@ -1101,7 +1106,7 @@ int db__message_write(struct mosquitto_db *db, struct mosquitto *context)
}
DL_FOREACH_SAFE(context->msgs_in.queued, tail, tmp){
if(context->msgs_out.inflight_maximum != 0 && msg_count >= context->msgs_out.inflight_maximum){
if(context->msgs_out.inflight_maximum != 0 && context->msgs_in.inflight_quota == 0){
break;
}
@@ -1120,7 +1125,7 @@ int db__message_write(struct mosquitto_db *db, struct mosquitto *context)
}
DL_FOREACH_SAFE(context->msgs_out.queued, tail, tmp){
if(context->msgs_out.inflight_maximum != 0 && msg_count >= context->msgs_out.inflight_maximum){
if(context->msgs_out.inflight_maximum != 0 && context->msgs_out.inflight_quota == 0){
break;
}
src/handle_auth.c
+2 -5
View File
@@ -25,6 +25,7 @@ Contributors:
#include "packet_mosq.h"
#include "property_mosq.h"
#include "send_mosq.h"
#include "will_mosq.h"
int handle__auth(struct mosquitto_db *db, struct mosquitto *context)
@@ -119,11 +120,7 @@ int handle__auth(struct mosquitto_db *db, struct mosquitto *context)
free(auth_data_out);
if(context->state == mosq_cs_authenticating && context->will){
/* Free will without sending if this is our first authentication attempt */
mosquitto_property_free_all(&context->will->properties);
mosquitto__free(context->will->msg.payload);
mosquitto__free(context->will->msg.topic);
mosquitto__free(context->will);
context->will = NULL;
will__clear(context);
}
if(rc == MOSQ_ERR_AUTH){
send__connack(db, context, 0, MQTT_RC_NOT_AUTHORIZED, NULL);
src/handle_connect.c
+38 -32
View File
@@ -30,6 +30,7 @@ Contributors:
#include "time_mosq.h"
#include "tls_mosq.h"
#include "util_mosq.h"
#include "will_mosq.h"
#ifdef WITH_WEBSOCKETS
# include <libwebsockets.h>
@@ -91,7 +92,7 @@ void connection_check_acl(struct mosquitto_db *db, struct mosquitto *context, st
msg_tail->store->qos, msg_tail->store->retain, MOSQ_ACL_READ) != MOSQ_ERR_SUCCESS){
DL_DELETE((*head), msg_tail);
db__msg_store_deref(db, &msg_tail->store);
db__msg_store_ref_dec(db, &msg_tail->store);
mosquitto_property_free_all(&msg_tail->properties);
mosquitto__free(msg_tail);
}
@@ -160,6 +161,8 @@ int connect__on_authorised(struct mosquitto_db *db, struct mosquitto *context, v
}
session_expiry__remove(found_context);
will_delay__remove(found_context);
will__clear(found_context);
found_context->clean_start = true;
found_context->session_expiry_interval = 0;
@@ -426,9 +429,16 @@ int handle__connect(struct mosquitto_db *db, struct mosquitto *context)
goto handle_connect_error;
}
context->protocol = mosq_p_mqtt31;
if((protocol_version&0x80) == 0x80){
context->is_bridge = true;
}
}else if(!strcmp(protocol_name, PROTOCOL_NAME)){
if((protocol_version&0x7F) == PROTOCOL_VERSION_v311){
context->protocol = mosq_p_mqtt311;
if((protocol_version&0x80) == 0x80){
context->is_bridge = true;
}
}else if((protocol_version&0x7F) == PROTOCOL_VERSION_v5){
context->protocol = mosq_p_mqtt5;
}else{
@@ -530,8 +540,12 @@ int handle__connect(struct mosquitto_db *db, struct mosquitto *context)
}else{
allow_zero_length_clientid = db->config->security_options.allow_zero_length_clientid;
}
if(clean_start == 0 || allow_zero_length_clientid == false){
send__connack(db, context, 0, CONNACK_REFUSED_IDENTIFIER_REJECTED, NULL);
if((context->protocol == mosq_p_mqtt311 && clean_start == 0) || allow_zero_length_clientid == false){
if(context->protocol == mosq_p_mqtt311){
send__connack(db, context, 0, CONNACK_REFUSED_IDENTIFIER_REJECTED, NULL);
}else{
send__connack(db, context, 0, MQTT_RC_UNSPECIFIED, NULL);
}
rc = MOSQ_ERR_PROTOCOL;
goto handle_connect_error;
}else{
@@ -576,25 +590,10 @@ int handle__connect(struct mosquitto_db *db, struct mosquitto *context)
if(username_flag){
rc = packet__read_string(&context->in_packet, &username, &slen);
if(rc == MOSQ_ERR_SUCCESS){
if(password_flag){
rc = packet__read_binary(&context->in_packet, (uint8_t **)&password, &slen);
if(rc == MOSQ_ERR_NOMEM){
rc = MOSQ_ERR_NOMEM;
goto handle_connect_error;
}else if(rc == MOSQ_ERR_PROTOCOL){
if(context->protocol == mosq_p_mqtt31){
/* Password flag given, but no password. Ignore. */
}else{
rc = MOSQ_ERR_PROTOCOL;
goto handle_connect_error;
}
}
}
}else if(rc == MOSQ_ERR_NOMEM){
if(rc == MOSQ_ERR_NOMEM){
rc = MOSQ_ERR_NOMEM;
goto handle_connect_error;
}else{
}else if(rc != MOSQ_ERR_SUCCESS){
if(context->protocol == mosq_p_mqtt31){
/* Username flag given, but no username. Ignore. */
username_flag = 0;
@@ -604,7 +603,7 @@ int handle__connect(struct mosquitto_db *db, struct mosquitto *context)
}
}
}else{
if(context->protocol == mosq_p_mqtt311 || context->protocol == mosq_p_mqtt5){
if(context->protocol == mosq_p_mqtt311 || context->protocol == mosq_p_mqtt31){
if(password_flag){
/* username_flag == 0 && password_flag == 1 is forbidden */
log__printf(NULL, MOSQ_LOG_ERR, "Protocol error from %s: password without username, closing connection.", client_id);
@@ -613,6 +612,21 @@ int handle__connect(struct mosquitto_db *db, struct mosquitto *context)
}
}
}
if(password_flag){
rc = packet__read_binary(&context->in_packet, (uint8_t **)&password, &slen);
if(rc == MOSQ_ERR_NOMEM){
rc = MOSQ_ERR_NOMEM;
goto handle_connect_error;
}else if(rc == MOSQ_ERR_PROTOCOL){
if(context->protocol == mosq_p_mqtt31){
/* Password flag given, but no password. Ignore. */
}else{
rc = MOSQ_ERR_PROTOCOL;
goto handle_connect_error;
}
}
}
if(context->in_packet.pos != context->in_packet.remaining_length){
/* Surplus data at end of packet, this must be an error. */
rc = MOSQ_ERR_PROTOCOL;
@@ -745,7 +759,7 @@ int handle__connect(struct mosquitto_db *db, struct mosquitto *context)
#endif /* FINAL_WITH_TLS_PSK */
}else{
#endif /* WITH_TLS */
if(username_flag){
if(username_flag || password_flag){
/* FIXME - these ensure the mosquitto_client_id() and
* mosquitto_client_username() functions work, but is hacky */
context->id = client_id;
@@ -776,9 +790,7 @@ int handle__connect(struct mosquitto_db *db, struct mosquitto *context)
context->password = password;
username = NULL; /* Avoid free() in error: below. */
password = NULL;
}
if(!username_flag){
}else{
if((db->config->per_listener_settings && context->listener->security_options.allow_anonymous == false)
|| (!db->config->per_listener_settings && db->config->security_options.allow_anonymous == false)){
@@ -829,13 +841,7 @@ int handle__connect(struct mosquitto_db *db, struct mosquitto *context)
return rc;
}else{
free(auth_data_out);
if(context->will){
mosquitto_property_free_all(&context->will->properties);
mosquitto__free(context->will->msg.payload);
mosquitto__free(context->will->msg.topic);
mosquitto__free(context->will);
context->will = NULL;
}
will__clear(context);
if(rc == MOSQ_ERR_AUTH){
send__connack(db, context, 0, MQTT_RC_NOT_AUTHORIZED, NULL);
mosquitto__free(context->id);
src/handle_disconnect.c
+2
View File
@@ -21,6 +21,7 @@ Contributors:
#include "packet_mosq.h"
#include "property_mosq.h"
#include "send_mosq.h"
#include "will_mosq.h"
int handle__disconnect(struct mosquitto_db *db, struct mosquitto *context)
@@ -66,6 +67,7 @@ int handle__disconnect(struct mosquitto_db *db, struct mosquitto *context)
if(reason_code == MQTT_RC_DISCONNECT_WITH_WILL_MSG){
context__set_state(context, mosq_cs_disconnect_with_will);
}else{
will__clear(context);
context__set_state(context, mosq_cs_disconnecting);
}
do_disconnect(db, context, MOSQ_ERR_SUCCESS);
src/logging.c
+10 -1
View File
@@ -373,10 +373,19 @@ int log__printf(struct mosquitto *mosq, int priority, const char *fmt, ...)
void log__internal(const char *fmt, ...)
{
va_list va;
char buf[200];
int len;
va_start(va, fmt);
log__vprintf(MOSQ_LOG_INTERNAL, fmt, va);
len = vsnprintf(buf, 200, fmt, va);
va_end(va);
if(len >= 200){
log__printf(NULL, MOSQ_LOG_INTERNAL, "Internal log buffer too short (%d)", len);
return;
}
log__printf(NULL, MOSQ_LOG_INTERNAL, "%s%s%s", "\e[32m", buf, "\e[0m");
}
int mosquitto_log_vprintf(int level, const char *fmt, va_list va)
src/loop.c
+16 -1
View File
@@ -614,12 +614,19 @@ void do_disconnect(struct mosquitto_db *db, struct mosquitto *context, int reaso
#ifdef WITH_EPOLL
struct epoll_event ev;
#endif
#ifdef WITH_WEBSOCKETS
bool is_duplicate = false;
#endif
if(context->state == mosq_cs_disconnected){
return;
}
#ifdef WITH_WEBSOCKETS
if(context->wsi){
if(context->state == mosq_cs_duplicate){
is_duplicate = true;
}
if(context->state != mosq_cs_disconnecting && context->state != mosq_cs_disconnect_with_will){
context__set_state(context, mosq_cs_disconnect_ws);
}
@@ -636,7 +643,15 @@ void do_disconnect(struct mosquitto_db *db, struct mosquitto *context, int reaso
context->sock = INVALID_SOCKET;
context->pollfd_index = -1;
}
context__remove_from_by_id(db, context);
if(is_duplicate){
/* This occurs if another client is taking over the same client id.
* It is important to remove this from the by_id hash here, so it
* doesn't leave us with multiple clients in the hash with the same
* id. Websockets doesn't actually close the connection here,
* unlike for normal clients, which means there is extra time when
* there could be two clients with the same id in the hash. */
context__remove_from_by_id(db, context);
}
}else
#endif
{
src/mosquitto_broker.h
+8
View File
@@ -17,6 +17,10 @@ Contributors:
#ifndef MOSQUITTO_BROKER_H
#define MOSQUITTO_BROKER_H
#ifdef __cplusplus
extern "C" {
#endif
#include <stdbool.h>
struct mosquitto;
@@ -161,4 +165,8 @@ const char *mosquitto_client_username(const struct mosquitto *client);
*/
int mosquitto_set_username(struct mosquitto *client, const char *username);
#ifdef __cplusplus
}
#endif
#endif
src/mosquitto_broker_internal.h
+3 -1
View File
@@ -619,7 +619,8 @@ int db__message_store(struct mosquitto_db *db, const struct mosquitto *source, u
int db__message_store_find(struct mosquitto *context, uint16_t mid, struct mosquitto_msg_store **stored);
void db__msg_store_add(struct mosquitto_db *db, struct mosquitto_msg_store *store);
void db__msg_store_remove(struct mosquitto_db *db, struct mosquitto_msg_store *store);
void db__msg_store_deref(struct mosquitto_db *db, struct mosquitto_msg_store **store);
void db__msg_store_ref_inc(struct mosquitto_msg_store *store);
void db__msg_store_ref_dec(struct mosquitto_db *db, struct mosquitto_msg_store **store);
void db__msg_store_clean(struct mosquitto_db *db);
void db__msg_store_compact(struct mosquitto_db *db);
int db__message_reconnect_reset(struct mosquitto_db *db, struct mosquitto *context);
@@ -740,6 +741,7 @@ void do_disconnect(struct mosquitto_db *db, struct mosquitto *context, int reaso
int will_delay__add(struct mosquitto *context);
void will_delay__check(struct mosquitto_db *db, time_t now);
void will_delay__send_all(struct mosquitto_db *db);
void will_delay__remove(struct mosquitto *mosq);
#endif
src/mosquitto_plugin.h
+20
View File
@@ -17,6 +17,10 @@ Contributors:
#ifndef MOSQUITTO_PLUGIN_H
#define MOSQUITTO_PLUGIN_H
#ifdef __cplusplus
extern "C" {
#endif
#define MOSQ_AUTH_PLUGIN_VERSION 4
#define MOSQ_ACL_NONE 0x00
@@ -77,6 +81,16 @@ struct mosquitto_acl_msg {
* denied.
*/
/* =========================================================================
*
* Helper Functions
*
* ========================================================================= */
/* There are functions that are available for plugin developers to use in
* mosquitto_broker.h, including logging and accessor functions.
*/
/* =========================================================================
*
@@ -295,4 +309,10 @@ int mosquitto_auth_psk_key_get(void *user_data, struct mosquitto *client, const
int mosquitto_auth_start(void *user_data, struct mosquitto *client, const char *method, bool reauth, const void *data_in, uint16_t data_in_len, void **data_out, uint16_t *data_out_len);
int mosquitto_auth_continue(void *user_data, struct mosquitto *client, const char *method, const void *data_in, uint16_t data_in_len, void **data_out, uint16_t *data_out_len);
#ifdef __cplusplus
}
#endif
#endif
src/net.c
+22
View File
@@ -368,6 +368,10 @@ static int mosquitto__tls_server_ctx(struct mosquitto__listener *listener)
#endif
#endif
#ifdef SSL_OP_NO_RENEGOTIATION
SSL_CTX_set_options(listener->ssl_ctx, SSL_OP_NO_RENEGOTIATION);
#endif
snprintf(buf, 256, "mosquitto-%d", listener->port);
SSL_CTX_set_session_id_context(listener->ssl_ctx, (unsigned char *)buf, strlen(buf));
@@ -534,6 +538,7 @@ int net__socket_listen(struct mosquitto__listener *listener)
return 1;
}
if(listener->tls_engine){
#if !defined(OPENSSL_NO_ENGINE)
engine = ENGINE_by_id(listener->tls_engine);
if(!engine){
log__printf(NULL, MOSQ_LOG_ERR, "Error loading %s engine\n", listener->tls_engine);
@@ -548,6 +553,7 @@ int net__socket_listen(struct mosquitto__listener *listener)
}
ENGINE_set_default(engine, ENGINE_METHOD_ALL);
ENGINE_free(engine); /* release the structural reference from ENGINE_by_id() */
#endif
}
/* FIXME user data? */
if(listener->require_certificate){
@@ -560,10 +566,13 @@ int net__socket_listen(struct mosquitto__listener *listener)
log__printf(NULL, MOSQ_LOG_ERR, "Error: Unable to load server certificate \"%s\". Check certfile.", listener->certfile);
net__print_error(MOSQ_LOG_ERR, "Error: %s");
COMPAT_CLOSE(sock);
#if !defined(OPENSSL_NO_ENGINE)
ENGINE_FINISH(engine);
#endif
return 1;
}
if(listener->tls_keyform == mosq_k_engine){
#if !defined(OPENSSL_NO_ENGINE)
UI_METHOD *ui_method = net__get_ui_method();
if(listener->tls_engine_kpass_sha1){
if(!ENGINE_ctrl_cmd(engine, ENGINE_SECRET_MODE, ENGINE_SECRET_MODE_SHA, NULL, NULL, 0)){
@@ -593,13 +602,16 @@ int net__socket_listen(struct mosquitto__listener *listener)
ENGINE_FINISH(engine);
return 1;
}
#endif
}else{
rc = SSL_CTX_use_PrivateKey_file(listener->ssl_ctx, listener->keyfile, SSL_FILETYPE_PEM);
if(rc != 1){
log__printf(NULL, MOSQ_LOG_ERR, "Error: Unable to load server key file \"%s\". Check keyfile.", listener->keyfile);
net__print_error(MOSQ_LOG_ERR, "Error: %s");
COMPAT_CLOSE(sock);
#if !defined(OPENSSL_NO_ENGINE)
ENGINE_FINISH(engine);
#endif
return 1;
}
}
@@ -608,7 +620,9 @@ int net__socket_listen(struct mosquitto__listener *listener)
log__printf(NULL, MOSQ_LOG_ERR, "Error: Server certificate/key are inconsistent.");
net__print_error(MOSQ_LOG_ERR, "Error: %s");
COMPAT_CLOSE(sock);
#if !defined(OPENSSL_NO_ENGINE)
ENGINE_FINISH(engine);
#endif
return 1;
}
/* Load CRLs if they exist. */
@@ -618,7 +632,9 @@ int net__socket_listen(struct mosquitto__listener *listener)
log__printf(NULL, MOSQ_LOG_ERR, "Error: Unable to obtain TLS store.");
net__print_error(MOSQ_LOG_ERR, "Error: %s");
COMPAT_CLOSE(sock);
#if !defined(OPENSSL_NO_ENGINE)
ENGINE_FINISH(engine);
#endif
return 1;
}
lookup = X509_STORE_add_lookup(store, X509_LOOKUP_file());
@@ -627,7 +643,9 @@ int net__socket_listen(struct mosquitto__listener *listener)
log__printf(NULL, MOSQ_LOG_ERR, "Error: Unable to load certificate revocation file \"%s\". Check crlfile.", listener->crlfile);
net__print_error(MOSQ_LOG_ERR, "Error: %s");
COMPAT_CLOSE(sock);
#if !defined(OPENSSL_NO_ENGINE)
ENGINE_FINISH(engine);
#endif
return 1;
}
X509_STORE_set_flags(store, X509_V_FLAG_CRL_CHECK);
@@ -644,7 +662,9 @@ int net__socket_listen(struct mosquitto__listener *listener)
if(mosquitto__tls_server_ctx(listener)){
COMPAT_CLOSE(sock);
#if !defined(OPENSSL_NO_ENGINE)
ENGINE_FINISH(engine);
#endif
return 1;
}
SSL_CTX_set_psk_server_callback(listener->ssl_ctx, psk_server_callback);
@@ -654,7 +674,9 @@ int net__socket_listen(struct mosquitto__listener *listener)
log__printf(NULL, MOSQ_LOG_ERR, "Error: Unable to set TLS PSK hint.");
net__print_error(MOSQ_LOG_ERR, "Error: %s");
COMPAT_CLOSE(sock);
#if !defined(OPENSSL_NO_ENGINE)
ENGINE_FINISH(engine);
#endif
return 1;
}
}
src/persist_read.c
+1 -1
View File
@@ -138,7 +138,7 @@ static int persist__client_msg_restore(struct mosquitto_db *db, struct P_client_
return 1;
}
cmsg->store = load->store;
cmsg->store->ref_count++;
db__msg_store_ref_inc(cmsg->store);
context = persist__find_or_add_context(db, chunk->client_id, 0);
if(!context){
src/security.c
+1 -1
View File
@@ -752,7 +752,7 @@ int mosquitto_psk_key_get(struct mosquitto_db *db, struct mosquitto *context, co
}
for(i=0; i<opts->auth_plugin_config_count; i++){
if(opts->auth_plugin_configs[i].plugin.version == 3
if(opts->auth_plugin_configs[i].plugin.version == 4
&& opts->auth_plugin_configs[i].plugin.psk_key_get_v4){
rc = opts->auth_plugin_configs[i].plugin.psk_key_get_v4(
src/subs.c
+7 -7
View File
@@ -146,14 +146,14 @@ static int subs__process(struct mosquitto_db *db, struct mosquitto__subhier *hie
}
#endif
if(hier->retained){
db__msg_store_deref(db, &hier->retained);
db__msg_store_ref_dec(db, &hier->retained);
#ifdef WITH_SYS_TREE
db->retained_count--;
#endif
}
if(stored->payloadlen){
hier->retained = stored;
hier->retained->ref_count++;
db__msg_store_ref_inc(hier->retained);
#ifdef WITH_SYS_TREE
db->retained_count++;
#endif
@@ -706,7 +706,7 @@ int sub__add(struct mosquitto_db *db, struct mosquitto *context, const char *sub
if(sub__topic_tokenise(sub, &tokens)) return 1;
if(!strcmp(tokens->topic, "$shared")){
if(!strcmp(tokens->topic, "$share")){
if(!tokens->next || !tokens->next->next){
sub__topic_tokens_free(tokens);
return MOSQ_ERR_PROTOCOL;
@@ -756,7 +756,7 @@ int sub__remove(struct mosquitto_db *db, struct mosquitto *context, const char *
if(sub__topic_tokenise(sub, &tokens)) return 1;
if(!strcmp(tokens->topic, "$shared")){
if(!strcmp(tokens->topic, "$share")){
if(!tokens->next || !tokens->next->next){
sub__topic_tokens_free(tokens);
return MOSQ_ERR_PROTOCOL;
@@ -803,7 +803,7 @@ int sub__messages_queue(struct mosquitto_db *db, const char *source_id, const ch
clients - this is required because websockets client calls
db__message_write(), which could remove the message if ref_count==0.
*/
(*stored)->ref_count++;
db__msg_store_ref_inc(*stored);
HASH_FIND(hh, db->subs, tokens->topic, tokens->topic_len, subhier);
if(subhier){
@@ -818,7 +818,7 @@ int sub__messages_queue(struct mosquitto_db *db, const char *source_id, const ch
sub__topic_tokens_free(tokens);
/* Remove our reference and free if needed. */
db__msg_store_deref(db, stored);
db__msg_store_ref_dec(db, stored);
return rc;
}
@@ -981,7 +981,7 @@ static int retain__process(struct mosquitto_db *db, struct mosquitto__subhier *b
struct mosquitto_msg_store *retained;
if(branch->retained->message_expiry_time > 0 && now > branch->retained->message_expiry_time){
db__msg_store_deref(db, &branch->retained);
db__msg_store_ref_dec(db, &branch->retained);
branch->retained = NULL;
#ifdef WITH_SYS_TREE
db->retained_count--;
src/will_delay.c
+15 -7
View File
@@ -24,12 +24,6 @@ Contributors:
#include "memory_mosq.h"
#include "time_mosq.h"
struct will_delay_list {
struct mosquitto *context;
struct will_delay_list *prev;
struct will_delay_list *next;
};
static struct will_delay_list *delay_list = NULL;
static time_t last_check = 0;
@@ -48,6 +42,7 @@ int will_delay__add(struct mosquitto *context)
if(!item) return MOSQ_ERR_NOMEM;
item->context = context;
context->will_delay_entry = item;
item->context->will_delay_time = time(NULL) + item->context->will_delay_interval;
DL_INSERT_INORDER(delay_list, item, will_delay__cmp);
@@ -64,6 +59,7 @@ void will_delay__send_all(struct mosquitto_db *db)
DL_FOREACH_SAFE(delay_list, item, tmp){
DL_DELETE(delay_list, item);
item->context->will_delay_interval = 0;
item->context->will_delay_entry = NULL;
context__send_will(db, item->context);
mosquitto__free(item);
}
@@ -82,8 +78,11 @@ void will_delay__check(struct mosquitto_db *db, time_t now)
if(item->context->will_delay_time < now){
DL_DELETE(delay_list, item);
item->context->will_delay_interval = 0;
item->context->will_delay_entry = NULL;
context__send_will(db, item->context);
context__add_to_disused(db, item->context);
if(item->context->session_expiry_interval == 0){
context__add_to_disused(db, item->context);
}
mosquitto__free(item);
}else{
return;
@@ -92,3 +91,12 @@ void will_delay__check(struct mosquitto_db *db, time_t now)
}
void will_delay__remove(struct mosquitto *mosq)
{
if(mosq->will_delay_entry != NULL){
DL_DELETE(delay_list, mosq->will_delay_entry);
mosq->will_delay_entry = NULL;
}
}
test/broker/01-connect-uname-password-denied-no-will.py
+1 -4
View File
@@ -31,9 +31,6 @@ mid = 1
subscribe_packet = mosq_test.gen_subscribe(mid, topic="will/test", qos=0)
suback_packet = mosq_test.gen_suback(mid, 0)
pingreq_packet = mosq_test.gen_pingreq()
pingresp_packet = mosq_test.gen_pingresp()
connect2_packet = mosq_test.gen_connect("connect-uname-pwd-test", keepalive=keepalive, username="user", password="password9")
connack2_packet = mosq_test.gen_connack(rc=5)
@@ -47,7 +44,7 @@ try:
sock2.close()
# If we receive a will here, this is an error
mosq_test.do_send_receive(sock1, pingreq_packet, pingresp_packet)
mosq_test.do_ping(sock1)
sock1.close()
rc = 0
test/broker/02-shared-qos0-v5.py
+14 -14
View File
@@ -3,10 +3,10 @@
# Test whether shared subscriptions work
# Client 1 subscribes to #, non shared. Should receive everything.
# Client 2 subscribes to $shared/one/share-test
# Client 3 subscribes to $shared/one/share-test and $shared/two/share-test
# Client 4 subscribes to $shared/two/share-test
# Client 5 subscribes to $shared/one/share-test
# Client 2 subscribes to $share/one/share-test
# Client 3 subscribes to $share/one/share-test and $share/two/share-test
# Client 4 subscribes to $share/two/share-test
# Client 5 subscribes to $share/one/share-test
# A publish to "share-test" should always go to client 1.
# The first publish should also go to client 2 (share one) and client 3 (share two)
@@ -37,19 +37,19 @@ connack5_packet = mosq_test.gen_connack(rc=0, proto_ver=5)
subscribe1_packet = mosq_test.gen_subscribe(mid, "#", 0, proto_ver=5)
suback1_packet = mosq_test.gen_suback(mid, 0, proto_ver=5)
subscribe2_packet = mosq_test.gen_subscribe(mid, "$shared/one/share-test", 0, proto_ver=5)
subscribe2_packet = mosq_test.gen_subscribe(mid, "$share/one/share-test", 0, proto_ver=5)
suback2_packet = mosq_test.gen_suback(mid, 0, proto_ver=5)
subscribe3a_packet = mosq_test.gen_subscribe(mid, "$shared/one/share-test", 0, proto_ver=5)
subscribe3a_packet = mosq_test.gen_subscribe(mid, "$share/one/share-test", 0, proto_ver=5)
suback3a_packet = mosq_test.gen_suback(mid, 0, proto_ver=5)
subscribe3b_packet = mosq_test.gen_subscribe(mid, "$shared/two/share-test", 0, proto_ver=5)
subscribe3b_packet = mosq_test.gen_subscribe(mid, "$share/two/share-test", 0, proto_ver=5)
suback3b_packet = mosq_test.gen_suback(mid, 0, proto_ver=5)
subscribe4_packet = mosq_test.gen_subscribe(mid, "$shared/two/share-test", 0, proto_ver=5)
subscribe4_packet = mosq_test.gen_subscribe(mid, "$share/two/share-test", 0, proto_ver=5)
suback4_packet = mosq_test.gen_suback(mid, 0, proto_ver=5)
subscribe5_packet = mosq_test.gen_subscribe(mid, "$shared/one/share-test", 0, proto_ver=5)
subscribe5_packet = mosq_test.gen_subscribe(mid, "$share/one/share-test", 0, proto_ver=5)
suback5_packet = mosq_test.gen_suback(mid, 0, proto_ver=5)
publish1_packet = mosq_test.gen_publish("share-test", qos=0, payload="message1", proto_ver=5)
@@ -60,19 +60,19 @@ mid = 2
unsubscribe1_packet = mosq_test.gen_unsubscribe(mid, "#", proto_ver=5)
unsuback1_packet = mosq_test.gen_unsuback(mid, proto_ver=5)
unsubscribe2_packet = mosq_test.gen_unsubscribe(mid, "$shared/one/share-test", proto_ver=5)
unsubscribe2_packet = mosq_test.gen_unsubscribe(mid, "$share/one/share-test", proto_ver=5)
unsuback2_packet = mosq_test.gen_unsuback(mid, proto_ver=5)
unsubscribe3a_packet = mosq_test.gen_unsubscribe(mid, "$shared/one/share-test", proto_ver=5)
unsubscribe3a_packet = mosq_test.gen_unsubscribe(mid, "$share/one/share-test", proto_ver=5)
unsuback3a_packet = mosq_test.gen_unsuback(mid, proto_ver=5)
unsubscribe3b_packet = mosq_test.gen_unsubscribe(mid, "$shared/two/share-test", proto_ver=5)
unsubscribe3b_packet = mosq_test.gen_unsubscribe(mid, "$share/two/share-test", proto_ver=5)
unsuback3b_packet = mosq_test.gen_unsuback(mid, proto_ver=5)
unsubscribe4_packet = mosq_test.gen_unsubscribe(mid, "$shared/two/share-test", proto_ver=5)
unsubscribe4_packet = mosq_test.gen_unsubscribe(mid, "$share/two/share-test", proto_ver=5)
unsuback4_packet = mosq_test.gen_unsuback(mid, proto_ver=5)
unsubscribe5_packet = mosq_test.gen_unsubscribe(mid, "$shared/one/share-test", proto_ver=5)
unsubscribe5_packet = mosq_test.gen_unsubscribe(mid, "$share/one/share-test", proto_ver=5)
unsuback5_packet = mosq_test.gen_unsuback(mid, proto_ver=5)

Some files were not shown because too many files have changed in this diff Show More