Publish from other nginx modules and send channels events to internal locations

.github/dependabot.yml

@@ -0,0 +1,6 @@
+version: 2
+updates:
+  - package-ecosystem: "github-actions"
+    directory: "/"
+    schedule:
+      interval: "daily"

.github/workflows/docker-publish.yml

@@ -0,0 +1,22 @@
+name: Docker
+on:
+  push:
+    branches:
+      - master
+  workflow_dispatch:
+jobs:
+  dispatch:
+    runs-on: ubuntu-latest
+    steps:
+      - env:
+          GITHUB_TOKEN: ${{ secrets.PUBLIC_REPO_ACCESS_TOKEN }}
+          INPUTS_CLIENT_PAYLOAD: '{"repository":${{ toJson(github.event.repository.name) }}}'
+          INPUTS_EVENT_TYPE: latest
+          INPUTS_REPOSITORY: ${{ github.repository_owner }}/${{ matrix.repo }}
+        uses: rekgrpth/github-repository-dispatch-shell-action@v1
+    strategy:
+      matrix:
+        repo:
+          - angie.docker
+          - freenginx.docker
+          - nginx.docker

.github/workflows/merge-upstream.yml

@@ -0,0 +1,15 @@
+name: Merge
+on:
+  schedule:
+    - cron: '0 19 * * *'
+  workflow_dispatch:
+jobs:
+  merge:
+    env:
+      GITHUB_TOKEN: ${{ secrets.PUBLIC_REPO_ACCESS_TOKEN }}
+    runs-on: ubuntu-latest
+    steps:
+      - uses: rekgrpth/git-clone-shell-action@v1
+      - env:
+          INPUTS_REPOSITORY: wandenberg/nginx-push-stream-module
+        uses: rekgrpth/git-fetch-upstream-merge-push-shell-action@v1

.keepalive

@@ -0,0 +1 @@
+1737227723

config

@@ -1,5 +1,5 @@
 ngx_addon_name=ngx_http_push_stream_module
-CORE_INCS="$CORE_INCS ${ngx_addon_dir}/src ${ngx_addon_dir}/include"
+#CORE_INCS="$CORE_INCS ${ngx_addon_dir}/src ${ngx_addon_dir}/include"
 
 if test -n "$ngx_module_link"; then
     ngx_module_type=HTTP

include/ngx_http_push_stream_module.h

@@ -127,6 +127,12 @@ typedef struct {
     ngx_str_t                       padding_by_user_agent;
     ngx_queue_t                    *paddings;
     ngx_http_complex_value_t       *allowed_origins;
+    ngx_http_complex_value_t       *channel_created_request_url;
+    ngx_http_complex_value_t       *channel_destroyed_request_url;
+    ngx_http_complex_value_t       *all_worker_clients_unsubscribed_request_url;
+    ngx_http_complex_value_t       *client_subscribed_request_url;
+    ngx_http_complex_value_t       *client_unsubscribed_request_url;
+    ngx_http_complex_value_t       *client_publish_request_url;
 } ngx_http_push_stream_loc_conf_t;
 
 // shared memory segment name

include/ngx_http_push_stream_module_ipc.h

@@ -34,8 +34,8 @@
 #ifndef NGX_HTTP_PUSH_STREAM_MODULE_IPC_H_
 #define NGX_HTTP_PUSH_STREAM_MODULE_IPC_H_
 
-#include <ngx_http_push_stream_module.h>
-#include <ngx_http_push_stream_module_subscriber.h>
+#include "ngx_http_push_stream_module.h"
+#include "ngx_http_push_stream_module_subscriber.h"
 
 #include <ngx_channel.h>
 

include/ngx_http_push_stream_module_publisher.h

@@ -26,7 +26,7 @@
 #ifndef NGX_HTTP_PUSH_STREAM_MODULE_PUBLISHER_H_
 #define NGX_HTTP_PUSH_STREAM_MODULE_PUBLISHER_H_
 
-#include <ngx_http_push_stream_module.h>
+#include "ngx_http_push_stream_module.h"
 
 static ngx_int_t    ngx_http_push_stream_channels_statistics_handler(ngx_http_request_t *r);
 static ngx_int_t    ngx_http_push_stream_publisher_handler(ngx_http_request_t *r);

include/ngx_http_push_stream_module_setup.h

@@ -26,13 +26,13 @@
 #ifndef NGX_HTTP_PUSH_STREAM_MODULE_SETUP_H_
 #define NGX_HTTP_PUSH_STREAM_MODULE_SETUP_H_
 
-#include <ngx_http_push_stream_module.h>
-#include <ngx_http_push_stream_rbtree_util.h>
-#include <ngx_http_push_stream_module_utils.h>
-#include <ngx_http_push_stream_module_ipc.h>
-#include <ngx_http_push_stream_module_publisher.h>
-#include <ngx_http_push_stream_module_subscriber.h>
-#include <ngx_http_push_stream_module_websocket.h>
+#include "ngx_http_push_stream_module.h"
+#include "ngx_http_push_stream_rbtree_util.h"
+#include "ngx_http_push_stream_module_utils.h"
+#include "ngx_http_push_stream_module_ipc.h"
+#include "ngx_http_push_stream_module_publisher.h"
+#include "ngx_http_push_stream_module_subscriber.h"
+#include "ngx_http_push_stream_module_websocket.h"
 
 #define NGX_HTTP_PUSH_STREAM_MESSAGE_BUFFER_CLEANUP_INTERVAL                5000     // 5 seconds
 static time_t NGX_HTTP_PUSH_STREAM_DEFAULT_SHM_MEMORY_CLEANUP_OBJECTS_TTL = 10;      // 10 seconds

include/ngx_http_push_stream_module_utils.h

@@ -26,8 +26,8 @@
 #ifndef NGX_HTTP_PUSH_STREAM_MODULE_UTILS_H_
 #define NGX_HTTP_PUSH_STREAM_MODULE_UTILS_H_
 
-#include <ngx_http_push_stream_module.h>
-#include <ngx_http_push_stream_module_ipc.h>
+#include "ngx_http_push_stream_module.h"
+#include "ngx_http_push_stream_module_ipc.h"
 
 typedef struct {
     ngx_queue_t           queue;
@@ -229,6 +229,7 @@ static ngx_str_t  NGX_HTTP_PUSH_STREAM_EVENT_TYPE_CHANNEL_CREATED = ngx_string("
 static ngx_str_t  NGX_HTTP_PUSH_STREAM_EVENT_TYPE_CHANNEL_DESTROYED = ngx_string("channel_destroyed");
 static ngx_str_t  NGX_HTTP_PUSH_STREAM_EVENT_TYPE_CLIENT_SUBSCRIBED = ngx_string("client_subscribed");
 static ngx_str_t  NGX_HTTP_PUSH_STREAM_EVENT_TYPE_CLIENT_UNSUBSCRIBED = ngx_string("client_unsubscribed");
+static ngx_str_t  NGX_HTTP_PUSH_STREAM_POST = ngx_string("POST");
 
 
 ngx_event_t         ngx_http_push_stream_memory_cleanup_event;
@@ -252,6 +253,7 @@ static ngx_int_t            ngx_http_push_stream_send_response_content_header(ng
 static ngx_int_t            ngx_http_push_stream_send_response(ngx_http_request_t *r, ngx_str_t *text, const ngx_str_t *content_type, ngx_int_t status_code);
 static ngx_int_t            ngx_http_push_stream_send_response_message(ngx_http_request_t *r, ngx_http_push_stream_channel_t *channel, ngx_http_push_stream_msg_t *msg, ngx_flag_t send_callback, ngx_flag_t send_separator);
 static ngx_int_t            ngx_http_push_stream_send_response_text(ngx_http_request_t *r, const u_char *text, uint len, ngx_flag_t last_buffer);
+static ngx_int_t            ngx_http_push_stream_output_filter(ngx_http_request_t *r, ngx_chain_t *in);
 static void                 ngx_http_push_stream_send_response_finalize(ngx_http_request_t *r);
 static void                 ngx_http_push_stream_send_response_finalize_for_longpolling_by_timeout(ngx_http_request_t *r);
 static ngx_int_t            ngx_http_push_stream_send_websocket_close_frame(ngx_http_request_t *r, ngx_uint_t http_status, const ngx_str_t *reason);
@@ -263,7 +265,8 @@ static void                 ngx_http_push_stream_complex_value(ngx_http_request_
 
 
 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_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_int_t                   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_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);
 static void                 ngx_http_push_stream_disconnect_timer_wake_handler(ngx_event_t *ev);
@@ -281,6 +284,7 @@ static ngx_str_t *          ngx_http_push_stream_create_str(ngx_pool_t *pool, ui
 
 static void                 ngx_http_push_stream_throw_the_message_away(ngx_http_push_stream_msg_t *msg, ngx_http_push_stream_shm_data_t *data);
 static ngx_int_t            ngx_http_push_stream_delete_channel(ngx_http_push_stream_main_conf_t *mcf, ngx_http_push_stream_channel_t *channel, u_char *text, size_t len, ngx_pool_t *temp_pool);
+ngx_int_t                   ngx_http_push_stream_delete_channel_my(ngx_log_t *log, ngx_str_t *id, u_char *text, size_t len, ngx_pool_t *temp_pool);
 static void                 ngx_http_push_stream_collect_expired_messages_data(ngx_http_push_stream_shm_data_t *data, ngx_flag_t force);
 static void                 ngx_http_push_stream_collect_expired_messages_and_empty_channels(ngx_flag_t force);
 static void                 ngx_http_push_stream_free_message_memory(ngx_slab_pool_t *shpool, ngx_http_push_stream_msg_t *msg);

include/ngx_http_push_stream_module_websocket.h

@@ -30,8 +30,8 @@
 #include <ngx_sha1.h>
 #endif
 
-#include <ngx_http_push_stream_module_utils.h>
-#include <ngx_http_push_stream_module_subscriber.h>
+#include "ngx_http_push_stream_module_utils.h"
+#include "ngx_http_push_stream_module_subscriber.h"
 
 static ngx_int_t    ngx_http_push_stream_websocket_handler(ngx_http_request_t *r);
 

include/ngx_http_push_stream_rbtree_util.h

@@ -34,7 +34,7 @@
 #ifndef NGX_HTTP_PUSH_STREAM_RBTREE_UTIL_H_
 #define NGX_HTTP_PUSH_STREAM_RBTREE_UTIL_H_
 
-static ngx_http_push_stream_channel_t *     ngx_http_push_stream_get_channel(ngx_str_t *id, ngx_log_t *log, ngx_http_push_stream_main_conf_t *mcf);
+static ngx_http_push_stream_channel_t *     ngx_http_push_stream_get_channel(ngx_str_t *id, ngx_log_t *log, ngx_http_push_stream_main_conf_t *mcf, ngx_http_request_t *r);
 static ngx_http_push_stream_channel_t *     ngx_http_push_stream_find_channel(ngx_str_t *id, ngx_log_t *log, ngx_http_push_stream_main_conf_t *mcf);
 
 static void         ngx_rbtree_generic_insert(ngx_rbtree_node_t *temp, ngx_rbtree_node_t *node, ngx_rbtree_node_t *sentinel, int (*compare) (const ngx_rbtree_node_t *left, const ngx_rbtree_node_t *right));

misc/tools/publisher.c

@@ -156,7 +156,7 @@ void
 write_message(Connection *connection, Statistics *stats)
 {
     char buffer[BUFFER_SIZE];
-    int len = 0, bytes_written = 0;
+    int len = 0;
 
     if ((connection->channel_id <= connection->channel_start) || (connection->channel_id > connection->channel_end)) {
         connection->channel_id = connection->channel_start;

misc/tools/subscriber.c

@@ -14,7 +14,7 @@ int
 main_program(int num_channels, int num_connections, const char *server_hostname, int server_port, int timeout)
 {
     struct sockaddr_in server_address;
-    int main_sd = -1, num_events = 0, i, j, event_mask, channels_per_connection, num, start_time = 0, iters_to_next_summary = 0;
+    int main_sd = -1, num_events = 0, i, j, event_mask, num, start_time = 0, iters_to_next_summary = 0;
     Connection *connections = NULL, *connection;
     Statistics stats = {0,0,0,0,0};
     int exitcode = EXIT_SUCCESS;
@@ -138,7 +138,7 @@ void
 subscribe_channels(Connection *connection, Statistics *stats)
 {
     char buffer[BUFFER_SIZE];
-    int len = 0, bytes_written = 0;
+    int len = 0;
     long i = 0;
 
     len = sprintf(buffer, "GET /sub");

src/ngx_http_push_stream_module.c

@@ -23,14 +23,14 @@
  * Authors: Wandenberg Peixoto <[email protected]>, Rogério Carvalho Schneider <[email protected]>
  */
 
-#include <ngx_http_push_stream_module.h>
-#include <ngx_http_push_stream_module_setup.c>
-#include <ngx_http_push_stream_rbtree_util.c>
-#include <ngx_http_push_stream_module_utils.c>
-#include <ngx_http_push_stream_module_ipc.c>
-#include <ngx_http_push_stream_module_publisher.c>
-#include <ngx_http_push_stream_module_subscriber.c>
-#include <ngx_http_push_stream_module_websocket.c>
+#include "ngx_http_push_stream_module.h"
+#include "ngx_http_push_stream_module_setup.c"
+#include "ngx_http_push_stream_rbtree_util.c"
+#include "ngx_http_push_stream_module_utils.c"
+#include "ngx_http_push_stream_module_ipc.c"
+#include "ngx_http_push_stream_module_publisher.c"
+#include "ngx_http_push_stream_module_subscriber.c"
+#include "ngx_http_push_stream_module_websocket.c"
 
 static ngx_str_t *
 ngx_http_push_stream_channel_info_formatted(ngx_pool_t *pool, const ngx_str_t *format, ngx_str_t *id, ngx_uint_t published_messages, ngx_uint_t stored_messages, ngx_uint_t subscribers)

src/ngx_http_push_stream_module.h

@@ -0,0 +1 @@
+../include/ngx_http_push_stream_module.h

src/ngx_http_push_stream_module_ipc.c

@@ -31,7 +31,7 @@
  * Modifications by: Wandenberg Peixoto <[email protected]>, Rogério Carvalho Schneider <[email protected]>
  */
 
-#include <ngx_http_push_stream_module_ipc.h>
+#include "ngx_http_push_stream_module_ipc.h"
 
 void ngx_http_push_stream_ipc_init_worker_data(ngx_http_push_stream_shm_data_t *data);
 static ngx_inline void ngx_http_push_stream_census_worker_subscribers_data(ngx_http_push_stream_shm_data_t *data);

src/ngx_http_push_stream_module_ipc.h

@@ -0,0 +1 @@
+../include/ngx_http_push_stream_module_ipc.h

src/ngx_http_push_stream_module_publisher.c

@@ -23,8 +23,8 @@
  * Authors: Wandenberg Peixoto <[email protected]>, Rogério Carvalho Schneider <[email protected]>
  */
 
-#include <ngx_http_push_stream_module_publisher.h>
-#include <ngx_http_push_stream_module_version.h>
+#include "ngx_http_push_stream_module_publisher.h"
+#include "ngx_http_push_stream_module_version.h"
 
 static ngx_int_t    ngx_http_push_stream_publisher_handle_after_read_body(ngx_http_request_t *r, ngx_http_client_body_handler_pt post_handler);
 
@@ -97,7 +97,7 @@ ngx_http_push_stream_publisher_handler(ngx_http_request_t *r)
 
         if (r->method & (NGX_HTTP_POST|NGX_HTTP_PUT)) {
             // create the channel if doesn't exist
-            requested_channel->channel = ngx_http_push_stream_get_channel(requested_channel->id, r->connection->log, mcf);
+            requested_channel->channel = ngx_http_push_stream_get_channel(requested_channel->id, r->connection->log, mcf, r);
             if (requested_channel->channel == NULL) {
                 return ngx_http_push_stream_send_only_header_response(r, NGX_HTTP_INTERNAL_SERVER_ERROR, NULL);
             }

src/ngx_http_push_stream_module_publisher.h

@@ -0,0 +1 @@
+../include/ngx_http_push_stream_module_publisher.h

src/ngx_http_push_stream_module_setup.c

@@ -23,7 +23,7 @@
  * Authors: Wandenberg Peixoto <[email protected]>, Rogério Carvalho Schneider <[email protected]>
  */
 
-#include <ngx_http_push_stream_module_setup.h>
+#include "ngx_http_push_stream_module_setup.h"
 
 ngx_uint_t ngx_http_push_stream_padding_max_len = 0;
 ngx_flag_t ngx_http_push_stream_enabled = 0;
@@ -249,6 +249,42 @@ static ngx_command_t    ngx_http_push_stream_commands[] = {
         NGX_HTTP_LOC_CONF_OFFSET,
         offsetof(ngx_http_push_stream_loc_conf_t, allow_connections_to_events_channel),
         NULL },
+    { ngx_string("push_stream_channel_created_request"),
+        NGX_HTTP_SRV_CONF|NGX_HTTP_LOC_CONF|NGX_HTTP_LIF_CONF|NGX_CONF_TAKE1,
+        ngx_http_set_complex_value_slot,
+        NGX_HTTP_LOC_CONF_OFFSET,
+        offsetof(ngx_http_push_stream_loc_conf_t, channel_created_request_url),
+        NULL } ,
+    { ngx_string("push_stream_channel_destroyed_request"),
+        NGX_HTTP_SRV_CONF|NGX_HTTP_LOC_CONF|NGX_HTTP_LIF_CONF|NGX_CONF_TAKE1,
+        ngx_http_set_complex_value_slot,
+        NGX_HTTP_LOC_CONF_OFFSET,
+        offsetof(ngx_http_push_stream_loc_conf_t, channel_destroyed_request_url),
+        NULL } ,
+    { ngx_string("push_stream_all_worker_clients_unsubscribed_request"),
+        NGX_HTTP_SRV_CONF|NGX_HTTP_LOC_CONF|NGX_HTTP_LIF_CONF|NGX_CONF_TAKE1,
+        ngx_http_set_complex_value_slot,
+        NGX_HTTP_LOC_CONF_OFFSET,
+        offsetof(ngx_http_push_stream_loc_conf_t, all_worker_clients_unsubscribed_request_url),
+        NULL } ,
+    { ngx_string("push_stream_client_subscribed_request"),
+        NGX_HTTP_SRV_CONF|NGX_HTTP_LOC_CONF|NGX_HTTP_LIF_CONF|NGX_CONF_TAKE1,
+        ngx_http_set_complex_value_slot,
+        NGX_HTTP_LOC_CONF_OFFSET,
+        offsetof(ngx_http_push_stream_loc_conf_t, client_subscribed_request_url),
+        NULL } ,
+    { ngx_string("push_stream_client_unsubscribed_request"),
+        NGX_HTTP_SRV_CONF|NGX_HTTP_LOC_CONF|NGX_HTTP_LIF_CONF|NGX_CONF_TAKE1,
+        ngx_http_set_complex_value_slot,
+        NGX_HTTP_LOC_CONF_OFFSET,
+        offsetof(ngx_http_push_stream_loc_conf_t, client_unsubscribed_request_url),
+        NULL } ,
+    { ngx_string("push_stream_client_publish_request"),
+        NGX_HTTP_SRV_CONF|NGX_HTTP_LOC_CONF|NGX_HTTP_LIF_CONF|NGX_CONF_TAKE1,
+        ngx_http_set_complex_value_slot,
+        NGX_HTTP_LOC_CONF_OFFSET,
+        offsetof(ngx_http_push_stream_loc_conf_t, client_publish_request_url),
+        NULL } ,
 
     ngx_null_command
 };
@@ -588,6 +624,12 @@ ngx_http_push_stream_create_loc_conf(ngx_conf_t *cf)
     ngx_str_null(&lcf->padding_by_user_agent);
     lcf->paddings = NULL;
     lcf->allowed_origins = NULL;
+    lcf->channel_created_request_url = NULL;
+    lcf->channel_destroyed_request_url = NULL;
+    lcf->all_worker_clients_unsubscribed_request_url = NULL;
+    lcf->client_subscribed_request_url = NULL;
+    lcf->client_unsubscribed_request_url = NULL;
+    lcf->client_publish_request_url = NULL;
 
     return lcf;
 }
@@ -638,6 +680,30 @@ ngx_http_push_stream_merge_loc_conf(ngx_conf_t *cf, void *parent, void *child)
         conf->allowed_origins = prev->allowed_origins ;
     }
 
+    if (conf->channel_created_request_url == NULL) {
+        conf->channel_created_request_url = prev->channel_created_request_url ;
+    }
+
+    if (conf->channel_destroyed_request_url == NULL) {
+        conf->channel_destroyed_request_url = prev->channel_destroyed_request_url ;
+    }
+
+    if (conf->all_worker_clients_unsubscribed_request_url == NULL) {
+        conf->all_worker_clients_unsubscribed_request_url = prev->all_worker_clients_unsubscribed_request_url ;
+    }
+
+    if (conf->client_subscribed_request_url == NULL) {
+        conf->client_subscribed_request_url = prev->client_subscribed_request_url ;
+    }
+
+    if (conf->client_unsubscribed_request_url == NULL) {
+        conf->client_unsubscribed_request_url = prev->client_unsubscribed_request_url ;
+    }
+
+    if (conf->client_publish_request_url == NULL) {
+        conf->client_publish_request_url = prev->client_publish_request_url ;
+    }
+
     if (conf->location_type == NGX_CONF_UNSET_UINT) {
         return NGX_CONF_OK;
     }
@@ -872,14 +938,14 @@ ngx_http_push_stream_subscriber(ngx_conf_t *cf, ngx_command_t *cmd, void *conf)
     }
 
     if (*field == NGX_HTTP_PUSH_STREAM_SUBSCRIBER_MODE_WEBSOCKET) {
-        char *rc = ngx_http_push_stream_setup_handler(cf, conf, &ngx_http_push_stream_websocket_handler);
 #if (NGX_HAVE_SHA1)
+        char *rc = ngx_http_push_stream_setup_handler(cf, conf, &ngx_http_push_stream_websocket_handler);
         if (rc == NGX_CONF_OK) {
             ngx_http_push_stream_loc_conf_t     *pslcf = conf;
             pslcf->location_type = NGX_HTTP_PUSH_STREAM_SUBSCRIBER_MODE_WEBSOCKET;
         }
 #else
-        rc = NGX_CONF_ERROR;
+        char *rc = NGX_CONF_ERROR;
         ngx_conf_log_error(NGX_LOG_ERR, cf, 0, "push stream module: push stream module: sha1 support is needed to use WebSocket");
 #endif
         return rc;
@@ -1131,7 +1197,7 @@ ngx_http_push_stream_init_shm_zone(ngx_shm_zone_t *shm_zone, void *data)
     d->mutex_round_robin = 0;
 
     if (mcf->events_channel_id.len > 0) {
-        if ((d->events_channel = ngx_http_push_stream_get_channel(&mcf->events_channel_id, ngx_cycle->log, mcf)) == NULL) {
+        if ((d->events_channel = ngx_http_push_stream_get_channel(&mcf->events_channel_id, ngx_cycle->log, mcf, NULL)) == NULL) {
             ngx_log_error(NGX_LOG_ERR, ngx_cycle->log, 0, "push stream module: unable to create events channel");
             return NGX_ERROR;
         }

src/ngx_http_push_stream_module_setup.h

@@ -0,0 +1 @@
+../include/ngx_http_push_stream_module_setup.h