Adding upstream version 2.4.59.upstream/2.4.59upstream

Signed-off-by: Daniel Baumann <daniel.baumann@progress-linux.org>

1 files changed, 1135 insertions, 385 deletions

diff --git a/modules/http2/h2_stream.c b/modules/http2/h2_stream.c
index 24ebc56..ee87555 100644
--- a/modules/http2/h2_stream.c
+++ b/modules/http2/h2_stream.c
@@ -17,34 +17,39 @@
 #include <assert.h>
 #include <stddef.h>
 
-#include <apr_strings.h>
+#include "apr.h"
+#include "apr_strings.h"
+#include "apr_lib.h"
+#include "apr_strmatch.h"
 
 #include <httpd.h>
 #include <http_core.h>
 #include <http_connection.h>
 #include <http_log.h>
+#include <http_protocol.h>
+#include <http_ssl.h>
 
 #include <nghttp2/nghttp2.h>
 
 #include "h2_private.h"
 #include "h2.h"
 #include "h2_bucket_beam.h"
-#include "h2_conn.h"
+#include "h2_c1.h"
 #include "h2_config.h"
-#include "h2_h2.h"
+#include "h2_protocol.h"
 #include "h2_mplx.h"
 #include "h2_push.h"
 #include "h2_request.h"
 #include "h2_headers.h"
 #include "h2_session.h"
 #include "h2_stream.h"
-#include "h2_task.h"
-#include "h2_ctx.h"
-#include "h2_task.h"
+#include "h2_c2.h"
+#include "h2_conn_ctx.h"
+#include "h2_c2.h"
 #include "h2_util.h"
 
 
-static const char *h2_ss_str(h2_stream_state_t state)
+static const char *h2_ss_str(const h2_stream_state_t state)
 {
     switch (state) {
         case H2_SS_IDLE:
@@ -68,7 +73,7 @@ static const char *h2_ss_str(h2_stream_state_t state)
     }
 }
 
-const char *h2_stream_state_str(h2_stream *stream) 
+const char *h2_stream_state_str(const h2_stream *stream)
 {
     return h2_ss_str(stream->state);
 }
@@ -120,7 +125,8 @@ static int trans_on_event[][H2_SS_MAX] = {
 { S_XXX, S_ERR,  S_ERR,  S_CL_L, S_CLS,  S_XXX,  S_XXX,  S_XXX, },/* EV_CLOSED_L*/
 { S_ERR, S_ERR,  S_ERR,  S_CL_R, S_ERR,  S_CLS,  S_NOP,  S_NOP, },/* EV_CLOSED_R*/
 { S_CLS, S_CLS,  S_CLS,  S_CLS,  S_CLS,  S_CLS,  S_NOP,  S_NOP, },/* EV_CANCELLED*/
-{ S_NOP, S_XXX,  S_XXX,  S_XXX,  S_XXX,  S_CLS,  S_CLN,  S_XXX, },/* EV_EOS_SENT*/
+{ S_NOP, S_XXX,  S_XXX,  S_XXX,  S_XXX,  S_CLS,  S_CLN,  S_NOP, },/* EV_EOS_SENT*/
+{ S_NOP, S_XXX,  S_CLS,  S_XXX,  S_XXX,  S_CLS,  S_XXX,  S_XXX, },/* EV_IN_ERROR*/
 };
 
 static int on_map(h2_stream_state_t state, int map[H2_SS_MAX])
@@ -142,7 +148,7 @@ static int on_frame(h2_stream_state_t state, int frame_type,
 {
     ap_assert(frame_type >= 0);
     ap_assert(state >= 0);
-    if (frame_type >= maxlen) {
+    if ((apr_size_t)frame_type >= maxlen) {
         return state; /* NOP, ignore unknown frame types */
     }
     return on_map(state, frame_map[frame_type]);
@@ -160,6 +166,7 @@ static int on_frame_recv(h2_stream_state_t state, int frame_type)
 
 static int on_event(h2_stream* stream, h2_stream_event_t ev)
 {
+    H2_STRM_ASSERT_MAGIC(stream, H2_STRM_MAGIC_OK);
     if (stream->monitor && stream->monitor->on_event) {
         stream->monitor->on_event(stream->monitor->ctx, stream, ev);
     }
@@ -169,10 +176,18 @@ static int on_event(h2_stream* stream, h2_stream_event_t ev)
     return stream->state;
 }
 
+static ssize_t stream_data_cb(nghttp2_session *ng2s,
+                              int32_t stream_id,
+                              uint8_t *buf,
+                              size_t length,
+                              uint32_t *data_flags,
+                              nghttp2_data_source *source,
+                              void *puser);
+
 static void H2_STREAM_OUT_LOG(int lvl, h2_stream *s, const char *tag)
 {
-    if (APLOG_C_IS_LEVEL(s->session->c, lvl)) {
-        conn_rec *c = s->session->c;
+    if (APLOG_C_IS_LEVEL(s->session->c1, lvl)) {
+        conn_rec *c = s->session->c1;
         char buffer[4 * 1024];
         apr_size_t len, bmax = sizeof(buffer)/sizeof(buffer[0]);
         
@@ -182,76 +197,116 @@ static void H2_STREAM_OUT_LOG(int lvl, h2_stream *s, const char *tag)
     }
 }
 
-static apr_status_t setup_input(h2_stream *stream) {
-    if (stream->input == NULL) {
-        int empty = (stream->input_eof 
-                     && (!stream->in_buffer 
-                         || APR_BRIGADE_EMPTY(stream->in_buffer)));
-        if (!empty) {
-            h2_beam_create(&stream->input, stream->pool, stream->id, 
-                           "input", H2_BEAM_OWNER_SEND, 0, 
-                           stream->session->s->timeout);
-            h2_beam_send_from(stream->input, stream->pool);
-        }
+static void stream_setup_input(h2_stream *stream)
+{
+    if (stream->input != NULL) return;
+    ap_assert(!stream->input_closed);
+    ap_log_cerror(APLOG_MARK, APLOG_TRACE2, 0, stream->session->c1,
+                  H2_STRM_MSG(stream, "setup input beam"));
+    h2_beam_create(&stream->input, stream->session->c1,
+                   stream->pool, stream->id,
+                   "input", 0, stream->session->s->timeout);
+}
+
+apr_status_t h2_stream_prepare_processing(h2_stream *stream)
+{
+    /* Right before processing starts, last chance to decide if
+     * there is need to an input beam. */
+    if (!stream->input_closed) {
+        stream_setup_input(stream);
     }
     return APR_SUCCESS;
 }
 
-static apr_status_t close_input(h2_stream *stream)
+static int input_buffer_is_empty(h2_stream *stream)
+{
+    return !stream->in_buffer || APR_BRIGADE_EMPTY(stream->in_buffer);
+}
+
+static apr_status_t input_flush(h2_stream *stream)
 {
-    conn_rec *c = stream->session->c;
     apr_status_t status = APR_SUCCESS;
+    apr_off_t written;
 
-    stream->input_eof = 1;
-    if (stream->input && h2_beam_is_closed(stream->input)) {
-        return APR_SUCCESS;
-    }
-    
-    ap_log_cerror(APLOG_MARK, APLOG_TRACE1, 0, stream->session->c,
-                  H2_STRM_MSG(stream, "closing input"));
-    if (stream->rst_error) {
-        return APR_ECONNRESET;
+    if (input_buffer_is_empty(stream)) goto cleanup;
+
+    ap_log_cerror(APLOG_MARK, APLOG_TRACE2, 0, stream->session->c1,
+                  H2_STRM_MSG(stream, "flush input"));
+    status = h2_beam_send(stream->input, stream->session->c1,
+                          stream->in_buffer, APR_BLOCK_READ, &written);
+    stream->in_last_write = apr_time_now();
+    if (APR_SUCCESS != status && h2_stream_is_at(stream, H2_SS_CLOSED_L)) {
+        ap_log_cerror(APLOG_MARK, APLOG_TRACE2, status, stream->session->c1,
+                      H2_STRM_MSG(stream, "send input error"));
+        h2_stream_dispatch(stream, H2_SEV_IN_ERROR);
     }
-    
-    if (stream->trailers && !apr_is_empty_table(stream->trailers)) {
-        apr_bucket *b;
-        h2_headers *r;
-        
-        if (!stream->in_buffer) {
-            stream->in_buffer = apr_brigade_create(stream->pool, c->bucket_alloc);
-        }
-        
-        r = h2_headers_create(HTTP_OK, stream->trailers, NULL, 
-            stream->in_trailer_octets, stream->pool);
-        stream->trailers = NULL;        
-        b = h2_bucket_headers_create(c->bucket_alloc, r);
-        APR_BRIGADE_INSERT_TAIL(stream->in_buffer, b);
-        
-        b = apr_bucket_eos_create(c->bucket_alloc);
-        APR_BRIGADE_INSERT_TAIL(stream->in_buffer, b);
-        
-        ap_log_cerror(APLOG_MARK, APLOG_TRACE2, 0, stream->session->c,
-                      H2_STRM_MSG(stream, "added trailers"));
-        h2_stream_dispatch(stream, H2_SEV_IN_DATA_PENDING);
+cleanup:
+    return status;
+}
+
+static void input_append_bucket(h2_stream *stream, apr_bucket *b)
+{
+    if (!stream->in_buffer) {
+        stream_setup_input(stream);
+        stream->in_buffer = apr_brigade_create(
+            stream->pool, stream->session->c1->bucket_alloc);
     }
-    if (stream->input) {
-        h2_stream_flush_input(stream);
-        return h2_beam_close(stream->input);
+    APR_BRIGADE_INSERT_TAIL(stream->in_buffer, b);
+}
+
+static void input_append_data(h2_stream *stream, const char *data, apr_size_t len)
+{
+    if (!stream->in_buffer) {
+        stream_setup_input(stream);
+        stream->in_buffer = apr_brigade_create(
+            stream->pool, stream->session->c1->bucket_alloc);
     }
-    return status;
+    apr_brigade_write(stream->in_buffer, NULL, NULL, data, len);
 }
 
-static apr_status_t close_output(h2_stream *stream)
+
+static apr_status_t close_input(h2_stream *stream)
 {
-    if (!stream->output || h2_beam_is_closed(stream->output)) {
-        return APR_SUCCESS;
+    conn_rec *c = stream->session->c1;
+    apr_status_t rv = APR_SUCCESS;
+    apr_bucket *b;
+
+    if (stream->input_closed) goto cleanup;
+
+    ap_log_cerror(APLOG_MARK, APLOG_TRACE1, 0, stream->session->c1,
+                  H2_STRM_MSG(stream, "closing input"));
+    if (!stream->rst_error
+        && stream->trailers_in
+        && !apr_is_empty_table(stream->trailers_in)) {
+        ap_log_cerror(APLOG_MARK, APLOG_TRACE2, 0, stream->session->c1,
+                      H2_STRM_MSG(stream, "adding trailers"));
+#if AP_HAS_RESPONSE_BUCKETS
+        b = ap_bucket_headers_create(stream->trailers_in,
+                                     stream->pool, c->bucket_alloc);
+#else
+        b = h2_bucket_headers_create(c->bucket_alloc,
+            h2_headers_create(HTTP_OK, stream->trailers_in, NULL,
+                              stream->in_trailer_octets, stream->pool));
+#endif
+        input_append_bucket(stream, b);
+        stream->trailers_in = NULL;
+    }
+
+    stream->input_closed = 1;
+    if (stream->input) {
+        b = apr_bucket_eos_create(c->bucket_alloc);
+        input_append_bucket(stream, b);
+        input_flush(stream);
+        h2_stream_dispatch(stream, H2_SEV_IN_DATA_PENDING);
+        ap_log_cerror(APLOG_MARK, APLOG_TRACE2, 0, stream->session->c1,
+                      H2_STRM_MSG(stream, "input flush + EOS"));
     }
-    ap_log_cerror(APLOG_MARK, APLOG_TRACE1, 0, stream->session->c,
-                  H2_STRM_MSG(stream, "closing output"));
-    return h2_beam_leave(stream->output);
+
+cleanup:
+    return rv;
 }
 
-static void on_state_enter(h2_stream *stream) 
+static void on_state_enter(h2_stream *stream)
 {
     if (stream->monitor && stream->monitor->on_state_enter) {
         stream->monitor->on_state_enter(stream->monitor->ctx, stream);
@@ -271,7 +326,7 @@ static void on_state_invalid(h2_stream *stream)
         stream->monitor->on_state_invalid(stream->monitor->ctx, stream);
     }
     /* stream got an event/frame invalid in its state */
-    ap_log_cerror(APLOG_MARK, APLOG_TRACE1, 0, stream->session->c,
+    ap_log_cerror(APLOG_MARK, APLOG_TRACE1, 0, stream->session->c1,
                   H2_STRM_MSG(stream, "invalid state event")); 
     switch (stream->state) {
         case H2_SS_OPEN:
@@ -288,17 +343,17 @@ static void on_state_invalid(h2_stream *stream)
 
 static apr_status_t transit(h2_stream *stream, int new_state)
 {
-    if (new_state == stream->state) {
+    if ((h2_stream_state_t)new_state == stream->state) {
         return APR_SUCCESS;
     }
     else if (new_state < 0) {
-        ap_log_cerror(APLOG_MARK, APLOG_WARNING, 0, stream->session->c, 
+        ap_log_cerror(APLOG_MARK, APLOG_WARNING, 0, stream->session->c1,
                       H2_STRM_LOG(APLOGNO(03081), stream, "invalid transition"));
         on_state_invalid(stream);
         return APR_EINVAL;
     }
     
-    ap_log_cerror(APLOG_MARK, APLOG_TRACE1, 0, stream->session->c, 
+    ap_log_cerror(APLOG_MARK, APLOG_TRACE1, 0, stream->session->c1,
                   H2_STRM_MSG(stream, "transit to [%s]"), h2_ss_str(new_state));
     stream->state = new_state;
     switch (new_state) {
@@ -312,14 +367,12 @@ static apr_status_t transit(h2_stream *stream, int new_state)
         case H2_SS_OPEN:
             break;
         case H2_SS_CLOSED_L:
-            close_output(stream);
             break;
         case H2_SS_CLOSED_R:
             close_input(stream);
             break;
         case H2_SS_CLOSED:
             close_input(stream);
-            close_output(stream);
             if (stream->out_buffer) {
                 apr_brigade_cleanup(stream->out_buffer);
             }
@@ -340,19 +393,20 @@ void h2_stream_dispatch(h2_stream *stream, h2_stream_event_t ev)
 {
     int new_state;
     
-    ap_log_cerror(APLOG_MARK, APLOG_TRACE2, 0, stream->session->c,
+    H2_STRM_ASSERT_MAGIC(stream, H2_STRM_MAGIC_OK);
+    ap_log_cerror(APLOG_MARK, APLOG_TRACE2, 0, stream->session->c1,
                   H2_STRM_MSG(stream, "dispatch event %d"), ev);
     new_state = on_event(stream, ev);
     if (new_state < 0) {
-        ap_log_cerror(APLOG_MARK, APLOG_WARNING, 0, stream->session->c, 
+        ap_log_cerror(APLOG_MARK, APLOG_WARNING, 0, stream->session->c1,
                       H2_STRM_LOG(APLOGNO(10002), stream, "invalid event %d"), ev);
         on_state_invalid(stream);
         AP_DEBUG_ASSERT(new_state > S_XXX);
         return;
     }
-    else if (new_state == stream->state) {
+    else if ((h2_stream_state_t)new_state == stream->state) {
         /* nop */
-        ap_log_cerror(APLOG_MARK, APLOG_TRACE2, 0, stream->session->c,
+        ap_log_cerror(APLOG_MARK, APLOG_TRACE2, 0, stream->session->c1,
                       H2_STRM_MSG(stream, "non-state event %d"), ev);
         return;
     }
@@ -365,9 +419,7 @@ void h2_stream_dispatch(h2_stream *stream, h2_stream_event_t ev)
 static void set_policy_for(h2_stream *stream, h2_request *r) 
 {
     int enabled = h2_session_push_enabled(stream->session);
-    stream->push_policy = h2_push_policy_determine(r->headers, stream->pool, 
-                                                   enabled);
-    r->serialize = h2_config_geti(stream->session->config, H2_CONF_SER_HEADERS);
+    stream->push_policy = h2_push_policy_determine(r->headers, stream->pool, enabled);
 }
 
 apr_status_t h2_stream_send_frame(h2_stream *stream, int ftype, int flags, size_t frame_len)
@@ -375,9 +427,10 @@ apr_status_t h2_stream_send_frame(h2_stream *stream, int ftype, int flags, size_
     apr_status_t status = APR_SUCCESS;
     int new_state, eos = 0;
 
+    H2_STRM_ASSERT_MAGIC(stream, H2_STRM_MAGIC_OK);
     new_state = on_frame_send(stream->state, ftype);
     if (new_state < 0) {
-        ap_log_cerror(APLOG_MARK, APLOG_TRACE1, 0, stream->session->c, 
+        ap_log_cerror(APLOG_MARK, APLOG_TRACE1, 0, stream->session->c1,
                       H2_STRM_MSG(stream, "invalid frame %d send"), ftype);
         AP_DEBUG_ASSERT(new_state > S_XXX);
         return transit(stream, new_state);
@@ -385,6 +438,12 @@ apr_status_t h2_stream_send_frame(h2_stream *stream, int ftype, int flags, size_
 
     ++stream->out_frames;
     stream->out_frame_octets += frame_len;
+    if(stream->c2) {
+      h2_conn_ctx_t *conn_ctx = h2_conn_ctx_get(stream->c2);
+      if(conn_ctx)
+        conn_ctx->bytes_sent = stream->out_frame_octets;
+    }
+
     switch (ftype) {
         case NGHTTP2_DATA:
             eos = (flags & NGHTTP2_FLAG_END_STREAM);
@@ -398,24 +457,18 @@ apr_status_t h2_stream_send_frame(h2_stream *stream, int ftype, int flags, size_
                 /* start pushed stream */
                 ap_assert(stream->request == NULL);
                 ap_assert(stream->rtmp != NULL);
-                status = h2_request_end_headers(stream->rtmp, stream->pool, 1, 0);
-                if (status != APR_SUCCESS) {
-                    return status;
-                }
-                set_policy_for(stream, stream->rtmp);
-                stream->request = stream->rtmp;
-                stream->rtmp = NULL;
+                status = h2_stream_end_headers(stream, 1, 0);
+                if (status != APR_SUCCESS) goto leave;
             break;
             
         default:
             break;
     }
-    ap_log_cerror(APLOG_MARK, APLOG_TRACE1, 0, stream->session->c, 
-                  H2_STRM_MSG(stream, "send frame %d, eos=%d"), ftype, eos);
     status = transit(stream, new_state);
     if (status == APR_SUCCESS && eos) {
         status = transit(stream, on_event(stream, H2_SEV_CLOSED_L));
     }
+leave:
     return status;
 }
 
@@ -424,9 +477,10 @@ apr_status_t h2_stream_recv_frame(h2_stream *stream, int ftype, int flags, size_
     apr_status_t status = APR_SUCCESS;
     int new_state, eos = 0;
 
+    H2_STRM_ASSERT_MAGIC(stream, H2_STRM_MAGIC_OK);
     new_state = on_frame_recv(stream->state, ftype);
     if (new_state < 0) {
-        ap_log_cerror(APLOG_MARK, APLOG_TRACE1, 0, stream->session->c, 
+        ap_log_cerror(APLOG_MARK, APLOG_TRACE1, 0, stream->session->c1,
                       H2_STRM_MSG(stream, "invalid frame %d recv"), ftype);
         AP_DEBUG_ASSERT(new_state > S_XXX);
         return transit(stream, new_state);
@@ -439,7 +493,7 @@ apr_status_t h2_stream_recv_frame(h2_stream *stream, int ftype, int flags, size_
             
         case NGHTTP2_HEADERS:
             eos = (flags & NGHTTP2_FLAG_END_STREAM);
-            if (stream->state == H2_SS_OPEN) {
+            if (h2_stream_is_at_or_past(stream, H2_SS_OPEN)) {
                 /* trailer HEADER */
                 if (!eos) {
                     h2_stream_rst(stream, H2_ERR_PROTOCOL_ERROR);
@@ -451,18 +505,13 @@ apr_status_t h2_stream_recv_frame(h2_stream *stream, int ftype, int flags, size_
                 ap_assert(stream->request == NULL);
                 if (stream->rtmp == NULL) {
                     /* This can only happen, if the stream has received no header
-                     * name/value pairs at all. The lastest nghttp2 version have become
+                     * name/value pairs at all. The latest nghttp2 version have become
                      * pretty good at detecting this early. In any case, we have
                      * to abort the connection here, since this is clearly a protocol error */
                     return APR_EINVAL;
                 }
-                status = h2_request_end_headers(stream->rtmp, stream->pool, eos, frame_len);
-                if (status != APR_SUCCESS) {
-                    return status;
-                }
-                set_policy_for(stream, stream->rtmp);
-                stream->request = stream->rtmp;
-                stream->rtmp = NULL;
+                status = h2_stream_end_headers(stream, eos, frame_len);
+                if (status != APR_SUCCESS) goto leave;
             }
             break;
             
@@ -473,22 +522,7 @@ apr_status_t h2_stream_recv_frame(h2_stream *stream, int ftype, int flags, size_
     if (status == APR_SUCCESS && eos) {
         status = transit(stream, on_event(stream, H2_SEV_CLOSED_R));
     }
-    return status;
-}
-
-apr_status_t h2_stream_flush_input(h2_stream *stream)
-{
-    apr_status_t status = APR_SUCCESS;
-    
-    if (stream->in_buffer && !APR_BRIGADE_EMPTY(stream->in_buffer)) {
-        setup_input(stream);
-        status = h2_beam_send(stream->input, stream->in_buffer, APR_BLOCK_READ);
-        stream->in_last_write = apr_time_now();
-    }
-    if (stream->input_eof 
-        && stream->input && !h2_beam_is_closed(stream->input)) {
-        status = h2_beam_close(stream->input);
-    }
+leave:
     return status;
 }
 
@@ -498,41 +532,59 @@ apr_status_t h2_stream_recv_DATA(h2_stream *stream, uint8_t flags,
     h2_session *session = stream->session;
     apr_status_t status = APR_SUCCESS;
     
+    H2_STRM_ASSERT_MAGIC(stream, H2_STRM_MAGIC_OK);
     stream->in_data_frames++;
     if (len > 0) {
-        if (APLOGctrace3(session->c)) {
+        if (APLOGctrace3(session->c1)) {
             const char *load = apr_pstrndup(stream->pool, (const char *)data, len);
-            ap_log_cerror(APLOG_MARK, APLOG_TRACE3, 0, session->c,
+            ap_log_cerror(APLOG_MARK, APLOG_TRACE3, 0, session->c1,
                           H2_STRM_MSG(stream, "recv DATA, len=%d: -->%s<--"), 
                           (int)len, load);
         }
         else {
-            ap_log_cerror(APLOG_MARK, APLOG_TRACE2, status, session->c,
+            ap_log_cerror(APLOG_MARK, APLOG_TRACE2, status, session->c1,
                           H2_STRM_MSG(stream, "recv DATA, len=%d"), (int)len);
         }
         stream->in_data_octets += len;
-        if (!stream->in_buffer) {
-            stream->in_buffer = apr_brigade_create(stream->pool, 
-                                                   session->c->bucket_alloc);
-        }
-        apr_brigade_write(stream->in_buffer, NULL, NULL, (const char *)data, len);
+        input_append_data(stream, (const char*)data, len);
+        input_flush(stream);
         h2_stream_dispatch(stream, H2_SEV_IN_DATA_PENDING);
     }
     return status;
 }
 
-static void prep_output(h2_stream *stream) {
-    conn_rec *c = stream->session->c;
-    if (!stream->out_buffer) {
-        stream->out_buffer = apr_brigade_create(stream->pool, c->bucket_alloc);
+#ifdef AP_DEBUG
+static apr_status_t stream_pool_destroy(void *data)
+{
+    h2_stream *stream = data;
+    switch (stream->magic) {
+    case H2_STRM_MAGIC_OK:
+        ap_log_cerror(APLOG_MARK, APLOG_ERR, 0, stream->session->c1,
+                      H2_STRM_MSG(stream, "was not destroyed explicitly"));
+        AP_DEBUG_ASSERT(0);
+        break;
+    case H2_STRM_MAGIC_SDEL:
+        /* stream has been explicitly destroyed, as it should */
+        H2_STRM_ASSIGN_MAGIC(stream, H2_STRM_MAGIC_PDEL);
+        break;
+    case H2_STRM_MAGIC_PDEL:
+        ap_log_cerror(APLOG_MARK, APLOG_ERR, 0, stream->session->c1,
+                      H2_STRM_MSG(stream, "already pool destroyed"));
+        AP_DEBUG_ASSERT(0);
+        break;
+    default:
+        AP_DEBUG_ASSERT(0);
     }
+    return APR_SUCCESS;
 }
+#endif
 
 h2_stream *h2_stream_create(int id, apr_pool_t *pool, h2_session *session,
                             h2_stream_monitor *monitor, int initiated_on)
 {
     h2_stream *stream = apr_pcalloc(pool, sizeof(h2_stream));
-    
+
+    H2_STRM_ASSIGN_MAGIC(stream, H2_STRM_MAGIC_OK);
     stream->id           = id;
     stream->initiated_on = initiated_on;
     stream->created      = apr_time_now();
@@ -540,15 +592,21 @@ h2_stream *h2_stream_create(int id, apr_pool_t *pool, h2_session *session,
     stream->pool         = pool;
     stream->session      = session;
     stream->monitor      = monitor;
-    stream->max_mem      = session->max_stream_mem;
-    
-#ifdef H2_NG2_LOCAL_WIN_SIZE
-    stream->in_window_size = 
-        nghttp2_session_get_stream_local_window_size(
-            stream->session->ngh2, stream->id);
+#ifdef AP_DEBUG
+    if (id) { /* stream 0 has special lifetime */
+        apr_pool_cleanup_register(pool, stream, stream_pool_destroy,
+                                  apr_pool_cleanup_null);
+    }
 #endif
 
-    ap_log_cerror(APLOG_MARK, APLOG_DEBUG, 0, session->c, 
+#ifdef H2_NG2_LOCAL_WIN_SIZE
+    if (id) {
+        stream->in_window_size =
+            nghttp2_session_get_stream_local_window_size(
+                stream->session->ngh2, stream->id);
+    }
+#endif
+    ap_log_cerror(APLOG_MARK, APLOG_DEBUG, 0, session->c1,
                   H2_STRM_LOG(APLOGNO(03082), stream, "created"));
     on_state_enter(stream);
     return stream;
@@ -556,59 +614,35 @@ h2_stream *h2_stream_create(int id, apr_pool_t *pool, h2_session *session,
 
 void h2_stream_cleanup(h2_stream *stream)
 {
-    apr_status_t status;
-    
+    /* Stream is done on c1. There might still be processing on a c2
+     * going on. The input/output beams get aborted and the stream's
+     * end of the in/out notifications get closed.
+     */
     ap_assert(stream);
+    H2_STRM_ASSERT_MAGIC(stream, H2_STRM_MAGIC_OK);
     if (stream->out_buffer) {
-        /* remove any left over output buckets that may still have
-         * references into request pools */
         apr_brigade_cleanup(stream->out_buffer);
     }
-    if (stream->input) {
-        h2_beam_abort(stream->input);
-        status = h2_beam_wait_empty(stream->input, APR_NONBLOCK_READ);
-        if (status == APR_EAGAIN) {
-            ap_log_cerror(APLOG_MARK, APLOG_TRACE2, 0, stream->session->c, 
-                          H2_STRM_MSG(stream, "wait on input drain"));
-            status = h2_beam_wait_empty(stream->input, APR_BLOCK_READ);
-            ap_log_cerror(APLOG_MARK, APLOG_TRACE2, status, stream->session->c, 
-                          H2_STRM_MSG(stream, "input drain returned"));
-        }
-    }
 }
 
 void h2_stream_destroy(h2_stream *stream)
 {
     ap_assert(stream);
-    ap_log_cerror(APLOG_MARK, APLOG_TRACE3, 0, stream->session->c, 
+    H2_STRM_ASSERT_MAGIC(stream, H2_STRM_MAGIC_OK);
+    ap_log_cerror(APLOG_MARK, APLOG_TRACE3, 0, stream->session->c1,
                   H2_STRM_MSG(stream, "destroy"));
+    H2_STRM_ASSIGN_MAGIC(stream, H2_STRM_MAGIC_SDEL);
     apr_pool_destroy(stream->pool);
 }
 
-apr_status_t h2_stream_prep_processing(h2_stream *stream)
-{
-    if (stream->request) {
-        const h2_request *r = stream->request;
-        ap_log_cerror(APLOG_MARK, APLOG_TRACE1, 0, stream->session->c,
-                      H2_STRM_MSG(stream, "schedule %s %s://%s%s chunked=%d"),
-                      r->method, r->scheme, r->authority, r->path, r->chunked);
-        setup_input(stream);
-        stream->scheduled = 1;
-        return APR_SUCCESS;
-    }
-    return APR_EINVAL;
-}
-
 void h2_stream_rst(h2_stream *stream, int error_code)
 {
+    H2_STRM_ASSERT_MAGIC(stream, H2_STRM_MAGIC_OK);
     stream->rst_error = error_code;
-    if (stream->input) {
-        h2_beam_abort(stream->input);
-    }
-    if (stream->output) {
-        h2_beam_leave(stream->output);
+    if (stream->c2) {
+        h2_c2_abort(stream->c2, stream->session->c1);
     }
-    ap_log_cerror(APLOG_MARK, APLOG_TRACE1, 0, stream->session->c,
+    ap_log_cerror(APLOG_MARK, APLOG_TRACE1, 0, stream->session->c1,
                   H2_STRM_MSG(stream, "reset, error=%d"), error_code);
     h2_stream_dispatch(stream, H2_SEV_CANCELLED);
 }
@@ -619,6 +653,7 @@ apr_status_t h2_stream_set_request_rec(h2_stream *stream,
     h2_request *req;
     apr_status_t status;
 
+    H2_STRM_ASSERT_MAGIC(stream, H2_STRM_MAGIC_OK);
     ap_assert(stream->request == NULL);
     ap_assert(stream->rtmp == NULL);
     if (stream->rst_error) {
@@ -640,6 +675,7 @@ apr_status_t h2_stream_set_request_rec(h2_stream *stream,
 
 void h2_stream_set_request(h2_stream *stream, const h2_request *r)
 {
+    H2_STRM_ASSERT_MAGIC(stream, H2_STRM_MAGIC_OK);
     ap_assert(stream->request == NULL);
     ap_assert(stream->rtmp == NULL);
     stream->rtmp = h2_request_clone(stream->pool, r);
@@ -647,43 +683,46 @@ void h2_stream_set_request(h2_stream *stream, const h2_request *r)
 
 static void set_error_response(h2_stream *stream, int http_status)
 {
-    if (!h2_stream_is_ready(stream)) {
-        conn_rec *c = stream->session->c;
-        apr_bucket *b;
-        h2_headers *response;
-        
-        response = h2_headers_die(http_status, stream->request, stream->pool);
-        prep_output(stream);
-        b = apr_bucket_eos_create(c->bucket_alloc);
-        APR_BRIGADE_INSERT_HEAD(stream->out_buffer, b);
-        b = h2_bucket_headers_create(c->bucket_alloc, response);
-        APR_BRIGADE_INSERT_HEAD(stream->out_buffer, b);
+    if (!h2_stream_is_ready(stream) && stream->rtmp) {
+        stream->rtmp->http_status = http_status;
     }
 }
 
 static apr_status_t add_trailer(h2_stream *stream,
                                 const char *name, size_t nlen,
-                                const char *value, size_t vlen)
+                                const char *value, size_t vlen,
+                                size_t max_field_len, int *pwas_added)
 {
-    conn_rec *c = stream->session->c;
+    conn_rec *c = stream->session->c1;
     char *hname, *hvalue;
+    const char *existing;
 
+    *pwas_added = 0;
     if (nlen == 0 || name[0] == ':') {
         ap_log_cerror(APLOG_MARK, APLOG_DEBUG, APR_EINVAL, c, 
                       H2_STRM_LOG(APLOGNO(03060), stream, 
                       "pseudo header in trailer"));
         return APR_EINVAL;
     }
-    if (h2_req_ignore_trailer(name, nlen)) {
+    if (h2_ignore_req_trailer(name, nlen)) {
         return APR_SUCCESS;
     }
-    if (!stream->trailers) {
-        stream->trailers = apr_table_make(stream->pool, 5);
+    if (!stream->trailers_in) {
+        stream->trailers_in = apr_table_make(stream->pool, 5);
     }
     hname = apr_pstrndup(stream->pool, name, nlen);
-    hvalue = apr_pstrndup(stream->pool, value, vlen);
     h2_util_camel_case_header(hname, nlen);
-    apr_table_mergen(stream->trailers, hname, hvalue);
+    existing = apr_table_get(stream->trailers_in, hname);
+    if (max_field_len 
+        && ((existing? strlen(existing)+2 : 0) + vlen + nlen + 2 > max_field_len)) {
+        /* "key: (oldval, )?nval" is too long */
+        return APR_EINVAL;
+    }
+    if (!existing) *pwas_added = 1;
+    hvalue = apr_pstrndup(stream->pool, value, vlen);
+    apr_table_mergen(stream->trailers_in, hname, hvalue);
+    ap_log_cerror(APLOG_MARK, APLOG_TRACE2, 0, c, 
+                  H2_STRM_MSG(stream, "added trailer '%s: %s'"), hname, hvalue);
     
     return APR_SUCCESS;
 }
@@ -693,274 +732,487 @@ apr_status_t h2_stream_add_header(h2_stream *stream,
                                   const char *value, size_t vlen)
 {
     h2_session *session = stream->session;
-    int error = 0;
-    apr_status_t status;
+    int error = 0, was_added = 0;
+    apr_status_t status = APR_SUCCESS;
     
-    if (stream->has_response) {
+    H2_STRM_ASSERT_MAGIC(stream, H2_STRM_MAGIC_OK);
+    if (stream->response) {
         return APR_EINVAL;    
     }
-    ++stream->request_headers_added;
+
     if (name[0] == ':') {
-        if ((vlen) > session->s->limit_req_line) {
+        if (vlen > APR_INT32_MAX || (int)vlen > session->s->limit_req_line) {
             /* pseudo header: approximation of request line size check */
-            ap_log_cerror(APLOG_MARK, APLOG_TRACE1, 0, session->c,
-                          H2_STRM_MSG(stream, "pseudo %s too long"), name);
+            if (!h2_stream_is_ready(stream)) {
+                ap_log_cerror(APLOG_MARK, APLOG_INFO, 0, session->c1,
+                              H2_STRM_LOG(APLOGNO(10178), stream,
+                                          "Request pseudo header exceeds "
+                                          "LimitRequestFieldSize: %s"), name);
+            }
             error = HTTP_REQUEST_URI_TOO_LARGE;
+            goto cleanup;
         }
     }
-    else if ((nlen + 2 + vlen) > session->s->limit_req_fieldsize) {
+    
+    if (session->s->limit_req_fields > 0 
+        && stream->request_headers_added > session->s->limit_req_fields) {
+        /* already over limit, count this attempt, but do not take it in */
+        ++stream->request_headers_added;
+    }
+    else if (H2_SS_IDLE == stream->state) {
+        if (!stream->rtmp) {
+            stream->rtmp = h2_request_create(stream->id, stream->pool,
+                                             NULL, NULL, NULL, NULL, NULL);
+        }
+        status = h2_request_add_header(stream->rtmp, stream->pool,
+                                       name, nlen, value, vlen,
+                                       session->s->limit_req_fieldsize, &was_added);
+        ap_log_cerror(APLOG_MARK, APLOG_TRACE2, status, session->c1,
+                      H2_STRM_MSG(stream, "add_header: '%.*s: %.*s"),
+                      (int)nlen, name, (int)vlen, value);
+        if (was_added) ++stream->request_headers_added;
+    }
+    else if (H2_SS_OPEN == stream->state) {
+        status = add_trailer(stream, name, nlen, value, vlen,
+                             session->s->limit_req_fieldsize, &was_added);
+        if (was_added) ++stream->request_headers_added;
+    }
+    else {
+        status = APR_EINVAL;
+        goto cleanup;
+    }
+    
+    if (APR_EINVAL == status) {
         /* header too long */
-        ap_log_cerror(APLOG_MARK, APLOG_TRACE1, 0, session->c,
-                      H2_STRM_MSG(stream, "header %s too long"), name);
+        if (!h2_stream_is_ready(stream)) {
+            ap_log_cerror(APLOG_MARK, APLOG_INFO, 0, session->c1,
+                          H2_STRM_LOG(APLOGNO(10180), stream,"Request header exceeds "
+                                      "LimitRequestFieldSize: %.*s"),
+                          (int)H2MIN(nlen, 80), name);
+        }
         error = HTTP_REQUEST_HEADER_FIELDS_TOO_LARGE;
+        goto cleanup;
     }
     
-    if (stream->request_headers_added > session->s->limit_req_fields + 4) {
-        /* too many header lines, include 4 pseudo headers */
-        if (stream->request_headers_added 
-            > session->s->limit_req_fields + 4 + 100) {
-            /* yeah, right */
+    if (session->s->limit_req_fields > 0 
+        && stream->request_headers_added > session->s->limit_req_fields) {
+        /* too many header lines */
+        if (stream->request_headers_added > session->s->limit_req_fields + 100) {
+            /* yeah, right, this request is way over the limit, say goodbye */
             h2_stream_rst(stream, H2_ERR_ENHANCE_YOUR_CALM);
             return APR_ECONNRESET;
         }
-        ap_log_cerror(APLOG_MARK, APLOG_TRACE1, 0, session->c,
-                      H2_STRM_MSG(stream, "too many header lines")); 
+        if (!h2_stream_is_ready(stream)) {
+            ap_log_cerror(APLOG_MARK, APLOG_INFO, 0, session->c1,
+                          H2_STRM_LOG(APLOGNO(10181), stream, "Number of request headers "
+                                      "exceeds LimitRequestFields"));
+        }
         error = HTTP_REQUEST_HEADER_FIELDS_TOO_LARGE;
+        goto cleanup;
     }
     
+cleanup:
     if (error) {
+        ++stream->request_headers_failed;
         set_error_response(stream, error);
         return APR_EINVAL; 
     }
-    else if (H2_SS_IDLE == stream->state) {
-        if (!stream->rtmp) {
-            stream->rtmp = h2_req_create(stream->id, stream->pool, 
-                                         NULL, NULL, NULL, NULL, NULL, 0);
-        }
-        status = h2_request_add_header(stream->rtmp, stream->pool,
-                                       name, nlen, value, vlen);
+    else if (status != APR_SUCCESS) {
+        ap_log_cerror(APLOG_MARK, APLOG_TRACE1, 0, session->c1,
+                      H2_STRM_MSG(stream, "header %s not accepted"), name);
+        h2_stream_dispatch(stream, H2_SEV_CANCELLED);
     }
-    else if (H2_SS_OPEN == stream->state) {
-        status = add_trailer(stream, name, nlen, value, vlen);
+    return status;
+}
+
+typedef struct {
+    apr_size_t maxlen;
+    const char *failed_key;
+} val_len_check_ctx;
+
+static int table_check_val_len(void *baton, const char *key, const char *value)
+{
+    val_len_check_ctx *ctx = baton;
+
+    if (strlen(value) <= ctx->maxlen) return 1;
+    ctx->failed_key = key;
+    return 0;
+}
+
+apr_status_t h2_stream_end_headers(h2_stream *stream, int eos, size_t raw_bytes)
+{
+    apr_status_t status;
+    val_len_check_ctx ctx;
+    int is_http_or_https;
+    h2_request *req = stream->rtmp;
+
+    H2_STRM_ASSERT_MAGIC(stream, H2_STRM_MAGIC_OK);
+    status = h2_request_end_headers(req, stream->pool, raw_bytes);
+    if (APR_SUCCESS != status || req->http_status != H2_HTTP_STATUS_UNSET) {
+        goto cleanup;
+    }
+
+    /* keep on returning APR_SUCCESS for error responses, so that we
+     * send it and do not RST the stream.
+     */
+    set_policy_for(stream, req);
+
+    ctx.maxlen = stream->session->s->limit_req_fieldsize;
+    ctx.failed_key = NULL;
+    apr_table_do(table_check_val_len, &ctx, req->headers, NULL);
+    if (ctx.failed_key) {
+        if (!h2_stream_is_ready(stream)) {
+            ap_log_cerror(APLOG_MARK, APLOG_INFO, 0, stream->session->c1,
+                          H2_STRM_LOG(APLOGNO(10230), stream,"Request header exceeds "
+                                      "LimitRequestFieldSize: %.*s"),
+                          (int)H2MIN(strlen(ctx.failed_key), 80), ctx.failed_key);
+        }
+        set_error_response(stream, HTTP_REQUEST_HEADER_FIELDS_TOO_LARGE);
+        goto cleanup;
+    }
+
+    /* http(s) scheme. rfc7540, ch. 8.1.2.3:
+     * This [:path] pseudo-header field MUST NOT be empty for "http" or "https"
+     * URIs; "http" or "https" URIs that do not contain a path component
+     * MUST include a value of '/'.  The exception to this rule is an
+     * OPTIONS request for an "http" or "https" URI that does not include
+     * a path component; these MUST include a ":path" pseudo-header field
+     * with a value of '*'
+     *
+     * All HTTP/2 requests MUST include exactly one valid value for the
+     * ":method", ":scheme", and ":path" pseudo-header fields, unless it is
+     * a CONNECT request.
+     */
+    is_http_or_https = (!req->scheme
+            || !(ap_cstr_casecmpn(req->scheme, "http", 4) != 0
+                 || (req->scheme[4] != '\0'
+                     && (apr_tolower(req->scheme[4]) != 's'
+                         || req->scheme[5] != '\0'))));
+
+    /* CONNECT. rfc7540, ch. 8.3:
+     * In HTTP/2, the CONNECT method is used to establish a tunnel over a
+     * single HTTP/2 stream to a remote host for similar purposes.  The HTTP
+     * header field mapping works as defined in Section 8.1.2.3 ("Request
+     * Pseudo-Header Fields"), with a few differences.  Specifically:
+     *   o  The ":method" pseudo-header field is set to "CONNECT".
+     *   o  The ":scheme" and ":path" pseudo-header fields MUST be omitted.
+     *   o  The ":authority" pseudo-header field contains the host and port to
+     *      connect to (equivalent to the authority-form of the request-target
+     *      of CONNECT requests (see [RFC7230], Section 5.3)).
+     */
+    if (!ap_cstr_casecmp(req->method, "CONNECT")) {
+        if (req->protocol) {
+            if (!strcmp("websocket", req->protocol)) {
+                if (!req->scheme || !req->path) {
+                    ap_log_cerror(APLOG_MARK, APLOG_INFO, 0, stream->session->c1,
+                                  H2_STRM_LOG(APLOGNO(10457), stream, "Request to websocket CONNECT "
+                                  "without :scheme or :path, sending 400 answer"));
+                    set_error_response(stream, HTTP_BAD_REQUEST);
+                    goto cleanup;
+                }
+            }
+            else {
+                /* do not know that protocol */
+                ap_log_cerror(APLOG_MARK, APLOG_DEBUG, 0, stream->session->c1, APLOGNO(10460)
+                              "':protocol: %s' header present in %s request",
+                              req->protocol, req->method);
+                set_error_response(stream, HTTP_NOT_IMPLEMENTED);
+                goto cleanup;
+            }
+        }
+        else if (req->scheme || req->path) {
+            ap_log_cerror(APLOG_MARK, APLOG_INFO, 0, stream->session->c1,
+                          H2_STRM_LOG(APLOGNO(10384), stream, "Request to CONNECT "
+                          "with :scheme or :path specified, sending 400 answer"));
+            set_error_response(stream, HTTP_BAD_REQUEST);
+            goto cleanup;
+        }
     }
-    else {
-        status = APR_EINVAL;
+    else if (is_http_or_https) {
+        if (!req->path) {
+            ap_log_cerror(APLOG_MARK, APLOG_INFO, 0, stream->session->c1,
+                          H2_STRM_LOG(APLOGNO(10385), stream, "Request for http(s) "
+                          "resource without :path, sending 400 answer"));
+            set_error_response(stream, HTTP_BAD_REQUEST);
+            goto cleanup;
+        }
+        if (!req->scheme) {
+            req->scheme = ap_ssl_conn_is_ssl(stream->session->c1)? "https" : "http";
+        }
     }
-    
-    if (status != APR_SUCCESS) {
-        ap_log_cerror(APLOG_MARK, APLOG_TRACE1, 0, session->c,
-                      H2_STRM_MSG(stream, "header %s not accepted"), name);
-        h2_stream_dispatch(stream, H2_SEV_CANCELLED);
+
+    if (req->scheme && (req->path && req->path[0] != '/')) {
+        /* We still have a scheme, which means we need to pass an absolute URI into
+         * our HTTP protocol handling and the missing '/' at the start will prevent
+         * us from doing so (as it then confuses path and authority). */
+        ap_log_cerror(APLOG_MARK, APLOG_INFO, 0, stream->session->c1,
+                      H2_STRM_LOG(APLOGNO(10379), stream, "Request :scheme '%s' and "
+                      "path '%s' do not allow creating an absolute URL. Failing "
+                      "request with 400."), req->scheme, req->path);
+        set_error_response(stream, HTTP_BAD_REQUEST);
+        goto cleanup;
+    }
+
+cleanup:
+    if (APR_SUCCESS == status) {
+        stream->request = req;
+        stream->rtmp = NULL;
+
+        if (APLOGctrace4(stream->session->c1)) {
+            int i;
+            const apr_array_header_t *t_h = apr_table_elts(req->headers);
+            const apr_table_entry_t *t_elt = (apr_table_entry_t *)t_h->elts;
+            ap_log_cerror(APLOG_MARK, APLOG_TRACE4, 0, stream->session->c1,
+                          H2_STRM_MSG(stream,"headers received from client:"));
+            for (i = 0; i < t_h->nelts; i++, t_elt++) {
+                ap_log_cerror(APLOG_MARK, APLOG_TRACE4, 0, stream->session->c1,
+                              H2_STRM_MSG(stream, "  %s: %s"),
+                              ap_escape_logitem(stream->pool, t_elt->key),
+                              ap_escape_logitem(stream->pool, t_elt->val));
+            }
+        }
     }
     return status;
 }
 
-static apr_bucket *get_first_headers_bucket(apr_bucket_brigade *bb)
+static apr_bucket *get_first_response_bucket(apr_bucket_brigade *bb)
 {
     if (bb) {
         apr_bucket *b = APR_BRIGADE_FIRST(bb);
         while (b != APR_BRIGADE_SENTINEL(bb)) {
+#if AP_HAS_RESPONSE_BUCKETS
+            if (AP_BUCKET_IS_RESPONSE(b)) {
+                return b;
+            }
+#else
             if (H2_BUCKET_IS_HEADERS(b)) {
                 return b;
             }
+#endif
             b = APR_BUCKET_NEXT(b);
         }
     }
     return NULL;
 }
 
-static apr_status_t add_buffered_data(h2_stream *stream, apr_off_t requested,
-                                      apr_off_t *plen, int *peos, int *is_all, 
-                                      h2_headers **pheaders)
+static void stream_do_error_bucket(h2_stream *stream, apr_bucket *b)
 {
-    apr_bucket *b, *e;
-    
-    *peos = 0;
-    *plen = 0;
-    *is_all = 0;
-    if (pheaders) {
-        *pheaders = NULL;
+    int err = ((ap_bucket_error *)(b->data))->status;
+    ap_log_cerror(APLOG_MARK, APLOG_TRACE1, 0, stream->session->c1,
+                  H2_STRM_MSG(stream, "error bucket received, err=%d"), err);
+    if (err >= 500) {
+        err = NGHTTP2_INTERNAL_ERROR;
     }
-
-    H2_STREAM_OUT_LOG(APLOG_TRACE2, stream, "add_buffered_data");
-    b = APR_BRIGADE_FIRST(stream->out_buffer);
-    while (b != APR_BRIGADE_SENTINEL(stream->out_buffer)) {
-        e = APR_BUCKET_NEXT(b);
-        if (APR_BUCKET_IS_METADATA(b)) {
-            if (APR_BUCKET_IS_FLUSH(b)) {
-                APR_BUCKET_REMOVE(b);
-                apr_bucket_destroy(b);
-            }
-            else if (APR_BUCKET_IS_EOS(b)) {
-                *peos = 1;
-                return APR_SUCCESS;
-            }
-            else if (H2_BUCKET_IS_HEADERS(b)) {
-                if (*plen > 0) {
-                    /* data before the response, can only return up to here */
-                    return APR_SUCCESS;
-                }
-                else if (pheaders) {
-                    *pheaders = h2_bucket_headers_get(b);
-                    APR_BUCKET_REMOVE(b);
-                    apr_bucket_destroy(b);
-                    ap_log_cerror(APLOG_MARK, APLOG_TRACE1, 0, stream->session->c,
-                                  H2_STRM_MSG(stream, "prep, -> response %d"), 
-                                  (*pheaders)->status);
-                    return APR_SUCCESS;
-                }
-                else {
-                    return APR_EAGAIN;
-                }
-            }
-        }
-        else if (b->length == 0) {
-            APR_BUCKET_REMOVE(b);
-            apr_bucket_destroy(b);
-        }
-        else {
-            ap_assert(b->length != (apr_size_t)-1);
-            *plen += b->length;
-            if (*plen >= requested) {
-                *plen = requested;
-                return APR_SUCCESS;
-            }
-        }
-        b = e;
+    else if (err >= 400) {
+        err = NGHTTP2_STREAM_CLOSED;
     }
-    *is_all = 1;
-    return APR_SUCCESS;
+    else {
+        err = NGHTTP2_PROTOCOL_ERROR;
+    }
+    h2_stream_rst(stream, err);
 }
 
-apr_status_t h2_stream_out_prepare(h2_stream *stream, apr_off_t *plen, 
-                                   int *peos, h2_headers **pheaders)
+static apr_status_t buffer_output_receive(h2_stream *stream)
 {
-    apr_status_t status = APR_SUCCESS;
-    apr_off_t requested, missing, max_chunk = H2_DATA_CHUNK_SIZE;
-    conn_rec *c;
-    int complete;
+    apr_status_t rv = APR_EAGAIN;
+    apr_off_t buf_len;
+    conn_rec *c1 = stream->session->c1;
+    apr_bucket *b, *e;
 
-    ap_assert(stream);
-    
+    if (!stream->output) {
+        goto cleanup;
+    }
     if (stream->rst_error) {
-        *plen = 0;
-        *peos = 1;
-        return APR_ECONNRESET;
+        rv = APR_ECONNRESET;
+        goto cleanup;
     }
-    
-    c = stream->session->c;
-    prep_output(stream);
 
-    /* determine how much we'd like to send. We cannot send more than
-     * is requested. But we can reduce the size in case the master
-     * connection operates in smaller chunks. (TSL warmup) */
-    if (stream->session->io.write_size > 0) {
-        max_chunk = stream->session->io.write_size - 9; /* header bits */ 
+    if (!stream->out_buffer) {
+        stream->out_buffer = apr_brigade_create(stream->pool, c1->bucket_alloc);
+        buf_len = 0;
     }
-    requested = (*plen > 0)? H2MIN(*plen, max_chunk) : max_chunk;
-    
-    /* count the buffered data until eos or a headers bucket */
-    status = add_buffered_data(stream, requested, plen, peos, &complete, pheaders);
-    
-    if (status == APR_EAGAIN) {
-        /* TODO: ugly, someone needs to retrieve the response first */
-        h2_mplx_keep_active(stream->session->mplx, stream);
-        ap_log_cerror(APLOG_MARK, APLOG_TRACE1, 0, c,
-                      H2_STRM_MSG(stream, "prep, response eagain"));
-        return status;
+    else {
+        /* if the brigade contains a file bucket, its normal report length
+         * might be megabytes, but the memory used is tiny. For buffering,
+         * we are only interested in the memory footprint. */
+        buf_len = h2_brigade_mem_size(stream->out_buffer);
     }
-    else if (status != APR_SUCCESS) {
-        return status;
+
+    if (buf_len > APR_INT32_MAX
+        || (apr_size_t)buf_len >= stream->session->max_stream_mem) {
+        /* we have buffered enough. No need to read more.
+         * However, we have now output pending for which we may not
+         * receive another poll event. We need to make sure that this
+         * stream is not suspended so we keep on processing output.
+         */
+        ap_log_cerror(APLOG_MARK, APLOG_TRACE1, rv, c1,
+                      H2_STRM_MSG(stream, "out_buffer, already has %ld length"),
+                      (long)buf_len);
+        rv = APR_SUCCESS;
+        goto cleanup;
     }
-    
-    if (pheaders && *pheaders) {
-        return APR_SUCCESS;
+
+    if (stream->output_eos) {
+        rv = APR_BRIGADE_EMPTY(stream->out_buffer)? APR_EOF : APR_SUCCESS;
     }
-    
-    /* If there we do not have enough buffered data to satisfy the requested
-     * length *and* we counted the _complete_ buffer (and did not stop in the middle
-     * because of meta data there), lets see if we can read more from the
-     * output beam */
-    missing = H2MIN(requested, stream->max_mem) - *plen;
-    if (complete && !*peos && missing > 0) {
-        apr_status_t rv = APR_EOF;
-        
-        if (stream->output) {
-            H2_STREAM_OUT_LOG(APLOG_TRACE2, stream, "pre");
-            rv = h2_beam_receive(stream->output, stream->out_buffer, 
-                                 APR_NONBLOCK_READ, stream->max_mem - *plen);
-            H2_STREAM_OUT_LOG(APLOG_TRACE2, stream, "post");
-        }
-        
-        if (rv == APR_SUCCESS) {
-            /* count the buffer again, now that we have read output */
-            status = add_buffered_data(stream, requested, plen, peos, &complete, pheaders);
-        }
-        else if (APR_STATUS_IS_EOF(rv)) {
-            apr_bucket *eos = apr_bucket_eos_create(c->bucket_alloc);
-            APR_BRIGADE_INSERT_TAIL(stream->out_buffer, eos);
-            *peos = 1;
-        }
-        else if (APR_STATUS_IS_EAGAIN(rv)) {
-            /* we set this is the status of this call only if there
-             * is no buffered data, see check below */
-        }
-        else {
-            /* real error reading. Give this back directly, even though
-             * we may have something buffered. */
-            status = rv;
+    else {
+        H2_STREAM_OUT_LOG(APLOG_TRACE2, stream, "pre");
+        rv = h2_beam_receive(stream->output, stream->session->c1, stream->out_buffer,
+                             APR_NONBLOCK_READ, stream->session->max_stream_mem - buf_len);
+        if (APR_SUCCESS != rv) {
+            if (APR_EAGAIN != rv) {
+                ap_log_cerror(APLOG_MARK, APLOG_TRACE1, rv, c1,
+                              H2_STRM_MSG(stream, "out_buffer, receive unsuccessful"));
+            }
         }
     }
-    
-    if (status == APR_SUCCESS) {
-        if (*peos || *plen) {
-            ap_log_cerror(APLOG_MARK, APLOG_TRACE1, 0, c,
-                          H2_STRM_MSG(stream, "prepare, len=%ld eos=%d"),
-                          (long)*plen, *peos);
-        }
-        else {
-            status = (stream->output && h2_beam_is_closed(stream->output))? APR_EOF : APR_EAGAIN;
-            ap_log_cerror(APLOG_MARK, APLOG_TRACE1, 0, c,
-                          H2_STRM_MSG(stream, "prepare, no data"));
+
+    /* get rid of buckets we have no need for */
+    if (!APR_BRIGADE_EMPTY(stream->out_buffer)) {
+        b = APR_BRIGADE_FIRST(stream->out_buffer);
+        while (b != APR_BRIGADE_SENTINEL(stream->out_buffer)) {
+            e = APR_BUCKET_NEXT(b);
+            if (APR_BUCKET_IS_METADATA(b)) {
+                if (APR_BUCKET_IS_FLUSH(b)) {  /* we flush any c1 data already */
+                    APR_BUCKET_REMOVE(b);
+                    apr_bucket_destroy(b);
+                }
+                else if (APR_BUCKET_IS_EOS(b)) {
+                    stream->output_eos = 1;
+                }
+                else if (AP_BUCKET_IS_ERROR(b)) {
+                    stream_do_error_bucket(stream, b);
+                    break;
+                }
+            }
+            else if (b->length == 0) {  /* zero length data */
+                APR_BUCKET_REMOVE(b);
+                apr_bucket_destroy(b);
+            }
+            b = e;
         }
     }
-    return status;
+    H2_STREAM_OUT_LOG(APLOG_TRACE2, stream, "out_buffer, after receive");
+
+cleanup:
+    return rv;
 }
 
-static int is_not_headers(apr_bucket *b)
+static int bucket_pass_to_c1(apr_bucket *b)
 {
-    return !H2_BUCKET_IS_HEADERS(b);
+#if AP_HAS_RESPONSE_BUCKETS
+    return !AP_BUCKET_IS_RESPONSE(b)
+           && !AP_BUCKET_IS_HEADERS(b)
+           && !APR_BUCKET_IS_EOS(b);
+#else
+    return !H2_BUCKET_IS_HEADERS(b) && !APR_BUCKET_IS_EOS(b);
+#endif
 }
 
 apr_status_t h2_stream_read_to(h2_stream *stream, apr_bucket_brigade *bb, 
                                apr_off_t *plen, int *peos)
 {
-    conn_rec *c = stream->session->c;
-    apr_status_t status = APR_SUCCESS;
+    apr_status_t rv = APR_SUCCESS;
 
+    H2_STRM_ASSERT_MAGIC(stream, H2_STRM_MAGIC_OK);
     if (stream->rst_error) {
         return APR_ECONNRESET;
     }
-    status = h2_append_brigade(bb, stream->out_buffer, plen, peos, is_not_headers);
-    if (status == APR_SUCCESS && !*peos && !*plen) {
-        status = APR_EAGAIN;
+    rv = h2_append_brigade(bb, stream->out_buffer, plen, peos, bucket_pass_to_c1);
+    if (APR_SUCCESS  == rv && !*peos && !*plen) {
+        rv = APR_EAGAIN;
     }
-    ap_log_cerror(APLOG_MARK, APLOG_TRACE2, status, c,
-                  H2_STRM_MSG(stream, "read_to, len=%ld eos=%d"),
-                  (long)*plen, *peos);
-    return status;
+    return rv;
 }
 
+static apr_status_t stream_do_trailers(h2_stream *stream)
+{
+    conn_rec *c1 = stream->session->c1;
+    int ngrv;
+    h2_ngheader *nh = NULL;
+    apr_bucket *b, *e;
+#if AP_HAS_RESPONSE_BUCKETS
+    ap_bucket_headers *headers = NULL;
+#else
+    h2_headers *headers = NULL;
+#endif
+    apr_status_t rv;
+
+    ap_assert(stream->response);
+    ap_assert(stream->out_buffer);
+
+    b = APR_BRIGADE_FIRST(stream->out_buffer);
+    while (b != APR_BRIGADE_SENTINEL(stream->out_buffer)) {
+        e = APR_BUCKET_NEXT(b);
+        if (APR_BUCKET_IS_METADATA(b)) {
+#if AP_HAS_RESPONSE_BUCKETS
+            if (AP_BUCKET_IS_HEADERS(b)) {
+                headers = b->data;
+#else /* AP_HAS_RESPONSE_BUCKETS */
+            if (H2_BUCKET_IS_HEADERS(b)) {
+                headers = h2_bucket_headers_get(b);
+#endif /* else AP_HAS_RESPONSE_BUCKETS */
+                APR_BUCKET_REMOVE(b);
+                apr_bucket_destroy(b);
+                ap_log_cerror(APLOG_MARK, APLOG_TRACE1, 0, c1,
+                              H2_STRM_MSG(stream, "process trailers"));
+                break;
+            }
+            else if (APR_BUCKET_IS_EOS(b)) {
+                break;
+            }
+        }
+        else {
+            break;
+        }
+        b = e;
+    }
 
+    if (!headers) {
+        rv = APR_EAGAIN;
+        goto cleanup;
+    }
+
+    rv = h2_res_create_ngtrailer(&nh, stream->pool, headers);
+    ap_log_cerror(APLOG_MARK, APLOG_DEBUG, rv, c1,
+                  H2_STRM_LOG(APLOGNO(03072), stream, "submit %d trailers"),
+                  (int)nh->nvlen);
+    if (APR_SUCCESS != rv) {
+        ap_log_cerror(APLOG_MARK, APLOG_DEBUG, rv, c1,
+                      H2_STRM_LOG(APLOGNO(10024), stream, "invalid trailers"));
+        h2_stream_rst(stream, NGHTTP2_PROTOCOL_ERROR);
+        goto cleanup;
+    }
+
+    ngrv = nghttp2_submit_trailer(stream->session->ngh2, stream->id, nh->nv, nh->nvlen);
+    if (nghttp2_is_fatal(ngrv)) {
+        rv = APR_EGENERAL;
+        h2_session_dispatch_event(stream->session,
+                                 H2_SESSION_EV_PROTO_ERROR, ngrv, nghttp2_strerror(rv));
+        ap_log_cerror(APLOG_MARK, APLOG_ERR, rv, c1,
+                      APLOGNO(02940) "submit_response: %s",
+                      nghttp2_strerror(rv));
+    }
+    stream->sent_trailers = 1;
+
+cleanup:
+    return rv;
+}
+
+#if AP_HAS_RESPONSE_BUCKETS
+apr_status_t h2_stream_submit_pushes(h2_stream *stream, ap_bucket_response *response)
+#else
 apr_status_t h2_stream_submit_pushes(h2_stream *stream, h2_headers *response)
+#endif
 {
     apr_status_t status = APR_SUCCESS;
     apr_array_header_t *pushes;
     int i;
     
+    H2_STRM_ASSERT_MAGIC(stream, H2_STRM_MAGIC_OK);
     pushes = h2_push_collect_update(stream, stream->request, response);
     if (pushes && !apr_is_empty_array(pushes)) {
-        ap_log_cerror(APLOG_MARK, APLOG_TRACE1, 0, stream->session->c,
+        ap_log_cerror(APLOG_MARK, APLOG_TRACE1, 0, stream->session->c1,
                       H2_STRM_MSG(stream, "found %d push candidates"),
                       pushes->nelts);
         for (i = 0; i < pushes->nelts; ++i) {
@@ -977,17 +1229,24 @@ apr_status_t h2_stream_submit_pushes(h2_stream *stream, h2_headers *response)
 
 apr_table_t *h2_stream_get_trailers(h2_stream *stream)
 {
+    H2_STRM_ASSERT_MAGIC(stream, H2_STRM_MAGIC_OK);
     return NULL;
 }
 
-const h2_priority *h2_stream_get_priority(h2_stream *stream, 
+#if AP_HAS_RESPONSE_BUCKETS
+const h2_priority *h2_stream_get_priority(h2_stream *stream,
+                                          ap_bucket_response *response)
+#else
+const h2_priority *h2_stream_get_priority(h2_stream *stream,
                                           h2_headers *response)
+#endif
 {
+    H2_STRM_ASSERT_MAGIC(stream, H2_STRM_MAGIC_OK);
     if (response && stream->initiated_on) {
         const char *ctype = apr_table_get(response->headers, "content-type");
         if (ctype) {
             /* FIXME: Not good enough, config needs to come from request->server */
-            return h2_config_get_priority(stream->session->config, ctype);
+            return h2_cconfig_get_priority(stream->session->c1, ctype);
         }
     }
     return NULL;
@@ -995,21 +1254,47 @@ const h2_priority *h2_stream_get_priority(h2_stream *stream,
 
 int h2_stream_is_ready(h2_stream *stream)
 {
-    if (stream->has_response) {
+    /* Have we sent a response or do we have the response in our buffer? */
+    H2_STRM_ASSERT_MAGIC(stream, H2_STRM_MAGIC_OK);
+    if (stream->response) {
         return 1;
     }
-    else if (stream->out_buffer && get_first_headers_bucket(stream->out_buffer)) {
+    else if (stream->out_buffer && get_first_response_bucket(stream->out_buffer)) {
         return 1;
     }
     return 0;
 }
 
-int h2_stream_was_closed(const h2_stream *stream)
+int h2_stream_wants_send_data(h2_stream *stream)
 {
-    switch (stream->state) {
+    H2_STRM_ASSERT_MAGIC(stream, H2_STRM_MAGIC_OK);
+    return h2_stream_is_ready(stream) &&
+           ((stream->out_buffer && !APR_BRIGADE_EMPTY(stream->out_buffer)) ||
+            (stream->output && !h2_beam_empty(stream->output)));
+}
+
+int h2_stream_is_at(const h2_stream *stream, h2_stream_state_t state)
+{
+    H2_STRM_ASSERT_MAGIC(stream, H2_STRM_MAGIC_OK);
+    return stream->state == state;
+}
+
+int h2_stream_is_at_or_past(const h2_stream *stream, h2_stream_state_t state)
+{
+    H2_STRM_ASSERT_MAGIC(stream, H2_STRM_MAGIC_OK);
+    switch (state) {
+        case H2_SS_IDLE:
+            return 1; /* by definition */
+        case H2_SS_RSVD_R: /*fall through*/
+        case H2_SS_RSVD_L: /*fall through*/
+        case H2_SS_OPEN:
+            return stream->state == state || stream->state >= H2_SS_OPEN;
+        case H2_SS_CLOSED_R: /*fall through*/
+        case H2_SS_CLOSED_L: /*fall through*/
         case H2_SS_CLOSED:
+            return stream->state == state || stream->state >= H2_SS_CLOSED;
         case H2_SS_CLEANUP:
-            return 1;
+            return stream->state == state;
         default:
             return 0;
     }
@@ -1019,6 +1304,7 @@ apr_status_t h2_stream_in_consumed(h2_stream *stream, apr_off_t amount)
 {
     h2_session *session = stream->session;
     
+    H2_STRM_ASSERT_MAGIC(stream, H2_STRM_MAGIC_OK);
     if (amount > 0) {
         apr_off_t consumed = amount;
         
@@ -1066,13 +1352,477 @@ apr_status_t h2_stream_in_consumed(h2_stream *stream, apr_off_t amount)
                 nghttp2_session_set_local_window_size(session->ngh2, 
                         NGHTTP2_FLAG_NONE, stream->id, win);
             } 
-            ap_log_cerror(APLOG_MARK, APLOG_TRACE2, 0, session->c,
-                          "h2_stream(%ld-%d): consumed %ld bytes, window now %d/%d",
-                          session->id, stream->id, (long)amount, 
-                          cur_size, stream->in_window_size);
+            ap_log_cerror(APLOG_MARK, APLOG_TRACE2, 0, session->c1,
+                          H2_STRM_MSG(stream, "consumed %ld bytes, window now %d/%d"),
+                          (long)amount, cur_size, stream->in_window_size);
         }
-#endif
+#endif /* #ifdef H2_NG2_LOCAL_WIN_SIZE */
     }
     return APR_SUCCESS;   
 }
 
+static apr_off_t output_data_buffered(h2_stream *stream, int *peos, int *pheader_blocked)
+{
+    /* How much data do we have in our buffers that we can write? */
+    apr_off_t buf_len = 0;
+    apr_bucket *b;
+
+    *peos = *pheader_blocked = 0;
+    if (stream->out_buffer) {
+        b = APR_BRIGADE_FIRST(stream->out_buffer);
+        while (b != APR_BRIGADE_SENTINEL(stream->out_buffer)) {
+            if (APR_BUCKET_IS_METADATA(b)) {
+                if (APR_BUCKET_IS_EOS(b)) {
+                    *peos = 1;
+                    break;
+                }
+#if AP_HAS_RESPONSE_BUCKETS
+                else if (AP_BUCKET_IS_RESPONSE(b)) {
+                    break;
+                }
+                else if (AP_BUCKET_IS_HEADERS(b)) {
+                    *pheader_blocked = 1;
+                    break;
+                }
+#else
+                else if (H2_BUCKET_IS_HEADERS(b)) {
+                    *pheader_blocked = 1;
+                    break;
+                }
+#endif
+            }
+            else {
+                buf_len += b->length;
+            }
+            b = APR_BUCKET_NEXT(b);
+        }
+    }
+    return buf_len;
+}
+
+static ssize_t stream_data_cb(nghttp2_session *ng2s,
+                              int32_t stream_id,
+                              uint8_t *buf,
+                              size_t length,
+                              uint32_t *data_flags,
+                              nghttp2_data_source *source,
+                              void *puser)
+{
+    h2_session *session = (h2_session *)puser;
+    conn_rec *c1 = session->c1;
+    apr_off_t buf_len;
+    int eos, header_blocked;
+    apr_status_t rv;
+    h2_stream *stream;
+
+    /* nghttp2 wants to send more DATA for the stream.
+     * we should have submitted the final response at this time
+     * after receiving output via stream_do_responses() */
+    ap_assert(session);
+    (void)ng2s;
+    (void)buf;
+    (void)source;
+    stream = nghttp2_session_get_stream_user_data(session->ngh2, stream_id);
+
+    if (!stream) {
+        ap_log_cerror(APLOG_MARK, APLOG_ERR, 0, c1,
+                      APLOGNO(02937)
+                      H2_SSSN_STRM_MSG(session, stream_id, "data_cb, stream not found"));
+        return NGHTTP2_ERR_CALLBACK_FAILURE;
+    }
+    H2_STRM_ASSERT_MAGIC(stream, H2_STRM_MAGIC_OK);
+    if (!stream->output || !stream->response || !stream->out_buffer) {
+        return NGHTTP2_ERR_DEFERRED;
+    }
+    if (stream->rst_error) {
+        return NGHTTP2_ERR_DEFERRED;
+    }
+    if (h2_c1_io_needs_flush(&session->io)) {
+        rv = h2_c1_io_pass(&session->io);
+        if (APR_STATUS_IS_EAGAIN(rv)) {
+            ap_log_cerror(APLOG_MARK, APLOG_TRACE1, 0, c1,
+                          H2_SSSN_STRM_MSG(session, stream_id, "suspending on c1 out needs flush"));
+            h2_stream_dispatch(stream, H2_SEV_OUT_C1_BLOCK);
+            return NGHTTP2_ERR_DEFERRED;
+        }
+        else if (rv) {
+            h2_session_dispatch_event(session, H2_SESSION_EV_CONN_ERROR, rv, NULL);
+            return NGHTTP2_ERR_CALLBACK_FAILURE;
+        }
+    }
+
+    /* determine how much we'd like to send. We cannot send more than
+     * is requested. But we can reduce the size in case the master
+     * connection operates in smaller chunks. (TSL warmup) */
+    if (stream->session->io.write_size > 0) {
+        apr_size_t chunk_len = stream->session->io.write_size - H2_FRAME_HDR_LEN;
+        if (length > chunk_len) {
+            length = chunk_len;
+        }
+    }
+    /* We allow configurable max DATA frame length. */
+    if (stream->session->max_data_frame_len > 0
+        && length > stream->session->max_data_frame_len) {
+      length = stream->session->max_data_frame_len;
+    }
+
+    /* How much data do we have in our buffers that we can write?
+     * if not enough, receive more. */
+    buf_len = output_data_buffered(stream, &eos, &header_blocked);
+    if (buf_len < (apr_off_t)length && !eos
+           && !header_blocked && !stream->rst_error) {
+        /* read more? */
+        ap_log_cerror(APLOG_MARK, APLOG_TRACE2, 0, c1,
+                      H2_SSSN_STRM_MSG(session, stream_id,
+                      "need more (read len=%ld, %ld in buffer)"),
+                      (long)length, (long)buf_len);
+        rv = buffer_output_receive(stream);
+        ap_log_cerror(APLOG_MARK, APLOG_TRACE1, rv, c1,
+                      H2_SSSN_STRM_MSG(session, stream_id,
+                      "buffer_output_received"));
+        if (APR_STATUS_IS_EAGAIN(rv)) {
+            /* currently, no more is available */
+        }
+        else if (APR_SUCCESS == rv) {
+            /* got some, re-assess */
+            buf_len = output_data_buffered(stream, &eos, &header_blocked);
+        }
+        else if (APR_EOF == rv) {
+            if (!stream->output_eos) {
+                /* Seeing APR_EOF without an EOS bucket received before indicates
+                 * that stream output is incomplete. Commonly, we expect to see
+                 * an ERROR bucket to have been generated. But faulty handlers
+                 * may not have generated one.
+                 * We need to RST the stream bc otherwise the client thinks
+                 * it is all fine. */
+                 ap_log_cerror(APLOG_MARK, APLOG_TRACE1, rv, c1,
+                               H2_SSSN_STRM_MSG(session, stream_id, "rst stream"));
+                 h2_stream_rst(stream, H2_ERR_STREAM_CLOSED);
+                 return NGHTTP2_ERR_DEFERRED;
+            }
+            ap_log_cerror(APLOG_MARK, APLOG_TRACE1, rv, c1,
+                          H2_SSSN_STRM_MSG(session, stream_id,
+                          "eof on receive (read len=%ld, %ld in buffer)"),
+                          (long)length, (long)buf_len);
+            eos = 1;
+            rv = APR_SUCCESS;
+        }
+        else if (APR_ECONNRESET == rv || APR_ECONNABORTED == rv) {
+            ap_log_cerror(APLOG_MARK, APLOG_DEBUG, rv, c1,
+                          H2_STRM_LOG(APLOGNO(10471), stream, "data_cb, reading data"));
+            h2_stream_rst(stream, H2_ERR_STREAM_CLOSED);
+            return NGHTTP2_ERR_DEFERRED;
+        }
+        else {
+            ap_log_cerror(APLOG_MARK, APLOG_ERR, rv, c1,
+                          H2_STRM_LOG(APLOGNO(02938), stream, "data_cb, reading data"));
+            h2_stream_rst(stream, H2_ERR_INTERNAL_ERROR);
+            return NGHTTP2_ERR_DEFERRED;
+        }
+    }
+
+    if (stream->rst_error) {
+        return NGHTTP2_ERR_DEFERRED;
+    }
+
+    if (buf_len == 0 && header_blocked) {
+        rv = stream_do_trailers(stream);
+        if (APR_SUCCESS != rv && !APR_STATUS_IS_EAGAIN(rv)) {
+            ap_log_cerror(APLOG_MARK, APLOG_ERR, rv, c1,
+                          H2_STRM_LOG(APLOGNO(10300), stream,
+                          "data_cb, error processing trailers"));
+            return NGHTTP2_ERR_CALLBACK_FAILURE;
+        }
+        length = 0;
+        eos = 0;
+    }
+    else if (buf_len > (apr_off_t)length) {
+        eos = 0;  /* Any EOS we have in the buffer does not apply yet */
+    }
+    else {
+        length = (size_t)buf_len;
+    }
+
+    if (length) {
+        ap_log_cerror(APLOG_MARK, APLOG_TRACE2, 0, c1,
+                      H2_STRM_MSG(stream, "data_cb, sending len=%ld, eos=%d"),
+                      (long)length, eos);
+        *data_flags |=  NGHTTP2_DATA_FLAG_NO_COPY;
+    }
+    else if (!eos && !stream->sent_trailers) {
+        /* We have not reached the end of DATA yet, DEFER sending */
+        ap_log_cerror(APLOG_MARK, APLOG_DEBUG, 0, c1,
+                      H2_STRM_LOG(APLOGNO(03071), stream, "data_cb, suspending"));
+        return NGHTTP2_ERR_DEFERRED;
+    }
+
+    if (eos) {
+        *data_flags |= NGHTTP2_DATA_FLAG_EOF;
+    }
+    return length;
+}
+
+static apr_status_t stream_do_response(h2_stream *stream)
+{
+    conn_rec *c1 = stream->session->c1;
+    apr_status_t rv = APR_EAGAIN;
+    int ngrv, is_empty = 0;
+    h2_ngheader *nh = NULL;
+    apr_bucket *b, *e;
+#if AP_HAS_RESPONSE_BUCKETS
+    ap_bucket_response *resp = NULL;
+#else
+    h2_headers *resp = NULL;
+#endif
+    nghttp2_data_provider provider, *pprovider = NULL;
+
+    H2_STRM_ASSERT_MAGIC(stream, H2_STRM_MAGIC_OK);
+    ap_assert(!stream->response);
+    ap_assert(stream->out_buffer);
+
+    b = APR_BRIGADE_FIRST(stream->out_buffer);
+    while (b != APR_BRIGADE_SENTINEL(stream->out_buffer)) {
+        e = APR_BUCKET_NEXT(b);
+        if (APR_BUCKET_IS_METADATA(b)) {
+#if AP_HAS_RESPONSE_BUCKETS
+            if (AP_BUCKET_IS_RESPONSE(b)) {
+                resp = b->data;
+#else /* AP_HAS_RESPONSE_BUCKETS */
+            if (H2_BUCKET_IS_HEADERS(b)) {
+                resp = h2_bucket_headers_get(b);
+#endif /* else AP_HAS_RESPONSE_BUCKETS */
+                APR_BUCKET_REMOVE(b);
+                apr_bucket_destroy(b);
+                ap_log_cerror(APLOG_MARK, APLOG_TRACE1, 0, c1,
+                              H2_STRM_MSG(stream, "process response %d"),
+                              resp->status);
+                is_empty = (e != APR_BRIGADE_SENTINEL(stream->out_buffer)
+                            && APR_BUCKET_IS_EOS(e));
+                break;
+            }
+            else if (APR_BUCKET_IS_EOS(b)) {
+                h2_stream_rst(stream, H2_ERR_INTERNAL_ERROR);
+                rv = APR_EINVAL;
+                goto cleanup;
+            }
+            else if (AP_BUCKET_IS_ERROR(b)) {
+                stream_do_error_bucket(stream, b);
+                rv = APR_EINVAL;
+                goto cleanup;
+            }
+        }
+        else {
+            /* data buckets before response headers, an error */
+            h2_stream_rst(stream, H2_ERR_INTERNAL_ERROR);
+            rv = APR_EINVAL;
+            goto cleanup;
+        }
+        b = e;
+    }
+
+    if (!resp) {
+        rv = APR_EAGAIN;
+        goto cleanup;
+    }
+
+    if (resp->status < 100) {
+        h2_stream_rst(stream, resp->status);
+        goto cleanup;
+    }
+
+    if (resp->status == HTTP_FORBIDDEN && resp->notes) {
+        const char *cause = apr_table_get(resp->notes, "ssl-renegotiate-forbidden");
+        if (cause) {
+            /* This request triggered a TLS renegotiation that is not allowed
+             * in HTTP/2. Tell the client that it should use HTTP/1.1 for this.
+             */
+            ap_log_cerror(APLOG_MARK, APLOG_DEBUG, resp->status, c1,
+                          H2_STRM_LOG(APLOGNO(03061), stream,
+                          "renegotiate forbidden, cause: %s"), cause);
+            h2_stream_rst(stream, H2_ERR_HTTP_1_1_REQUIRED);
+            goto cleanup;
+        }
+    }
+
+    ap_log_cerror(APLOG_MARK, APLOG_DEBUG, 0, c1,
+                  H2_STRM_LOG(APLOGNO(03073), stream,
+                  "submit response %d"), resp->status);
+
+    /* If this stream is not a pushed one itself,
+     * and HTTP/2 server push is enabled here,
+     * and the response HTTP status is not sth >= 400,
+     * and the remote side has pushing enabled,
+     * -> find and perform any pushes on this stream
+     *    *before* we submit the stream response itself.
+     *    This helps clients avoid opening new streams on Link
+     *    resp that get pushed right afterwards.
+     *
+     * *) the response code is relevant, as we do not want to
+     *    make pushes on 401 or 403 codes and friends.
+     *    And if we see a 304, we do not push either
+     *    as the client, having this resource in its cache, might
+     *    also have the pushed ones as well.
+     */
+    if (!stream->initiated_on
+        && !stream->response
+        && stream->request && stream->request->method
+        && !strcmp("GET", stream->request->method)
+        && (resp->status < 400)
+        && (resp->status != 304)
+        && h2_session_push_enabled(stream->session)) {
+        /* PUSH is possible and enabled on server, unless the request
+         * denies it, submit resources to push */
+        const char *s = apr_table_get(resp->notes, H2_PUSH_MODE_NOTE);
+        if (!s || strcmp(s, "0")) {
+            ap_log_cerror(APLOG_MARK, APLOG_TRACE1, 0, c1,
+                          H2_STRM_MSG(stream, "submit pushes, note=%s"), s);
+            h2_stream_submit_pushes(stream, resp);
+        }
+    }
+
+    if (!stream->pref_priority) {
+        stream->pref_priority = h2_stream_get_priority(stream, resp);
+    }
+    h2_session_set_prio(stream->session, stream, stream->pref_priority);
+
+    if (resp->status == 103
+        && !h2_config_sgeti(stream->session->s, H2_CONF_EARLY_HINTS)) {
+        /* suppress sending this to the client, it might have triggered
+         * pushes and served its purpose nevertheless */
+        rv = APR_SUCCESS;
+        goto cleanup;
+    }
+    if (resp->status >= 200) {
+        stream->response = resp;
+    }
+
+    if (!is_empty) {
+        memset(&provider, 0, sizeof(provider));
+        provider.source.fd = stream->id;
+        provider.read_callback = stream_data_cb;
+        pprovider = &provider;
+    }
+
+    rv = h2_res_create_ngheader(&nh, stream->pool, resp);
+    if (APR_SUCCESS != rv) {
+        ap_log_cerror(APLOG_MARK, APLOG_DEBUG, rv, c1,
+                      H2_STRM_LOG(APLOGNO(10025), stream, "invalid response"));
+        h2_stream_rst(stream, NGHTTP2_PROTOCOL_ERROR);
+        goto cleanup;
+    }
+
+    ngrv = nghttp2_submit_response(stream->session->ngh2, stream->id,
+                                   nh->nv, nh->nvlen, pprovider);
+    if (nghttp2_is_fatal(ngrv)) {
+        rv = APR_EGENERAL;
+        h2_session_dispatch_event(stream->session,
+                                 H2_SESSION_EV_PROTO_ERROR, ngrv, nghttp2_strerror(rv));
+        ap_log_cerror(APLOG_MARK, APLOG_ERR, rv, c1,
+                      APLOGNO(10402) "submit_response: %s",
+                      nghttp2_strerror(rv));
+        goto cleanup;
+    }
+
+    if (stream->initiated_on) {
+        ++stream->session->pushes_submitted;
+    }
+    else {
+        ++stream->session->responses_submitted;
+    }
+
+cleanup:
+    return rv;
+}
+
+static void stream_do_responses(h2_stream *stream)
+{
+    h2_session *session = stream->session;
+    conn_rec *c1 = session->c1;
+    apr_status_t rv;
+
+    ap_assert(!stream->response);
+    ap_log_cerror(APLOG_MARK, APLOG_TRACE2, 0, c1,
+                  H2_STRM_MSG(stream, "do_response"));
+    rv = buffer_output_receive(stream);
+    ap_log_cerror(APLOG_MARK, APLOG_TRACE1, rv, c1,
+                  H2_SSSN_STRM_MSG(session, stream->id,
+                  "buffer_output_received2"));
+    if (APR_SUCCESS != rv && APR_EAGAIN != rv) {
+        h2_stream_rst(stream, NGHTTP2_PROTOCOL_ERROR);
+    }
+    else {
+        /* process all headers sitting at the buffer head. */
+        do {
+            rv = stream_do_response(stream);
+        } while (APR_SUCCESS == rv
+                 && !stream->rst_error
+                 && !stream->response);
+    }
+}
+
+void h2_stream_on_output_change(h2_stream *stream)
+{
+    conn_rec *c1 = stream->session->c1;
+    apr_status_t rv = APR_EAGAIN;
+
+    /* stream->pout_recv_write signalled a change. Check what has happend, read
+     * from it and act on seeing a response/data. */
+    H2_STRM_ASSERT_MAGIC(stream, H2_STRM_MAGIC_OK);
+    if (!stream->output) {
+        /* c2 has not assigned the output beam to the stream (yet). */
+        ap_log_cerror(APLOG_MARK, APLOG_WARNING, 0, c1,
+                      H2_STRM_MSG(stream, "read_output, no output beam registered"));
+    }
+    else if (h2_stream_is_at_or_past(stream, H2_SS_CLOSED)) {
+        ap_log_cerror(APLOG_MARK, APLOG_DEBUG, rv, c1,
+                      H2_STRM_LOG(APLOGNO(10301), stream, "already closed"));
+    }
+    else if (h2_stream_is_at(stream, H2_SS_CLOSED_L)) {
+        /* We have delivered a response to a stream that was not closed
+         * by the client. This could be a POST with body that we negate
+         * and we need to RST_STREAM to end if. */
+        ap_log_cerror(APLOG_MARK, APLOG_DEBUG, 0, c1,
+                      H2_STRM_LOG(APLOGNO(10313), stream, "remote close missing"));
+        h2_stream_rst(stream, H2_ERR_NO_ERROR);
+    }
+    else {
+        /* stream is not closed, a change in output happened. There are
+         * two modes of operation here:
+         * 1) the final response has been submitted. nghttp2 is invoking
+         *    stream_data_cb() to progress the stream. This handles DATA,
+         *    trailers, EOS and ERRORs.
+         *    When stream_data_cb() runs out of things to send, it returns
+         *    NGHTTP2_ERR_DEFERRED and nghttp2 *suspends* further processing
+         *    until we tell it to resume.
+         * 2) We have not seen the *final* response yet. The stream can not
+         *    send any response DATA. The nghttp2 stream_data_cb() is not
+         *    invoked. We need to receive output, expecting not DATA but
+         *    RESPONSEs (intermediate may arrive) and submit those. On
+         *    the final response, nghttp2 will start calling stream_data_cb().
+         */
+        if (stream->response) {
+            nghttp2_session_resume_data(stream->session->ngh2, stream->id);
+            ap_log_cerror(APLOG_MARK, APLOG_TRACE2, 0, c1,
+                          H2_STRM_MSG(stream, "resumed"));
+        }
+        else {
+            stream_do_responses(stream);
+            if (!stream->rst_error) {
+                nghttp2_session_resume_data(stream->session->ngh2, stream->id);
+            }
+        }
+    }
+}
+
+void h2_stream_on_input_change(h2_stream *stream)
+{
+    H2_STRM_ASSERT_MAGIC(stream, H2_STRM_MAGIC_OK);
+    ap_assert(stream->input);
+    h2_beam_report_consumption(stream->input);
+    if (h2_stream_is_at(stream, H2_SS_CLOSED_L)
+        && !h2_mplx_c1_stream_is_running(stream->session->mplx, stream)) {
+        ap_log_cerror(APLOG_MARK, APLOG_DEBUG, 0, stream->session->c1,
+                      H2_STRM_LOG(APLOGNO(10026), stream, "remote close missing"));
+        h2_stream_rst(stream, H2_ERR_NO_ERROR);
+    }
+}