send old messages together when using jsonp

include/ngx_http_push_stream_module_utils.h

@@ -209,8 +209,9 @@ static const ngx_str_t  NGX_HTTP_PUSH_STREAM_EVENTSOURCE_CONTENT_TYPE = ngx_stri
 static const ngx_str_t  NGX_HTTP_PUSH_STREAM_EVENTSOURCE_PING_MESSAGE_CHUNK = ngx_string("6" CRLF ": -1" CRLF CRLF);
 
 static const ngx_str_t  NGX_HTTP_PUSH_STREAM_LAST_CHUNK = ngx_string("0" CRLF CRLF);
-static const ngx_str_t  NGX_HTTP_PUSH_STREAM_CALLBACK_INIT_CHUNK = ngx_string("3" CRLF "(" CRLF CRLF);
-static const ngx_str_t  NGX_HTTP_PUSH_STREAM_CALLBACK_END_CHUNK = ngx_string("4" CRLF ");" CRLF CRLF);
+static const ngx_str_t  NGX_HTTP_PUSH_STREAM_CALLBACK_INIT_CHUNK = ngx_string("2" CRLF "([" CRLF);
+static const ngx_str_t  NGX_HTTP_PUSH_STREAM_CALLBACK_MID_CHUNK = ngx_string("1" CRLF "," CRLF);
+static const ngx_str_t  NGX_HTTP_PUSH_STREAM_CALLBACK_END_CHUNK = ngx_string("5" CRLF "]);" CRLF CRLF);
 
 static const ngx_str_t  NGX_HTTP_PUSH_STREAM_PADDING_BY_USER_AGENT_PATTERN = ngx_string("([^:]+),(\\d+),(\\d+)");
 
@@ -234,7 +235,7 @@ static ngx_str_t *          ngx_http_push_stream_format_message(ngx_http_push_st
 static ngx_str_t *          ngx_http_push_stream_apply_template_to_each_line(ngx_str_t *text, const ngx_str_t *message_template, ngx_pool_t *temp_pool);
 static ngx_int_t            ngx_http_push_stream_send_response_content_header(ngx_http_request_t *r, ngx_http_push_stream_loc_conf_t *pslcf);
 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);
+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 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);

include/ngx_http_push_stream_module_version.h

@@ -27,6 +27,6 @@
 #define NGX_HTTP_PUSH_STREAM_MODULE_VERSION_H_
 
 static const ngx_str_t  NGX_HTTP_PUSH_STREAM_TAG = ngx_string("0.3.3");
-static const ngx_str_t  NGX_HTTP_PUSH_STREAM_COMMIT = ngx_string("cd264714638b797333f70d8f4432901507e053c5");
+static const ngx_str_t  NGX_HTTP_PUSH_STREAM_COMMIT = ngx_string("bc4aeda8fa33e5806abe97cbee93f9fa045af834");
 
 #endif /* NGX_HTTP_PUSH_STREAM_MODULE_VERSION_H_ */

src/ngx_http_push_stream_module_ipc.c

@@ -399,12 +399,12 @@ ngx_http_push_stream_respond_to_subscribers(ngx_http_push_stream_channel_t *chan
                 ngx_http_send_header(subscriber->request);
 
                 ngx_http_push_stream_send_response_content_header(subscriber->request, ngx_http_get_module_loc_conf(subscriber->request, ngx_http_push_stream_module));
-                ngx_http_push_stream_send_response_message(subscriber->request, channel, msg);
+                ngx_http_push_stream_send_response_message(subscriber->request, channel, msg, 1, 0);
                 ngx_http_push_stream_send_response_finalize(subscriber->request);
 
                 cur = prev;
             } else {
-                if (ngx_http_push_stream_send_response_message(subscriber->request, channel, msg) != NGX_OK) {
+                if (ngx_http_push_stream_send_response_message(subscriber->request, channel, msg, 0, 0) != NGX_OK) {
                     ngx_http_push_stream_queue_elem_t *prev = (ngx_http_push_stream_queue_elem_t *) ngx_queue_prev(&cur->queue);
                     ngx_http_push_stream_send_response_finalize(subscriber->request);
                     cur = prev;

src/ngx_http_push_stream_module_subscriber.c

@@ -262,6 +262,11 @@ ngx_http_push_stream_subscriber_polling_handler(ngx_http_request_t *r, ngx_http_
         return NGX_HTTP_INTERNAL_SERVER_ERROR;
     }
 
+    if (ctx->callback != NULL) {
+        ngx_http_push_stream_send_response_text(r, ctx->callback->data, ctx->callback->len, 0);
+        ngx_http_push_stream_send_response_text(r, NGX_HTTP_PUSH_STREAM_CALLBACK_INIT_CHUNK.data, NGX_HTTP_PUSH_STREAM_CALLBACK_INIT_CHUNK.len, 0);
+    }
+
     cur = channels_ids;
     while ((cur = (ngx_http_push_stream_requested_channel_t *) ngx_queue_next(&cur->queue)) != channels_ids) {
         channel = ngx_http_push_stream_find_channel_locked(cur->id, r->connection->log);
@@ -273,6 +278,10 @@ ngx_http_push_stream_subscriber_polling_handler(ngx_http_request_t *r, ngx_http_
         ngx_http_push_stream_send_old_messages(r, channel, cur->backtrack_messages, if_modified_since, tag, greater_message_time, greater_message_tag, last_event_id);
     }
 
+    if (ctx->callback != NULL) {
+        ngx_http_push_stream_send_response_text(r, NGX_HTTP_PUSH_STREAM_CALLBACK_END_CHUNK.data, NGX_HTTP_PUSH_STREAM_CALLBACK_END_CHUNK.len, 0);
+    }
+
     if (cf->footer_template.len > 0) {
         ngx_http_push_stream_send_response_text(r, cf->footer_template.data, cf->footer_template.len, 0);
     }
@@ -584,7 +593,7 @@ ngx_http_push_stream_send_old_messages(ngx_http_request_t *r, ngx_http_push_stre
             // positioning at first message, and send the others
             while ((qtd > 0) && (!message->deleted) && ((message = (ngx_http_push_stream_msg_t *) ngx_queue_next(&message->queue)) != message_sentinel)) {
                 if (start == 0) {
-                    ngx_http_push_stream_send_response_message(r, channel, message);
+                    ngx_http_push_stream_send_response_message(r, channel, message, 0, 1);
                     qtd--;
                 } else {
                     start--;
@@ -606,7 +615,7 @@ ngx_http_push_stream_send_old_messages(ngx_http_request_t *r, ngx_http_push_stre
                 }
 
                 if (found && (((greater_message_time == 0) && (greater_message_tag == -1)) || (greater_message_time > message->time) || ((greater_message_time == message->time) && (greater_message_tag >= message->tag)))) {
-                    ngx_http_push_stream_send_response_message(r, channel, message);
+                    ngx_http_push_stream_send_response_message(r, channel, message, 0, 1);
                 }
             }
         }

src/ngx_http_push_stream_module_utils.c

@@ -110,7 +110,7 @@ ngx_http_push_stream_delete_unrecoverable_channels(ngx_http_push_stream_shm_data
                                         ngx_http_push_stream_send_response_content_header(subscriber->request, ngx_http_get_module_loc_conf(subscriber->request, ngx_http_push_stream_module));
                                     }
 
-                                    ngx_http_push_stream_send_response_message(subscriber->request, channel, channel->channel_deleted_message);
+                                    ngx_http_push_stream_send_response_message(subscriber->request, channel, channel->channel_deleted_message, 1, 1);
 
                                     break;
                                 }
@@ -433,10 +433,11 @@ ngx_http_push_stream_send_response_content_header(ngx_http_request_t *r, ngx_htt
 }
 
 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_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)
 {
     ngx_http_push_stream_loc_conf_t       *pslcf = ngx_http_get_module_loc_conf(r, ngx_http_push_stream_module);
     ngx_http_push_stream_subscriber_ctx_t *ctx = ngx_http_get_module_ctx(r, ngx_http_push_stream_module);
+    ngx_flag_t                             use_jsonp = (ctx != NULL) && (ctx->callback != NULL);
     ngx_int_t rc = NGX_OK;
 
     if (pslcf->eventsource_support) {
@@ -452,7 +453,7 @@ ngx_http_push_stream_send_response_message(ngx_http_request_t *r, ngx_http_push_
     if (rc == NGX_OK) {
         ngx_str_t *str = ngx_http_push_stream_get_formatted_message(r, channel, msg, r->pool);
         if (str != NULL) {
-            if ((rc == NGX_OK) && (ctx != NULL) && (ctx->callback != NULL)) {
+            if ((rc == NGX_OK) && use_jsonp && send_callback) {
                 rc = ngx_http_push_stream_send_response_text(r, ctx->callback->data, ctx->callback->len, 0);
                 if (rc == NGX_OK) {
                     rc = ngx_http_push_stream_send_response_text(r, NGX_HTTP_PUSH_STREAM_CALLBACK_INIT_CHUNK.data, NGX_HTTP_PUSH_STREAM_CALLBACK_INIT_CHUNK.len, 0);
@@ -463,8 +464,14 @@ ngx_http_push_stream_send_response_message(ngx_http_request_t *r, ngx_http_push_
                 rc = ngx_http_push_stream_send_response_text(r, str->data, str->len, 0);
             }
 
-            if ((rc == NGX_OK) && (ctx != NULL) && (ctx->callback != NULL)) {
-                rc = ngx_http_push_stream_send_response_text(r, NGX_HTTP_PUSH_STREAM_CALLBACK_END_CHUNK.data, NGX_HTTP_PUSH_STREAM_CALLBACK_END_CHUNK.len, 0);
+            if ((rc == NGX_OK) && use_jsonp) {
+                if (send_separator) {
+                    rc = ngx_http_push_stream_send_response_text(r, NGX_HTTP_PUSH_STREAM_CALLBACK_MID_CHUNK.data, NGX_HTTP_PUSH_STREAM_CALLBACK_MID_CHUNK.len, 0);
+                }
+
+                if (send_callback) {
+                    rc = ngx_http_push_stream_send_response_text(r, NGX_HTTP_PUSH_STREAM_CALLBACK_END_CHUNK.data, NGX_HTTP_PUSH_STREAM_CALLBACK_END_CHUNK.len, 0);
+                }
             }
 
             if (rc == NGX_OK) {
@@ -946,7 +953,7 @@ ngx_http_push_stream_ping_timer_wake_handler(ngx_event_t *ev)
     } else if (pslcf->location_type == NGX_HTTP_PUSH_STREAM_WEBSOCKET_MODE) {
         rc = ngx_http_push_stream_send_response_text(r, NGX_HTTP_PUSH_STREAM_WEBSOCKET_PING_LAST_FRAME_BYTE, sizeof(NGX_HTTP_PUSH_STREAM_WEBSOCKET_PING_LAST_FRAME_BYTE), 1);
     } else {
-        rc = ngx_http_push_stream_send_response_message(r, NULL, ngx_http_push_stream_ping_msg);
+        rc = ngx_http_push_stream_send_response_message(r, NULL, ngx_http_push_stream_ping_msg, 1, 1);
     }
 
     if (rc != NGX_OK) {

test/base_test_case.rb

@@ -224,11 +224,17 @@ module BaseTestCase
     TCPSocket.open(nginx_host, nginx_port)
   end
 
-  def read_response(socket)
-    response = socket.readpartial(1)
+  def read_response(socket, wait_for=nil)
+    response ||= socket.readpartial(1)
     while (tmp = socket.read_nonblock(256))
       response += tmp
     end
+  rescue Errno::EAGAIN => e
+    headers, body = (response || "").split("\r\n\r\n", 2)
+    if !wait_for.nil? && (body.nil? || body.empty? || !body.include?(wait_for))
+      IO.select([socket])
+      retry
+    end
   ensure
     fail("Any response") if response.nil?
     headers, body = response.split("\r\n\r\n", 2)

test/test_measure_memory.rb

@@ -6,7 +6,7 @@ class TestMeasureMemory < Test::Unit::TestCase
   @@message_estimate_size = 174
   @@channel_estimate_size = 536
   @@subscriber_estimate_size = 230
-  @@subscriber_estimate_system_size = 6500
+  @@subscriber_estimate_system_size = 6600
 
   def global_configuration
     @max_reserved_memory = "2m"

test/test_send_signals.rb

@@ -34,50 +34,23 @@ class TestSendSignals < Test::Unit::TestCase
           pub_1.callback {
             assert_equal(200, pub_1.response_header.status, "Don't get channels statistics")
             assert_not_equal(0, pub_1.response_header.content_length, "Don't received channels statistics")
-            begin
-              resp_1 = JSON.parse(pub_1.response)
-              assert(resp_1.has_key?("channels"), "Didn't received the correct answer with channels info")
-              assert_equal(1, resp_1["channels"].to_i, "Didn't create channel")
-              assert_equal(1, resp_1["by_worker"].count, "Didn't return infos by_worker")
-              pid = resp_1["by_worker"][0]['pid'].to_i
+            resp_1 = JSON.parse(pub_1.response)
+            assert(resp_1.has_key?("channels"), "Didn't received the correct answer with channels info")
+            assert_equal(1, resp_1["channels"].to_i, "Didn't create channel")
+            assert_equal(1, resp_1["by_worker"].count, "Didn't return infos by_worker")
+            pid = resp_1["by_worker"][0]['pid'].to_i
 
-              # send reload signal
-              `#{ nginx_executable } -c #{ config_filename } -s reload > /dev/null 2>&1`
-
-              # publish a message
-              pub_2 = EventMachine::HttpRequest.new(nginx_address + '/pub?id=' + channel.to_s).post :head => headers, :body => body, :timeout => 30
-              pub_2.callback {
-                # add new subscriber
-                sub_2 = EventMachine::HttpRequest.new(nginx_address + '/sub/' + channel.to_s + '.b1').get :head => headers, :timeout => 30
-                sub_2.stream { |chunk|
-                  response2 = response2 + chunk
-                  if response2.strip == @header_template
-                    # check statistics again
-                    pub_3 = EventMachine::HttpRequest.new(nginx_address + '/channels-stats').get :head => headers, :timeout => 30
-                    pub_3.callback {
-
-                      resp_2 = JSON.parse(pub_3.response)
-                      assert(resp_2.has_key?("channels"), "Didn't received the correct answer with channels info")
-                      assert_equal(1, resp_2["channels"].to_i, "Didn't create channel")
-                      assert_equal(1, resp_2["published_messages"].to_i, "Didn't create messages")
-                      assert_equal(2, resp_2["subscribers"].to_i, "Didn't create subscribers")
-                      assert_equal(2, resp_2["by_worker"].count, "Didn't return infos by_worker")
-
-                    }
-                  end
-                }
-              }
-            rescue JSON::ParserError
-              fail("Didn't receive a valid response")
-              EventMachine.stop
-            end
+            # send reload signal
+            `#{ nginx_executable } -c #{ config_filename } -s reload > /dev/null 2>&1`
           }
         end
       }
 
+
+      conectted_after_reloaded = false
       i = 0
       # check if first worker die
-      EM.add_periodic_timer(1) do
+      EM.add_periodic_timer(0.5) do
 
         # check statistics again
         pub_4 = EventMachine::HttpRequest.new(nginx_address + '/channels-stats').get :head => headers, :timeout => 30
@@ -85,7 +58,34 @@ class TestSendSignals < Test::Unit::TestCase
           resp_3 = JSON.parse(pub_4.response)
           assert(resp_3.has_key?("by_worker"), "Didn't received the correct answer with channels info")
 
-          if resp_3["by_worker"].count == 1
+          if resp_3["by_worker"].count == 2 && !conectted_after_reloaded
+            conectted_after_reloaded = true
+
+            # publish a message
+            pub_2 = EventMachine::HttpRequest.new(nginx_address + '/pub?id=' + channel.to_s).post :head => headers, :body => body, :timeout => 30
+            pub_2.callback {
+              # add new subscriber
+              sub_2 = EventMachine::HttpRequest.new(nginx_address + '/sub/' + channel.to_s + '.b1').get :head => headers, :timeout => 30
+              sub_2.stream { |chunk|
+                response2 = response2 + chunk
+                if response2.strip == @header_template
+                  # check statistics again
+                  pub_3 = EventMachine::HttpRequest.new(nginx_address + '/channels-stats').get :head => headers, :timeout => 30
+                  pub_3.callback {
+
+                    resp_2 = JSON.parse(pub_3.response)
+                    assert(resp_2.has_key?("channels"), "Didn't received the correct answer with channels info")
+                    assert_equal(1, resp_2["channels"].to_i, "Didn't create channel")
+                    assert_equal(1, resp_2["published_messages"].to_i, "Didn't create messages")
+                    assert_equal(2, resp_2["subscribers"].to_i, "Didn't create subscribers")
+                    assert_equal(2, resp_2["by_worker"].count, "Didn't return infos by_worker")
+                  }
+                end
+              }
+            }
+          end
+
+          if resp_3["by_worker"].count == 1 && conectted_after_reloaded
             assert_equal(1, resp_3["channels"].to_i, "Didn't create channel")
             assert_equal(1, resp_3["published_messages"].to_i, "Didn't create messages")
             assert_equal(1, resp_3["subscribers"].to_i, "Didn't create subscribers")
@@ -97,7 +97,7 @@ class TestSendSignals < Test::Unit::TestCase
           end
 
           i = i + 1
-          if i == 60
+          if i == 120
             fail("Worker didn't die in 60 seconds")
             EventMachine.stop
           end

test/test_subscriber_long_polling.rb

@@ -585,7 +585,7 @@ class TestSubscriberLongPolling < Test::Unit::TestCase
 
       sub_1 = EventMachine::HttpRequest.new(nginx_address + '/sub/' + channel.to_s + '?callback=' + callback_function_name).get :head => headers, :timeout => 30
       sub_1.callback {
-        assert_equal("#{callback_function_name}\r\n(\r\n#{body}\r\n);\r\n", sub_1.response, "Wrong message")
+        assert_equal("#{callback_function_name}\r\n([#{body}\r\n]);\r\n", sub_1.response, "Wrong message")
         EventMachine.stop
       }
 
@@ -595,4 +595,26 @@ class TestSubscriberLongPolling < Test::Unit::TestCase
     }
   end
 
+  def test_return_old_messages_using_function_name_specified_in_callback_parameter_grouping_in_one_answer
+    headers = {'accept' => 'application/json'}
+    channel = 'ch_test_return_old_messages_using_function_name_specified_in_callback_parameter_grouping_in_one_answer'
+    body = 'body'
+    response = ""
+    callback_function_name = "callback_function"
+
+    EventMachine.run {
+
+      publish_message_inline(channel, {'accept' => 'text/html'}, body)
+      publish_message_inline(channel, {'accept' => 'text/html'}, body + "1")
+
+      sub_1 = EventMachine::HttpRequest.new(nginx_address + '/sub/' + channel.to_s + '.b2' + '?callback=' + callback_function_name).get :head => headers, :timeout => 30
+      sub_1.callback {
+        assert_equal("#{callback_function_name}\r\n([#{body}\r\n,#{body + "1"}\r\n,]);\r\n", sub_1.response, "Wrong message")
+        EventMachine.stop
+      }
+
+      add_test_timeout
+    }
+  end
+
 end

test/test_subscriber_polling.rb

@@ -593,7 +593,7 @@ class TestSubscriberPolling < Test::Unit::TestCase
 
       sub_1 = EventMachine::HttpRequest.new(nginx_address + '/sub/' + channel.to_s + '?callback=' + callback_function_name).get :head => headers, :timeout => 30
       sub_1.callback {
-        assert_equal("#{callback_function_name}\r\n(\r\n#{body}\r\n);\r\n", sub_1.response, "Wrong message")
+        assert_equal("#{callback_function_name}\r\n([#{body}\r\n,]);\r\n", sub_1.response, "Wrong message")
         EventMachine.stop
       }
 

test/test_websocket.rb

@@ -238,7 +238,7 @@ class TestWebSocket < Test::Unit::TestCase
 
     socket = TCPSocket.open(nginx_host, nginx_port)
     socket.print("#{request}\r\n")
-    headers, body = read_response(socket)
+    headers, body = read_response(socket, "aaa")
     assert(body.start_with?("\201~\377\377aaa"), "Wrong response")
   end
 
@@ -258,7 +258,7 @@ class TestWebSocket < Test::Unit::TestCase
 
     socket = TCPSocket.open(nginx_host, nginx_port)
     socket.print("#{request}\r\n")
-    headers, body = read_response(socket)
+    headers, body = read_response(socket, "aaa")
     assert(body.start_with?("\201\177\000\000\000\000\000\001\000\000aaa"), "Wrong response")
   end