tests/index.html tests/mootools-1.2.4.js tests/test.js tests/test.rb @@ -177,6 +177,15 @@ static ngx_chain_t * ngx_http_push_create_output_chain_general(ngx_buf_t *buf, n (h)->prev = x #endif +#ifndef ngx_queue_next +#define ngx_queue_next(q) \ + (q)->next +#endif + +#ifndef NGX_FILE_OWNER_ACCESS +#define NGX_FILE_OWNER_ACCESS 0600 +#endif + //string constants //headers const ngx_str_t NGX_HTTP_PUSH_HEADER_ETAG = ngx_string("Etag"); src/ngx_http_push_module.h @@ -146,7 +146,6 @@ static ngx_inline void ngx_http_push_process_worker_message(void) { sentinel = &worker_messages[ngx_process_slot]; worker_msg = sentinel; - ngx_queue_init(&sentinel->queue); //reset the worker message sentinel while((worker_msg = (ngx_http_push_worker_msg_t *)ngx_queue_next(&worker_msg->queue)) != sentinel) { if(worker_msg->pid!=ngx_pid) { //that's quite bad you see. a previous worker died with an undelivered message. @@ -186,6 +185,7 @@ static ngx_inline void ngx_http_push_process_worker_message(void) { ngx_http_push_respond_to_subscribers(channel, subscriber_sentinel, msg, status_code, status_line); ngx_shmtx_lock(&shpool->mutex); } + ngx_queue_init(&sentinel->queue); //reset the worker message sentinel ngx_shmtx_unlock(&shpool->mutex); return; } src/ngx_http_push_module_proletariat.c @@ -190,6 +190,27 @@ static char *ngx_http_push_subscriber(ngx_conf_t *cf, ngx_command_t *cmd, void * return ngx_http_push_setup_handler(cf, conf, &ngx_http_push_subscriber_handler); } +//great justice appears to be at hand +static ngx_int_t ngx_http_push_movezig_channel_locked(ngx_http_push_channel_t * channel, ngx_slab_pool_t * shpool) { + ngx_queue_t *sentinel = &channel->message_queue->queue; + ngx_http_push_msg_t *msg=NULL; + while(!ngx_queue_empty(sentinel)) { + msg = ngx_queue_data(ngx_queue_head(sentinel), ngx_http_push_msg_t, queue); + ngx_http_push_force_delete_message_locked(channel, msg, shpool); + } + return NGX_OK; +} + +static void ngx_http_push_exit_master(ngx_cycle_t *cycle) { + ngx_slab_pool_t *shpool = (ngx_slab_pool_t *) ngx_http_push_shm_zone->shm.addr; + + //destroy channel tree in shared memory + ngx_shmtx_lock(&shpool->mutex); + ngx_http_push_walk_rbtree(ngx_http_push_movezig_channel_locked); + ngx_shmtx_unlock(&shpool->mutex); +} + + static ngx_command_t ngx_http_push_commands[] = { @@ -319,6 +340,6 @@ ngx_module_t ngx_http_push_module = { NULL, /* init thread */ NULL, /* exit thread */ NULL, /* exit process */ - NULL, /* exit master */ + ngx_http_push_exit_master, /* exit master */ NGX_MODULE_V1_PADDING }; src/ngx_http_push_module_setup.c @@ -199,6 +199,22 @@ static void ngx_rbtree_generic_insert( ngx_rbt_red(node); } +#define ngx_http_push_walk_rbtree(apply) \ + ngx_http_push_rbtree_walker(&((ngx_http_push_shm_data_t *) ngx_http_push_shm_zone->data)->tree, (ngx_slab_pool_t *)ngx_http_push_shm_zone->shm.addr, apply, ((ngx_http_push_shm_data_t *) ngx_http_push_shm_zone->data)->tree.root) + +static void ngx_http_push_rbtree_walker(ngx_rbtree_t *tree, ngx_slab_pool_t *shpool, ngx_int_t (*apply)(ngx_http_push_channel_t * channel, ngx_slab_pool_t * shpool), ngx_rbtree_node_t *node) { + ngx_rbtree_node_t *sentinel = tree->sentinel; + + if(node!=sentinel) { + apply((ngx_http_push_channel_t *)node, shpool); + if(node->left!=NULL) { + ngx_http_push_rbtree_walker(tree, shpool, apply, node->left); + } + if(node->right!=NULL) { + ngx_http_push_rbtree_walker(tree, shpool, apply, node->right); + } + } +} static void ngx_http_push_rbtree_insert(ngx_rbtree_node_t *temp, ngx_rbtree_node_t *node, ngx_rbtree_node_t *sentinel) { src/ngx_http_push_rbtree_util.c @@ -4,13 +4,13 @@ worker_processes 3; #debug_points stop; -error_log logs/error.log debug; +error_log logs/error.log; #error_log logs/error.log notice; #error_log logs/error.log info; pid logs/nginx.pid; #daemon off; -working_directory /home/leop/verk/nginx-testing/core/; +working_directory core/; events { @@ -18,29 +18,27 @@ events { } http { - include mime.types; default_type application/octet-stream; sendfile on; keepalive_timeout 65; push_authorized_channels_only off; - push_max_channel_id_length 3; server { - listen 8098; - push_min_message_recipients 1; - push_min_message_buffer_length 5; - push_max_message_buffer_length 20; - location = /listen { - set $push_channel_id $arg_channel; - push_subscriber; - push_subscriber_concurrency broadcast; - push_channel_group steve; - } - location /send { - default_type text/plain; - set $push_channel_id $arg_channel; - push_channel_group steve; - push_publisher; - push_message_timeout 1h; + listen 8082; + root /home/leop/sandbox/nginx_push/tests; + location /broadcast { + location = /broadcast/sub { + set $push_channel_id $arg_channel; + push_subscriber; + push_subscriber_concurrency broadcast; + push_channel_group broadcast; + } + location = /broadcast/pub { + set $push_channel_id $arg_channel; + push_publisher; + push_min_message_buffer_length 5; + push_max_message_buffer_length 20; + push_channel_group broadcast; + } } } } tests/nginx.conf @@ -2,7 +2,7 @@ print("The test 'framework' used here is still a touch wonky. A test failure nee require "httpest" --fyi, the test "framework" used here is still wonky. math.randomseed(os.time()) local request=httpest.request -local sendurl, listenurl = "http://localhost:8098/send?channel=%s", "http://localhost:8098/listen?channel=%s" +local sendurl, listenurl = "http://localhost:8082/broadcast/pub?channel=%s", "http://localhost:8082/broadcast/sub?channel=%s" local function send(channel, message, callback) assert(request{ url=sendurl:format(channel), tests/test.lua 3b178f4d69b301b23c43e59978b0668f377cdb7d 4d0bdf3f5b90e436a26e17cac5676dc19dedd416 Carson Baker carson@carsonbaker.org http://github.com/carsonbaker/nginx_http_push_module/commit/786177aff0fc0474145708d510eb53500c371699 786177aff0fc0474145708d510eb53500c371699 2009-11-04T22:04:06-08:00 2009-11-04T22:04:06-08:00 Merge branch 'master' of git://github.com/slact/nginx_http_push_module 3fd6fb36198e0b5d4a494887bb3711c8ab23ddff Carson Baker carson@carsonbaker.org