apply pull openresty#4

Author: cryptofuture

src/ngx_http_rds_json_filter_module.c

@@ -243,6 +243,7 @@ ngx_http_rds_json_header_filter(ngx_http_request_t *r)
     ctx->tag = (ngx_buf_tag_t) &ngx_http_rds_json_filter_module;
 
     ctx->state = state_expect_header;
+    ctx->handler = ngx_http_rds_json_process_header;
 
     ctx->header_sent = 0;
 
@@ -299,6 +300,22 @@ ngx_http_rds_json_body_filter(ngx_http_request_t *r, ngx_chain_t *in)
     ngx_log_debug0(NGX_LOG_DEBUG_HTTP, r->connection->log, 0,
                    "rds json body filter postponed header sending");
 
+        if (ctx->handler) {
+            rc = ctx->handler(r, in, ctx);
+        } else {
+            /* status done */
+
+            ngx_log_debug0(NGX_LOG_DEBUG_HTTP, r->connection->log, 0,
+                    "rds json body filter discarding unexpected trailing buffers");
+
+            /* mark the remaining bufs as consumed */
+
+            ngx_http_rds_json_discard_bufs(r->pool, in);
+
+            return NGX_OK;
+        }
+
+#if 0
     switch (ctx->state) {
     case state_expect_header:
         rc = ngx_http_rds_json_process_header(r, in, ctx);
@@ -341,7 +358,7 @@ ngx_http_rds_json_body_filter(ngx_http_request_t *r, ngx_chain_t *in)
 
         break;
     }
-
+#endif
     dd("body filter rc: %d", (int) rc);
 
     if (rc == NGX_ERROR || rc >= NGX_HTTP_SPECIAL_RESPONSE) {

src/ngx_http_rds_json_filter_module.h

@@ -78,7 +78,12 @@ typedef enum {
 } ngx_http_rds_json_state_t;
 
 
-typedef struct {
+typedef struct ngx_http_rds_json_ctx_s  ngx_http_rds_json_ctx_t;
+
+typedef ngx_int_t (*rds_json_process_handler_pt)(ngx_http_request_t *r,
+        ngx_chain_t *in, ngx_http_rds_json_ctx_t *ctx);
+
+struct ngx_http_rds_json_ctx_s {
     ngx_http_rds_json_state_t            state;
 
     ngx_str_t                           *col_name;
@@ -106,10 +111,13 @@ typedef struct {
 
     uint32_t                             field_data_rest;
 
+    rds_json_process_handler_pt          handler;
+
     ngx_flag_t                           header_sent:1;
     ngx_flag_t                           seen_stream_end:1;
     ngx_flag_t                           generated_col_names:1;
-} ngx_http_rds_json_ctx_t;
+};
+
 
 
 #endif /* NGX_HTTP_RDS_JSON_FILTER_MODULE_H */

src/ngx_http_rds_json_processor.c

@@ -73,6 +73,7 @@ ngx_http_rds_json_process_header(ngx_http_request_t *r,
         }
 
         ctx->state = state_done;
+        ctx->handler = NULL;
 
         /* now we send the postponed response header */
         if (!ctx->header_sent) {
@@ -104,6 +105,7 @@ ngx_http_rds_json_process_header(ngx_http_request_t *r,
     }
 
     ctx->state = state_expect_col;
+    ctx->handler = ngx_http_rds_json_process_col;
     ctx->cur_col = 0;
     ctx->col_count = header.col_count;
 
@@ -189,6 +191,7 @@ ngx_http_rds_json_process_col(ngx_http_request_t *r, ngx_chain_t *in,
         dd("end of column list");
 
         ctx->state = state_expect_row;
+        ctx->handler = ngx_http_rds_json_process_row;
         ctx->row = 0;
 
         dd("output \"[\"");
@@ -285,6 +288,7 @@ ngx_http_rds_json_process_row(ngx_http_request_t *r, ngx_chain_t *in,
     if (*b->pos++ == 0) {
         /* end of row list */
         ctx->state = state_done;
+        ctx->handler = NULL;
 
         if (b->pos != b->last) {
             ngx_log_error(NGX_LOG_ERR, r->connection->log, 0,
@@ -318,6 +322,7 @@ ngx_http_rds_json_process_row(ngx_http_request_t *r, ngx_chain_t *in,
     ctx->row++;
     ctx->cur_col = 0;
     ctx->state = state_expect_field;
+    ctx->handler = ngx_http_rds_json_process_field;
 
     if (b->pos == b->last) {
         in = in->next;
@@ -417,6 +422,7 @@ ngx_http_rds_json_process_field(ngx_http_request_t *r, ngx_chain_t *in,
             dd("process field: need to read more field data");
 
             ctx->state = state_expect_more_field_data;
+            ctx->handler = ngx_http_rds_json_process_more_field_data;
 
             return ngx_http_rds_json_process_more_field_data(r, in, ctx);
         }
@@ -427,6 +433,7 @@ ngx_http_rds_json_process_field(ngx_http_request_t *r, ngx_chain_t *in,
             dd("reached the end of the current row");
 
             ctx->state = state_expect_row;
+            ctx->handler = ngx_http_rds_json_process_row;
 
             return ngx_http_rds_json_process_row(r, in, ctx);
         }
@@ -494,13 +501,15 @@ ngx_http_rds_json_process_more_field_data(ngx_http_request_t *r,
             dd("process more field data: reached the end of the current row");
 
             ctx->state = state_expect_row;
+            ctx->handler = ngx_http_rds_json_process_row;
 
             return ngx_http_rds_json_process_row(r, in, ctx);
         }
 
         dd("proces more field data: read the next field");
 
         ctx->state = state_expect_field;
+        ctx->handler = ngx_http_rds_json_process_field;
 
         return ngx_http_rds_json_process_field(r, in, ctx);
     }