|
|
|
@ -121,6 +121,7 @@ void packet__cleanup_all_no_locks(struct mosquitto *mosq)
|
|
|
|
|
mosquitto__FREE(packet);
|
|
|
|
|
}
|
|
|
|
|
mosq->out_packet_count = 0;
|
|
|
|
|
mosq->out_packet_bytes = 0;
|
|
|
|
|
mosq->out_packet_last = NULL;
|
|
|
|
|
|
|
|
|
|
packet__cleanup(&mosq->in_packet);
|
|
|
|
@ -134,6 +135,21 @@ void packet__cleanup_all(struct mosquitto *mosq)
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
static void packet__queue_append(struct mosquitto *mosq, struct mosquitto__packet *packet)
|
|
|
|
|
{
|
|
|
|
|
pthread_mutex_lock(&mosq->out_packet_mutex);
|
|
|
|
|
if(mosq->out_packet){
|
|
|
|
|
mosq->out_packet_last->next = packet;
|
|
|
|
|
}else{
|
|
|
|
|
mosq->out_packet = packet;
|
|
|
|
|
}
|
|
|
|
|
mosq->out_packet_last = packet;
|
|
|
|
|
mosq->out_packet_count++;
|
|
|
|
|
mosq->out_packet_bytes += packet->packet_length;
|
|
|
|
|
pthread_mutex_unlock(&mosq->out_packet_mutex);
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
int packet__queue(struct mosquitto *mosq, struct mosquitto__packet *packet)
|
|
|
|
|
{
|
|
|
|
|
#ifndef WITH_BROKER
|
|
|
|
@ -148,15 +164,7 @@ int packet__queue(struct mosquitto *mosq, struct mosquitto__packet *packet)
|
|
|
|
|
packet->pos = WS_PACKET_OFFSET;
|
|
|
|
|
packet->to_process = packet->packet_length - WS_PACKET_OFFSET;
|
|
|
|
|
|
|
|
|
|
pthread_mutex_lock(&mosq->out_packet_mutex);
|
|
|
|
|
if(mosq->out_packet){
|
|
|
|
|
mosq->out_packet_last->next = packet;
|
|
|
|
|
}else{
|
|
|
|
|
mosq->out_packet = packet;
|
|
|
|
|
}
|
|
|
|
|
mosq->out_packet_last = packet;
|
|
|
|
|
mosq->out_packet_count++;
|
|
|
|
|
pthread_mutex_unlock(&mosq->out_packet_mutex);
|
|
|
|
|
packet__queue_append(mosq, packet);
|
|
|
|
|
|
|
|
|
|
lws_callback_on_writable(mosq->wsi);
|
|
|
|
|
return MOSQ_ERR_SUCCESS;
|
|
|
|
@ -173,14 +181,7 @@ int packet__queue(struct mosquitto *mosq, struct mosquitto__packet *packet)
|
|
|
|
|
packet->to_process = packet->packet_length - WS_PACKET_OFFSET;
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
pthread_mutex_lock(&mosq->out_packet_mutex);
|
|
|
|
|
if(mosq->out_packet){
|
|
|
|
|
mosq->out_packet_last->next = packet;
|
|
|
|
|
}else{
|
|
|
|
|
mosq->out_packet = packet;
|
|
|
|
|
}
|
|
|
|
|
mosq->out_packet_last = packet;
|
|
|
|
|
pthread_mutex_unlock(&mosq->out_packet_mutex);
|
|
|
|
|
packet__queue_append(mosq, packet);
|
|
|
|
|
|
|
|
|
|
#ifdef WITH_BROKER
|
|
|
|
|
return packet__write(mosq);
|
|
|
|
@ -225,11 +226,13 @@ struct mosquitto__packet *packet__get_next_out(struct mosquitto *mosq)
|
|
|
|
|
|
|
|
|
|
pthread_mutex_lock(&mosq->out_packet_mutex);
|
|
|
|
|
if(mosq->out_packet){
|
|
|
|
|
mosq->out_packet_count--;
|
|
|
|
|
mosq->out_packet_bytes -= mosq->out_packet->packet_length;
|
|
|
|
|
|
|
|
|
|
mosq->out_packet = mosq->out_packet->next;
|
|
|
|
|
if(!mosq->out_packet){
|
|
|
|
|
mosq->out_packet_last = NULL;
|
|
|
|
|
}
|
|
|
|
|
mosq->out_packet_count--;
|
|
|
|
|
packet = mosq->out_packet;
|
|
|
|
|
}
|
|
|
|
|
pthread_mutex_unlock(&mosq->out_packet_mutex);
|
|
|
|
|