adding function ngx_http_push_stream_free_worker_message_memory_loked

include/ngx_http_push_stream_module_utils.h

@@ -254,6 +254,7 @@ static void                 ngx_http_push_stream_delete_channel(ngx_str_t *id, n
 static void                 ngx_http_push_stream_collect_expired_messages(ngx_http_push_stream_shm_data_t *data, ngx_slab_pool_t *shpool, ngx_rbtree_node_t *node, ngx_flag_t force);
 static void                 ngx_http_push_stream_collect_expired_messages_and_empty_channels(ngx_http_push_stream_shm_data_t *data, ngx_slab_pool_t *shpool, ngx_rbtree_node_t *node, ngx_flag_t force);
 static void                 ngx_http_push_stream_free_message_memory_locked(ngx_slab_pool_t *shpool, ngx_http_push_stream_msg_t *msg);
+static void                 ngx_http_push_stream_free_worker_message_memory_locked(ngx_slab_pool_t *shpool, ngx_http_push_stream_worker_msg_t *worker_msg);
 static ngx_int_t            ngx_http_push_stream_free_memory_of_expired_messages_and_channels(ngx_flag_t force);
 static ngx_inline void      ngx_http_push_stream_ensure_qtd_of_messages_locked(ngx_http_push_stream_channel_t *channel, ngx_uint_t max_messages, ngx_flag_t expired);
 static ngx_inline void      ngx_http_push_stream_delete_worker_channel(void);
@@ -263,7 +264,6 @@ static ngx_http_push_stream_content_subtype_t *     ngx_http_push_stream_match_c
 static ngx_http_push_stream_line_t *                ngx_http_push_stream_split_by_crlf(ngx_str_t *msg, ngx_pool_t *temp_pool);
 static ngx_str_t *                                  ngx_http_push_stream_join_with_crlf(ngx_http_push_stream_line_t *lines, ngx_pool_t *temp_pool);
 
-
 static ngx_str_t *          ngx_http_push_stream_get_formatted_current_time(ngx_pool_t *pool);
 static ngx_str_t *          ngx_http_push_stream_get_formatted_hostname(ngx_pool_t *pool);
 

include/ngx_http_push_stream_module_version.h

@@ -27,6 +27,6 @@
 #define NGX_HTTP_PUSH_STREAM_MODULE_VERSION_H_
 
 static const ngx_str_t  NGX_HTTP_PUSH_STREAM_TAG = ngx_string("0.3.2");
-static const ngx_str_t  NGX_HTTP_PUSH_STREAM_COMMIT = ngx_string("e72b082850848b16264d4096641fc8f1a639d330");
+static const ngx_str_t  NGX_HTTP_PUSH_STREAM_COMMIT = ngx_string("f7ea53db0f1122817cdf2586ae4e14ad12d34cfd");
 
 #endif /* NGX_HTTP_PUSH_STREAM_MODULE_VERSION_H_ */

src/ngx_http_push_stream_module_ipc.c

@@ -172,8 +172,7 @@ ngx_http_push_stream_clean_worker_data()
     ngx_shmtx_lock(&shpool->mutex);
     if (data->ipc[ngx_process_slot].messages_queue != NULL) {
         while ((cur_msg = (ngx_http_push_stream_worker_msg_t *) ngx_queue_next(&data->ipc[ngx_process_slot].messages_queue->queue)) != data->ipc[ngx_process_slot].messages_queue) {
-            ngx_queue_remove(&cur_msg->queue);
-            ngx_slab_free_locked(shpool, cur_msg);
+            ngx_http_push_stream_free_worker_message_memory_locked(shpool, cur_msg);
         }
     }
 
@@ -316,12 +315,7 @@ ngx_http_push_stream_process_worker_message(void)
 
         // free worker_msg already sent
         ngx_shmtx_lock(&shpool->mutex);
-        worker_msg->msg->workers_ref_count--;
-        if ((worker_msg->msg->workers_ref_count <= 0) && worker_msg->msg->deleted) {
-            worker_msg->msg->expires = ngx_time() + ngx_http_push_stream_module_main_conf->shm_cleanup_objects_ttl;
-        }
-        ngx_queue_remove(&worker_msg->queue);
-        ngx_slab_free_locked(shpool, worker_msg);
+        ngx_http_push_stream_free_worker_message_memory_locked(shpool, worker_msg);
         ngx_shmtx_unlock(&shpool->mutex);
     }
 }

src/ngx_http_push_stream_module_utils.c

@@ -762,6 +762,18 @@ ngx_http_push_stream_free_message_memory_locked(ngx_slab_pool_t *shpool, ngx_htt
 }
 
 
+static void
+ngx_http_push_stream_free_worker_message_memory_locked(ngx_slab_pool_t *shpool, ngx_http_push_stream_worker_msg_t *worker_msg)
+{
+    worker_msg->msg->workers_ref_count--;
+    if ((worker_msg->msg->workers_ref_count <= 0) && worker_msg->msg->deleted) {
+        worker_msg->msg->expires = ngx_time() + ngx_http_push_stream_module_main_conf->shm_cleanup_objects_ttl;
+    }
+    ngx_queue_remove(&worker_msg->queue);
+    ngx_slab_free_locked(shpool, worker_msg);
+}
+
+
 static void
 ngx_http_push_stream_mark_message_to_delete_locked(ngx_http_push_stream_msg_t *msg)
 {
@@ -783,7 +795,7 @@ ngx_http_push_stream_timer_set(ngx_msec_t timer_interval, ngx_event_t *event, ng
             ngx_shmtx_lock(&shpool->mutex);
             if (event->handler == NULL) {
                 event->handler = event_handler;
-                event->data = NULL;
+                event->data = event; //set event as data to avoid error when running on debug mode (on log event)
                 event->log = ngx_cycle->log;
                 ngx_http_push_stream_timer_reset(timer_interval, event);
             }