Author: rhuijben
Date: Fri Oct 30 13:35:07 2015
New Revision: 1711471
URL: http://svn.apache.org/viewvc?rev=1711471&view=rev
Log:
Add a simple bucket that reads a prefix from a stream before the actual data
starts. This prefix is than handed to a callback before the actual reading
starts.
This bucket makes it easy to hand of the stream containing the actual data
before some fixed size header is read and handle the result event-driven.
(Something that is very useful in the http2 protocol handling for standard
frames)
* buckets/prefix_buckets.c
New file.
(prefix_context_t): New struct.
(serf_bucket_prefix_create): New function.
(read_prefix,
serf_prefix_read,
serf_prefix_read_iovec,
serf_prefix_peek,
serf_prefix_get_remaining,
serf_prefix_set_config,
serf_prefix_destroy): New static functions.
(serf_bucket_type_prefix): New bucket type.
* serf_bucket_types.h
(serf_bucket_type_prefix): New bucket type.
(SERF_BUCKET_IS_PREFIX): New define.
(serf_bucket_prefix_handler_t): New typedef.
(serf_bucket_prefix_create): New function.
* test/test_buckets.c
(test_random_eagain_in_response): Property undefine BODY.
(prefix_cb): New struct.
(prefix_callback,
test_prefix_buckets): New function.
(test_buckets): Add test_prefix_buckets.
Added:
serf/trunk/buckets/prefix_buckets.c (with props)
Modified:
serf/trunk/serf_bucket_types.h
serf/trunk/test/test_buckets.c
Added: serf/trunk/buckets/prefix_buckets.c
URL:
http://svn.apache.org/viewvc/serf/trunk/buckets/prefix_buckets.c?rev=1711471&view=auto
==============================================================================
--- serf/trunk/buckets/prefix_buckets.c (added)
+++ serf/trunk/buckets/prefix_buckets.c Fri Oct 30 13:35:07 2015
@@ -0,0 +1,244 @@
+/* ====================================================================
+ * 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 <apr_pools.h>
+
+#include "serf.h"
+#include "serf_bucket_util.h"
+
+typedef struct prefix_context_t {
+ serf_bucket_t *stream;
+ apr_size_t prefix_len;
+ apr_size_t read_len;
+
+ serf_bucket_prefix_handler_t handler;
+ void *handler_baton;
+
+ char *buffer;
+} prefix_context_t;
+
+
+serf_bucket_t *serf_bucket_prefix_create(serf_bucket_t *stream,
+ apr_size_t prefix_len,
+ serf_bucket_prefix_handler_t handler,
+ void *handler_baton,
+ serf_bucket_alloc_t *allocator)
+{
+ prefix_context_t *ctx = serf_bucket_mem_alloc(allocator, sizeof(*ctx));
+
+ ctx->stream = stream;
+ ctx->prefix_len = prefix_len;
+ ctx->read_len = 0;
+
+ ctx->handler = handler;
+ ctx->handler_baton = handler_baton;
+
+ ctx->buffer = NULL;
+
+ return serf_bucket_create(&serf_bucket_type_prefix, allocator, ctx);
+}
+
+static apr_status_t read_prefix(serf_bucket_t *bucket)
+{
+ prefix_context_t *ctx = bucket->data;
+ const char *data;
+ apr_size_t len;
+ apr_status_t status;
+
+ if (!ctx->read_len) {
+
+ /* Perhaps we can handle this without copying any data? */
+ status = serf_bucket_read(ctx->stream, ctx->prefix_len, &data,
+ &len);
+
+ if (SERF_BUCKET_READ_ERROR(status))
+ return status;
+
+ if (APR_STATUS_IS_EOF(status) || (len == ctx->prefix_len)) {
+ apr_status_t cb_status;
+
+ /* Prefix reading is done */
+ ctx->prefix_len = 0;
+
+ cb_status = ctx->handler(ctx->handler_baton, ctx->stream,
+ data, len);
+
+ if (SERF_BUCKET_READ_ERROR(cb_status))
+ return cb_status;
+
+ return status;
+ }
+ else if (len == 0) {
+ /* Nothing read at all. Try again later */
+ return APR_EAGAIN;
+ }
+
+ /* Create a buffer to hold what we already read */
+ ctx->buffer = serf_bucket_mem_alloc(bucket->allocator,
ctx->prefix_len);
+ memcpy(ctx->buffer, data, len);
+ ctx->read_len = len;
+
+ if (status)
+ return status;
+
+ /* Else: Try filling the rest of the buffer */
+ }
+
+ while (TRUE) {
+
+ status = serf_bucket_read(ctx->stream, ctx->prefix_len - ctx->read_len,
+ &data, &len);
+
+ if (SERF_BUCKET_READ_ERROR(status))
+ return status;
+
+ memcpy(ctx->buffer + ctx->read_len, data, len);
+ ctx->read_len += len;
+
+ if (APR_STATUS_IS_EOF(status) || (ctx->prefix_len == ctx->read_len)) {
+ apr_status_t cb_status;
+
+ /* Prefix reading is done */
+ ctx->prefix_len = 0;
+
+ cb_status = ctx->handler(ctx->handler_baton, ctx->stream,
+ ctx->buffer, ctx->read_len);
+
+ serf_bucket_mem_free(bucket->allocator, ctx->buffer);
+ ctx->buffer = NULL;
+
+ if (SERF_BUCKET_READ_ERROR(cb_status))
+ return cb_status;
+
+ return status;
+ }
+ else if (status)
+ return status;
+ }
+}
+
+static apr_status_t serf_prefix_read(serf_bucket_t *bucket,
+ apr_size_t requested,
+ const char **data,
+ apr_size_t *len)
+{
+ prefix_context_t *ctx = bucket->data;
+
+ if (ctx->prefix_len) {
+ apr_status_t status = read_prefix(bucket);
+
+ if (status) {
+ *len = 0;
+ return status;
+ }
+ }
+
+ return serf_bucket_read(ctx->stream, requested, data, len);
+}
+
+static apr_status_t serf_prefix_read_iovec(serf_bucket_t *bucket,
+ apr_size_t requested,
+ int vecs_size,
+ struct iovec *vecs,
+ int *vecs_used)
+{
+ prefix_context_t *ctx = bucket->data;
+
+ if (ctx->prefix_len) {
+ apr_status_t status = read_prefix(bucket);
+
+ if (status) {
+ *vecs_used = 0;
+ return status;
+ }
+ }
+
+ return serf_bucket_read_iovec(ctx->stream, requested, vecs_size,
+ vecs, vecs_used);
+}
+
+static apr_status_t serf_prefix_peek(serf_bucket_t *bucket,
+ const char **data,
+ apr_size_t *len)
+{
+ prefix_context_t *ctx = bucket->data;
+
+ if (ctx->prefix_len) {
+ apr_status_t status = read_prefix(bucket);
+
+ if (status) {
+ *len = 0;
+ return status;
+ }
+ }
+
+ return serf_bucket_peek(ctx->stream, data, len);
+}
+
+static apr_uint64_t serf_prefix_get_remaining(serf_bucket_t *bucket)
+{
+ prefix_context_t *ctx = bucket->data;
+ apr_uint64_t remaining;
+
+ remaining = serf_bucket_get_remaining(ctx->stream);
+
+ if (remaining != SERF_LENGTH_UNKNOWN && ctx->prefix_len) {
+ remaining -= (ctx->prefix_len - ctx->read_len);
+ }
+
+ return remaining;
+}
+
+static apr_status_t serf_prefix_set_config(serf_bucket_t *bucket,
+ serf_config_t *config)
+{
+ prefix_context_t *ctx = bucket->data;
+
+ return serf_bucket_set_config(ctx->stream, config);
+}
+
+static void serf_prefix_destroy(serf_bucket_t *bucket)
+{
+ prefix_context_t *ctx = bucket->data;
+
+ if (ctx->buffer)
+ serf_bucket_mem_free(bucket->allocator, ctx->buffer);
+
+ serf_default_destroy_and_data(bucket);
+}
+
+/* ### need to implement */
+#define serf_prefix_readline NULL
+
+const serf_bucket_type_t serf_bucket_type_prefix = {
+ "prefix",
+ serf_prefix_read,
+ serf_prefix_readline, /* #### */
+ serf_prefix_read_iovec,
+ serf_default_read_for_sendfile,
+ serf_buckets_are_v2,
+ serf_prefix_peek,
+ serf_prefix_destroy,
+ serf_default_read_bucket,
+ serf_prefix_get_remaining,
+ serf_prefix_set_config
+};
+
+
Propchange: serf/trunk/buckets/prefix_buckets.c
------------------------------------------------------------------------------
svn:eol-style = native
Modified: serf/trunk/serf_bucket_types.h
URL:
http://svn.apache.org/viewvc/serf/trunk/serf_bucket_types.h?rev=1711471&r1=1711470&r2=1711471&view=diff
==============================================================================
--- serf/trunk/serf_bucket_types.h (original)
+++ serf/trunk/serf_bucket_types.h Fri Oct 30 13:35:07 2015
@@ -766,6 +766,32 @@ serf_bucket_t *serf_bucket_copy_create(
/* ==================================================================== */
+/* Reads a specific number of bytes from an inner bucket and forwards that
+ to a callback, before reading the rest of the bucket normally.
+
+ If the internal bucket is at EOF before the prefix length is read, the
+ callback is called with whatever is read before returning EOF.
+ */
+
+extern const serf_bucket_type_t serf_bucket_type_prefix;
+#define SERF_BUCKET_IS_PREFIX(b) SERF_BUCKET_CHECK((b), prefix)
+
+/* Callback for the prefix handler */
+typedef apr_status_t (*serf_bucket_prefix_handler_t)(
+ void *baton,
+ serf_bucket_t *stream,
+ const char *data,
+ apr_size_t len);
+
+serf_bucket_t *serf_bucket_prefix_create(
+ serf_bucket_t *stream,
+ apr_size_t prefix_len,
+ serf_bucket_prefix_handler_t handler,
+ void *handler_baton,
+ serf_bucket_alloc_t *allocator);
+
+/* ==================================================================== */
+
/* ### do we need a PIPE bucket type? they are simple apr_file_t objects */
Modified: serf/trunk/test/test_buckets.c
URL:
http://svn.apache.org/viewvc/serf/trunk/test/test_buckets.c?rev=1711471&r1=1711470&r2=1711471&view=diff
==============================================================================
--- serf/trunk/test/test_buckets.c (original)
+++ serf/trunk/test/test_buckets.c Fri Oct 30 13:35:07 2015
@@ -1371,8 +1371,8 @@ static void test_random_eagain_in_respon
CuAssert(tc, "Read less data than expected.", strlen(ptr) == 0);
}
apr_pool_destroy(iter_pool);
-
}
+#undef BODY
static void test_dechunk_buckets(CuTest *tc)
{
@@ -1813,6 +1813,85 @@ static void test_linebuf_fetch_crlf(CuTe
}
+typedef struct prefix_cb
+{
+ serf_bucket_alloc_t *allocator;
+ apr_size_t len;
+ char *data;
+} prefix_cb;
+
+/* Implements serf_bucket_prefix_handler_t */
+static apr_status_t prefix_callback(void *baton,
+ serf_bucket_t *inner_bucket,
+ const char *data,
+ apr_size_t len)
+{
+ prefix_cb *pb = baton;
+
+ pb->len = len;
+ pb->data = serf_bstrmemdup(pb->allocator, data, len);
+ return APR_SUCCESS;
+}
+
+
+static void test_prefix_buckets(CuTest *tc)
+{
+ test_baton_t *tb = tc->testBaton;
+ serf_bucket_alloc_t *alloc;
+ prefix_cb pb;
+ serf_bucket_t *agg, *bkt, *prefix;
+ const char *BODY = "12345678901234567890";
+ const char *data;
+ apr_size_t len;
+
+ alloc = serf_bucket_allocator_create(tb->pool, NULL, NULL);
+ pb.allocator = alloc;
+
+ agg = serf_bucket_aggregate_create(alloc);
+
+ /* First try reading less than first chunk */
+ bkt = SERF_BUCKET_SIMPLE_STRING(BODY, alloc);
+ serf_bucket_aggregate_append(agg, bkt);
+ bkt = SERF_BUCKET_SIMPLE_STRING(BODY, alloc);
+ serf_bucket_aggregate_append(agg, bkt);
+
+ prefix = serf_bucket_prefix_create(agg, 15,
+ prefix_callback, &pb, alloc);
+
+ pb.data = NULL;
+ read_and_check_bucket(tc, prefix, "6789012345678901234567890");
+ CuAssertIntEquals(tc, 15, pb.len);
+ CuAssertStrEquals(tc, "123456789012345", pb.data);
+
+ /* Then more than first chunk*/
+ bkt = SERF_BUCKET_SIMPLE_STRING(BODY, alloc);
+ serf_bucket_aggregate_append(agg, bkt);
+ bkt = SERF_BUCKET_SIMPLE_STRING(BODY, alloc);
+ serf_bucket_aggregate_append(agg, bkt);
+
+ prefix = serf_bucket_prefix_create(agg, 25,
+ prefix_callback, &pb, alloc);
+
+ pb.data = NULL;
+ read_and_check_bucket(tc, prefix, "678901234567890");
+ CuAssertIntEquals(tc, 25, pb.len);
+ CuAssertStrEquals(tc, "1234567890123456789012345", pb.data);
+
+ /* And an early EOF */
+ bkt = SERF_BUCKET_SIMPLE_STRING(BODY, alloc);
+ serf_bucket_aggregate_append(agg, bkt);
+
+ prefix = serf_bucket_prefix_create(agg, 25,
+ prefix_callback, &pb, alloc);
+
+ CuAssertIntEquals(tc, APR_EOF,
+ serf_bucket_read(prefix, SERF_READ_ALL_AVAIL,
+ &data, &len));
+ CuAssertIntEquals(tc, 0, len);
+ CuAssertIntEquals(tc, 20, pb.len);
+ CuAssertStrEquals(tc, "12345678901234567890", pb.data);
+}
+
/* Basic test for unframe buckets. */
static void test_http2_unframe_buckets(CuTest *tc)
{
@@ -2291,6 +2370,7 @@ CuSuite *test_buckets(void)
SUITE_ADD_TEST(suite, test_linebuf_fetch_crlf);
SUITE_ADD_TEST(suite, test_dechunk_buckets);
SUITE_ADD_TEST(suite, test_deflate_buckets);
+ SUITE_ADD_TEST(suite, test_prefix_buckets);
SUITE_ADD_TEST(suite, test_http2_unframe_buckets);
SUITE_ADD_TEST(suite, test_http2_unpad_buckets);
SUITE_ADD_TEST(suite, test_hpack_huffman_decode);
