diff --git a/src/ngx_http_rds_json_filter_module.c b/src/ngx_http_rds_json_filter_module.c
index c746607..a548f08 100644
--- a/src/ngx_http_rds_json_filter_module.c
+++ b/src/ngx_http_rds_json_filter_module.c
@@ -239,6 +239,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;
 
@@ -295,6 +296,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);
@@ -336,7 +353,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) {
diff --git a/src/ngx_http_rds_json_filter_module.h b/src/ngx_http_rds_json_filter_module.h
index 18019b6..db80c8a 100644
--- a/src/ngx_http_rds_json_filter_module.h
+++ b/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 */
diff --git a/src/ngx_http_rds_json_processor.c b/src/ngx_http_rds_json_processor.c
index 0400afb..1e1f9c0 100644
--- a/src/ngx_http_rds_json_processor.c
+++ b/src/ngx_http_rds_json_processor.c
@@ -71,6 +71,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) {
@@ -102,6 +103,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;
 
@@ -187,6 +189,7 @@ ngx_http_rds_json_process_col(ngx_http_request_t *r,
         dd("end of column list");
 
         ctx->state = state_expect_row;
+        ctx->handler = ngx_http_rds_json_process_row;
         ctx->row = 0;
 
         dd("output \"[\"");
@@ -284,6 +287,7 @@ ngx_http_rds_json_process_row(ngx_http_request_t *r,
     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,
@@ -313,6 +317,7 @@ ngx_http_rds_json_process_row(ngx_http_request_t *r,
     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;
@@ -412,6 +417,7 @@ ngx_http_rds_json_process_field(ngx_http_request_t *r,
             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);
         }
@@ -422,6 +428,7 @@ ngx_http_rds_json_process_field(ngx_http_request_t *r,
             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);
         }
@@ -491,6 +498,7 @@ 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);
         }
@@ -498,6 +506,7 @@ ngx_http_rds_json_process_more_field_data(ngx_http_request_t *r,
         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);
     }