Skip to content

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

change end of line for eventsource from CRLF to LF

parent 9b9dee58
Hide whitespace changes
Inline Side-by-side
Showing with 135 additions and 50 deletions
include/ngx_http_push_stream_module_utils.h
View file @ e8a6de63
......@@ -207,13 +207,13 @@ static const ngx_str_t NGX_HTTP_PUSH_STREAM_TOKEN_MESSAGE_TAG = ngx_string("~ta
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(": " CRLF);
static const ngx_str_t NGX_HTTP_PUSH_STREAM_EVENTSOURCE_COMMENT_TEMPLATE = ngx_string(": ~text~" CRLF);
static const ngx_str_t NGX_HTTP_PUSH_STREAM_EVENTSOURCE_DEFAULT_HEADER_TEMPLATE = ngx_string(": \n");
static const ngx_str_t NGX_HTTP_PUSH_STREAM_EVENTSOURCE_COMMENT_TEMPLATE = ngx_string(": ~text~\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_ID_TEMPLATE = ngx_string("id: ~event-id~" CRLF);
static const ngx_str_t NGX_HTTP_PUSH_STREAM_EVENTSOURCE_EVENT_TEMPLATE = ngx_string("event: ~event-type~" CRLF);
static const ngx_str_t NGX_HTTP_PUSH_STREAM_EVENTSOURCE_ID_TEMPLATE = ngx_string("id: ~event-id~\n");
static const ngx_str_t NGX_HTTP_PUSH_STREAM_EVENTSOURCE_EVENT_TEMPLATE = ngx_string("event: ~event-type~\n");
static const ngx_str_t NGX_HTTP_PUSH_STREAM_EVENTSOURCE_CONTENT_TYPE = ngx_string("text/event-stream; charset=utf-8");
static const ngx_str_t NGX_HTTP_PUSH_STREAM_EVENTSOURCE_PING_MESSAGE_CHUNK = ngx_string(": -1" CRLF);
static const ngx_str_t NGX_HTTP_PUSH_STREAM_EVENTSOURCE_PING_MESSAGE_CHUNK = ngx_string(": -1\n");
static const ngx_str_t NGX_HTTP_PUSH_STREAM_CALLBACK_INIT_CHUNK = ngx_string("([");
static const ngx_str_t NGX_HTTP_PUSH_STREAM_CALLBACK_MID_CHUNK = ngx_string(",");
......
misc/Gemfile
View file @ e8a6de63
......@@ -16,6 +16,7 @@ group :test do
gem 'thin', '~> 1.5.1'
gem 'net-http-persistent', '~> 2.9', :require => 'net/http/persistent'
gem 'websocket-eventmachine-client'
gem 'em-eventsource'
gem 'byebug', '~> 1.3.1'
end
......
misc/Gemfile.lock
View file @ e8a6de63
......@@ -3,7 +3,7 @@ GEM
specs:
Platform (0.4.0)
RedCloth (4.2.9)
addressable (2.3.6)
addressable (2.3.7)
byebug (1.3.1)
columnize (~> 0.3.6)
debugger-linecache (~> 1.2.0)
......@@ -14,6 +14,9 @@ GEM
daemons (1.1.9)
debugger-linecache (1.2.0)
diff-lcs (1.2.5)
em-eventsource (0.2.0)
em-http-request (>= 1.0.0)
eventmachine (>= 1.0.0.beta3)
em-http-request (1.0.3)
addressable (>= 2.2.3)
cookiejar
......@@ -22,7 +25,7 @@ GEM
http_parser.rb (>= 0.5.3)
em-socksify (0.3.0)
eventmachine (>= 1.0.0.beta.4)
eventmachine (1.0.3)
eventmachine (1.0.7)
execjs (2.0.2)
ffi (1.9.6)
filewatcher (0.3.4)
......@@ -94,6 +97,7 @@ PLATFORMS
DEPENDENCIES
RedCloth (~> 4.2.9)
byebug (~> 1.3.1)
em-eventsource
em-http-request (~> 1.0.3)
filewatcher
github-markup (~> 0.7.5)
......
misc/spec/subscriber/event_source_spec.rb
View file @ e8a6de63
......@@ -16,10 +16,15 @@ describe "Subscriber Event Source" do
nginx_run_server(config.merge(:header_template => "header")) do |conf|
EventMachine.run do
source = EventMachine::EventSource.new(nginx_address + '/sub/' + channel.to_s)
source.open do
EventMachine.stop
end
sub = EventMachine::HttpRequest.new(nginx_address + '/sub/' + channel.to_s).get
sub.stream do |chunk|
sub.response_header["CONTENT_TYPE"].should eql("text/event-stream; charset=utf-8")
EventMachine.stop
source.start
end
end
end
......@@ -32,7 +37,7 @@ describe "Subscriber Event Source" do
EventMachine.run do
sub = EventMachine::HttpRequest.new(nginx_address + '/sub/' + channel.to_s).get
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")
chunk.should eql(": header line 1\n: header line 2\n: header line 3\n: header line 4\n")
EventMachine.stop
end
end
......@@ -46,7 +51,7 @@ describe "Subscriber Event Source" do
EventMachine.run do
sub = EventMachine::HttpRequest.new(nginx_address + '/sub/' + channel.to_s).get
sub.stream do |chunk|
chunk.should eql(": header line 1\\nheader line 2\r\n")
chunk.should eql(": header line 1\\nheader line 2\n")
EventMachine.stop
end
end
......@@ -64,7 +69,7 @@ describe "Subscriber Event Source" do
response += chunk
end
sub.callback do
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")
response.should eql(": \n: footer line 1\n: footer line 2\n: footer line 3\n: footer line 4\n")
EventMachine.stop
end
end
......@@ -82,7 +87,7 @@ describe "Subscriber Event Source" do
response += chunk
end
sub.callback do
response.should eql(": \r\n: footer line 1\\nfooter line 2\r\n")
response.should eql(": \n: footer line 1\\nfooter line 2\n")
EventMachine.stop
end
end
......@@ -96,16 +101,23 @@ describe "Subscriber Event Source" do
nginx_run_server(config) do |conf|
EventMachine.run do
source = EventMachine::EventSource.new(nginx_address + '/sub/_' + channel.to_s)
source.message do |message|
message.should eql(body)
publish_message_inline(channel, headers, body)
end
source.start
sub = EventMachine::HttpRequest.new(nginx_address + '/sub/' + channel.to_s).get
sub.stream do |chunk|
response += chunk
if response.include?("data: ")
response.should eql(": \r\ndata: #{body}\r\n\r\n")
response.should eql(": \ndata: #{body}\n\n")
EventMachine.stop
end
end
publish_message_inline(channel, headers, body)
publish_message_inline("_#{channel}", headers, body)
end
end
end
......@@ -117,16 +129,23 @@ describe "Subscriber Event Source" do
nginx_run_server(config) do |conf|
EventMachine.run do
source = EventMachine::EventSource.new(nginx_address + '/sub/_' + channel.to_s)
source.message do |message|
message.should eql(body)
publish_message_inline(channel, headers, body)
end
source.start
sub = EventMachine::HttpRequest.new(nginx_address + '/sub/' + channel.to_s).get
sub.stream do |chunk|
response += chunk
if response.include?("data: ")
response.should eql(": \r\ndata: #{body}\r\n\r\n")
response.should eql(": \ndata: #{body}\n\n")
EventMachine.stop
end
end
publish_message_inline(channel, headers, body)
publish_message_inline("_#{channel}", headers, body)
end
end
end
......@@ -139,16 +158,24 @@ describe "Subscriber Event Source" do
nginx_run_server(config) do |conf|
EventMachine.run do
source = EventMachine::EventSource.new(nginx_address + '/sub/_' + channel.to_s)
source.message do |message|
message.should eql(body)
source.last_event_id.should eql(event_id)
publish_message_inline(channel, headers.merge('Event-Id' => event_id), body)
end
source.start
sub = EventMachine::HttpRequest.new(nginx_address + '/sub/' + channel.to_s).get
sub.stream do |chunk|
response += chunk
if response.include?("data: ")
response.should eql(": \r\nid: #{event_id}\r\ndata: #{body}\r\n\r\n")
response.should eql(": \nid: #{event_id}\ndata: #{body}\n\n")
EventMachine.stop
end
end
publish_message_inline(channel, headers.merge('Event-Id' => event_id), body)
publish_message_inline("_#{channel}", headers.merge('Event-Id' => event_id), body)
end
end
end
......@@ -161,16 +188,23 @@ describe "Subscriber Event Source" do
nginx_run_server(config) do |conf|
EventMachine.run do
source = EventMachine::EventSource.new(nginx_address + '/sub/_' + channel.to_s)
source.on event_type do |message|
message.should eql(body)
publish_message_inline(channel, headers.merge('Event-type' => event_type), body)
end
source.start
sub = EventMachine::HttpRequest.new(nginx_address + '/sub/' + channel.to_s).get
sub.stream do |chunk|
response += chunk
if response.include?("data: ")
response.should eql(": \r\nevent: #{event_type}\r\ndata: #{body}\r\n\r\n")
response.should eql(": \nevent: #{event_type}\ndata: #{body}\n\n")
EventMachine.stop
end
end
publish_message_inline(channel, headers.merge('Event-type' => event_type), body)
publish_message_inline("_#{channel}", headers.merge('Event-type' => event_type), body)
end
end
end
......@@ -182,16 +216,23 @@ describe "Subscriber Event Source" do
nginx_run_server(config.merge(:message_template => '{\"id\":\"~id~\", \"message\":\"~text~\"}')) do |conf|
EventMachine.run do
source = EventMachine::EventSource.new(nginx_address + '/sub/_' + channel.to_s)
source.message do |message|
message.should eql(%({"id":"1", "message":"#{body}"}))
publish_message_inline(channel, headers, body)
end
source.start
sub = EventMachine::HttpRequest.new(nginx_address + '/sub/' + channel.to_s).get
sub.stream do |chunk|
response += chunk
if response.include?("data: ")
response.should eql(%(: \r\ndata: {"id":"1", "message":"#{body}"}\r\n\r\n))
response.should eql(%(: \ndata: {"id":"1", "message":"#{body}"}\n\n))
EventMachine.stop
end
end
publish_message_inline(channel, headers, body)
publish_message_inline("_#{channel}", headers, body)
end
end
end
......@@ -203,16 +244,23 @@ describe "Subscriber Event Source" do
nginx_run_server(config.merge(:message_template => '{\"id\":\"~id~\", \"message\":\"~text~\"}')) do |conf|
EventMachine.run do
source = EventMachine::EventSource.new(nginx_address + '/sub/_' + channel.to_s)
source.message do |message|
message.should eql(%({"id":"1", "message":"#{body}"}))
publish_message_inline(channel, headers, body)
end
source.start
sub = EventMachine::HttpRequest.new(nginx_address + '/sub/' + channel.to_s).get
sub.stream do |chunk|
response += chunk
if response.include?("data: ")
response.should eql(%(: \r\ndata: {"id":"1", "message":"#{body}"}\r\n\r\n))
response.should eql(%(: \ndata: {"id":"1", "message":"#{body}"}\n\n))
EventMachine.stop
end
end
publish_message_inline(channel, headers, body)
publish_message_inline("_#{channel}", headers, body)
end
end
end
......@@ -225,16 +273,24 @@ describe "Subscriber Event Source" do
nginx_run_server(config.merge(:message_template => '{\"id\":\"~id~\", \"message\":\"~text~\"}')) do |conf|
EventMachine.run do
source = EventMachine::EventSource.new(nginx_address + '/sub/_' + channel.to_s)
source.message do |message|
message.should eql(%({"id":"1", "message":"#{body}"}))
source.last_event_id.should eql(event_id)
publish_message_inline(channel, headers.merge('Event-Id' => event_id), body)
end
source.start
sub = EventMachine::HttpRequest.new(nginx_address + '/sub/' + channel.to_s).get
sub.stream do |chunk|
response += chunk
if response.include?("data: ")
response.should eql(%(: \r\nid: #{event_id}\r\ndata: {"id":"1", "message":"#{body}"}\r\n\r\n))
response.should eql(%(: \nid: #{event_id}\ndata: {"id":"1", "message":"#{body}"}\n\n))
EventMachine.stop
end
end
publish_message_inline(channel, headers.merge('Event-Id' => event_id), body)
publish_message_inline("_#{channel}", headers.merge('Event-Id' => event_id), body)
end
end
end
......@@ -247,16 +303,23 @@ describe "Subscriber Event Source" do
nginx_run_server(config.merge(:message_template => '{\"id\":\"~id~\", \"message\":\"~text~\"}')) do |conf|
EventMachine.run do
source = EventMachine::EventSource.new(nginx_address + '/sub/_' + channel.to_s)
source.on event_type do |message|
message.should eql(%({"id":"1", "message":"#{body}"}))
publish_message_inline(channel, headers.merge('Event-type' => event_type), body)
end
source.start
sub = EventMachine::HttpRequest.new(nginx_address + '/sub/' + channel.to_s).get
sub.stream do |chunk|
response += chunk
if response.include?("data: ")
response.should eql(%(: \r\nevent: #{event_type}\r\ndata: {"id":"1", "message":"#{body}"}\r\n\r\n))
response.should eql(%(: \nevent: #{event_type}\ndata: {"id":"1", "message":"#{body}"}\n\n))
EventMachine.stop
end
end
publish_message_inline(channel, headers.merge('Event-type' => event_type), body)
publish_message_inline("_#{channel}", headers.merge('Event-type' => event_type), body)
end
end
end
......@@ -267,15 +330,22 @@ describe "Subscriber Event Source" do
nginx_run_server(config) do |conf|
EventMachine.run do
source = EventMachine::EventSource.new(nginx_address + '/sub/_' + channel.to_s)
source.message do |message|
message.should eql("line 1\nline 2\nline 3\nline 4")
publish_message_inline(channel, headers, body)
end
source.start
sub = EventMachine::HttpRequest.new(nginx_address + '/sub/' + channel.to_s).get
sub.stream do |chunk|
if chunk.include?("line 4")
chunk.should eql("data: line 1\r\ndata: line 2\r\ndata: line 3\r\ndata: line 4\r\n\r\n")
chunk.should eql("data: line 1\ndata: line 2\ndata: line 3\ndata: line 4\n\n")
EventMachine.stop
end
end
publish_message_inline(channel, headers, body)
publish_message_inline("_#{channel}", headers, body)
end
end
end
......@@ -286,15 +356,22 @@ describe "Subscriber Event Source" do
nginx_run_server(config) do |conf|
EventMachine.run do
source = EventMachine::EventSource.new(nginx_address + '/sub/_' + channel.to_s)
source.message do |message|
message.should eql(body)
publish_message_inline(channel, headers, body)
end
source.start
sub = EventMachine::HttpRequest.new(nginx_address + '/sub/' + channel.to_s).get
sub.stream do |chunk|
if chunk.include?("line 2")
chunk.should eql("data: line 1\\nline 2\r\n\r\n")
chunk.should eql("data: line 1\\nline 2\n\n")
EventMachine.stop
end
end
publish_message_inline(channel, headers, body)
publish_message_inline("_#{channel}", headers, body)
end
end
end
......@@ -307,7 +384,7 @@ describe "Subscriber Event Source" do
sub = EventMachine::HttpRequest.new(nginx_address + '/sub/' + channel.to_s).get
sub.stream do |chunk|
if chunk.include?("-1")
chunk.should eql(": -1\r\n")
chunk.should eql(": -1\n")
EventMachine.stop
end
end
......@@ -335,14 +412,14 @@ describe "Subscriber Event Source" do
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.should eql(": header\ndata: msg #{body}\n\n: footer\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")
response.should eql(": header\ndata: msg #{body}\n\n: footer\n")
EventMachine.stop
end
end
......
misc/spec/subscriber/padding_by_user_agent_spec.rb
View file @ e8a6de63
......@@ -198,9 +198,9 @@ describe "Subscriber Padding by user agent" do
describe "for EventSource mode" do
let(:config) { default_config.merge(:subscriber_mode => "eventsource") }
let(:padding_pattern) { /(:::)+\r\n$/ }
let(:header_delta) { 4 }
let(:body_delta) { 10 }
let(:padding_pattern) { /(:::)+\n$/ }
let(:header_delta) { 3 }
let(:body_delta) { 8 }
it_should_behave_like "apply padding"
end
......
misc/spec/subscriber/receive_old_message_spec.rb
View file @ e8a6de63
......@@ -12,6 +12,8 @@ describe "Receive old messages" do
}
end
let(:eol) { "\r\n" }
shared_examples_for "can receive old messages" do
it "should receive old messages in a multi channel subscriber using backtrack" do
channel_1 = 'ch_test_retreive_old_messages_in_multichannel_subscribe_1'
......@@ -34,7 +36,7 @@ describe "Receive old messages" do
response_headers['ETAG'].to_s.should_not eql("")
end
lines = response.split("\r\n")
lines = response.split(eol)
lines[0].should eql('HEADER')
line = JSON.parse(lines[1])
line['channel'].should eql(channel_2.to_s)
......@@ -93,7 +95,7 @@ describe "Receive old messages" do
response_headers['ETAG'].to_s.should_not eql("")
end
lines = response.split("\r\n")
lines = response.split(eol)
lines[0].should eql('HEADER')
line = JSON.parse(lines[1])
......@@ -148,7 +150,7 @@ describe "Receive old messages" do
response_headers['ETAG'].to_s.should_not eql("")
end
lines = response.split("\r\n")
lines = response.split(eol)
lines[0].should eql('HEADER')
line = JSON.parse(lines[1])
......@@ -244,7 +246,7 @@ describe "Receive old messages" do
response_headers['ETAG'].to_s.should eql("1")
end
response.should eql("msg 1\r\n")
response.should eql("msg 1#{eol}")
end
end
end
......@@ -304,11 +306,11 @@ describe "Receive old messages" do
sub_1 = EventMachine::HttpRequest.new(url).get :head => request_headers
sub_1.stream do |chunk|
response += chunk
lines = response.split("\r\n").map {|line| line.gsub(/^: /, "").gsub(/^data: /, "").gsub(/^id: .*/, "") }.delete_if{|line| line.empty?}.compact
lines = response.split(eol).map {|line| line.gsub(/^: /, "").gsub(/^data: /, "").gsub(/^id: .*/, "") }.delete_if{|line| line.empty?}.compact
if lines.length >= number_expected_lines
EventMachine.stop
block.call("#{lines.join("\r\n")}\r\n", sub_1.response_header) unless block.nil?
block.call("#{lines.join(eol)}#{eol}", sub_1.response_header) unless block.nil?
end
end
end
......@@ -334,6 +336,7 @@ describe "Receive old messages" do
context "in event source mode" do
let(:subscriber_mode) { "eventsource" }
let(:eol) { "\n" }
it_should_behave_like "can receive old messages"
end
......
src/ngx_http_push_stream_module_setup.c
View file @ e8a6de63
......@@ -407,8 +407,8 @@ ngx_http_push_stream_postconfig(ngx_conf_t *cf)
}
ngx_memset(aux->data, ':', padding_max_len);
padding_max_len -= 2;
ngx_memcpy(aux->data + padding_max_len, CRLF, 2);
padding_max_len -= 1;
ngx_memcpy(aux->data + padding_max_len, "\n", 1);
ngx_int_t i, len = ngx_http_push_stream_padding_max_len;
for (i = steps; i >= 0; i--) {
......@@ -651,14 +651,14 @@ ngx_http_push_stream_merge_loc_conf(ngx_conf_t *cf, void *parent, void *child)
// 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 + ngx_strlen(CRLF));
ngx_str_t *template = ngx_http_push_stream_create_str(cf->pool, NGX_HTTP_PUSH_STREAM_EVENTSOURCE_MESSAGE_PREFIX.len + aux->len + 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);
ngx_memcpy(last, "\n", 1);
conf->message_template.data = template->data;
conf->message_template.len = template->len;
......
src/ngx_http_push_stream_module_utils.c
View file @ e8a6de63
......@@ -323,8 +323,8 @@ ngx_http_push_stream_convert_char_to_msg_on_shared(ngx_http_push_stream_main_con
}
ngx_str_t *tmp = ngx_http_push_stream_join_with_crlf(lines, temp_pool);
if ((aux = ngx_http_push_stream_create_str(temp_pool, tmp->len + 2)) != NULL) {
ngx_sprintf(aux->data, "%V" CRLF, tmp);
if ((aux = ngx_http_push_stream_create_str(temp_pool, tmp->len + 1)) != NULL) {
ngx_sprintf(aux->data, "%V\n", tmp);
}
} else {
aux = ngx_http_push_stream_format_message(channel, msg, &msg->raw, cur->template, temp_pool);
......
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