returning to use ngx_http_chunked_filter_module on version 1.1.4+

CHANGELOG.textile

 h1(#changelog). Changelog
 
+* Back to use Nginx chunked filter
+
 h2. Version 0.3.5
 
 * Use JSONP when port numbers don't match.
@@ -133,7 +135,7 @@ h2. Version 0.2.2
 * Fixing memory allocation for channel id
 * Fixing response for detailed channels statistics when have a big number of channels
 * Creating different timers to clear the memory, one for expired messages and another for empty channels
-* Disable Nginx chuncked filter for module locations
+* Disable Nginx chunked filter for module locations
 
 h2. Version 0.2.1
 

README.textile

@@ -174,11 +174,6 @@ h1(#installation). Installation <a name="installation" href="#">&nbsp;</a>
     sudo /usr/local/nginx/sbin/nginx -c $NGINX_PUSH_STREAM_MODULE_PATH/misc/nginx.conf
 </pre>
 
-h1(#attention). Attention
-
-This module controls everything needed to send the messages to subscribers.
-So it disable Nginx's chuncked filter to reduce memory consumption in streaming connections.
-
 h1(#memory-usage). Memory usage
 
 Just as information is listed below the minimum amount of memory used for each object:

include/ngx_http_push_stream_module.h

@@ -264,8 +264,6 @@ static const ngx_str_t  NGX_HTTP_PUSH_STREAM_HEADER_EXPLAIN = ngx_string("X-Ngin
 static const ngx_str_t  NGX_HTTP_PUSH_STREAM_HEADER_MODE = ngx_string("X-Nginx-PushStream-Mode");
 static const ngx_str_t  NGX_HTTP_PUSH_STREAM_HEADER_TAG = ngx_string("X-Nginx-PushStream-Tag");
 static const ngx_str_t  NGX_HTTP_PUSH_STREAM_HEADER_COMMIT = ngx_string("X-Nginx-PushStream-Commit");
-static const ngx_str_t  NGX_HTTP_PUSH_STREAM_HEADER_TRANSFER_ENCODING = ngx_string("Transfer-Encoding");
-static const ngx_str_t  NGX_HTTP_PUSH_STREAM_HEADER_CHUNCKED = ngx_string("chunked");
 static const ngx_str_t  NGX_HTTP_PUSH_STREAM_HEADER_ETAG = ngx_string("Etag");
 static const ngx_str_t  NGX_HTTP_PUSH_STREAM_HEADER_IF_NONE_MATCH = ngx_string("If-None-Match");
 static const ngx_str_t  NGX_HTTP_PUSH_STREAM_HEADER_VARY = ngx_string("Vary");

include/ngx_http_push_stream_module_utils.h

@@ -209,12 +209,11 @@ static const ngx_str_t  NGX_HTTP_PUSH_STREAM_EVENTSOURCE_MESSAGE_PREFIX = ngx_st
 static const ngx_str_t  NGX_HTTP_PUSH_STREAM_EVENTSOURCE_ID_TEMPLATE = ngx_string("id: ~event-id~");
 static const ngx_str_t  NGX_HTTP_PUSH_STREAM_EVENTSOURCE_EVENT_TEMPLATE = ngx_string("event: ~event-type~");
 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("6" CRLF ": -1" CRLF CRLF);
+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_LAST_CHUNK = ngx_string("0" 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_CALLBACK_INIT_CHUNK = ngx_string("([");
+static const ngx_str_t  NGX_HTTP_PUSH_STREAM_CALLBACK_MID_CHUNK = ngx_string(",");
+static const ngx_str_t  NGX_HTTP_PUSH_STREAM_CALLBACK_END_CHUNK = ngx_string("]);" CRLF);
 static const ngx_str_t  NGX_HTTP_PUSH_STREAM_CALLBACK_CONTENT_TYPE = ngx_string("application/javascript");
 
 static const ngx_str_t  NGX_HTTP_PUSH_STREAM_PADDING_BY_USER_AGENT_PATTERN = ngx_string("([^:]+),(\\d+),(\\d+)");

misc/spec/mix/measure_memory_spec.rb

@@ -15,8 +15,8 @@ describe "Measure Memory" do
     }
   end
 
-  message_estimate_size = 175
-  channel_estimate_size = 272
+  message_estimate_size = 168
+  channel_estimate_size = 270
   subscriber_estimate_size = 160
   subscriber_estimate_system_size = 7000
 
@@ -44,7 +44,7 @@ describe "Measure Memory" do
 
           resp = JSON.parse(pub_2.response)
           expected_message = shared_size / (message_estimate_size + body.size)
-          resp["published_messages"].to_i.should be_within(20).of(expected_message)
+          resp["published_messages"].to_i.should be_within(80).of(expected_message)
           EventMachine.stop
         end
       end

misc/spec/subscriber/websocket_spec.rb

@@ -274,7 +274,7 @@ describe "Subscriber WebSocket" do
       socket_2 = open_socket(nginx_host, nginx_port)
       socket_2.print("#{request_2}\r\n")
       headers_2, body_2 = read_response_on_socket(socket_2, '}')
-      body_2.should eql("11\r\n{\"text\":\"#{body}\"}\r\n\r\n")
+      body_2.should eql("{\"text\":\"#{body}\"}\r\n")
     end
   end
 

src/ngx_http_push_stream_module_ipc.c

@@ -406,7 +406,6 @@ ngx_http_push_stream_respond_to_subscribers(ngx_http_push_stream_channel_t *chan
             ngx_http_push_stream_subscription_t *subscription = ngx_queue_data(cur, ngx_http_push_stream_subscription_t, channel_worker_queue);
             ngx_http_push_stream_subscriber_t *subscriber = subscription->subscriber;
             if (subscriber->longpolling) {
-                ngx_http_push_stream_add_response_header(subscriber->request, &NGX_HTTP_PUSH_STREAM_HEADER_TRANSFER_ENCODING, &NGX_HTTP_PUSH_STREAM_HEADER_CHUNCKED);
                 ngx_http_push_stream_add_polling_headers(subscriber->request, msg->time, msg->tag, subscriber->request->pool);
                 ngx_http_send_header(subscriber->request);
 

src/ngx_http_push_stream_module_setup.c

@@ -764,8 +764,6 @@ ngx_http_push_stream_setup_handler(ngx_conf_t *cf, void *conf, ngx_int_t (*handl
     psmcf->enabled = 1;
     clcf->handler = handler;
     clcf->if_modified_since = NGX_HTTP_IMS_OFF;
-    // disable chunked_filter_module for streaming connections
-    clcf->chunked_transfer_encoding = 0;
 
     return NGX_CONF_OK;
 }

src/ngx_http_push_stream_module_subscriber.c

@@ -118,7 +118,6 @@ ngx_http_push_stream_subscriber_handler(ngx_http_request_t *r)
         return NGX_HTTP_INTERNAL_SERVER_ERROR;
     }
 
-    ngx_http_push_stream_add_response_header(r, &NGX_HTTP_PUSH_STREAM_HEADER_TRANSFER_ENCODING, &NGX_HTTP_PUSH_STREAM_HEADER_CHUNCKED);
     ngx_http_send_header(r);
 
     // sending response content header
@@ -264,7 +263,6 @@ ngx_http_push_stream_subscriber_polling_handler(ngx_http_request_t *r, ngx_http_
     r->headers_out.status = NGX_HTTP_OK;
     r->headers_out.content_length_n = -1;
 
-    ngx_http_push_stream_add_response_header(r, &NGX_HTTP_PUSH_STREAM_HEADER_TRANSFER_ENCODING, &NGX_HTTP_PUSH_STREAM_HEADER_CHUNCKED);
     ngx_http_send_header(r);
 
     // sending response content header
@@ -297,7 +295,7 @@ ngx_http_push_stream_subscriber_polling_handler(ngx_http_request_t *r, ngx_http_
         ngx_http_push_stream_send_response_text(r, cf->footer_template.data, cf->footer_template.len, 0);
     }
 
-    ngx_http_push_stream_send_response_text(r, NGX_HTTP_PUSH_STREAM_LAST_CHUNK.data, NGX_HTTP_PUSH_STREAM_LAST_CHUNK.len, 1);
+    ngx_http_send_special(r, NGX_HTTP_LAST | NGX_HTTP_FLUSH);
 
     return NGX_OK;
 }

src/ngx_http_push_stream_module_utils.c

@@ -96,7 +96,6 @@ ngx_http_push_stream_delete_channels(ngx_http_push_stream_shm_data_t *data, ngx_
                         ngx_shmtx_unlock(&shpool->mutex);
 
                         if (subscriber->longpolling) {
-                            ngx_http_push_stream_add_response_header(subscriber->request, &NGX_HTTP_PUSH_STREAM_HEADER_TRANSFER_ENCODING, &NGX_HTTP_PUSH_STREAM_HEADER_CHUNCKED);
                             ngx_http_push_stream_add_polling_headers(subscriber->request, ngx_time(), 0, subscriber->request->pool);
                             ngx_http_send_header(subscriber->request);
 
@@ -518,11 +517,7 @@ ngx_http_push_stream_output_filter(ngx_http_request_t *r, ngx_chain_t *in)
     rc = ngx_http_output_filter(r, in);
 
     if ((ctx = ngx_http_get_module_ctx(r, ngx_http_push_stream_module)) != NULL) {
-        #if defined nginx_version && nginx_version >= 1001004
-            ngx_chain_update_chains(r->pool, &ctx->free, &ctx->busy, &in, (ngx_buf_tag_t) &ngx_http_push_stream_module);
-        #else
-            ngx_chain_update_chains(&ctx->free, &ctx->busy, &in, (ngx_buf_tag_t) &ngx_http_push_stream_module);
-        #endif
+        ngx_chain_update_chains(r->pool, &ctx->free, &ctx->busy, &in, (ngx_buf_tag_t) &ngx_http_push_stream_module);
     }
 
     return rc;
@@ -635,7 +630,7 @@ ngx_http_push_stream_send_response_finalize(ngx_http_request_t *r)
         if (pslcf->location_type == NGX_HTTP_PUSH_STREAM_WEBSOCKET_MODE) {
             rc = ngx_http_push_stream_send_response_text(r, NGX_HTTP_PUSH_STREAM_WEBSOCKET_CLOSE_LAST_FRAME_BYTE, sizeof(NGX_HTTP_PUSH_STREAM_WEBSOCKET_CLOSE_LAST_FRAME_BYTE), 1);
         } else {
-            rc = ngx_http_push_stream_send_response_text(r, NGX_HTTP_PUSH_STREAM_LAST_CHUNK.data, NGX_HTTP_PUSH_STREAM_LAST_CHUNK.len, 1);
+            rc = ngx_http_send_special(r, NGX_HTTP_LAST | NGX_HTTP_FLUSH);
         }
     }
 
@@ -1241,10 +1236,9 @@ ngx_http_push_stream_get_formatted_chunk(const u_char *text, off_t len, ngx_pool
 {
     ngx_str_t            *chunk;
 
-    /* the "0000000000000000" is 64-bit hexadimal string */
-    chunk = ngx_http_push_stream_create_str(temp_pool, sizeof("0000000000000000" CRLF CRLF CRLF) + len);
+    chunk = ngx_http_push_stream_create_str(temp_pool, sizeof(CRLF) + len);
     if (chunk != NULL) {
-        ngx_sprintf(chunk->data, "%xO" CRLF "%*s" CRLF CRLF, len + sizeof(CRLF) - 1, (size_t) len, text);
+        ngx_sprintf(chunk->data, "%*s" CRLF, (size_t) len, text);
         chunk->len = ngx_strlen(chunk->data);
     }
     return chunk;