Skip to content

N
nginx-push-stream-module
Commit 085a3b1f authored by Wandenberg's avatar Wandenberg
Browse files
Options

avoid reapply formatter to header, message or footer template when inside an... 

avoid reapply formatter to header, message or footer template when inside an if on event source mode
parent 0013ac3b
Hide whitespace changes
Inline Side-by-side
Showing with 117 additions and 69 deletions
CHANGELOG.textile
View file @ 085a3b1f
h1(#changelog). Changelog h1(#changelog). Changelog
* Avoid reapply formatter to header, message or footer template when inside an if on event source mode
* Added support for OPTIONS method on publisher location * Added support for OPTIONS method on publisher location
* Unified longPollingTimeout and timeout configurations on javascript client * Unified longPollingTimeout and timeout configurations on javascript client
* Renamed some javascript client configurations * Renamed some javascript client configurations
......
include/ngx_http_push_stream_module_utils.h
View file @ 085a3b1f
...@@ -203,6 +203,7 @@ static const ngx_str_t NGX_HTTP_PUSH_STREAM_TOKEN_MESSAGE_TEXT = ngx_string("~t ...@@ -203,6 +203,7 @@ static const ngx_str_t NGX_HTTP_PUSH_STREAM_TOKEN_MESSAGE_TEXT = ngx_string("~t
static const ngx_str_t NGX_HTTP_PUSH_STREAM_TOKEN_MESSAGE_TAG = ngx_string("~tag~"); static const ngx_str_t NGX_HTTP_PUSH_STREAM_TOKEN_MESSAGE_TAG = ngx_string("~tag~");
static const ngx_str_t NGX_HTTP_PUSH_STREAM_TOKEN_MESSAGE_TIME = ngx_string("~time~"); static const ngx_str_t NGX_HTTP_PUSH_STREAM_TOKEN_MESSAGE_TIME = ngx_string("~time~");
static const ngx_str_t NGX_HTTP_PUSH_STREAM_EVENTSOURCE_COMMENT_PREFIX = ngx_string(": ");
static const ngx_str_t NGX_HTTP_PUSH_STREAM_EVENTSOURCE_DEFAULT_HEADER_TEMPLATE = ngx_string(": \r\n"); static const ngx_str_t NGX_HTTP_PUSH_STREAM_EVENTSOURCE_DEFAULT_HEADER_TEMPLATE = ngx_string(": \r\n");
static const ngx_str_t NGX_HTTP_PUSH_STREAM_EVENTSOURCE_COMMENT_TEMPLATE = ngx_string(": ~text~\r\n"); static const ngx_str_t NGX_HTTP_PUSH_STREAM_EVENTSOURCE_COMMENT_TEMPLATE = ngx_string(": ~text~\r\n");
static const ngx_str_t NGX_HTTP_PUSH_STREAM_EVENTSOURCE_MESSAGE_PREFIX = ngx_string("data: "); static const ngx_str_t NGX_HTTP_PUSH_STREAM_EVENTSOURCE_MESSAGE_PREFIX = ngx_string("data: ");
......
misc/spec/subscriber/event_source_spec.rb
View file @ 085a3b1f
...@@ -32,7 +32,7 @@ describe "Subscriber Event Source" do ...@@ -32,7 +32,7 @@ describe "Subscriber Event Source" do
EventMachine.run do EventMachine.run do
sub = EventMachine::HttpRequest.new(nginx_address + '/sub/' + channel.to_s).get sub = EventMachine::HttpRequest.new(nginx_address + '/sub/' + channel.to_s).get
sub.stream do |chunk| sub.stream do |chunk|
chunk.should eql(": header line 1\r\n: header line 2\r\n: header line 3\r\n: header line 4\r\n\r\n") chunk.should eql(": header line 1\r\n: header line 2\r\n: header line 3\r\n: header line 4\r\n")
EventMachine.stop EventMachine.stop
end end
end end
...@@ -46,7 +46,7 @@ describe "Subscriber Event Source" do ...@@ -46,7 +46,7 @@ describe "Subscriber Event Source" do
EventMachine.run do EventMachine.run do
sub = EventMachine::HttpRequest.new(nginx_address + '/sub/' + channel.to_s).get sub = EventMachine::HttpRequest.new(nginx_address + '/sub/' + channel.to_s).get
sub.stream do |chunk| sub.stream do |chunk|
chunk.should eql(": header line 1\\nheader line 2\r\n\r\n") chunk.should eql(": header line 1\\nheader line 2\r\n")
EventMachine.stop EventMachine.stop
end end
end end
...@@ -64,7 +64,7 @@ describe "Subscriber Event Source" do ...@@ -64,7 +64,7 @@ describe "Subscriber Event Source" do
response += chunk response += chunk
end end
sub.callback do sub.callback do
response.should eql(": \r\n\r\n: footer line 1\r\n: footer line 2\r\n: footer line 3\r\n: footer line 4\r\n\r\n") response.should eql(": \r\n: footer line 1\r\n: footer line 2\r\n: footer line 3\r\n: footer line 4\r\n")
EventMachine.stop EventMachine.stop
end end
end end
...@@ -82,7 +82,7 @@ describe "Subscriber Event Source" do ...@@ -82,7 +82,7 @@ describe "Subscriber Event Source" do
response += chunk response += chunk
end end
sub.callback do sub.callback do
response.should eql(": \r\n\r\n: footer line 1\\nfooter line 2\r\n\r\n") response.should eql(": \r\n: footer line 1\\nfooter line 2\r\n")
EventMachine.stop EventMachine.stop
end end
end end
...@@ -100,7 +100,7 @@ describe "Subscriber Event Source" do ...@@ -100,7 +100,7 @@ describe "Subscriber Event Source" do
sub.stream do |chunk| sub.stream do |chunk|
response += chunk response += chunk
if response.include?("data: ") if response.include?("data: ")
response.should eql(": \r\n\r\ndata: #{body}\r\n\r\n") response.should eql(": \r\ndata: #{body}\r\n\r\n")
EventMachine.stop EventMachine.stop
end end
end end
...@@ -121,7 +121,7 @@ describe "Subscriber Event Source" do ...@@ -121,7 +121,7 @@ describe "Subscriber Event Source" do
sub.stream do |chunk| sub.stream do |chunk|
response += chunk response += chunk
if response.include?("data: ") if response.include?("data: ")
response.should eql(": \r\n\r\ndata: #{body}\r\n\r\n") response.should eql(": \r\ndata: #{body}\r\n\r\n")
EventMachine.stop EventMachine.stop
end end
end end
...@@ -143,7 +143,7 @@ describe "Subscriber Event Source" do ...@@ -143,7 +143,7 @@ describe "Subscriber Event Source" do
sub.stream do |chunk| sub.stream do |chunk|
response += chunk response += chunk
if response.include?("data: ") if response.include?("data: ")
response.should eql(": \r\n\r\nid: #{event_id}\r\ndata: #{body}\r\n\r\n") response.should eql(": \r\nid: #{event_id}\r\ndata: #{body}\r\n\r\n")
EventMachine.stop EventMachine.stop
end end
end end
...@@ -165,7 +165,7 @@ describe "Subscriber Event Source" do ...@@ -165,7 +165,7 @@ describe "Subscriber Event Source" do
sub.stream do |chunk| sub.stream do |chunk|
response += chunk response += chunk
if response.include?("data: ") if response.include?("data: ")
response.should eql(": \r\n\r\nevent: #{event_type}\r\ndata: #{body}\r\n\r\n") response.should eql(": \r\nevent: #{event_type}\r\ndata: #{body}\r\n\r\n")
EventMachine.stop EventMachine.stop
end end
end end
...@@ -186,7 +186,7 @@ describe "Subscriber Event Source" do ...@@ -186,7 +186,7 @@ describe "Subscriber Event Source" do
sub.stream do |chunk| sub.stream do |chunk|
response += chunk response += chunk
if response.include?("data: ") if response.include?("data: ")
response.should eql(%(: \r\n\r\ndata: {"id":"1", "message":"#{body}"}\r\n\r\n)) response.should eql(%(: \r\ndata: {"id":"1", "message":"#{body}"}\r\n\r\n))
EventMachine.stop EventMachine.stop
end end
end end
...@@ -207,7 +207,7 @@ describe "Subscriber Event Source" do ...@@ -207,7 +207,7 @@ describe "Subscriber Event Source" do
sub.stream do |chunk| sub.stream do |chunk|
response += chunk response += chunk
if response.include?("data: ") if response.include?("data: ")
response.should eql(%(: \r\n\r\ndata: {"id":"1", "message":"#{body}"}\r\n\r\n)) response.should eql(%(: \r\ndata: {"id":"1", "message":"#{body}"}\r\n\r\n))
EventMachine.stop EventMachine.stop
end end
end end
...@@ -229,7 +229,7 @@ describe "Subscriber Event Source" do ...@@ -229,7 +229,7 @@ describe "Subscriber Event Source" do
sub.stream do |chunk| sub.stream do |chunk|
response += chunk response += chunk
if response.include?("data: ") if response.include?("data: ")
response.should eql(%(: \r\n\r\nid: #{event_id}\r\ndata: {"id":"1", "message":"#{body}"}\r\n\r\n)) response.should eql(%(: \r\nid: #{event_id}\r\ndata: {"id":"1", "message":"#{body}"}\r\n\r\n))
EventMachine.stop EventMachine.stop
end end
end end
...@@ -251,7 +251,7 @@ describe "Subscriber Event Source" do ...@@ -251,7 +251,7 @@ describe "Subscriber Event Source" do
sub.stream do |chunk| sub.stream do |chunk|
response += chunk response += chunk
if response.include?("data: ") if response.include?("data: ")
response.should eql(%(: \r\n\r\nevent: #{event_type}\r\ndata: {"id":"1", "message":"#{body}"}\r\n\r\n)) response.should eql(%(: \r\nevent: #{event_type}\r\ndata: {"id":"1", "message":"#{body}"}\r\n\r\n))
EventMachine.stop EventMachine.stop
end end
end end
...@@ -314,4 +314,45 @@ describe "Subscriber Event Source" do ...@@ -314,4 +314,45 @@ describe "Subscriber Event Source" do
end end
end end
end end
it "should not reaplly formatter to header, message or footer template when inside an if" do
channel = 'ch_test_not_reaplly_formatter_on_header_message_footer_template'
body = 'test message'
response = ''
extra_location = %(
location ~ /ev/(.*) {
push_stream_subscriber "eventsource";
push_stream_channels_path "$1";
if ($arg_tests = "on") {
push_stream_channels_path "test_$1";
}
}
)
nginx_run_server(config.merge(:extra_location => extra_location, :header_template => "header", :message_template => "msg ~text~", :footer_template => "footer", :subscriber_connection_ttl => '1s')) do |conf|
EventMachine.run do
sub = EventMachine::HttpRequest.new(nginx_address + '/ev/' + channel.to_s).get
sub.stream do |chunk|
response += chunk
if response.include?("footer")
response.should eql(": header\r\ndata: msg #{body}\r\n\r\n: footer\r\n")
response = ''
sub_1 = EventMachine::HttpRequest.new(nginx_address + '/ev/' + channel.to_s + '?tests=on').get
sub_1.stream do |chunk_1|
response += chunk_1
if response.include?("footer")
response.should eql(": header\r\ndata: msg #{body}\r\n\r\n: footer\r\n")
EventMachine.stop
end
end
publish_message_inline("test_" + channel, headers, body)
end
end
publish_message_inline(channel, headers, body)
end
end
end
end end
src/ngx_http_push_stream_module_setup.c
View file @ 085a3b1f
...@@ -602,47 +602,85 @@ ngx_http_push_stream_merge_loc_conf(ngx_conf_t *cf, void *parent, void *child) ...@@ -602,47 +602,85 @@ ngx_http_push_stream_merge_loc_conf(ngx_conf_t *cf, void *parent, void *child)
// changing properties for event source support // changing properties for event source support
if (conf->location_type == NGX_HTTP_PUSH_STREAM_SUBSCRIBER_MODE_EVENTSOURCE) { if (conf->location_type == NGX_HTTP_PUSH_STREAM_SUBSCRIBER_MODE_EVENTSOURCE) {
// formatting header template // formatting header template
if (ngx_strncmp(conf->header_template.data, NGX_HTTP_PUSH_STREAM_EVENTSOURCE_COMMENT_PREFIX.data, NGX_HTTP_PUSH_STREAM_EVENTSOURCE_COMMENT_PREFIX.len) != 0) {
if (conf->header_template.len > 0) {
ngx_str_t *aux = ngx_http_push_stream_apply_template_to_each_line(&conf->header_template, &NGX_HTTP_PUSH_STREAM_EVENTSOURCE_COMMENT_TEMPLATE, cf->pool);
if (aux == NULL) {
ngx_conf_log_error(NGX_LOG_ERR, cf, 0, "push_stream_message_module failed to apply template to header message.");
return NGX_CONF_ERROR;
}
conf->header_template.data = aux->data;
conf->header_template.len = aux->len;
} else {
conf->header_template.data = NGX_HTTP_PUSH_STREAM_EVENTSOURCE_DEFAULT_HEADER_TEMPLATE.data;
conf->header_template.len = NGX_HTTP_PUSH_STREAM_EVENTSOURCE_DEFAULT_HEADER_TEMPLATE.len;
}
}
// formatting message template
if (ngx_strncmp(conf->message_template.data, NGX_HTTP_PUSH_STREAM_EVENTSOURCE_MESSAGE_PREFIX.data, NGX_HTTP_PUSH_STREAM_EVENTSOURCE_MESSAGE_PREFIX.len) != 0) {
ngx_str_t *aux = (conf->message_template.len > 0) ? &conf->message_template : (ngx_str_t *) &NGX_HTTP_PUSH_STREAM_TOKEN_MESSAGE_TEXT;
ngx_str_t *template = ngx_http_push_stream_create_str(cf->pool, NGX_HTTP_PUSH_STREAM_EVENTSOURCE_MESSAGE_PREFIX.len + aux->len + sizeof(CRLF) -1);
if (template == NULL) {
ngx_conf_log_error(NGX_LOG_ERR, cf, 0, "push stream module: unable to allocate memory to append message prefix to message template");
return NGX_CONF_ERROR;
}
u_char *last = ngx_copy(template->data, NGX_HTTP_PUSH_STREAM_EVENTSOURCE_MESSAGE_PREFIX.data, NGX_HTTP_PUSH_STREAM_EVENTSOURCE_MESSAGE_PREFIX.len);
last = ngx_copy(last, aux->data, aux->len);
ngx_memcpy(last, CRLF, 2);
conf->message_template.data = template->data;
conf->message_template.len = template->len;
}
// formatting footer template
if (ngx_strncmp(conf->footer_template.data, NGX_HTTP_PUSH_STREAM_EVENTSOURCE_COMMENT_PREFIX.data, NGX_HTTP_PUSH_STREAM_EVENTSOURCE_COMMENT_PREFIX.len) != 0) {
if (conf->footer_template.len > 0) {
ngx_str_t *aux = ngx_http_push_stream_apply_template_to_each_line(&conf->footer_template, &NGX_HTTP_PUSH_STREAM_EVENTSOURCE_COMMENT_TEMPLATE, cf->pool);
if (aux == NULL) {
ngx_conf_log_error(NGX_LOG_ERR, cf, 0, "push_stream_message_module failed to apply template to footer message.");
return NGX_CONF_ERROR;
}
conf->footer_template.data = aux->data;
conf->footer_template.len = aux->len;
}
}
} else {
// formatting header and footer template for chunk transfer
if (conf->header_template.len > 0) { if (conf->header_template.len > 0) {
ngx_str_t *aux = ngx_http_push_stream_apply_template_to_each_line(&conf->header_template, &NGX_HTTP_PUSH_STREAM_EVENTSOURCE_COMMENT_TEMPLATE, cf->pool); ngx_str_t *aux = NULL;
if (conf->location_type == NGX_HTTP_PUSH_STREAM_SUBSCRIBER_MODE_WEBSOCKET) {
aux = ngx_http_push_stream_get_formatted_websocket_frame(conf->header_template.data, conf->header_template.len, cf->pool);
} else {
aux = ngx_http_push_stream_get_formatted_chunk(conf->header_template.data, conf->header_template.len, cf->pool);
}
if (aux == NULL) { if (aux == NULL) {
ngx_conf_log_error(NGX_LOG_ERR, cf, 0, "push_stream_message_module failed to apply template to header message."); ngx_conf_log_error(NGX_LOG_ERR, cf, 0, "push stream module: unable to allocate memory to format header template");
return NGX_CONF_ERROR; return NGX_CONF_ERROR;
} }
conf->header_template.data = aux->data; conf->header_template.data = aux->data;
conf->header_template.len = aux->len; conf->header_template.len = aux->len;
} else {
conf->header_template.data = NGX_HTTP_PUSH_STREAM_EVENTSOURCE_DEFAULT_HEADER_TEMPLATE.data;
conf->header_template.len = NGX_HTTP_PUSH_STREAM_EVENTSOURCE_DEFAULT_HEADER_TEMPLATE.len;
}
// formatting message template
ngx_str_t *aux = (conf->message_template.len > 0) ? &conf->message_template : (ngx_str_t *) &NGX_HTTP_PUSH_STREAM_TOKEN_MESSAGE_TEXT;
ngx_str_t *template = ngx_http_push_stream_create_str(cf->pool, NGX_HTTP_PUSH_STREAM_EVENTSOURCE_MESSAGE_PREFIX.len + aux->len + sizeof(CRLF) -1);
if (template == NULL) {
ngx_conf_log_error(NGX_LOG_ERR, cf, 0, "push stream module: unable to allocate memory to append message prefix to message template");
return NGX_CONF_ERROR;
} }
u_char *last = ngx_copy(template->data, NGX_HTTP_PUSH_STREAM_EVENTSOURCE_MESSAGE_PREFIX.data, NGX_HTTP_PUSH_STREAM_EVENTSOURCE_MESSAGE_PREFIX.len);
last = ngx_copy(last, aux->data, aux->len);
ngx_memcpy(last, CRLF, 2);
conf->message_template.data = template->data;
conf->message_template.len = template->len;
// formatting footer template
if (conf->footer_template.len > 0) { if (conf->footer_template.len > 0) {
ngx_str_t *aux = ngx_http_push_stream_apply_template_to_each_line(&conf->footer_template, &NGX_HTTP_PUSH_STREAM_EVENTSOURCE_COMMENT_TEMPLATE, cf->pool); ngx_str_t *aux = NULL;
if (conf->location_type == NGX_HTTP_PUSH_STREAM_SUBSCRIBER_MODE_WEBSOCKET) {
aux = ngx_http_push_stream_get_formatted_websocket_frame(conf->footer_template.data, conf->footer_template.len, cf->pool);
} else {
aux = ngx_http_push_stream_get_formatted_chunk(conf->footer_template.data, conf->footer_template.len, cf->pool);
}
if (aux == NULL) { if (aux == NULL) {
ngx_conf_log_error(NGX_LOG_ERR, cf, 0, "push_stream_message_module failed to apply template to footer message."); ngx_conf_log_error(NGX_LOG_ERR, cf, 0, "push stream module: unable to allocate memory to format footer template");
return NGX_CONF_ERROR; return NGX_CONF_ERROR;
} }
conf->footer_template.data = aux->data; conf->footer_template.data = aux->data;
conf->footer_template.len = aux->len; conf->footer_template.len = aux->len;
} }
} }
// sanity checks // sanity checks
// ping message interval cannot be zero // ping message interval cannot be zero
if ((conf->ping_message_interval != NGX_CONF_UNSET_MSEC) && (conf->ping_message_interval == 0)) { if ((conf->ping_message_interval != NGX_CONF_UNSET_MSEC) && (conf->ping_message_interval == 0)) {
...@@ -686,39 +724,6 @@ ngx_http_push_stream_merge_loc_conf(ngx_conf_t *cf, void *parent, void *child) ...@@ -686,39 +724,6 @@ ngx_http_push_stream_merge_loc_conf(ngx_conf_t *cf, void *parent, void *child)
return NGX_CONF_ERROR; return NGX_CONF_ERROR;
} }
// formatting header and footer template for chunk transfer
if (conf->header_template.len > 0) {
ngx_str_t *aux = NULL;
if (conf->location_type == NGX_HTTP_PUSH_STREAM_SUBSCRIBER_MODE_WEBSOCKET) {
aux = ngx_http_push_stream_get_formatted_websocket_frame(conf->header_template.data, conf->header_template.len, cf->pool);
} else {
aux = ngx_http_push_stream_get_formatted_chunk(conf->header_template.data, conf->header_template.len, cf->pool);
}
if (aux == NULL) {
ngx_conf_log_error(NGX_LOG_ERR, cf, 0, "push stream module: unable to allocate memory to format header template");
return NGX_CONF_ERROR;
}
conf->header_template.data = aux->data;
conf->header_template.len = aux->len;
}
if (conf->footer_template.len > 0) {
ngx_str_t *aux = NULL;
if (conf->location_type == NGX_HTTP_PUSH_STREAM_SUBSCRIBER_MODE_WEBSOCKET) {
aux = ngx_http_push_stream_get_formatted_websocket_frame(conf->footer_template.data, conf->footer_template.len, cf->pool);
} else {
aux = ngx_http_push_stream_get_formatted_chunk(conf->footer_template.data, conf->footer_template.len, cf->pool);
}
if (aux == NULL) {
ngx_conf_log_error(NGX_LOG_ERR, cf, 0, "push stream module: unable to allocate memory to format footer template");
return NGX_CONF_ERROR;
}
conf->footer_template.data = aux->data;
conf->footer_template.len = aux->len;
}
if ((conf->location_type == NGX_HTTP_PUSH_STREAM_SUBSCRIBER_MODE_LONGPOLLING) || if ((conf->location_type == NGX_HTTP_PUSH_STREAM_SUBSCRIBER_MODE_LONGPOLLING) ||
(conf->location_type == NGX_HTTP_PUSH_STREAM_SUBSCRIBER_MODE_POLLING) || (conf->location_type == NGX_HTTP_PUSH_STREAM_SUBSCRIBER_MODE_POLLING) ||
(conf->location_type == NGX_HTTP_PUSH_STREAM_SUBSCRIBER_MODE_STREAMING) || (conf->location_type == NGX_HTTP_PUSH_STREAM_SUBSCRIBER_MODE_STREAMING) ||
......
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment

Technounit-GROUP 2023