/* ====================================================================
 *    Licensed to the Apache Software Foundation (ASF) under one
 *    or more contributor license agreements.  See the NOTICE file
 *    distributed with this work for additional information
 *    regarding copyright ownership.  The ASF licenses this file
 *    to you under the Apache License, Version 2.0 (the
 *    "License"); you may not use this file except in compliance
 *    with the License.  You may obtain a copy of the License at
 *
 *      http://www.apache.org/licenses/LICENSE-2.0
 *
 *    Unless required by applicable law or agreed to in writing,
 *    software distributed under the License is distributed on an
 *    "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
 *    KIND, either express or implied.  See the License for the
 *    specific language governing permissions and limitations
 *    under the License.
 * ====================================================================
 */

#include <stdlib.h>

#include <apr_pools.h>
#include <apr_strings.h>

#include "serf.h"
#include "serf_bucket_util.h"
#include "serf_private.h"

#include "protocols/http2_buckets.h"
#include "protocols/http2_protocol.h"

struct serf_http2_stream_data_t
{
    serf_request_t *request; /* May be NULL as streams may outlive requests */
    serf_incoming_request_t *in_request;
    serf_bucket_t *response_agg;
    serf_hpack_table_t *tbl;
    serf_bucket_t *data_tail;
    bool resetted;
};

serf_http2_stream_t *
serf_http2__stream_create(serf_http2_protocol_t *h2,
                          apr_int32_t streamid,
                          apr_uint32_t lr_window,
                          apr_uint32_t rl_window,
                          serf_bucket_alloc_t *alloc)
{
    serf_http2_stream_t *stream = serf_bucket_mem_alloc(alloc,
                                                        sizeof(*stream));

    stream->h2 = h2;
    stream->alloc = alloc;

    stream->next = stream->prev = NULL;

    /* Delay creating this? */
    stream->data = serf_bucket_mem_alloc(alloc, sizeof(*stream->data));
    stream->data->request = NULL;
    stream->data->in_request = NULL;
    stream->data->response_agg = NULL;
    stream->data->tbl = NULL;
    stream->data->data_tail = NULL;
    stream->data->resetted = false;

    stream->lr_window = lr_window;
    stream->rl_window = rl_window;

    stream->rl_window_upd_below = 1024 * 1024;
    stream->rl_window_upd_to = 16 * 1024 * 1024;

    if (streamid >= 0)
        stream->streamid = streamid;
    else
        stream->streamid = -1; /* Undetermined yet */

    stream->status = (streamid >= 0) ? H2S_IDLE : H2S_INIT;
    stream->new_reserved_stream = NULL;

    stream->prev_writable = stream->next_writable = NULL;

    return stream;
}

void
serf_http2__stream_pre_cleanup(serf_http2_stream_t *stream)
{
    if (stream->data) {
        if (stream->data->data_tail) {
            serf_bucket_destroy(stream->data->data_tail);
            stream->data->data_tail = NULL;
        }
    }
}

void
serf_http2__stream_cleanup(serf_http2_stream_t *stream)
{
    if (stream->data) {
        if (stream->data->response_agg)
            serf_bucket_destroy(stream->data->response_agg);

        serf_bucket_mem_free(stream->alloc, stream->data);
        stream->data = NULL;
    }
    serf_bucket_mem_free(stream->alloc, stream);
}

static apr_status_t stream_send_headers(serf_http2_stream_t *stream,
                                        serf_bucket_t *hpack,
                                        apr_size_t max_payload_size,
                                        bool end_stream,
                                        bool priority)
{
    apr_status_t status;
    bool first_frame = true;

    /* And now schedule the packet for writing. Note that it is required
    by the HTTP/2 spec to send HEADERS and CONTINUATION directly after
    each other, without other frames inbetween. */
    while (hpack != NULL)
    {
        serf_bucket_t *next;
        apr_uint64_t remaining;

        /* hpack buckets implement get_remaining. And if they didn't adding the
        framing around them would apply some reads that fix the buckets.

        So we can ignore the theoretical endless loop here for two different
        reasons
        */
        remaining = serf_bucket_get_remaining(hpack);

        if (remaining > max_payload_size) {
            serf_bucket_split_create(&next, &hpack, hpack,
                                     max_payload_size - (max_payload_size / 4),
                                     max_payload_size);
        }
        else
        {
            next = hpack;
            hpack = NULL;
        }

        next = serf__bucket_http2_frame_create(next,
                                               first_frame
                                               ? HTTP2_FRAME_TYPE_HEADERS
                                               : HTTP2_FRAME_TYPE_CONTINUATION,
                                               (end_stream
                                                ? HTTP2_FLAG_END_STREAM
                                                : 0)
                                               | ((hpack != NULL)
                                                  ? 0
                                                  : HTTP2_FLAG_END_HEADERS)
                                               | (priority
                                                  ? HTTP2_FLAG_PRIORITY
                                                  : 0),
                                               &stream->streamid,
                                               serf_http2__allocate_stream_id,
                                               stream,
                                               max_payload_size,
                                               next->allocator);
        status = serf_http2__enqueue_frame(stream->h2, next, TRUE);

        if (SERF_BUCKET_READ_ERROR(status))
            return status; /* Connection dead */

        first_frame = false; /* Continue with 'continuation' frames */
    }

    return APR_SUCCESS;
}

typedef struct window_allocate_info_t
{
    serf_http2_stream_t *stream;
    serf_bucket_t *bkt;
    apr_size_t allocated;
} window_allocate_info_t;

static apr_status_t data_write_started(void *baton,
                                       apr_uint64_t bytes_read)
{
    window_allocate_info_t *wai = baton;

    bytes_read = serf_bucket_get_remaining(wai->bkt);

    /* Handles unavailable for free */
    if (bytes_read <= wai->allocated) {
        /* Nice, we can return something now */
        apr_size_t to_much = (wai->allocated - (apr_size_t)bytes_read);

        serf_http2__return_window(wai->stream->h2, wai->stream, to_much);

        wai->allocated = 0;
    }
    return APR_SUCCESS;
}

static apr_status_t data_write_done(void *baton,
                                    apr_uint64_t bytes_read)
{
    window_allocate_info_t *wai = baton;

    if (wai->allocated && bytes_read <= wai->allocated) {
        /* Nice, we can return something now */
        apr_size_t to_much = (wai->allocated - (apr_size_t)bytes_read);
        wai->stream->lr_window += to_much;

        serf_http2__return_window(wai->stream->h2, wai->stream, to_much);

        wai->allocated = 0;
    }

    serf_bucket_mem_free(wai->stream->alloc, wai);
    return APR_SUCCESS;
}


static apr_status_t stream_send_data(serf_http2_stream_t *stream,
                                     serf_bucket_t *data)
{
    apr_uint64_t remaining;
    serf_bucket_t *next;
    apr_size_t prefix_len;
    bool end_stream;
    apr_status_t status;

    SERF_H2_assert(stream->status == H2S_OPEN
                   || stream->status == H2S_HALFCLOSED_REMOTE);
    SERF_H2_assert(!stream->data->data_tail || (data ==
                                                stream->data->data_tail));

    /* Sending DATA frames over HTTP/2 is not easy as this usually requires
       handling windowing, priority, etc. This code will improve over time */
    stream->data->data_tail = NULL;

    if (!data)
        remaining = 0;
    else
        remaining = serf_bucket_get_remaining(data);

    /* If the stream decided we are already done */
    if (remaining == 0) {
        if (stream->status == H2S_OPEN)
            stream->status = H2S_HALFCLOSED_LOCAL;
        else
            stream->status = H2S_CLOSED;

        serf_bucket_destroy(data);

        next = serf__bucket_http2_frame_create(NULL, HTTP2_FRAME_TYPE_DATA,
                                               HTTP2_FLAG_END_STREAM,
                                               &stream->streamid,
                                               serf_http2__allocate_stream_id,
                                               stream, 0, stream->alloc);
        return serf_http2__enqueue_frame(stream->h2, next, false);
    }

    prefix_len = serf_http2__alloc_window(stream->h2, stream,
                                          (remaining >= APR_SIZE_MAX)
                                          ? SERF_READ_ALL_AVAIL
                                          : (apr_size_t)remaining);

    if (prefix_len == 0) {
        /* No window left */
        stream->data->data_tail = data;

        /* Write more later */
        serf_http2__ensure_writable(stream);

        return APR_SUCCESS;
    }

    if (prefix_len < remaining) {
        window_allocate_info_t *wai;
        serf_bucket_split_create(&data, &stream->data->data_tail, data,
                                 MIN(prefix_len, 1024), prefix_len);

        wai = serf_bucket_mem_alloc(stream->alloc, sizeof(*wai));
        wai->stream = stream;
        wai->bkt = data;
        wai->allocated = prefix_len;

        data = serf__bucket_event_create(data, wai,
                                         data_write_started,
                                         data_write_done, NULL, stream->alloc);
        end_stream = false;

        serf_http2__ensure_writable(stream);
    }
    else {
        end_stream = true;

        if (stream->status == H2S_OPEN)
            stream->status = H2S_HALFCLOSED_LOCAL;
        else
            stream->status = H2S_CLOSED;
    }

    next = serf__bucket_http2_frame_create(data, HTTP2_FRAME_TYPE_DATA,
                                           end_stream ? HTTP2_FLAG_END_STREAM
                                                      : 0,
                                           &stream->streamid,
                                           serf_http2__allocate_stream_id,
                                           stream, prefix_len,
                                           data->allocator);

    status = serf_http2__enqueue_frame(stream->h2, next, TRUE);

    return status;
}

apr_status_t
serf_http2__stream_write_data(serf_http2_stream_t *stream)
{
    SERF_H2_assert(stream->status == H2S_OPEN
                   || stream->status == H2S_HALFCLOSED_REMOTE);
    SERF_H2_assert(stream->data->data_tail != NULL);

    return stream_send_data(stream, stream->data->data_tail);
}

static apr_status_t destroy_request_bucket(void *baton,
                                           apr_uint64_t bytes_read)
{
    serf_request_t *request = baton;

    serf_bucket_destroy(request->req_bkt);
    request->req_bkt = NULL;
    request->writing = SERF_WRITING_FINISHED;
    return APR_SUCCESS;
}

apr_status_t
serf_http2__stream_setup_next_request(serf_http2_stream_t *stream,
                                      serf_connection_t *conn,
                                      apr_size_t max_payload_size,
                                      serf_hpack_table_t *hpack_tbl)
{
    serf_request_t *request = conn->unwritten_reqs;
    apr_status_t status;
    serf_bucket_t *hpack;
    serf_bucket_t *body;
    bool end_stream;
    bool priority = false;

    SERF_H2_assert(request != NULL);
    if (!request)
        return APR_EGENERAL;

    stream->data->request = request;
    request->protocol_baton = stream;

    if (!request->req_bkt) {
        status = serf__setup_request(request);
        if (status)
            return status;
    }

    conn->unwritten_reqs = request->next;
    if (conn->unwritten_reqs_tail == request)
        conn->unwritten_reqs = conn->unwritten_reqs_tail = NULL;

    request->next = NULL;

    serf__link_requests(&conn->written_reqs, &conn->written_reqs_tail,
                        request);
    conn->nr_of_written_reqs++;
    conn->nr_of_unwritten_reqs--;

    serf__bucket_request_read(request->req_bkt, &body, NULL, NULL);
    status = serf__bucket_hpack_create_from_request(
                            &hpack, hpack_tbl,
                            request->req_bkt,
                            request->conn->host_info.scheme,
                            request->allocator);
    if (status)
        return status;

    if (request->depends_on && request->depends_on->protocol_baton)
    {
        serf_http2_stream_t *ds = request->depends_on->protocol_baton;

        if (ds->streamid >= 0) {
            serf_bucket_t *agg;
            unsigned char priority_data[5];

            agg = serf_bucket_aggregate_create(request->allocator);

            priority_data[0] = (ds->streamid >> 24) & 0x7F;
            /* bit 7 of [0] is the exclusive flag */
            priority_data[1] = (ds->streamid >> 16) & 0xFF;
            priority_data[2] = (ds->streamid >> 8) & 0xFF;
            priority_data[3] = ds->streamid & 0xFF;
            priority_data[4] = request->dep_priority >> 8;

            serf_bucket_aggregate_append(
                agg,
                serf_bucket_simple_copy_create((void *)priority_data,
                                               5, request->allocator));

            serf_bucket_aggregate_append(agg, hpack);
            hpack = agg;

            priority = true;
        }
    }

    if (!body) {
        serf_bucket_destroy(request->req_bkt);
        request->req_bkt = NULL;
        end_stream = true;
    }
    else
        end_stream = false;

    status = stream_send_headers(stream, hpack, max_payload_size,
                                 end_stream, priority);
    if (status)
        return status;

    if (end_stream) {
        stream->status = H2S_HALFCLOSED_LOCAL; /* Headers sent; no body */
        return APR_SUCCESS;
    }

    /* Yuck... we are not allowed to destroy body */
    body = serf_bucket_barrier_create(body, request->allocator);

    /* Setup an event bucket to destroy the actual request bucket when
       the body is done */
    body = serf__bucket_event_create(body, request,
                                     NULL, NULL, destroy_request_bucket,
                                     request->allocator);

    stream->status = H2S_OPEN; /* Headers sent. Body to go */
    request->writing = SERF_WRITING_STARTED;
    return stream_send_data(stream, body);
}

apr_status_t
serf_http2__stream_reset(serf_http2_stream_t *stream,
                         apr_status_t reason,
                         bool local_reset)
{
    stream->status = H2S_CLOSED;

    if (stream->streamid < 0 || stream->data->resetted)
        return APR_SUCCESS;

    stream->data->resetted = true;

    if (local_reset)
        return serf_http2__enqueue_stream_reset(stream->h2,
                                                stream->streamid,
                                                reason);

    return APR_SUCCESS;
}

void serf_http2__stream_cancel_request(serf_http2_stream_t *stream,
                                       serf_request_t *rq,
                                       apr_status_t reason)
{
    if (stream->streamid < 0)
        return; /* Never hit the wire */
    else if (stream->status == H2S_CLOSED)
        return; /* We are already detached */

    if (reason < SERF_ERROR_HTTP2_NO_ERROR
        || reason > SERF_ERROR_HTTP2_HTTP_1_1_REQUIRED)
    {
        reason = SERF_ERROR_HTTP2_CANCEL;
    }

    stream->data->request = rq; /* Might have changed! */

    /* Let the other party know we don't want anything */
    serf_http2__stream_reset(stream, reason, true);

}

void serf_http2__stream_prioritize_request(serf_http2_stream_t *stream,
                                           serf_request_t *rq,
                                           bool exclusive)
{
    if (stream->streamid < 0)
        return; /* Never hit the wire */
    else if (stream->status == H2S_CLOSED)
        return; /* We are already detached */

    /* Ignore for now. We start by handling this at setup */
}


static apr_status_t
stream_response_eof(void *baton,
                    serf_bucket_t *aggregate_bucket)
{
    serf_http2_stream_t *stream = baton;

    if (stream->data->resetted)
        return APR_EAGAIN;

    switch (stream->status)
    {
        case H2S_CLOSED:
        case H2S_HALFCLOSED_REMOTE:
            return APR_EOF;
        default:
            return APR_EAGAIN;
    }
}

static int set_hpack_header(void *baton,
                            const char *key,
                            const char *value)
{
    serf_bucket_t *hpack = baton;

    serf__bucket_hpack_setc(hpack, key, value);
    return 0;
}

static apr_status_t
http2_stream_enqueue_response(serf_incoming_request_t *request,
                              void *enqueue_baton,
                              serf_bucket_t *response_bkt)
{
    serf_http2_stream_t *stream = enqueue_baton;
    serf_bucket_t *hpack;
    serf_bucket_t *headers;
    serf_bucket_t *h1_response;
    serf_status_line sline;
    apr_status_t status;

    /* OK, this could be implemented using knowledge of the buckets, in
       a 100% more efficient, but I don't want to introduce new bucket
       types for this yet. Let's just read everything the http/1 way
       and put it in HTTP/2 appropriate places */
    h1_response = serf_bucket_response_create(response_bkt,
                                              stream->alloc);
    do
    {
        status = serf_bucket_response_status(h1_response, &sline);
    } while (status != APR_SUCCESS);

    if (status != APR_SUCCESS)
        return APR_EGENERAL; /* Can't read statusline. No EAGAIN support before
                                the body (yet) */

    hpack = serf__bucket_hpack_create(stream->data->tbl, stream->alloc);
    serf__bucket_hpack_setc(hpack, ":status",
                            apr_itoa(stream->data->in_request->pool,
                                     sline.code));

    do
    {
        status = serf_bucket_response_wait_for_headers(h1_response);
    } while (status != APR_SUCCESS);

    if (status != APR_SUCCESS)
        return APR_EGENERAL; /* Can't read body. No EAGAIN support before
                                the body (yet) */

    headers = serf_bucket_response_get_headers(h1_response);

    serf_bucket_headers_do(headers, set_hpack_header, hpack);

    status = stream_send_headers(stream, hpack,
                                 serf_http2__max_payload_size(stream->h2),
                                 false /* eos */, false /* priority */);

    if (status)
        return status;

    return stream_send_data(stream, response_bkt);
}

static apr_status_t
stream_setup_response(serf_http2_stream_t *stream,
                      serf_config_t *config)
{
    serf_bucket_t *agg;
    apr_status_t status;

    agg = serf_bucket_aggregate_create(stream->alloc);
    serf_bucket_aggregate_hold_open(agg, stream_response_eof, stream);

    serf_bucket_set_config(agg, config);
    stream->data->response_agg = agg;

    if (stream->data->request) {
        serf_request_t *request = stream->data->request;

        if (!request->resp_bkt) {
            apr_pool_t *scratch_pool = request->respool; /* ### Pass scratch pool */

            request->resp_bkt = request->acceptor(request, agg,
                                                  request->acceptor_baton,
                                                  scratch_pool);
        }
    }
    else {
        serf_incoming_request_t *in_request = stream->data->in_request;

        if (!in_request) {
            serf_incoming_request_setup_t req_setup;
            void *req_setup_baton;

            status = serf_http2__setup_incoming_request(&in_request, &req_setup,
                                                        &req_setup_baton,
                                                        stream->h2);

            if (status)
                return status;

            stream->data->in_request = in_request;

            status = req_setup(&in_request->req_bkt, agg,
                               in_request, req_setup_baton,
                               &in_request->handler,
                               &in_request->handler_baton,
                               &in_request->response_setup,
                               &in_request->response_setup_baton,
                               in_request->pool);

            if (status)
                return status;

            stream->status = H2S_OPEN;

            in_request->enqueue_response = http2_stream_enqueue_response;
            in_request->enqueue_baton = stream;
        }
    }

    return APR_SUCCESS;
}

static apr_status_t
stream_promise_done(void *baton,
                    serf_bucket_t *done_agg)
{
    serf_http2_stream_t *parent_stream = baton;
    serf_http2_stream_t *stream = parent_stream->new_reserved_stream;

    SERF_H2_assert(stream != NULL);
    SERF_H2_assert(stream->status == H2S_RESERVED_REMOTE);
    parent_stream->new_reserved_stream = NULL; /* End of PUSH_PROMISE */

    /* Anything else? */


    /* ### Absolute minimal implementation.
           Just sending that we are not interested in the initial SETTINGS
           would be the easier approach. */
    serf_http2__stream_reset(stream, SERF_ERROR_HTTP2_REFUSED_STREAM, TRUE);


    /* Exit condition:
       * Either we should accept the stream and are ready to receive
         HEADERS and DATA on it.
       * Or we aren't and reject the stream
     */
    SERF_H2_assert(stream->status == H2S_CLOSED
                   || stream->data->request != NULL);

    /* We must return a proper error or EOF here! */
    return APR_EOF;
}

serf_bucket_t *
serf_http2__stream_handle_hpack(serf_http2_stream_t *stream,
                                serf_bucket_t *bucket,
                                unsigned char frametype,
                                bool end_stream,
                                apr_size_t max_entry_size,
                                serf_hpack_table_t *hpack_tbl,
                                serf_config_t *config,
                                serf_bucket_alloc_t *allocator)
{
    if (frametype == HTTP2_FRAME_TYPE_HEADERS) {

        if (!stream->data->response_agg)
            stream_setup_response(stream, config);

        stream->data->tbl = hpack_tbl;

        bucket = serf__bucket_hpack_decode_create(bucket, max_entry_size,
                                                  hpack_tbl, allocator);

        serf_bucket_aggregate_append(stream->data->response_agg, bucket);

        if (end_stream) {

            if (stream->status == H2S_HALFCLOSED_LOCAL)
                stream->status = H2S_CLOSED;
            else
                stream->status = H2S_HALFCLOSED_REMOTE;
        }
        return NULL; /* We want to drain the bucket ourselves */
    }
    else
    {
        serf_bucket_t *agg;
        SERF_H2_assert(frametype == HTTP2_FRAME_TYPE_PUSH_PROMISE);

        /* First create the HPACK decoder as requested */

     /* TODO: Store key+value somewhere to allow asking the application
             if it is interested in the promised stream.

             Most likely it is not interested *yet* as the HTTP/2 spec
             recommends pushing promised items *before* the stream that
             references them.

             So we probably want to store the request anyway, to allow
             matching this against a later added outgoing request.
     */
        bucket = serf__bucket_hpack_decode_create(bucket, max_entry_size,
                                                  hpack_tbl, allocator);

        /* And now wrap around it the easiest way to get an EOF callback */
        agg = serf_bucket_aggregate_create(allocator);
        serf_bucket_aggregate_append(agg, bucket);

        serf_bucket_aggregate_hold_open(agg, stream_promise_done, stream);

        /* And return the aggregate, so the bucket will be drained for us */
        return agg;
    }
}

serf_bucket_t *
serf_http2__stream_handle_data(serf_http2_stream_t *stream,
                               serf_bucket_t *bucket,
                               unsigned char frametype,
                               bool end_stream,
                               serf_config_t *config,
                               serf_bucket_alloc_t *allocator)
{
    if (!stream->data->response_agg)
        stream_setup_response(stream, config);

    serf_bucket_aggregate_append(stream->data->response_agg, bucket);

    if (end_stream) {

        if (stream->status == H2S_HALFCLOSED_LOCAL)
            stream->status = H2S_CLOSED;
        else
            stream->status = H2S_HALFCLOSED_REMOTE;
    }

    return NULL;
}

apr_status_t
serf_http2__stream_processor(void *baton,
                             serf_http2_protocol_t *h2,
                             serf_bucket_t *bucket)
{
    serf_http2_stream_t *stream = baton;
    serf_http2_stream_data_t *sd = stream->data;
    apr_status_t status = APR_SUCCESS;

    SERF_H2_assert(stream->data->response_agg != NULL
                   && !stream->data->resetted);

    if (sd->request) {

        SERF_H2_assert(sd->request->resp_bkt != NULL);

        /* Response handlers are expected to read until they get some error,
           but at least some implementations assume that just returning
           APR_SUCCESS will have them called again, as that used to work as
           an APR_EAGAIN like system in HTTP/1.

           But we can't just fall back with HTTP/2, as we might still have
           some part of the frame open (good case), or we might have completed
           the frame and are never called again. */
        do {
            status = serf__handle_response(sd->request,
                                           sd->request->respool);
        } while (status == APR_SUCCESS);

        if (sd->resetted) {
            status = APR_EOF;
        }
        else if (!APR_STATUS_IS_EOF(status)
                 && !SERF_BUCKET_READ_ERROR(status))
        {
            return status;
        }

          /* Ok, the request thinks is done, let's handle the bookkeeping,
             to remove it from the outstanding requests */
        {
            serf_connection_t *conn = serf_request_get_conn(sd->request);
            serf_request_t **rq = &conn->written_reqs;
            serf_request_t *last = NULL;

            while (*rq && (*rq != sd->request)) {
                last = *rq;
                rq = &last->next;
            }

            if (*rq)
            {
                (*rq) = sd->request->next;

                if (conn->written_reqs_tail == sd->request)
                    conn->written_reqs_tail = last;

                conn->nr_of_written_reqs--;
            }

            serf__destroy_request(sd->request);
            stream->data->request = NULL;
        }

        if (SERF_BUCKET_READ_ERROR(status)) {

            if (stream->status != H2S_CLOSED) {
              /* Tell the other side that we are no longer interested
                 to receive more data */
                serf_http2__stream_reset(stream, status, TRUE);
            }

            return status;
        }

        SERF_H2_assert(APR_STATUS_IS_EOF(status));

        /* Even though the request reported that it is done, we might not
           have read all the data that we should (*cough* padding *cough*),
           or perhaps an invalid 'Content-Length' value; maybe both.

           This may even handle not-interested - return EOF cases, but that
           would have broken the pipeline for HTTP/1.1.
           */

        /* ### For now, fall through and eat whatever is left.
               Usually this is 0 bytes */

        status = APR_SUCCESS;
    }
    else if (stream->data->in_request) {
        serf_incoming_request_t *request = stream->data->in_request;

        SERF_H2_assert(request->req_bkt != NULL);

        status = request->handler(request, request->req_bkt,
                                  request->handler_baton,
                                  request->pool);

        if (!APR_STATUS_IS_EOF(status)
            && !SERF_BUCKET_READ_ERROR(status))
            return status;

        if (APR_STATUS_IS_EOF(status)) {
            status = serf_incoming_response_create(request);

            if (status)
                return status;
        }

        if (SERF_BUCKET_READ_ERROR(status)) {

            if (stream->status != H2S_CLOSED) {
                /* Tell the other side that we are no longer interested
                to receive more data */
                serf_http2__stream_reset(stream, status, TRUE);
            }

            return status;
        }
    }

    while (!status)
    {
        struct iovec vecs[SERF__STD_IOV_COUNT];
        int vecs_used;

        /* Drain the bucket as efficiently as possible */
        status = serf_bucket_read_iovec(stream->data->response_agg,
                                        SERF_READ_ALL_AVAIL, COUNT_OF(vecs),
                                        vecs, &vecs_used);

        if (vecs_used) {
          /* We have data... What should we do with it? */
        }
    }

    if ((APR_STATUS_IS_EOF(status) || sd->resetted)
        && (stream->status == H2S_CLOSED
            || stream->status == H2S_HALFCLOSED_REMOTE))
    {
      /* If there was a request, it is already gone, so we can now safely
         destroy our aggregate which may include everything upto the http2
         frames */
        serf_bucket_destroy(stream->data->response_agg);
        stream->data->response_agg = NULL;
        status = APR_EOF;
    }

    return status;
}