Skip to content

HTTPS clone URL

Subversion checkout URL

You can clone with HTTPS or Subversion.

Download ZIP
Browse files

added push_channel_group setting

  • Loading branch information...
commit 0c4be3ed7df08cb8a15f7327be0bfabaacd06111 1 parent 7700e87
@slact authored
View
10 README
@@ -78,7 +78,15 @@ push_store_messages [ on | off ]
context: http, server, location
Whether or not message queuing is enabled. "Off" is equivalent to the setting
push_channel_buffer_length 0;
-
+
+push_channel_group [ string ]
+ default: (none)
+ context: server, location
+ As settings are bound to locations and urls instead of individual channels,
+ it is useful to be able to have channels that can be reached only from some
+ locations and never others. That's where this setting comes in. Think of it
+ as a prefix string for the channel id.
+
push_queue_messages [ on | off ] (DEPRECATED)
default: on
context: http, server, location
View
43 src/ngx_http_push_module.c
@@ -115,20 +115,31 @@ static ngx_http_push_msg_t * ngx_http_push_find_message_locked(ngx_http_push_cha
return NULL;
}
-static ngx_int_t ngx_http_push_set_channel_id(ngx_str_t *id, ngx_http_request_t *r, ngx_http_push_loc_conf_t *cf) {
+static ngx_str_t * ngx_http_push_get_channel_id(ngx_http_request_t *r, ngx_http_push_loc_conf_t *cf) {
ngx_http_variable_value_t *vv = ngx_http_get_indexed_variable(r, cf->index);
+ ngx_str_t *prefix;
+ size_t len;
+ ngx_str_t *id;
if (vv == NULL || vv->not_found || vv->len == 0) {
ngx_log_error(NGX_LOG_ERR, r->connection->log, 0,
- "push module: the \"$push_channel_id\" variable is required but is not set");
- return NGX_ERROR;
+ "push module: the $push_channel_id variable is required but is not set");
+ return NULL;
}
- //no need to copy anything? ok...
- id->data=vv->data;
- id->len=vv->len;
- return NGX_OK;
+ prefix = &cf->channel_group;
+ len = prefix->len + 1 + vv->len;
+ if((id = ngx_palloc(r->pool, sizeof(*id) + len))==NULL) {
+ ngx_log_error(NGX_LOG_ERR, r->connection->log, 0,
+ "push module: unable to allocate memory for $push_channel_id string");
+ return NULL;
+ }
+ id->len=len;
+ id->data=(u_char *)(id+1);
+ ngx_memcpy(id->data, prefix->data, prefix->len);
+ id->data[prefix->len]='/';
+ ngx_memcpy(id->data + prefix->len + 1, vv->data, id->len);
+ return id;
}
-
#define NGX_HTTP_PUSH_MAKE_ETAG(message_tag, etag, alloc_func, pool) \
etag = alloc_func(pool, sizeof(*etag) + NGX_INT_T_LEN); \
if(etag!=NULL) { \
@@ -146,7 +157,7 @@ static ngx_int_t ngx_http_push_set_channel_id(ngx_str_t *id, ngx_http_request_t
static ngx_int_t ngx_http_push_subscriber_handler(ngx_http_request_t *r) {
ngx_http_push_loc_conf_t *cf = ngx_http_get_module_loc_conf(r, ngx_http_push_module);
ngx_slab_pool_t *shpool = (ngx_slab_pool_t *) ngx_http_push_shm_zone->shm.addr;
- ngx_str_t id;
+ ngx_str_t *id;
ngx_http_push_channel_t *channel;
ngx_http_push_msg_t *msg;
ngx_int_t msg_search_outcome;
@@ -159,13 +170,14 @@ static ngx_int_t ngx_http_push_subscriber_handler(ngx_http_request_t *r) {
return NGX_HTTP_NOT_ALLOWED;
}
- if(ngx_http_push_set_channel_id(&id, r, cf) !=NGX_OK) {
+ if((id=ngx_http_push_get_channel_id(r, cf)) == NULL) {
return NGX_HTTP_INTERNAL_SERVER_ERROR;
}
//get the channel and check channel authorization while we're at it.
ngx_shmtx_lock(&shpool->mutex);
- channel = (cf->authorize_channel==1 ? ngx_http_push_find_channel : ngx_http_push_get_channel)(&id, &((ngx_http_push_shm_data_t *) ngx_http_push_shm_zone->data)->tree, shpool, r->connection->log);
+ channel = (cf->authorize_channel==1 ? ngx_http_push_find_channel : ngx_http_push_get_channel)(id, &((ngx_http_push_shm_data_t *) ngx_http_push_shm_zone->data)->tree, shpool, r->connection->log);
+
if (channel==NULL) {
//unable to allocate channel OR channel not found
ngx_shmtx_unlock(&shpool->mutex);
@@ -422,7 +434,7 @@ static ngx_int_t ngx_http_push_broadcast_locked(ngx_http_push_channel_t *channel
//OK?...
static void ngx_http_push_publisher_body_handler(ngx_http_request_t * r) {
- ngx_str_t id;
+ ngx_str_t *id;
ngx_http_push_loc_conf_t *cf = ngx_http_get_module_loc_conf(r, ngx_http_push_module);
ngx_slab_pool_t *shpool = (ngx_slab_pool_t *) ngx_http_push_shm_zone->shm.addr;
ngx_buf_t *buf = NULL, *buf_copy;
@@ -433,19 +445,20 @@ static void ngx_http_push_publisher_body_handler(ngx_http_request_t * r) {
ngx_uint_t subscribers = 0;
ngx_uint_t messages = 0;
- NGX_HTTP_PUSH_PUBLISHER_CHECK(ngx_http_push_set_channel_id(&id, r, cf), NGX_ERROR, r, "can't determine channel id")
+ NGX_HTTP_PUSH_PUBLISHER_CHECK((id = ngx_http_push_get_channel_id(r, cf)), NULL, r, "can't determine channel id")
ngx_shmtx_lock(&shpool->mutex);
//POST requests will need a channel created if it doesn't yet exist.
if(method==NGX_HTTP_POST || method==NGX_HTTP_PUT) {
- channel = ngx_http_push_get_channel(&id, &((ngx_http_push_shm_data_t *) ngx_http_push_shm_zone->data)->tree, shpool, r->connection->log);
+ channel = ngx_http_push_get_channel(id, &((ngx_http_push_shm_data_t *) ngx_http_push_shm_zone->data)->tree, shpool, r->connection->log);
NGX_HTTP_PUSH_PUBLISHER_CHECK_LOCKED(channel, NULL, r, "push module: unable to allocate memory for new channel", shpool);
}
//no other request method needs that.
else{
//just find the channel. if it's not there, NULL.
- channel = ngx_http_push_find_channel(&id, &((ngx_http_push_shm_data_t *) ngx_http_push_shm_zone->data)->tree, shpool, r->connection->log);
+ channel = ngx_http_push_find_channel(id, &((ngx_http_push_shm_data_t *) ngx_http_push_shm_zone->data)->tree, shpool, r->connection->log);
}
+ ngx_pfree(ngx_http_push_pool, id);
if(channel!=NULL) {
subscribers = channel->subscribers;
View
4 src/ngx_http_push_module.h
@@ -58,6 +58,8 @@ typedef struct {
ngx_int_t authorize_channel;
ngx_int_t store_messages;
ngx_int_t min_message_recipients;
+ ngx_str_t channel_group;
+ ngx_int_t max_channel_id_length;
} ngx_http_push_loc_conf_t;
//message queue
@@ -134,7 +136,7 @@ static ngx_inline void ngx_http_push_free_message_locked(ngx_http_push_msg_t *ms
static ngx_http_push_msg_t * ngx_http_push_find_message_locked(ngx_http_push_channel_t *channel, ngx_http_request_t *r, ngx_int_t *status);
//channel
-static ngx_int_t ngx_http_push_set_channel_id(ngx_str_t *id, ngx_http_request_t *r, ngx_http_push_loc_conf_t *cf);
+static ngx_str_t * ngx_http_push_get_channel_id(ngx_http_request_t *r, ngx_http_push_loc_conf_t *cf);
static ngx_int_t ngx_http_push_channel_info(ngx_http_request_t *r, ngx_uint_t message_queue_size, ngx_uint_t subscriber_queue_size, time_t last_seen);
//subscriber
View
9 src/ngx_http_push_module_setup.c
@@ -102,6 +102,7 @@ static void * ngx_http_push_create_loc_conf(ngx_conf_t *cf) {
lcf->authorize_channel=NGX_CONF_UNSET;
lcf->store_messages=NGX_CONF_UNSET;
lcf->min_message_recipients=NGX_CONF_UNSET;
+ lcf->channel_group.data=NULL;
return lcf;
}
@@ -115,6 +116,7 @@ static char * ngx_http_push_merge_loc_conf(ngx_conf_t *cf, void *parent, void *c
ngx_conf_merge_value(conf->authorize_channel, prev->authorize_channel, 0);
ngx_conf_merge_value(conf->store_messages, prev->store_messages, 1);
ngx_conf_merge_value(conf->min_message_recipients, prev->min_message_recipients, NGX_HTTP_PUSH_MIN_MESSAGE_RECIPIENTS);
+ ngx_conf_merge_str_value(conf->channel_group, prev->channel_group, "");
//sanity checks
if(conf->max_messages < conf->min_messages) {
//min/max buffer size makes sense?
@@ -257,6 +259,13 @@ static ngx_command_t ngx_http_push_commands[] = {
offsetof(ngx_http_push_loc_conf_t, store_messages),
NULL },
+ { ngx_string("push_channel_group"),
+ NGX_HTTP_SRV_CONF|NGX_HTTP_LOC_CONF|NGX_CONF_TAKE1,
+ ngx_conf_set_str_slot,
+ NGX_HTTP_LOC_CONF_OFFSET,
+ offsetof(ngx_http_push_loc_conf_t, channel_group),
+ NULL },
+
//deprecated and misleading. remove no earlier than november 2009.
{ ngx_string("push_buffer_size"),
NGX_HTTP_MAIN_CONF|NGX_CONF_TAKE1,
Please sign in to comment.
Something went wrong with that request. Please try again.