LibMosquitto 源码分析之三：主要数据结构

Posted 2021-04-05 思源互联科技

主要数据结构 客户端状态

该状态为用户连接成功并通讯 CONNECT 之后结果;

enum mosquitto_client_state {
mosq_cs_new = 0,
mosq_cs_connected = 1,
mosq_cs_disconnecting = 2,// mosquitto_disconnect时设置mosq_cs_connect_async = 3,// mosquitto_connect_bind_async,

异步线程来connect _mosquitto_thread_main(需要WITH_THREADING)mosq_cs_connect_pending = 4//没用到

};

消息状态

消息发送与接收流程用,关注 mosq_ms_wait_for_xxxx 状态,客户端处理此类 消息;
消息处理流程可参考协议处理流程部分;

enum mosquitto_msg_state {。
mosq_ms_invalid = 0,
mosq_ms_publish_qos0 = 1,
mosq_ms_publish_qos1 = 2,
mosq_ms_wait_for_puback = 3,//Oos==1时,发送PUBLISH后等待PUBACK

返回

mosq_ms_wait_for_pubrec = 5, //Oos==2时,发送PUBLISH后,等待PUBREC返回

mosq_ms_resend_pubrel = 6,

mosq_ms_wait_for_pubrel = 7, //Oos==2时,发送PUBREC后等待PUBREL返回

mosq_ms_resend_pubcomp = 8,

mosq_ms_wait_for_pubcomp = 9, //Oos==2时,发送PUBREL后等待PUBCOMP返回

mosq_ms_send_pubrec = 10,

mosq_ms_queued = 11};

数据包、数据包队列

发送数据(组包后)或者接受数据后(解包前)状态

struct _mosquitto_packet{
uint8_t *payload;
struct _mosquitto_packet *next;
uint32_t remaining_mult;
uint32_t remaining_length;
uint32_t packet_length;
uint32_t to_process;//发送进度,记录还未发送多少字节,缺省为

packet_length
uint32_t pos;//组包或者发送时用到,发送时记录发送到什么位置uint16_t mid;//消息id,当Qos==0 时回调on_publish时用uint8_t command;
int8_t remaining_count;

};

mosq_ms_publish_qos2 = 4,

消息

struct mosquitto_message{int mid;

char *topic;void *payload;int payloadlen;int qos;
bool retain;

};

消息队列

struct mosquitto_message_all{
struct mosquitto_message_all *next;time_t timestamp;//时间,记录本地软件tick时间//enum mosquitto_msg_direction direction;enum mosquitto_msg_state state;
bool dup;
struct mosquitto_message msg;

};

会话相关属性(上下文)

struct mosquitto {
mosq_sock_t sock;
mosq_sock_t sockpairR, sockpairW;// socket管道通知:非阻塞模式

时,通知用,在mosquitto_loop 调用发送,enum _mosquitto_protocol protocol;char *address;
char *id;//客户端ID

char *username;
char *password;
uint16_t keepalive;
uint16_t last_mid; //最后一个消息id,发消息后++enum mosquitto_client_state state;

time_t last_msg_in;time_t last_msg_out;

time_t ping_t;
struct _mosquitto_packet in_packet;//接收数据包用struct _mosquitto_packet *current_out_packet;struct _mosquitto_packet *out_packet;//发送数据包队列struct mosquitto_message *will;

#ifdef WITH_TLS

SSL *ssl;
SSL_CTX *ssl_ctx;
char *tls_cafile;
char *tls_capath;
char *tls_certfile;
char *tls_keyfile;
int (*tls_pw_callback)(char *buf, int size, int rwflag, void

*userdata);
char *tls_version;
char *tls_ciphers;
char *tls_psk;
char *tls_psk_identity;int tls_cert_reqs;
bool tls_insecure;

#endif

bool want_write;

bool want_connect;
#if defined(WITH_THREADING) && !defined(WITH_BROKER)

pthread_mutex_t callback_mutex; pthread_mutex_t log_callback_mutex; pthread_mutex_t msgtime_mutex; pthread_mutex_t out_packet_mutex; pthread_mutex_t current_out_packet_mutex; pthread_mutex_t state_mutex; pthread_mutex_t in_message_mutex; pthread_mutex_t out_message_mutex; pthread_mutex_t mid_mutex;

pthread_t thread_id;

#endif

bool clean_session;

void *userdata;
bool in_callback;
unsigned int message_retry;
time_t last_retry_check;
struct mosquitto_message_all *in_messages;//收到消息队列struct mosquitto_message_all *in_messages_last;struct mosquitto_message_all *out_messages;发送消息队列

struct mosquitto_message_all *out_messages_last;

void (*on_connect)(struct mosquitto *, void *userdata, intrc);

void (*on_disconnect)(struct mosquitto *, void *userdata, intrc);

void (*on_publish)(struct mosquitto *, void *userdata, intmid);

void (*on_message)(struct mosquitto *, void *userdata, const struct mosquitto_message *message);

void (*on_subscribe)(struct mosquitto *, void *userdata, intmid, int qos_count, const int *granted_qos);

void (*on_unsubscribe)(struct mosquitto *, void *userdata, intmid);

void (*on_log)(struct mosquitto *, void *userdata, int level,const char *str);

//void (*on_error)();

char *host;
int port;
int in_queue_len; //收到消息队列长度
int out_queue_len;//发送消息队列长度
char *bind_address;
unsigned int reconnect_delay;
unsigned int reconnect_delay_max;
bool reconnect_exponential_backoff;
bool threaded;
int inflight_messages; //对于Qos>0的消息,记录没有完成交互记录int max_inflight_messages;

};