adding dynamic callback parameter to implement jsonp in a easy way

include/ngx_http_push_stream_module.h

@@ -182,6 +182,7 @@ typedef struct {
     ngx_chain_t                        *free;
     ngx_chain_t                        *busy;
     ngx_http_push_stream_padding_t     *padding;
+    ngx_str_t                          *callback;
 } ngx_http_push_stream_subscriber_ctx_t;
 
 // messages to worker processes
@@ -253,6 +254,7 @@ static const ngx_str_t NGX_HTTP_PUSH_STREAM_CHANNEL_DELETED = ngx_string("Channe
 static ngx_str_t        NGX_HTTP_PUSH_STREAM_EMPTY = ngx_string("");
 static const ngx_str_t  NGX_HTTP_PUSH_STREAM_BACKTRACK_SEP = ngx_string(".b");
 static const ngx_str_t  NGX_HTTP_PUSH_STREAM_SLASH = ngx_string("/");
+static const ngx_str_t  NGX_HTTP_PUSH_STREAM_CALLBACK = ngx_string("callback");
 
 static const ngx_str_t  NGX_HTTP_PUSH_STREAM_DATE_FORMAT_ISO_8601 = ngx_string("%4d-%02d-%02dT%02d:%02d:%02d");
 

include/ngx_http_push_stream_module_utils.h

@@ -209,6 +209,8 @@ 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_PADDING_BY_USER_AGENT_PATTERN = ngx_string("([^:]+),(\\d+),(\\d+)");
 

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.2");
-static const ngx_str_t  NGX_HTTP_PUSH_STREAM_COMMIT = ngx_string("842d509527c19bb844b1454788d518dec1321877");
+static const ngx_str_t  NGX_HTTP_PUSH_STREAM_COMMIT = ngx_string("395dad0853cde872b656f79580f4bc884d9fcec5");
 
 #endif /* NGX_HTTP_PUSH_STREAM_MODULE_VERSION_H_ */

src/ngx_http_push_stream_module_subscriber.c

@@ -142,6 +142,7 @@ ngx_http_push_stream_subscriber_polling_handler(ngx_http_request_t *r, ngx_http_
 {
     ngx_http_push_stream_loc_conf_t                *cf = ngx_http_get_module_loc_conf(r, ngx_http_push_stream_module);
     ngx_slab_pool_t                                *shpool = (ngx_slab_pool_t *)ngx_http_push_stream_shm_zone->shm.addr;
+    ngx_http_push_stream_subscriber_ctx_t          *ctx = ngx_http_get_module_ctx(r, ngx_http_push_stream_module);
     ngx_http_push_stream_requested_channel_t       *cur;
     ngx_http_push_stream_subscriber_t              *worker_subscriber;
     ngx_http_push_stream_channel_t                 *channel;
@@ -151,6 +152,7 @@ ngx_http_push_stream_subscriber_polling_handler(ngx_http_request_t *r, ngx_http_
     time_t                                          greater_message_time;
     ngx_int_t                                       greater_message_tag;
     ngx_flag_t                                      has_message_to_send = 0;
+    ngx_str_t                                       callback_function_name;
 
     if (cf->last_received_message_tag != NULL) {
         ngx_http_push_stream_complex_value(r, cf->last_received_message_tag, &vv_etag);
@@ -159,6 +161,14 @@ ngx_http_push_stream_subscriber_polling_handler(ngx_http_request_t *r, ngx_http_
         etag = ngx_http_push_stream_get_header(r, &NGX_HTTP_PUSH_STREAM_HEADER_IF_NONE_MATCH);
     }
 
+    if (ngx_http_arg(r, NGX_HTTP_PUSH_STREAM_CALLBACK.data, NGX_HTTP_PUSH_STREAM_CALLBACK.len, &callback_function_name) == NGX_OK) {
+        ngx_http_push_stream_unescape_uri(&callback_function_name);
+        if ((ctx->callback = ngx_http_push_stream_get_formatted_chunk(callback_function_name.data, callback_function_name.len, r->pool)) == NULL) {
+            ngx_log_error(NGX_LOG_ERR, r->connection->log, 0, "push stream module: unable to allocate memory for callback function name");
+            return NGX_HTTP_INTERNAL_SERVER_ERROR;
+        }
+    }
+
     tag = ((etag != NULL) && ((tag = ngx_atoi(etag->data, etag->len)) != NGX_ERROR)) ? ngx_abs(tag) : -1;
 
     greater_message_tag = tag;

src/ngx_http_push_stream_module_utils.c

@@ -432,7 +432,8 @@ 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_loc_conf_t *pslcf = ngx_http_get_module_loc_conf(r, ngx_http_push_stream_module);
+    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_int_t rc = NGX_OK;
 
     if (pslcf->eventsource_support) {
@@ -445,10 +446,24 @@ ngx_http_push_stream_send_response_message(ngx_http_request_t *r, ngx_http_push_
         }
     }
 
-    if (rc != NGX_ERROR) {
+    if (rc == NGX_OK) {
         ngx_str_t *str = ngx_http_push_stream_get_formatted_message(r, channel, msg, r->pool);
         if (str != NULL) {
-            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, 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);
+                }
+            }
+
+            if (rc == NGX_OK) {
+                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) {
                 rc = ngx_http_push_stream_send_response_padding(r, str->len, 0);
             }
@@ -1141,6 +1156,15 @@ ngx_http_push_stream_add_request_context(ngx_http_request_t *r)
         return NULL;
     }
 
+    ctx->busy = NULL;
+    ctx->free = NULL;
+    ctx->disconnect_timer = NULL;
+    ctx->ping_timer = NULL;
+    ctx->subscriber = NULL;
+    ctx->longpolling = 0;
+    ctx->padding = NULL;
+    ctx->callback = NULL;
+
     // set a cleaner to request
     cln->handler = (ngx_pool_cleanup_pt) ngx_http_push_stream_cleanup_request_context;
     cln->data = r;

test/test_subscriber_long_polling.rb

@@ -574,4 +574,25 @@ class TestSubscriberLongPolling < Test::Unit::TestCase
     }
   end
 
+  def test_return_message_using_function_name_specified_in_callback_parameter
+    headers = {'accept' => 'application/json'}
+    channel = 'ch_test_return_message_using_function_name_specified_in_callback_parameter'
+    body = 'body'
+    response = ""
+    callback_function_name = "callback_function"
+
+    EventMachine.run {
+
+      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")
+        EventMachine.stop
+      }
+
+      publish_message_inline(channel, {'accept' => 'text/html'}, body)
+
+      add_test_timeout
+    }
+  end
+
 end

test/test_subscriber_polling.rb

@@ -581,4 +581,24 @@ class TestSubscriberPolling < Test::Unit::TestCase
     }
   end
 
+  def test_return_message_using_function_name_specified_in_callback_parameter_when_polling
+    headers = {'accept' => 'application/json'}
+    channel = 'ch_test_return_message_using_function_name_specified_in_callback_parameter_when_polling'
+    body = 'body'
+    response = ""
+    callback_function_name = "callback_function"
+
+    EventMachine.run {
+      publish_message_inline(channel, {'accept' => 'text/html'}, body)
+
+      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")
+        EventMachine.stop
+      }
+
+      add_test_timeout
+    }
+  end
+
 end