[PATCH 07 of 15] Upstream: pipe length and input_filter_init in buffered mode

[Maxim Dounin]

# HG changeset patch
# User Maxim Dounin <mdounin at mdounin.ru>
# Date 1297818715 -10800
# Node ID 657ec1945f55f25b5fa4913b1acd2f8f310f9283
# Parent  3e730e31d8eb54db2d79079b3eadbcc8bba2e914
Upstream: pipe length and input_filter_init in buffered mode.

As long as ngx_event_pipe() has more data read from upstream than specified
in p->length it's passed to input filter even if buffer isn't yet full.  This
allows to process data with known length without relying on connection close
to signal data end.

By default p->length is set to -1 in upstream module, i.e. end of data is
indicated by connection close.  To set it from per-protocol handlers upstream
input_filter_init() now called in buffered mode (as well as in
unbuffered mode).

diff --git a/src/event/ngx_event_pipe.c b/src/event/ngx_event_pipe.c
--- a/src/event/ngx_event_pipe.c
+++ b/src/event/ngx_event_pipe.c
@@ -392,8 +392,31 @@ ngx_event_pipe_read_upstream(ngx_event_p
                        cl->buf->file_last - cl->buf->file_pos);
     }
 
+    ngx_log_debug1(NGX_LOG_DEBUG_EVENT, p->log, 0,
+                   "pipe length: %O", p->length);
+
 #endif
 
+    if (p->free_raw_bufs && p->length != -1) {
+        cl = p->free_raw_bufs;
+
+        if (cl->buf->last - cl->buf->pos >= p->length) {
+
+            /* STUB */ cl->buf->num = p->num++;
+
+            if (p->input_filter(p, cl->buf) == NGX_ERROR) {
+                 return NGX_ABORT;
+            }
+
+            p->free_raw_bufs = cl->next;
+        }
+    }
+
+    if (p->length == 0) {
+        p->upstream_done = 1;
+        p->read = 1;
+    }
+
     if ((p->upstream_eof || p->upstream_error) && p->free_raw_bufs) {
 
         /* STUB */ p->free_raw_bufs->buf->num = p->num++;
@@ -848,6 +871,12 @@ ngx_event_pipe_copy_input_filter(ngx_eve
     }
     p->last_in = &cl->next;
 
+    if (p->length == -1) {
+        return NGX_OK;
+    }
+
+    p->length -= b->last - b->pos;
+
     return NGX_OK;
 }
 
diff --git a/src/event/ngx_event_pipe.h b/src/event/ngx_event_pipe.h
--- a/src/event/ngx_event_pipe.h
+++ b/src/event/ngx_event_pipe.h
@@ -65,6 +65,7 @@ struct ngx_event_pipe_s {
     ssize_t            busy_size;
 
     off_t              read_length;
+    off_t              length;
 
     off_t              max_temp_file_size;
     ssize_t            temp_file_write_size;
diff --git a/src/http/ngx_http_upstream.c b/src/http/ngx_http_upstream.c
--- a/src/http/ngx_http_upstream.c
+++ b/src/http/ngx_http_upstream.c
@@ -2304,6 +2304,15 @@ ngx_http_upstream_send_response(ngx_http
     p->send_timeout = clcf->send_timeout;
     p->send_lowat = clcf->send_lowat;
 
+    p->length = -1;
+
+    if (u->input_filter_init
+        && u->input_filter_init(p->input_ctx) != NGX_OK)
+    {
+        ngx_http_upstream_finalize_request(r, u, 0);
+        return;
+    }
+
     u->read_event_handler = ngx_http_upstream_process_upstream;
     r->write_event_handler = ngx_http_upstream_process_downstream;