notify only current worker

Author: RekGRpth

include/ngx_http_push_stream_module_ipc.h

@@ -50,7 +50,7 @@ ngx_socket_t    ngx_http_push_stream_socketpairs[NGX_MAX_PROCESSES][2];
 
 static ngx_int_t    ngx_http_push_stream_register_worker_message_handler(ngx_cycle_t *cycle);
 
-static void    ngx_http_push_stream_broadcast(ngx_http_push_stream_channel_t *channel, ngx_http_push_stream_msg_t *msg, ngx_log_t *log, ngx_http_push_stream_main_conf_t *mcf);
+static void    ngx_http_push_stream_broadcast(ngx_http_push_stream_channel_t *channel, ngx_http_push_stream_msg_t *msg, ngx_log_t *log, ngx_http_push_stream_main_conf_t *mcf, ngx_pid_t *pid);
 
 static ngx_int_t        ngx_http_push_stream_alert_worker(ngx_pid_t pid, ngx_int_t slot, ngx_log_t *log, ngx_channel_t command);
 #define ngx_http_push_stream_alert_worker_check_messages(pid, slot, log) ngx_http_push_stream_alert_worker(pid, slot, log, NGX_CMD_HTTP_PUSH_STREAM_CHECK_MESSAGES)

include/ngx_http_push_stream_module_utils.h

@@ -262,9 +262,8 @@ ngx_chain_t *               ngx_http_push_stream_get_buf(ngx_http_request_t *r);
 static void                 ngx_http_push_stream_unescape_uri(ngx_str_t *value);
 static void                 ngx_http_push_stream_complex_value(ngx_http_request_t *r, ngx_http_complex_value_t *val, ngx_str_t *value);
 
-
-ngx_int_t                   ngx_http_push_stream_add_msg_to_channel(ngx_http_push_stream_main_conf_t *mcf, ngx_log_t *log, ngx_http_push_stream_channel_t *channel, u_char *text, size_t len, ngx_str_t *event_id, ngx_str_t *event_type, ngx_flag_t store_messages, ngx_pool_t *temp_pool);
-void                        ngx_http_push_stream_add_msg_to_channel_my(ngx_log_t *log, ngx_str_t *id, ngx_str_t *text, ngx_str_t *event_id, ngx_str_t *event_type, ngx_flag_t store_messages, ngx_pool_t *temp_pool);
+ngx_int_t                   ngx_http_push_stream_add_msg_to_channel(ngx_http_push_stream_main_conf_t *mcf, ngx_log_t *log, ngx_http_push_stream_channel_t *channel, u_char *text, size_t len, ngx_str_t *event_id, ngx_str_t *event_type, ngx_flag_t store_messages, ngx_pool_t *temp_pool, ngx_pid_t *pid);
+void                        ngx_http_push_stream_add_msg_to_channel_my(ngx_log_t *log, ngx_str_t *id, ngx_str_t *text, ngx_str_t *event_id, ngx_str_t *event_type, ngx_flag_t store_messages, ngx_pool_t *temp_pool, ngx_pid_t *pid);
 ngx_int_t                   ngx_http_push_stream_send_event(ngx_http_push_stream_main_conf_t *mcf, ngx_log_t *log, ngx_http_push_stream_channel_t *channel, ngx_str_t *event_id, ngx_pool_t *temp_pool, ngx_http_request_t *r, ngx_http_complex_value_t *uri);
 
 static void                 ngx_http_push_stream_ping_timer_wake_handler(ngx_event_t *ev);

src/ngx_http_push_stream_module_ipc.c

@@ -463,7 +463,7 @@ ngx_http_push_stream_send_worker_message(ngx_http_push_stream_channel_t *channel
 
 
 static void
-ngx_http_push_stream_broadcast(ngx_http_push_stream_channel_t *channel, ngx_http_push_stream_msg_t *msg, ngx_log_t *log, ngx_http_push_stream_main_conf_t *mcf)
+ngx_http_push_stream_broadcast(ngx_http_push_stream_channel_t *channel, ngx_http_push_stream_msg_t *msg, ngx_log_t *log, ngx_http_push_stream_main_conf_t *mcf, ngx_pid_t *pid)
 {
     // subscribers are queued up in a local pool. Queue heads, however, are located
     // in shared memory, identified by pid.
@@ -474,12 +474,14 @@ ngx_http_push_stream_broadcast(ngx_http_push_stream_channel_t *channel, ngx_http
     ngx_shmtx_lock(channel->mutex);
     for (q = ngx_queue_head(&channel->workers_with_subscribers); q != ngx_queue_sentinel(&channel->workers_with_subscribers); q = ngx_queue_next(q)) {
         worker = ngx_queue_data(q, ngx_http_push_stream_pid_queue_t, queue);
+        if (pid && *pid != worker->pid) continue;
         ngx_http_push_stream_send_worker_message(channel, &worker->subscriptions, worker->pid, worker->slot, msg, &queue_was_empty[worker->slot], log, mcf);
     }
     ngx_shmtx_unlock(channel->mutex);
 
     for (q = ngx_queue_head(&channel->workers_with_subscribers); q != ngx_queue_sentinel(&channel->workers_with_subscribers); q = ngx_queue_next(q)) {
         worker = ngx_queue_data(q, ngx_http_push_stream_pid_queue_t, queue);
+        if (pid && *pid != worker->pid) continue;
         // interprocess communication breakdown
         if (queue_was_empty[worker->slot] && (ngx_http_push_stream_alert_worker_check_messages(worker->pid, worker->slot, log) != NGX_OK)) {
             ngx_log_error(NGX_LOG_ERR, log, 0, "push stream module: error communicating with worker process, pid: %P, slot: %d", worker->pid, worker->slot);

src/ngx_http_push_stream_module_publisher.c

@@ -277,7 +277,7 @@ ngx_http_push_stream_publisher_body_handler(ngx_http_request_t *r)
     for (q = ngx_queue_head(&ctx->requested_channels->queue); q != ngx_queue_sentinel(&ctx->requested_channels->queue); q = ngx_queue_next(q)) {
         requested_channel = ngx_queue_data(q, ngx_http_push_stream_requested_channel_t, queue);
 
-        if (ngx_http_push_stream_add_msg_to_channel(mcf, r->connection->log, requested_channel->channel, buf->pos, ngx_buf_size(buf), event_id, event_type, cf->store_messages, r->pool) != NGX_OK) {
+        if (ngx_http_push_stream_add_msg_to_channel(mcf, r->connection->log, requested_channel->channel, buf->pos, ngx_buf_size(buf), event_id, event_type, cf->store_messages, r->pool, NULL) != NGX_OK) {
             ngx_http_finalize_request(r, NGX_HTTP_INTERNAL_SERVER_ERROR);
             return;
         }

src/ngx_http_push_stream_module_utils.c

@@ -375,7 +375,7 @@ ngx_http_push_stream_convert_char_to_msg_on_shared(ngx_http_push_stream_main_con
 }
 
 void
-ngx_http_push_stream_add_msg_to_channel_my(ngx_log_t *log, ngx_str_t *id, ngx_str_t *text, ngx_str_t *event_id, ngx_str_t *event_type, ngx_flag_t store_messages, ngx_pool_t *temp_pool)
+ngx_http_push_stream_add_msg_to_channel_my(ngx_log_t *log, ngx_str_t *id, ngx_str_t *text, ngx_str_t *event_id, ngx_str_t *event_type, ngx_flag_t store_messages, ngx_pool_t *temp_pool, ngx_pid_t *pid)
 {
 //    ngx_log_error(NGX_LOG_ERR, log, 0, "id = %V", id);
 //    ngx_log_error(NGX_LOG_ERR, log, 0, "ngx_http_push_stream_global_shm_zone = %p", ngx_http_push_stream_global_shm_zone);
@@ -389,15 +389,15 @@ ngx_http_push_stream_add_msg_to_channel_my(ngx_log_t *log, ngx_str_t *id, ngx_st
         ngx_http_push_stream_channel_t *channel = ngx_http_push_stream_find_channel(id, log, mcf);
 //        ngx_log_error(NGX_LOG_ERR, log, 0, "channel = %p", channel);
         if (channel != NULL) {
-            if (ngx_http_push_stream_add_msg_to_channel(mcf, log, channel, text->data, text->len, event_id, event_type, store_messages, temp_pool) != NGX_OK) {
+            if (ngx_http_push_stream_add_msg_to_channel(mcf, log, channel, text->data, text->len, event_id, event_type, store_messages, temp_pool, pid) != NGX_OK) {
                 ngx_log_error(NGX_LOG_ERR, log, 0, "ngx_http_push_stream_add_msg_to_channel != NGX_OK");
             }
         }
     }
 }
 
 ngx_int_t
-ngx_http_push_stream_add_msg_to_channel(ngx_http_push_stream_main_conf_t *mcf, ngx_log_t *log, ngx_http_push_stream_channel_t *channel, u_char *text, size_t len, ngx_str_t *event_id, ngx_str_t *event_type, ngx_flag_t store_messages, ngx_pool_t *temp_pool)
+ngx_http_push_stream_add_msg_to_channel(ngx_http_push_stream_main_conf_t *mcf, ngx_log_t *log, ngx_http_push_stream_channel_t *channel, u_char *text, size_t len, ngx_str_t *event_id, ngx_str_t *event_type, ngx_flag_t store_messages, ngx_pool_t *temp_pool, ngx_pid_t *pid)
 {
     ngx_http_push_stream_shm_data_t        *data = mcf->shm_data;
     ngx_http_push_stream_msg_t             *msg;
@@ -459,7 +459,7 @@ ngx_http_push_stream_add_msg_to_channel(ngx_http_push_stream_main_conf_t *mcf, n
     }
 
     // send an alert to workers
-    ngx_http_push_stream_broadcast(channel, msg, log, mcf);
+    ngx_http_push_stream_broadcast(channel, msg, log, mcf, pid);
 
     // turn on timer to cleanup buffer of old messages
     ngx_http_push_stream_buffer_cleanup_timer_set();
@@ -483,7 +483,7 @@ ngx_http_push_stream_send_event(ngx_http_push_stream_main_conf_t *mcf, ngx_log_t
         ngx_str_t *event = ngx_http_push_stream_create_str(temp_pool, len);
         if (event != NULL) {
             ngx_sprintf(event->data, NGX_HTTP_PUSH_STREAM_EVENT_TEMPLATE, event_type, &channel->id);
-            ngx_http_push_stream_add_msg_to_channel(mcf, log, data->events_channel, event->data, ngx_strlen(event->data), NULL, event_type, 1, temp_pool);
+            ngx_http_push_stream_add_msg_to_channel(mcf, log, data->events_channel, event->data, ngx_strlen(event->data), NULL, event_type, 1, temp_pool, NULL);
         }
 
         if ((received_temp_pool == NULL) && (temp_pool != NULL)) {

src/ngx_http_push_stream_module_websocket.c

@@ -416,7 +416,7 @@ ngx_http_push_stream_websocket_reading(ngx_http_request_t *r)
                                     }
                                 }
                             } else {
-                                if (ngx_http_push_stream_add_msg_to_channel(mcf, r->connection->log, subscription->channel, ctx->frame->payload, ctx->frame->payload_len, NULL, NULL, cf->store_messages, ctx->temp_pool) != NGX_OK) {
+                                if (ngx_http_push_stream_add_msg_to_channel(mcf, r->connection->log, subscription->channel, ctx->frame->payload, ctx->frame->payload_len, NULL, NULL, cf->store_messages, ctx->temp_pool, NULL) != NGX_OK) {
                                     goto finalize;
                                 }
                             }
@@ -529,7 +529,7 @@ static ngx_int_t ngx_http_push_stream_post_subrequest_handler(ngx_http_request_t
     ngx_http_push_stream_subscription_t *subscription = data;
     if (r->out != NULL) {
         if (r->out->buf != NULL) {
-            if (ngx_http_push_stream_add_msg_to_channel(mcf, r->connection->log, subscription->channel, r->out->buf->pos, ngx_buf_size(r->out->buf), NULL, NULL, cf->store_messages, r->pool) != NGX_OK) {
+            if (ngx_http_push_stream_add_msg_to_channel(mcf, r->connection->log, subscription->channel, r->out->buf->pos, ngx_buf_size(r->out->buf), NULL, NULL, cf->store_messages, r->pool, NULL) != NGX_OK) {
                 ngx_log_error(NGX_LOG_ERR, r->connection->log, 0, "ngx_http_push_stream_add_msg_to_channel != NGX_OK");
             }
         }