static apr_status_t CaseFilterInFilter(ap_filter_t *f,
                                       apr_bucket_brigade *pbbOut,
                                       ap_input_mode_t eMode,
                                       apr_read_type_e eBlock,
                                       apr_off_t nBytes)
{
    request_rec *r = f->r;
    conn_rec *c = r->connection;
    CaseFilterInContext *pCtx;
    apr_status_t ret;

    if (!(pCtx = f->ctx)) {
        f->ctx = pCtx = apr_palloc(r->pool, sizeof *pCtx);
        pCtx->pbbTmp = apr_brigade_create(r->pool, c->bucket_alloc);
    }

    if (APR_BRIGADE_EMPTY(pCtx->pbbTmp)) {
        ret = ap_get_brigade(f->next, pCtx->pbbTmp, eMode, eBlock, nBytes);

        if (eMode == AP_MODE_EATCRLF || ret != APR_SUCCESS)
            return ret;
    }

    while (!APR_BRIGADE_EMPTY(pCtx->pbbTmp)) {
        apr_bucket *pbktIn = APR_BRIGADE_FIRST(pCtx->pbbTmp);
        apr_bucket *pbktOut;
        const char *data;
        apr_size_t len;
        char *buf;
        apr_size_t n;

        /* It is tempting to do this...
         * APR_BUCKET_REMOVE(pB);
         * APR_BRIGADE_INSERT_TAIL(pbbOut,pB);
         * and change the case of the bucket data, but that would be wrong
         * for a file or socket buffer, for example...
         */

        if (APR_BUCKET_IS_EOS(pbktIn)) {
            APR_BUCKET_REMOVE(pbktIn);
            APR_BRIGADE_INSERT_TAIL(pbbOut, pbktIn);
            break;
        }

        ret=apr_bucket_read(pbktIn, &data, &len, eBlock);
        if (ret != APR_SUCCESS)
            return ret;

        buf = ap_malloc(len);
        for (n=0 ; n < len ; ++n) {
            buf[n] = apr_toupper(data[n]);
        }

        pbktOut = apr_bucket_heap_create(buf, len, 0, c->bucket_alloc);
        APR_BRIGADE_INSERT_TAIL(pbbOut, pbktOut);
        apr_bucket_delete(pbktIn);
    }

    return APR_SUCCESS;
}

apr_status_t h2_beam_send(h2_bucket_beam *beam, 
                          apr_bucket_brigade *red_brigade, 
                          apr_read_type_e block)
{
    apr_bucket *bred;
    apr_status_t status = APR_SUCCESS;
    h2_beam_lock bl;

    /* Called from the red thread to add buckets to the beam */
    if (enter_yellow(beam, &bl) == APR_SUCCESS) {
        r_purge_reds(beam);
        
        if (beam->aborted) {
            status = APR_ECONNABORTED;
        }
        else if (red_brigade) {
            int force_report = !APR_BRIGADE_EMPTY(red_brigade); 
            while (!APR_BRIGADE_EMPTY(red_brigade)
                   && status == APR_SUCCESS) {
                bred = APR_BRIGADE_FIRST(red_brigade);
                status = append_bucket(beam, bred, block, beam->red_pool, &bl);
            }
            report_production(beam, force_report);
            if (beam->m_cond) {
                apr_thread_cond_broadcast(beam->m_cond);
            }
        }
        report_consumption(beam, 0);
        leave_yellow(beam, &bl);
    }
    return status;
}

apr_status_t h2_stream_read_to(h2_stream *stream, apr_bucket_brigade *bb, 
                               apr_size_t *plen, int *peos)
{
    apr_status_t status = APR_SUCCESS;

    if (stream->rst_error) {
        return APR_ECONNRESET;
    }
    
    if (APR_BRIGADE_EMPTY(stream->bbout)) {
        apr_size_t tlen = *plen;
        int eos;
        status = h2_mplx_out_read_to(stream->session->mplx, stream->id, 
                                     stream->bbout, &tlen, &eos);
    }
    
    if (status == APR_SUCCESS && !APR_BRIGADE_EMPTY(stream->bbout)) {
        status = h2_transfer_brigade(bb, stream->bbout, stream->pool, 
                                     plen, peos);
    }
    else {
        *plen = 0;
        *peos = 0;
    }

    if (status == APR_SUCCESS && !*peos && !*plen) {
        status = APR_EAGAIN;
    }
    ap_log_cerror(APLOG_MARK, APLOG_TRACE1, status, stream->session->c,
                  "h2_stream(%ld-%d): read_to, len=%ld eos=%d",
                  stream->session->id, stream->id, (long)*plen, *peos);
    return status;
}

apr_status_t h2_beam_shutdown(h2_bucket_beam *beam, apr_read_type_e block)
{
    apr_status_t status;
    h2_beam_lock bl;
    
    if ((status = enter_yellow(beam, &bl)) == APR_SUCCESS) {
        r_purge_reds(beam);
        h2_blist_cleanup(&beam->red);
        beam_close(beam);
        report_consumption(beam, 0);
        
        while (status == APR_SUCCESS 
               && (!H2_BPROXY_LIST_EMPTY(&beam->proxies)
                   || (beam->green && !APR_BRIGADE_EMPTY(beam->green)))) {
            if (block == APR_NONBLOCK_READ || !bl.mutex) {
                status = APR_EAGAIN;
                break;
            }
            if (beam->m_cond) {
                apr_thread_cond_broadcast(beam->m_cond);
            }
            status = wait_cond(beam, bl.mutex);
        }
        leave_yellow(beam, &bl);
    }
    return status;
}

APU_DECLARE(apr_status_t) apr_brigade_puts(apr_bucket_brigade *bb,
                                           apr_brigade_flush flush, void *ctx,
                                           const char *str)
{
    apr_size_t len = strlen(str);
    apr_bucket *bkt = APR_BRIGADE_LAST(bb);
    if (!APR_BRIGADE_EMPTY(bb) && APR_BUCKET_IS_HEAP(bkt)) {
        /* If there is enough space available in a heap bucket
         * at the end of the brigade, copy the string directly
         * into the heap bucket
         */
        apr_bucket_heap *h = bkt->data;
        apr_size_t bytes_avail = h->alloc_len - bkt->length;

        if (bytes_avail >= len) {
            char *buf = h->base + bkt->start + bkt->length;
            memcpy(buf, str, len);
            bkt->length += len;
            return APR_SUCCESS;
        }
    }

    /* If the string could not be copied into an existing heap
     * bucket, delegate the work to apr_brigade_write(), which
     * knows how to grow the brigade
     */
    return apr_brigade_write(bb, flush, ctx, str, len);
}

apr_status_t h2_stream_prep_read(h2_stream *stream, 
                                 apr_size_t *plen, int *peos)
{
    apr_status_t status = APR_SUCCESS;
    const char *src;
    
    if (stream->bbout && !APR_BRIGADE_EMPTY(stream->bbout)) {
        src = "stream";
        status = h2_util_bb_avail(stream->bbout, plen, peos);
        if (status == APR_SUCCESS && !*peos && !*plen) {
            apr_brigade_cleanup(stream->bbout);
            return h2_stream_prep_read(stream, plen, peos);
        }
    }
    else {
        src = "mplx";
        status = h2_mplx_out_readx(stream->m, stream->id, 
                                   NULL, NULL, plen, peos);
    }
    if (status == APR_SUCCESS && !*peos && !*plen) {
        status = APR_EAGAIN;
    }
    ap_log_cerror(APLOG_MARK, APLOG_TRACE1, status, stream->m->c,
                  "h2_stream(%ld-%d): prep_read %s, len=%ld eos=%d",
                  stream->m->id, stream->id, 
                  src, (long)*plen, *peos);
    return status;
}

static apr_status_t send_brigade_blocking(apr_socket_t *s,
                                          apr_bucket_brigade *bb,
                                          apr_size_t *bytes_written,
                                          conn_rec *c)
{
    apr_status_t rv;

    rv = APR_SUCCESS;
    while (!APR_BRIGADE_EMPTY(bb)) {
        rv = send_brigade_nonblocking(s, bb, bytes_written, c);
        if (rv != APR_SUCCESS) {
            if (APR_STATUS_IS_EAGAIN(rv)) {
                /* Wait until we can send more data */
                apr_int32_t nsds;
                apr_interval_time_t timeout;
                apr_pollfd_t pollset;

                pollset.p = c->pool;
                pollset.desc_type = APR_POLL_SOCKET;
                pollset.reqevents = APR_POLLOUT;
                pollset.desc.s = s;
                apr_socket_timeout_get(s, &timeout);
                rv = apr_poll(&pollset, 1, &nsds, timeout);
                if (rv != APR_SUCCESS) {
                    break;
                }
            }
            else {
                break;
            }
        }
    }
    return rv;
}

static apr_status_t pass_out(apr_bucket_brigade *bb, void *ctx) 
{
    h2_conn_io *io = (h2_conn_io*)ctx;
    apr_status_t status;
    apr_off_t bblen;
    
    if (APR_BRIGADE_EMPTY(bb)) {
        return APR_SUCCESS;
    }
    
    ap_update_child_status(io->connection->sbh, SERVER_BUSY_WRITE, NULL);
    status = apr_brigade_length(bb, 0, &bblen);
    if (status == APR_SUCCESS) {
        ap_log_cerror(APLOG_MARK, APLOG_TRACE1, 0, io->connection,
                      "h2_conn_io(%ld): pass_out brigade %ld bytes",
                      io->connection->id, (long)bblen);
        status = ap_pass_brigade(io->connection->output_filters, bb);
        if (status == APR_SUCCESS) {
            io->bytes_written += (apr_size_t)bblen;
            io->last_write = apr_time_now();
        }
        apr_brigade_cleanup(bb);
    }
    return status;
}

APU_DECLARE(apr_bucket_brigade *) apr_brigade_split_ex(apr_bucket_brigade *b,
                                                       apr_bucket *e,
                                                       apr_bucket_brigade *a)
{
    apr_bucket *f;

    if (!a) {
        a = apr_brigade_create(b->p, b->bucket_alloc);
    }
    else if (!APR_BRIGADE_EMPTY(a)) {
        apr_brigade_cleanup(a);
    }
    /* Return an empty brigade if there is nothing left in 
     * the first brigade to split off 
     */
    if (e != APR_BRIGADE_SENTINEL(b)) {
        f = APR_RING_LAST(&b->list);
        APR_RING_UNSPLICE(e, f, link);
        APR_RING_SPLICE_HEAD(&a->list, e, f, apr_bucket, link);
    }

    APR_BRIGADE_CHECK_CONSISTENCY(a);
    APR_BRIGADE_CHECK_CONSISTENCY(b);

    return a;
}

apr_status_t h2_stream_set_response(h2_stream *stream, h2_response *response,
                                    apr_bucket_brigade *bb)
{
    apr_status_t status = APR_SUCCESS;
    
    stream->response = response;
    if (bb && !APR_BRIGADE_EMPTY(bb)) {
        int move_all = INT_MAX;
        /* we can move file handles from h2_mplx into this h2_stream as many
         * as we want, since the lifetimes are the same and we are not freeing
         * the ones in h2_mplx->io before this stream is done. */
        status = h2_util_move(stream->bbout, bb, 16 * 1024, &move_all,  
                              "h2_stream_set_response");
    }
    if (APLOGctrace1(stream->session->c)) {
        apr_size_t len = 0;
        int eos = 0;
        h2_util_bb_avail(stream->bbout, &len, &eos);
        ap_log_cerror(APLOG_MARK, APLOG_DEBUG, status, stream->session->c,
                      "h2_stream(%ld-%d): set_response(%s), len=%ld, eos=%d", 
                      stream->session->id, stream->id, response->status,
                      (long)len, (int)eos);
    }
    return status;
}

apr_status_t h2_io_in_read(h2_io *io, apr_bucket_brigade *bb, 
                           apr_size_t maxlen)
{
    apr_off_t start_len = 0;
    apr_bucket *last;
    apr_status_t status;

    if (io->rst_error) {
        return APR_ECONNABORTED;
    }
    
    if (!io->bbin || APR_BRIGADE_EMPTY(io->bbin)) {
        return io->eos_in? APR_EOF : APR_EAGAIN;
    }
    
    apr_brigade_length(bb, 1, &start_len);
    last = APR_BRIGADE_LAST(bb);
    status = h2_util_move(bb, io->bbin, maxlen, NULL, "h2_io_in_read");
    if (status == APR_SUCCESS) {
        apr_bucket *nlast = APR_BRIGADE_LAST(bb);
        apr_off_t end_len = 0;
        apr_brigade_length(bb, 1, &end_len);
        if (last == nlast) {
            return APR_EAGAIN;
        }
        io->input_consumed += (end_len - start_len);
    }
    return status;
}

apr_status_t h2_conn_io_consider_flush(h2_conn_io *io)
{
    apr_status_t status = APR_SUCCESS;
    int flush_now = 0;
    
    /* The HTTP/1.1 network output buffer/flush behaviour does not
     * give optimal performance in the HTTP/2 case, as the pattern of
     * buckets (data/eor/eos) is different.
     * As long as we do not have found out the "best" way to deal with
     * this, force a flush at least every WRITE_BUFFER_SIZE amount
     * of data which seems to work nicely.
     */
    if (io->unflushed) {
        apr_off_t len = 0;
        if (!APR_BRIGADE_EMPTY(io->output)) {
            apr_brigade_length(io->output, 0, &len);
        }
        len += io->buflen;
        flush_now = (len >= WRITE_BUFFER_SIZE);
    }
    
    if (flush_now) {
        return h2_conn_io_flush(io);
    }
    return status;
}

/*
 * This function assumes that either ctx->buffered_bb == NULL, or
 * ctx->buffered_bb is empty, or ctx->buffered_bb == bb
 */
static void setaside_remaining_output(ap_filter_t *f,
                                      core_output_filter_ctx_t *ctx,
                                      apr_bucket_brigade *bb,
                                      conn_rec *c)
{
    if (bb == NULL) {
        return;
    }
    remove_empty_buckets(bb);
    if (!APR_BRIGADE_EMPTY(bb)) {
        c->data_in_output_filters = 1;
        if (bb != ctx->buffered_bb) {
            if (!ctx->deferred_write_pool) {
                apr_pool_create(&ctx->deferred_write_pool, c->pool);
                apr_pool_tag(ctx->deferred_write_pool, "deferred_write");
            }
            ap_save_brigade(f, &(ctx->buffered_bb), &bb,
                            ctx->deferred_write_pool);
        }
    }
    else if (ctx->deferred_write_pool) {
        /*
         * There are no more requests in the pipeline. We can just clear the
         * pool.
         */
        apr_pool_clear(ctx->deferred_write_pool);
    }
}

apr_status_t h2_stream_set_response(h2_stream *stream, h2_response *response,
                                    apr_bucket_brigade *bb)
{
    apr_status_t status = APR_SUCCESS;
    if (!output_open(stream)) {
        ap_log_cerror(APLOG_MARK, APLOG_DEBUG, 0, stream->session->c,
                      "h2_stream(%ld-%d): output closed", 
                      stream->session->id, stream->id);
        return APR_ECONNRESET;
    }
    
    stream->response = response;
    if (bb && !APR_BRIGADE_EMPTY(bb)) {
        int move_all = INT_MAX;
        /* we can move file handles from h2_mplx into this h2_stream as many
         * as we want, since the lifetimes are the same and we are not freeing
         * the ones in h2_mplx->io before this stream is done. */
        H2_STREAM_OUT(APLOG_TRACE2, stream, "h2_stream set_response_pre");
        status = h2_util_move(stream->bbout, bb, 16 * 1024, &move_all,  
                              "h2_stream_set_response");
        H2_STREAM_OUT(APLOG_TRACE2, stream, "h2_stream set_response_post");
    }
    
    ap_log_cerror(APLOG_MARK, APLOG_DEBUG, status, stream->session->c,
                  "h2_stream(%ld-%d): set_response(%d)", 
                  stream->session->id, stream->id, response->http_status);
    return status;
}

apr_status_t h2_stream_readx(h2_stream *stream, 
                             h2_io_data_cb *cb, void *ctx,
                             apr_size_t *plen, int *peos)
{
    if (stream->bbout && !APR_BRIGADE_EMPTY(stream->bbout)) {
        return h2_util_bb_readx(stream->bbout, cb, ctx, plen, peos);
    }
    return h2_mplx_out_readx(stream->m, stream->id, 
                             cb, ctx, plen, peos);
}

apr_status_t h2_stream_read_to(h2_stream *stream, apr_bucket_brigade *bb, 
                               apr_off_t *plen, int *peos)
{
    apr_status_t status = APR_SUCCESS;
    apr_table_t *trailers = NULL;

    H2_STREAM_OUT(APLOG_TRACE2, stream, "h2_stream read_to_pre");
    if (stream->rst_error) {
        return APR_ECONNRESET;
    }
    
    if (APR_BRIGADE_EMPTY(stream->bbout)) {
        apr_off_t tlen = *plen;
        int eos;
        status = h2_mplx_out_read_to(stream->session->mplx, stream->id, 
                                     stream->bbout, &tlen, &eos, &trailers);
    }
    
    if (status == APR_SUCCESS && !APR_BRIGADE_EMPTY(stream->bbout)) {
        status = h2_transfer_brigade(bb, stream->bbout, stream->pool, 
                                     plen, peos);
    }
    else {
        *plen = 0;
        *peos = 0;
    }

    if (trailers && stream->response) {
        ap_log_cerror(APLOG_MARK, APLOG_TRACE1, status, stream->session->c,
                      "h2_stream(%ld-%d): read_to, saving trailers",
                      stream->session->id, stream->id);
        h2_response_set_trailers(stream->response, trailers);
    }
    
    if (status == APR_SUCCESS && !*peos && !*plen) {
        status = APR_EAGAIN;
    }
    H2_STREAM_OUT(APLOG_TRACE2, stream, "h2_stream read_to_post");
    ap_log_cerror(APLOG_MARK, APLOG_TRACE1, status, stream->session->c,
                  "h2_stream(%ld-%d): read_to, len=%ld eos=%d",
                  stream->session->id, stream->id, (long)*plen, *peos);
    return status;
}

apr_status_t h2_conn_io_read(h2_conn_io *io,
                             apr_read_type_e block,
                             h2_conn_io_on_read_cb on_read_cb,
                             void *puser)
{
    apr_status_t status;
    int done = 0;
    ap_log_cerror(APLOG_MARK, APLOG_TRACE2, 0, io->connection,
                  "h2_conn_io: try read, block=%d", block);
    
    if (!APR_BRIGADE_EMPTY(io->input)) {
        /* Seems something is left from a previous read, lets
         * satisfy our caller with the data we already have. */
        status = h2_conn_io_bucket_read(io, block, on_read_cb, puser, &done);
        apr_brigade_cleanup(io->input);
        if (status != APR_SUCCESS || done) {
            return status;
        }
    }

    /* We only do a blocking read when we have no streams to process. So,
     * in httpd scoreboard lingo, we are in a KEEPALIVE connection state.
     * When reading non-blocking, we do have streams to process and update
     * child with NULL request. That way, any current request information
     * in the scoreboard is preserved.
     */
    if (block == APR_BLOCK_READ) {
        ap_update_child_status_from_conn(io->connection->sbh, 
                                         SERVER_BUSY_KEEPALIVE, 
                                         io->connection);
    }
    else {
        ap_update_child_status(io->connection->sbh, SERVER_BUSY_READ, NULL);
    }

    /* TODO: replace this with a connection filter itself, so that we
     * no longer need to transfer incoming buckets to our own brigade. 
     */
    status = ap_get_brigade(io->connection->input_filters,
                            io->input, AP_MODE_READBYTES,
                            block, 64 * 4096);
    switch (status) {
        case APR_SUCCESS:
            return h2_conn_io_bucket_read(io, block, on_read_cb, puser, &done);
        case APR_EOF:
        case APR_EAGAIN:
            break;
        default:
            ap_log_cerror(APLOG_MARK, APLOG_DEBUG, status, io->connection,
                          "h2_conn_io: error reading");
            break;
    }
    return status;
}

int h2_beam_empty(h2_bucket_beam *beam)
{
    int empty = 1;
    h2_beam_lock bl;
    
    if (enter_yellow(beam, &bl) == APR_SUCCESS) {
        empty = (H2_BLIST_EMPTY(&beam->red) 
                 && (!beam->green || APR_BRIGADE_EMPTY(beam->green)));
        leave_yellow(beam, &bl);
    }
    return empty;
}

static apr_status_t out_write(h2_mplx *m, h2_io *io, 
                              ap_filter_t* f, apr_bucket_brigade *bb,
                              apr_table_t *trailers,
                              struct apr_thread_cond_t *iowait)
{
    apr_status_t status = APR_SUCCESS;
    /* We check the memory footprint queued for this stream_id
     * and block if it exceeds our configured limit.
     * We will not split buckets to enforce the limit to the last
     * byte. After all, the bucket is already in memory.
     */
    while (status == APR_SUCCESS 
           && !APR_BRIGADE_EMPTY(bb) 
           && !is_aborted(m, &status)) {
        
        status = h2_io_out_write(io, bb, m->stream_max_mem, trailers,
                                 &m->tx_handles_reserved);
        /* Wait for data to drain until there is room again or
         * stream timeout expires */
        h2_io_signal_init(io, H2_IO_WRITE, m->stream_timeout, iowait);
        while (status == APR_SUCCESS
               && !APR_BRIGADE_EMPTY(bb) 
               && iowait
               && (m->stream_max_mem <= h2_io_out_length(io))
               && !is_aborted(m, &status)) {
            trailers = NULL;
            if (f) {
                ap_log_cerror(APLOG_MARK, APLOG_TRACE1, status, f->c,
                              "h2_mplx(%ld-%d): waiting for out drain", 
                              m->id, io->id);
            }
            status = h2_io_signal_wait(m, io);
        }
        h2_io_signal_exit(io);
    }
    apr_brigade_cleanup(bb);
    
    return status;
}

static apr_status_t h2_conn_io_bucket_read(h2_conn_io *io,
                                           apr_read_type_e block,
                                           h2_conn_io_on_read_cb on_read_cb,
                                           void *puser, int *pdone)
{
    apr_status_t status = APR_SUCCESS;
    apr_size_t readlen = 0;
    *pdone = 0;
    
    while (status == APR_SUCCESS && !*pdone
           && !APR_BRIGADE_EMPTY(io->input)) {
        
        apr_bucket* bucket = APR_BRIGADE_FIRST(io->input);
        if (APR_BUCKET_IS_METADATA(bucket)) {
            /* we do nothing regarding any meta here */
        }
        else {
            const char *bucket_data = NULL;
            apr_size_t bucket_length = 0;
            status = apr_bucket_read(bucket, &bucket_data,
                                     &bucket_length, block);
            
            if (status == APR_SUCCESS && bucket_length > 0) {
                if (APLOGctrace2(io->connection)) {
                    char buffer[32];
                    h2_util_hex_dump(buffer, sizeof(buffer)/sizeof(buffer[0]),
                                     bucket_data, bucket_length);
                    ap_log_cerror(APLOG_MARK, APLOG_TRACE2, 0, io->connection,
                                  "h2_conn_io(%ld): read %d bytes: %s",
                                  io->connection->id, (int)bucket_length, buffer);
                }
                
                if (bucket_length > 0) {
                    apr_size_t consumed = 0;
                    status = on_read_cb(bucket_data, bucket_length,
                                        &consumed, pdone, puser);
                    if (status == APR_SUCCESS && bucket_length > consumed) {
                        /* We have data left in the bucket. Split it. */
                        status = apr_bucket_split(bucket, consumed);
                    }
                    readlen += consumed;
                }
            }
        }
        apr_bucket_delete(bucket);
    }
    if (readlen == 0 && status == APR_SUCCESS && block == APR_NONBLOCK_READ) {
        return APR_EAGAIN;
    }
    return status;
}