Mirrors_Framework/mosquitto
1
0
Fork 0
mirror of https://github.com/eclipse-mosquitto/mosquitto.git synced 2026-02-06 02:52:07 +08:00
Code Issues Packages Projects Releases Wiki Activity

Refactor client structs.

Browse Source
This commit is contained in:
Roger A. Light
2022-12-23 17:22:10 +00:00
parent 5364410615
commit 1081661f74
5 changed files with 130 additions and 128 deletions
Download Patch File Download Diff File Expand all files Collapse all files

52
include/mosquitto_broker.h
View File

@@ -238,19 +238,12 @@ struct mosquitto_evt_persist_restore {
void *future[8]; void *future[8];
}; };
/* Data for the MOSQ_EVT_PERSIST_CLIENT_ADD/_DELETE/_UPDATE event */
/* NOTE: The persistence interface is currently marked as unstable, which means struct mosquitto_client {
* it may change in a future minor release. */ char *client_id;
struct mosquitto_evt_persist_client { char *username;
void *future; char *auth_method;
const char *client_id; struct mosquitto_message_v5 *will;
const char *username;
const char *auth_method;
const struct mosquitto_message_v5 *will;
char *plugin_client_id;
char *plugin_username;
char *plugin_auth_method;
struct mosquitto_message_v5 *plugin_will;
time_t will_delay_time; /* update */ time_t will_delay_time; /* update */
time_t session_expiry_time; /* update */ time_t session_expiry_time; /* update */
uint32_t will_delay_interval; uint32_t will_delay_interval;
@@ -263,6 +256,15 @@ struct mosquitto_evt_persist_client {
void *future2[8]; void *future2[8];
}; };
/* Data for the MOSQ_EVT_PERSIST_CLIENT_ADD/_DELETE/_UPDATE event */
/* NOTE: The persistence interface is currently marked as unstable, which means
* it may change in a future minor release. */
struct mosquitto_evt_persist_client {
void *future;
struct mosquitto_client data;
void *future2[8];
};
struct mosquitto_subscription { struct mosquitto_subscription {
char *client_id; char *client_id;
@@ -890,7 +892,7 @@ mosq_EXPORT int mosquitto_broker_node_id_set(uint16_t id);
* client * client
* client->listener_port - the listener port that this client last connected to * client->listener_port - the listener port that this client last connected to
* *
* All other members of struct mosquitto_evt_persist_client are unused. * All other members of struct mosquitto_client are unused.
* *
* Returns: * Returns:
* MOSQ_ERR_SUCCESS - on success * MOSQ_ERR_SUCCESS - on success
@@ -898,7 +900,7 @@ mosq_EXPORT int mosquitto_broker_node_id_set(uint16_t id);
* client with the same ID already exists. * client with the same ID already exists.
* MOSQ_ERR_NOMEM - on out of memory * MOSQ_ERR_NOMEM - on out of memory
*/ */
mosq_EXPORT int mosquitto_persist_client_add(struct mosquitto_evt_persist_client *client); mosq_EXPORT int mosquitto_persist_client_add(struct mosquitto_client *client);
/* Function: mosquitto_persist_client_update /* Function: mosquitto_persist_client_update
@@ -925,14 +927,14 @@ mosq_EXPORT int mosquitto_persist_client_add(struct mosquitto_evt_persist_client
* client * client
* client->listener_port - the listener port that this client last connected to * client->listener_port - the listener port that this client last connected to
* *
* All other members of struct mosquitto_evt_persist_client are unused. * All other members of struct mosquitto_client are unused.
* *
* Returns: * Returns:
* MOSQ_ERR_SUCCESS - on success * MOSQ_ERR_SUCCESS - on success
* MOSQ_ERR_INVAL - if client or client->plugin_client_id is NULL * MOSQ_ERR_INVAL - if client or client->plugin_client_id is NULL
* MOSQ_ERR_NOT_FOUND - the client is not found * MOSQ_ERR_NOT_FOUND - the client is not found
*/ */
mosq_EXPORT int mosquitto_persist_client_update(struct mosquitto_evt_persist_client *client); mosq_EXPORT int mosquitto_persist_client_update(struct mosquitto_client *client);
/* Function: mosquitto_persist_client_delete /* Function: mosquitto_persist_client_delete
@@ -968,14 +970,14 @@ mosq_EXPORT int mosquitto_persist_client_delete(const char *client_id);
* client_msg->subscription_identifier - the MQTT v5 subscription identifier, * client_msg->subscription_identifier - the MQTT v5 subscription identifier,
* for outgoing messages only. * for outgoing messages only.
* *
* All other members of struct mosquitto_evt_persist_client_msg are unused. * All other members of struct mosquitto_client_msg are unused.
* *
* Returns: * Returns:
* MOSQ_ERR_SUCCESS - on success * MOSQ_ERR_SUCCESS - on success
* MOSQ_ERR_INVAL - if client_msg or client_msg->plugin_client_id is NULL * MOSQ_ERR_INVAL - if client_msg or client_msg->plugin_client_id is NULL
* MOSQ_ERR_NOT_FOUND - the client or base message is not found * MOSQ_ERR_NOT_FOUND - the client or base message is not found
*/ */
mosq_EXPORT int mosquitto_persist_client_msg_add(struct mosquitto_evt_persist_client_msg *client_msg); mosq_EXPORT int mosquitto_persist_client_msg_add(struct mosquitto_client_msg *client_msg);
/* Function: mosquitto_persist_client_msg_delete /* Function: mosquitto_persist_client_msg_delete
@@ -991,14 +993,14 @@ mosq_EXPORT int mosquitto_persist_client_msg_add(struct mosquitto_evt_persist_cl
* client_msg->direction - the direction of the message from the perspective * client_msg->direction - the direction of the message from the perspective
* of the broker (mosq_bmd_in / mosq_bmd_out) * of the broker (mosq_bmd_in / mosq_bmd_out)
* *
* All other members of struct mosquitto_evt_persist_client_msg are unused. * All other members of struct mosquitto_client_msg are unused.
* *
* Returns: * Returns:
* MOSQ_ERR_SUCCESS - on success * MOSQ_ERR_SUCCESS - on success
* MOSQ_ERR_INVAL - if client_msg or client_msg->plugin_client_id is NULL * MOSQ_ERR_INVAL - if client_msg or client_msg->plugin_client_id is NULL
* MOSQ_ERR_NOT_FOUND - the client is not found * MOSQ_ERR_NOT_FOUND - the client is not found
*/ */
mosq_EXPORT int mosquitto_persist_client_msg_delete(struct mosquitto_evt_persist_client_msg *client_msg); mosq_EXPORT int mosquitto_persist_client_msg_delete(struct mosquitto_client_msg *client_msg);
/* Function: mosquitto_persist_client_msg_update /* Function: mosquitto_persist_client_msg_update
@@ -1015,14 +1017,14 @@ mosq_EXPORT int mosquitto_persist_client_msg_delete(struct mosquitto_evt_persist
* of the broker (mosq_bmd_in / mosq_bmd_out) * of the broker (mosq_bmd_in / mosq_bmd_out)
* client_msg->state - the new state of the message * client_msg->state - the new state of the message
* *
* All other members of struct mosquitto_evt_persist_client_msg are unused. * All other members of struct mosquitto_client_msg are unused.
* *
* Returns: * Returns:
* MOSQ_ERR_SUCCESS - on success * MOSQ_ERR_SUCCESS - on success
* MOSQ_ERR_INVAL - if client_msg or client_msg->plugin_client_id is NULL * MOSQ_ERR_INVAL - if client_msg or client_msg->plugin_client_id is NULL
* MOSQ_ERR_NOT_FOUND - the client is not found * MOSQ_ERR_NOT_FOUND - the client is not found
*/ */
mosq_EXPORT int mosquitto_persist_client_msg_update(struct mosquitto_evt_persist_client_msg *client_msg); mosq_EXPORT int mosquitto_persist_client_msg_update(struct mosquitto_client_msg *client_msg);
/* Function: mosquitto_persist_client_msg_clear /* Function: mosquitto_persist_client_msg_clear
@@ -1036,14 +1038,14 @@ mosq_EXPORT int mosquitto_persist_client_msg_update(struct mosquitto_evt_persist
* the perspective of the broker (mosq_bmd_in / mosq_bmd_out / mosq_bmd_all) * the perspective of the broker (mosq_bmd_in / mosq_bmd_out / mosq_bmd_all)
* client_msg->state - the new state of the message * client_msg->state - the new state of the message
* *
* All other members of struct mosquitto_evt_persist_client_msg are unused. * All other members of struct mosquitto_client_msg are unused.
* *
* Returns: * Returns:
* MOSQ_ERR_SUCCESS - on success * MOSQ_ERR_SUCCESS - on success
* MOSQ_ERR_INVAL - if client_msg or client_msg->plugin_client_id is NULL * MOSQ_ERR_INVAL - if client_msg or client_msg->plugin_client_id is NULL
* MOSQ_ERR_NOT_FOUND - the client is not found * MOSQ_ERR_NOT_FOUND - the client is not found
*/ */
mosq_EXPORT int mosquitto_persist_client_msg_clear(struct mosquitto_evt_persist_client_msg *client_msg); mosq_EXPORT int mosquitto_persist_client_msg_clear(struct mosquitto_client_msg *client_msg);
/* Function: mosquitto_persist_base_msg_add /* Function: mosquitto_persist_base_msg_add
* *

36
plugins/persist-sqlite/clients.c
View File

@@ -33,11 +33,11 @@ int persist_sqlite__client_add_cb(int event, void *event_data, void *userdata)
UNUSED(event); UNUSED(event);
if(sqlite3_bind_text(ms->client_add_stmt, 1, if(sqlite3_bind_text(ms->client_add_stmt, 1,
ed->client_id, (int)strlen(ed->client_id), SQLITE_STATIC) == SQLITE_OK){ ed->data.client_id, (int)strlen(ed->data.client_id), SQLITE_STATIC) == SQLITE_OK){
if(ed->username){ if(ed->data.username){
sqlite3_bind_text(ms->client_add_stmt, 2, sqlite3_bind_text(ms->client_add_stmt, 2,
ed->username, (int)strlen(ed->username), ed->data.username, (int)strlen(ed->data.username),
SQLITE_STATIC); SQLITE_STATIC);
}else{ }else{
sqlite3_bind_null(ms->client_add_stmt, 2); sqlite3_bind_null(ms->client_add_stmt, 2);
@@ -45,14 +45,14 @@ int persist_sqlite__client_add_cb(int event, void *event_data, void *userdata)
now = time(NULL); now = time(NULL);
if(sqlite3_bind_int64(ms->client_add_stmt, 3, now) == SQLITE_OK if(sqlite3_bind_int64(ms->client_add_stmt, 3, now) == SQLITE_OK
&& sqlite3_bind_int64(ms->client_add_stmt, 4, ed->will_delay_time) == SQLITE_OK && sqlite3_bind_int64(ms->client_add_stmt, 4, ed->data.will_delay_time) == SQLITE_OK
&& sqlite3_bind_int64(ms->client_add_stmt, 5, ed->session_expiry_time) == SQLITE_OK && sqlite3_bind_int64(ms->client_add_stmt, 5, ed->data.session_expiry_time) == SQLITE_OK
&& sqlite3_bind_int(ms->client_add_stmt, 6, ed->listener_port) == SQLITE_OK && sqlite3_bind_int(ms->client_add_stmt, 6, ed->data.listener_port) == SQLITE_OK
&& sqlite3_bind_int(ms->client_add_stmt, 7, (int)ed->max_packet_size) == SQLITE_OK && sqlite3_bind_int(ms->client_add_stmt, 7, (int)ed->data.max_packet_size) == SQLITE_OK
&& sqlite3_bind_int(ms->client_add_stmt, 8, ed->max_qos) == SQLITE_OK && sqlite3_bind_int(ms->client_add_stmt, 8, ed->data.max_qos) == SQLITE_OK
&& sqlite3_bind_int(ms->client_add_stmt, 9, ed->retain_available) == SQLITE_OK && sqlite3_bind_int(ms->client_add_stmt, 9, ed->data.retain_available) == SQLITE_OK
&& sqlite3_bind_int(ms->client_add_stmt, 10, (int)ed->session_expiry_interval) == SQLITE_OK && sqlite3_bind_int(ms->client_add_stmt, 10, (int)ed->data.session_expiry_interval) == SQLITE_OK
&& sqlite3_bind_int(ms->client_add_stmt, 11, (int)ed->will_delay_interval) == SQLITE_OK && sqlite3_bind_int(ms->client_add_stmt, 11, (int)ed->data.will_delay_interval) == SQLITE_OK
){ ){
ms->event_count++; ms->event_count++;
@@ -78,7 +78,7 @@ int persist_sqlite__client_remove_cb(int event, void *event_data, void *userdata
UNUSED(event); UNUSED(event);
if(sqlite3_bind_text(ms->subscription_clear_stmt, 1, if(sqlite3_bind_text(ms->subscription_clear_stmt, 1,
ed->client_id, (int)strlen(ed->client_id), SQLITE_STATIC) == SQLITE_OK){ ed->data.client_id, (int)strlen(ed->data.client_id), SQLITE_STATIC) == SQLITE_OK){
ms->event_count++; ms->event_count++;
rc = sqlite3_step(ms->subscription_clear_stmt); rc = sqlite3_step(ms->subscription_clear_stmt);
@@ -90,7 +90,7 @@ int persist_sqlite__client_remove_cb(int event, void *event_data, void *userdata
} }
} }
if(sqlite3_bind_text(ms->client_remove_stmt, 1, if(sqlite3_bind_text(ms->client_remove_stmt, 1,
ed->client_id, (int)strlen(ed->client_id), SQLITE_STATIC) == SQLITE_OK){ ed->data.client_id, (int)strlen(ed->data.client_id), SQLITE_STATIC) == SQLITE_OK){
ms->event_count++; ms->event_count++;
rc = sqlite3_step(ms->client_remove_stmt); rc = sqlite3_step(ms->client_remove_stmt);
@@ -101,7 +101,7 @@ int persist_sqlite__client_remove_cb(int event, void *event_data, void *userdata
rc = MOSQ_ERR_UNKNOWN; rc = MOSQ_ERR_UNKNOWN;
} }
} }
persist_sqlite__client_msg_clear(ms, ed->client_id); persist_sqlite__client_msg_clear(ms, ed->data.client_id);
return rc; return rc;
} }
@@ -115,10 +115,10 @@ int persist_sqlite__client_update_cb(int event, void *event_data, void *userdata
UNUSED(event); UNUSED(event);
if(sqlite3_bind_int64(ms->client_update_stmt, 1, ed->session_expiry_time) == SQLITE_OK if(sqlite3_bind_int64(ms->client_update_stmt, 1, ed->data.session_expiry_time) == SQLITE_OK
&& sqlite3_bind_int64(ms->client_update_stmt, 2, ed->will_delay_time) == SQLITE_OK && sqlite3_bind_int64(ms->client_update_stmt, 2, ed->data.will_delay_time) == SQLITE_OK
&& sqlite3_bind_text(ms->client_update_stmt, 3, ed->client_id, && sqlite3_bind_text(ms->client_update_stmt, 3, ed->data.client_id,
(int)strlen(ed->client_id), SQLITE_STATIC) == SQLITE_OK (int)strlen(ed->data.client_id), SQLITE_STATIC) == SQLITE_OK
){ ){
ms->event_count++; ms->event_count++;

28
plugins/persist-sqlite/restore.c
View File

@@ -179,7 +179,7 @@ static int client_restore(struct mosquitto_sqlite *ms)
{ {
sqlite3_stmt *stmt; sqlite3_stmt *stmt;
int rc; int rc;
struct mosquitto_evt_persist_client client; struct mosquitto_client client;
long count = 0, failed = 0; long count = 0, failed = 0;
const char *str; const char *str;
@@ -201,11 +201,11 @@ static int client_restore(struct mosquitto_sqlite *ms)
while(sqlite3_step(stmt) == SQLITE_ROW){ while(sqlite3_step(stmt) == SQLITE_ROW){
str = (const char *)sqlite3_column_text(stmt, 0); str = (const char *)sqlite3_column_text(stmt, 0);
if(str){ if(str){
client.plugin_client_id = strdup(str); client.client_id = strdup(str);
} }
str = (const char *)sqlite3_column_text(stmt, 1); str = (const char *)sqlite3_column_text(stmt, 1);
if(str){ if(str){
client.plugin_username = strdup(str); client.username = strdup(str);
} }
client.will_delay_time = (time_t)sqlite3_column_int64(stmt, 2); client.will_delay_time = (time_t)sqlite3_column_int64(stmt, 2);
client.session_expiry_time = (time_t)sqlite3_column_int64(stmt, 3); client.session_expiry_time = (time_t)sqlite3_column_int64(stmt, 3);
@@ -339,7 +339,7 @@ static int base_msg_restore(struct mosquitto_sqlite *ms)
static int client_msg_restore(struct mosquitto_sqlite *ms) static int client_msg_restore(struct mosquitto_sqlite *ms)
{ {
sqlite3_stmt *stmt; sqlite3_stmt *stmt;
struct mosquitto_evt_persist_client_msg client_msg; struct mosquitto_client_msg client_msg;
int rc; int rc;
long count = 0, failed = 0; long count = 0, failed = 0;
@@ -355,16 +355,16 @@ static int client_msg_restore(struct mosquitto_sqlite *ms)
memset(&client_msg, 0, sizeof(client_msg)); memset(&client_msg, 0, sizeof(client_msg));
while(sqlite3_step(stmt) == SQLITE_ROW){ while(sqlite3_step(stmt) == SQLITE_ROW){
client_msg.data.client_id = (const char *)sqlite3_column_text(stmt, 0); client_msg.client_id = (const char *)sqlite3_column_text(stmt, 0);
client_msg.data.cmsg_id = (uint64_t)sqlite3_column_int64(stmt, 1); client_msg.cmsg_id = (uint64_t)sqlite3_column_int64(stmt, 1);
client_msg.data.store_id = (uint64_t)sqlite3_column_int64(stmt, 2); client_msg.store_id = (uint64_t)sqlite3_column_int64(stmt, 2);
client_msg.data.dup = sqlite3_column_int(stmt, 3); client_msg.dup = sqlite3_column_int(stmt, 3);
client_msg.data.direction = (uint8_t)sqlite3_column_int(stmt, 4); client_msg.direction = (uint8_t)sqlite3_column_int(stmt, 4);
client_msg.data.mid = (uint16_t)sqlite3_column_int(stmt, 5); client_msg.mid = (uint16_t)sqlite3_column_int(stmt, 5);
client_msg.data.qos = (uint8_t)sqlite3_column_int(stmt, 6); client_msg.qos = (uint8_t)sqlite3_column_int(stmt, 6);
client_msg.data.retain = sqlite3_column_int(stmt, 7); client_msg.retain = sqlite3_column_int(stmt, 7);
client_msg.data.state = (uint8_t)sqlite3_column_int(stmt, 8); client_msg.state = (uint8_t)sqlite3_column_int(stmt, 8);
client_msg.data.subscription_identifier = (uint32_t)sqlite3_column_int(stmt, 9); client_msg.subscription_identifier = (uint32_t)sqlite3_column_int(stmt, 9);
rc = mosquitto_persist_client_msg_add(&client_msg); rc = mosquitto_persist_client_msg_add(&client_msg);
if(rc == MOSQ_ERR_SUCCESS){ if(rc == MOSQ_ERR_SUCCESS){

50
src/plugin_persist.c
View File

@@ -59,21 +59,21 @@ void plugin_persist__handle_client_add(struct mosquitto *context)
opts = &db.config->security_options; opts = &db.config->security_options;
memset(&event_data, 0, sizeof(event_data)); memset(&event_data, 0, sizeof(event_data));
event_data.client_id = context->id; event_data.data.client_id = context->id;
event_data.username = context->username; event_data.data.username = context->username;
event_data.auth_method = context->auth_method; event_data.data.auth_method = context->auth_method;
event_data.will_delay_time = context->will_delay_time; event_data.data.will_delay_time = context->will_delay_time;
event_data.session_expiry_time = context->session_expiry_time; event_data.data.session_expiry_time = context->session_expiry_time;
event_data.will_delay_interval = context->will_delay_interval; event_data.data.will_delay_interval = context->will_delay_interval;
event_data.session_expiry_interval = context->session_expiry_interval; event_data.data.session_expiry_interval = context->session_expiry_interval;
if(context->listener){ if(context->listener){
event_data.listener_port = context->listener->port; event_data.data.listener_port = context->listener->port;
}else{ }else{
event_data.listener_port = 0; event_data.data.listener_port = 0;
} }
event_data.max_qos = context->max_qos; event_data.data.max_qos = context->max_qos;
event_data.retain_available = context->retain_available; event_data.data.retain_available = context->retain_available;
event_data.max_packet_size = context->maximum_packet_size; event_data.data.max_packet_size = context->maximum_packet_size;
DL_FOREACH(opts->plugin_callbacks.persist_client_add, cb_base){ DL_FOREACH(opts->plugin_callbacks.persist_client_add, cb_base){
cb_base->cb(MOSQ_EVT_PERSIST_CLIENT_ADD, &event_data, cb_base->userdata); cb_base->cb(MOSQ_EVT_PERSIST_CLIENT_ADD, &event_data, cb_base->userdata);
@@ -95,21 +95,21 @@ void plugin_persist__handle_client_update(struct mosquitto *context)
opts = &db.config->security_options; opts = &db.config->security_options;
memset(&event_data, 0, sizeof(event_data)); memset(&event_data, 0, sizeof(event_data));
event_data.client_id = context->id; event_data.data.client_id = context->id;
event_data.username = context->username; event_data.data.username = context->username;
event_data.auth_method = context->auth_method; event_data.data.auth_method = context->auth_method;
event_data.will_delay_time = context->will_delay_time; event_data.data.will_delay_time = context->will_delay_time;
event_data.session_expiry_time = context->session_expiry_time; event_data.data.session_expiry_time = context->session_expiry_time;
event_data.will_delay_interval = context->will_delay_interval; event_data.data.will_delay_interval = context->will_delay_interval;
event_data.session_expiry_interval = context->session_expiry_interval; event_data.data.session_expiry_interval = context->session_expiry_interval;
if(context->listener){ if(context->listener){
event_data.listener_port = context->listener->port; event_data.data.listener_port = context->listener->port;
}else{ }else{
event_data.listener_port = 0; event_data.data.listener_port = 0;
} }
event_data.max_qos = context->max_qos; event_data.data.max_qos = context->max_qos;
event_data.retain_available = context->retain_available; event_data.data.retain_available = context->retain_available;
event_data.max_packet_size = context->maximum_packet_size; event_data.data.max_packet_size = context->maximum_packet_size;
DL_FOREACH(opts->plugin_callbacks.persist_client_update, cb_base){ DL_FOREACH(opts->plugin_callbacks.persist_client_update, cb_base){
cb_base->cb(MOSQ_EVT_PERSIST_CLIENT_UPDATE, &event_data, cb_base->userdata); cb_base->cb(MOSQ_EVT_PERSIST_CLIENT_UPDATE, &event_data, cb_base->userdata);
@@ -133,7 +133,7 @@ void plugin_persist__handle_client_delete(struct mosquitto *context)
} }
opts = &db.config->security_options; opts = &db.config->security_options;
memset(&event_data, 0, sizeof(event_data)); memset(&event_data, 0, sizeof(event_data));
event_data.client_id = context->id; event_data.data.client_id = context->id;
DL_FOREACH(opts->plugin_callbacks.persist_client_delete, cb_base){ DL_FOREACH(opts->plugin_callbacks.persist_client_delete, cb_base){
cb_base->cb(MOSQ_EVT_PERSIST_CLIENT_DELETE, &event_data, cb_base->userdata); cb_base->cb(MOSQ_EVT_PERSIST_CLIENT_DELETE, &event_data, cb_base->userdata);

92
src/plugin_public.c
View File

@@ -461,7 +461,7 @@ BROKER_EXPORT int mosquitto_apply_on_all_clients(int (*FUNC_client_functor)(cons
return rc; return rc;
} }
BROKER_EXPORT int mosquitto_persist_client_add(struct mosquitto_evt_persist_client *client) BROKER_EXPORT int mosquitto_persist_client_add(struct mosquitto_client *client)
{ {
struct mosquitto *context; struct mosquitto *context;
int i; int i;
@@ -470,13 +470,13 @@ BROKER_EXPORT int mosquitto_persist_client_add(struct mosquitto_evt_persist_clie
if(client == NULL){ if(client == NULL){
return MOSQ_ERR_INVAL; return MOSQ_ERR_INVAL;
} }
if(client->plugin_client_id == NULL){ if(client->client_id == NULL){
rc = MOSQ_ERR_INVAL; rc = MOSQ_ERR_INVAL;
goto error; goto error;
} }
context = NULL; context = NULL;
HASH_FIND(hh_id, db.contexts_by_id, client->plugin_client_id, strlen(client->plugin_client_id), context); HASH_FIND(hh_id, db.contexts_by_id, client->client_id, strlen(client->client_id), context);
if(context){ if(context){
rc = MOSQ_ERR_INVAL; rc = MOSQ_ERR_INVAL;
goto error; goto error;
@@ -488,12 +488,12 @@ BROKER_EXPORT int mosquitto_persist_client_add(struct mosquitto_evt_persist_clie
goto error; goto error;
} }
context->id = client->plugin_client_id; context->id = client->client_id;
client->plugin_client_id = NULL; client->client_id = NULL;
context->username = client->plugin_username; context->username = client->username;
client->plugin_username = NULL; client->username = NULL;
context->auth_method = client->plugin_auth_method; context->auth_method = client->auth_method;
client->plugin_auth_method = NULL; client->auth_method = NULL;
context->clean_start = false; context->clean_start = false;
context->will_delay_time = client->will_delay_time; context->will_delay_time = client->will_delay_time;
@@ -519,14 +519,14 @@ BROKER_EXPORT int mosquitto_persist_client_add(struct mosquitto_evt_persist_clie
return MOSQ_ERR_SUCCESS; return MOSQ_ERR_SUCCESS;
error: error:
SAFE_FREE(client->plugin_client_id); SAFE_FREE(client->client_id);
SAFE_FREE(client->plugin_username); SAFE_FREE(client->username);
SAFE_FREE(client->plugin_auth_method); SAFE_FREE(client->auth_method);
return rc; return rc;
} }
BROKER_EXPORT int mosquitto_persist_client_update(struct mosquitto_evt_persist_client *client) BROKER_EXPORT int mosquitto_persist_client_update(struct mosquitto_client *client)
{ {
struct mosquitto *context; struct mosquitto *context;
int i; int i;
@@ -548,8 +548,8 @@ BROKER_EXPORT int mosquitto_persist_client_update(struct mosquitto_evt_persist_c
} }
mosquitto_free(context->username); mosquitto_free(context->username);
context->username = client->plugin_username; context->username = client->username;
client->plugin_username = NULL; client->username = NULL;
context->clean_start = false; context->clean_start = false;
context->will_delay_time = client->will_delay_time; context->will_delay_time = client->will_delay_time;
@@ -572,7 +572,7 @@ BROKER_EXPORT int mosquitto_persist_client_update(struct mosquitto_evt_persist_c
return MOSQ_ERR_SUCCESS; return MOSQ_ERR_SUCCESS;
error: error:
SAFE_FREE(client->plugin_username); SAFE_FREE(client->username);
return rc; return rc;
} }
@@ -611,33 +611,33 @@ static struct mosquitto__base_msg *find_store_msg(uint64_t store_id)
return base_msg; return base_msg;
} }
BROKER_EXPORT int mosquitto_persist_client_msg_add(struct mosquitto_evt_persist_client_msg *client_msg) BROKER_EXPORT int mosquitto_persist_client_msg_add(struct mosquitto_client_msg *client_msg)
{ {
struct mosquitto *context; struct mosquitto *context;
struct mosquitto__base_msg *base_msg; struct mosquitto__base_msg *base_msg;
if(client_msg == NULL || client_msg->data.client_id == NULL){ if(client_msg == NULL || client_msg->client_id == NULL){
return MOSQ_ERR_INVAL; return MOSQ_ERR_INVAL;
} }
HASH_FIND(hh_id, db.contexts_by_id, client_msg->data.client_id, strlen(client_msg->data.client_id), context); HASH_FIND(hh_id, db.contexts_by_id, client_msg->client_id, strlen(client_msg->client_id), context);
if(context == NULL){ if(context == NULL){
return MOSQ_ERR_NOT_FOUND; return MOSQ_ERR_NOT_FOUND;
} }
base_msg = find_store_msg(client_msg->data.store_id); base_msg = find_store_msg(client_msg->store_id);
if(base_msg == NULL){ if(base_msg == NULL){
return MOSQ_ERR_NOT_FOUND; return MOSQ_ERR_NOT_FOUND;
} }
if(client_msg->data.direction == mosq_md_out){ if(client_msg->direction == mosq_md_out){
if(client_msg->data.qos > 0){ if(client_msg->qos > 0){
context->last_mid = client_msg->data.mid; context->last_mid = client_msg->mid;
} }
return db__message_insert_outgoing(context, client_msg->data.cmsg_id, client_msg->data.mid, return db__message_insert_outgoing(context, client_msg->cmsg_id, client_msg->mid,
client_msg->data.qos, client_msg->data.retain, client_msg->qos, client_msg->retain,
base_msg, client_msg->data.subscription_identifier, false, false); base_msg, client_msg->subscription_identifier, false, false);
}else if(client_msg->data.direction == mosq_md_in){ }else if(client_msg->direction == mosq_md_in){
return db__message_insert_incoming(context, client_msg->data.cmsg_id, base_msg, false); return db__message_insert_incoming(context, client_msg->cmsg_id, base_msg, false);
}else{ }else{
return MOSQ_ERR_INVAL; return MOSQ_ERR_INVAL;
} }
@@ -645,21 +645,21 @@ BROKER_EXPORT int mosquitto_persist_client_msg_add(struct mosquitto_evt_persist_
} }
BROKER_EXPORT int mosquitto_persist_client_msg_delete(struct mosquitto_evt_persist_client_msg *client_msg) BROKER_EXPORT int mosquitto_persist_client_msg_delete(struct mosquitto_client_msg *client_msg)
{ {
struct mosquitto *context; struct mosquitto *context;
if(client_msg == NULL || client_msg->data.client_id == NULL) return MOSQ_ERR_INVAL; if(client_msg == NULL || client_msg->client_id == NULL) return MOSQ_ERR_INVAL;
HASH_FIND(hh_id, db.contexts_by_id, client_msg->data.client_id, strlen(client_msg->data.client_id), context); HASH_FIND(hh_id, db.contexts_by_id, client_msg->client_id, strlen(client_msg->client_id), context);
if(context == NULL){ if(context == NULL){
return MOSQ_ERR_NOT_FOUND; return MOSQ_ERR_NOT_FOUND;
} }
if(client_msg->data.direction == mosq_md_out){ if(client_msg->direction == mosq_md_out){
return db__message_delete_outgoing(context, client_msg->data.mid, client_msg->data.state, client_msg->data.qos); return db__message_delete_outgoing(context, client_msg->mid, client_msg->state, client_msg->qos);
}else if(client_msg->data.direction == mosq_md_in){ }else if(client_msg->direction == mosq_md_in){
return db__message_remove_incoming(context, client_msg->data.mid); return db__message_remove_incoming(context, client_msg->mid);
}else{ }else{
return MOSQ_ERR_INVAL; return MOSQ_ERR_INVAL;
} }
@@ -667,20 +667,20 @@ BROKER_EXPORT int mosquitto_persist_client_msg_delete(struct mosquitto_evt_persi
} }
BROKER_EXPORT int mosquitto_persist_client_msg_update(struct mosquitto_evt_persist_client_msg *client_msg) BROKER_EXPORT int mosquitto_persist_client_msg_update(struct mosquitto_client_msg *client_msg)
{ {
struct mosquitto *context; struct mosquitto *context;
if(client_msg == NULL || client_msg->data.client_id == NULL) return MOSQ_ERR_INVAL; if(client_msg == NULL || client_msg->client_id == NULL) return MOSQ_ERR_INVAL;
HASH_FIND(hh_id, db.contexts_by_id, client_msg->data.client_id, strlen(client_msg->data.client_id), context); HASH_FIND(hh_id, db.contexts_by_id, client_msg->client_id, strlen(client_msg->client_id), context);
if(context == NULL){ if(context == NULL){
return MOSQ_ERR_NOT_FOUND; return MOSQ_ERR_NOT_FOUND;
} }
if(client_msg->data.direction == mosq_md_out){ if(client_msg->direction == mosq_md_out){
db__message_update_outgoing(context, client_msg->data.mid, client_msg->data.state, client_msg->data.qos, false); db__message_update_outgoing(context, client_msg->mid, client_msg->state, client_msg->qos, false);
}else if(client_msg->data.direction == mosq_md_in){ }else if(client_msg->direction == mosq_md_in){
// FIXME db__message_update_incoming(context, client_msg->mid, client_msg->state, client_msg->qos, false); // FIXME db__message_update_incoming(context, client_msg->mid, client_msg->state, client_msg->qos, false);
}else{ }else{
return MOSQ_ERR_INVAL; return MOSQ_ERR_INVAL;
@@ -689,20 +689,20 @@ BROKER_EXPORT int mosquitto_persist_client_msg_update(struct mosquitto_evt_persi
} }
BROKER_EXPORT int mosquitto_persist_client_msg_clear(struct mosquitto_evt_persist_client_msg *client_msg) BROKER_EXPORT int mosquitto_persist_client_msg_clear(struct mosquitto_client_msg *client_msg)
{ {
struct mosquitto *context; struct mosquitto *context;
if(client_msg == NULL || client_msg->data.client_id == NULL) return MOSQ_ERR_INVAL; if(client_msg == NULL || client_msg->client_id == NULL) return MOSQ_ERR_INVAL;
HASH_FIND(hh_id, db.contexts_by_id, client_msg->data.client_id, strlen(client_msg->data.client_id), context); HASH_FIND(hh_id, db.contexts_by_id, client_msg->client_id, strlen(client_msg->client_id), context);
if(context == NULL){ if(context == NULL){
return MOSQ_ERR_NOT_FOUND; return MOSQ_ERR_NOT_FOUND;
} }
if(client_msg->data.direction == mosq_bmd_in || client_msg->data.direction == mosq_bmd_all){ if(client_msg->direction == mosq_bmd_in || client_msg->direction == mosq_bmd_all){
db__messages_delete_incoming(context); db__messages_delete_incoming(context);
}else if(client_msg->data.direction == mosq_bmd_out || client_msg->data.direction == mosq_bmd_all){ }else if(client_msg->direction == mosq_bmd_out || client_msg->direction == mosq_bmd_all){
db__messages_delete_outgoing(context); db__messages_delete_outgoing(context);
} }
return MOSQ_ERR_SUCCESS; return MOSQ_ERR_SUCCESS;