fix bug which make subscribers unresponsive on some race conditions when the...

fix bug which make subscribers unresponsive on some race conditions when the workers does not receives an alert

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("73e503c8a3f678dabd55a93a23a226c7730fa7be");
+static const ngx_str_t  NGX_HTTP_PUSH_STREAM_COMMIT = ngx_string("1ee1e3af467be23b909d7ddf7ca493fe33e79c24");
 
 #endif /* NGX_HTTP_PUSH_STREAM_MODULE_VERSION_H_ */

src/ngx_http_push_stream_module_ipc.c

@@ -374,9 +374,9 @@ ngx_http_push_stream_broadcast(ngx_http_push_stream_channel_t *channel, ngx_http
     ngx_shmtx_unlock(&shpool->mutex);
 
     cur = sentinel;
-    while (((cur = (ngx_http_push_stream_pid_queue_t *) ngx_queue_next(&cur->queue)) != sentinel) && queue_was_empty[cur->slot]) {
+    while ((cur = (ngx_http_push_stream_pid_queue_t *) ngx_queue_next(&cur->queue)) != sentinel) {
         // interprocess communication breakdown
-        if (ngx_http_push_stream_alert_worker_check_messages(cur->pid, cur->slot, log) != NGX_OK) {
+        if (queue_was_empty[cur->slot] && (ngx_http_push_stream_alert_worker_check_messages(cur->pid, cur->slot, log) != NGX_OK)) {
             ngx_log_error(NGX_LOG_ERR, log, 0, "push stream module: error communicating with worker process, pid: %P, slot: %d", cur->pid, cur->slot);
         }
     }