# HG changeset patch # User Maxim Dounin # Date 1316113247 0 # Node ID b66712cde67d3d7aaac61252df24336f3773f0f1 # Parent dbddec65fdab46e3d97a3fa0f84b4e8ede67b2b5 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;