| /* ==================================================================== |
| * 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; |
| } |
