fix memory leak which happens after a worker dies or the server receive the SIGHUP (reload) signal

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.4");
-static const ngx_str_t  NGX_HTTP_PUSH_STREAM_COMMIT = ngx_string("ab5f17888393a57a3191f0f4ea631be717e6a30b");
+static const ngx_str_t  NGX_HTTP_PUSH_STREAM_COMMIT = ngx_string("7e70cc1cd5f5070d37ca23d7961b4afafebe3668");
 
 #endif /* NGX_HTTP_PUSH_STREAM_MODULE_VERSION_H_ */

src/ngx_http_push_stream_module_ipc.c

@@ -161,6 +161,24 @@ ngx_http_push_stream_ipc_init_worker()
 }
 
 
+static ngx_int_t
+ngx_http_push_stream_unsubscribe_worker_locked(ngx_http_push_stream_channel_t *channel, ngx_slab_pool_t *shpool)
+{
+    ngx_http_push_stream_pid_queue_t        *sentinel = &channel->workers_with_subscribers;
+    ngx_http_push_stream_pid_queue_t        *cur = sentinel;
+
+    while ((cur = (ngx_http_push_stream_pid_queue_t *) ngx_queue_next(&cur->queue)) != sentinel) {
+        if ((cur->pid == ngx_pid) || (cur->slot == ngx_process_slot)) {
+            ngx_queue_remove(&cur->queue);
+            ngx_slab_free_locked(shpool, cur);
+            break;
+        }
+    }
+
+    return NGX_OK;
+}
+
+
 static void
 ngx_http_push_stream_clean_worker_data()
 {
@@ -180,6 +198,8 @@ ngx_http_push_stream_clean_worker_data()
         ngx_queue_init(&data->ipc[ngx_process_slot].subscribers_sentinel->queue);
     }
 
+    ngx_http_push_stream_walk_rbtree(ngx_http_push_stream_unsubscribe_worker_locked);
+
     ngx_shmtx_unlock(&shpool->mutex);
 
     data->ipc[ngx_process_slot].pid = -1;

test/base_test_case.rb

@@ -4,6 +4,7 @@ require 'erb'
 require 'fileutils'
 require 'ruby-debug'
 require 'test/unit'
+require 'eventmachine'
 require 'em-http'
 require 'json'
 require 'socket'
@@ -175,6 +176,18 @@ module BaseTestCase
     self.send(:global_configuration) if self.respond_to?(:global_configuration)
   end
 
+  def publish_message_inline_with_callbacks(channel, headers, body, callbacks = {})
+    pub = EventMachine::HttpRequest.new(nginx_address + '/pub?id=' + channel.to_s).post :head => headers, :body => body, :timeout => 30
+    pub.callback do
+      if pub.response_header.status == 200
+        callbacks[:success].call(pub.response_header.status, pub.response) unless callbacks[:success].nil?
+      else
+        callbacks[:error].call(pub.response_header.status, pub.response) unless callbacks[:error].nil?
+      end
+    end
+    pub
+  end
+
   def publish_message(channel, headers, body)
     EventMachine.run {
       pub = publish_message_inline(channel, headers, body) do
@@ -183,6 +196,7 @@ module BaseTestCase
         assert_equal(channel, response["channel"].to_s, "Channel was not recognized")
         EventMachine.stop
       end
+      add_test_timeout
     }
   end