/* ====================================================================
* 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 <apr_strings.h>
#include <apr_lib.h>
#include <apr_date.h>
#include "serf.h"
#include "serf_bucket_util.h"
#include "serf_bucket_types.h"
#include <stdlib.h>
/* This is an implementation of Bidirectional Web Transfer Protocol (BWTP)
* See:
* http://bwtp.wikidot.com/
*/
typedef struct {
int channel;
int open;
int type; /* 0 = header, 1 = message */ /* TODO enum? */
const char *phrase;
serf_bucket_t *headers;
char req_line[1000];
} frame_context_t;
typedef struct {
serf_bucket_t *stream;
serf_bucket_t *body; /* Pointer to the stream wrapping the body. */
serf_bucket_t *headers; /* holds parsed headers */
enum {
STATE_STATUS_LINE, /* reading status line */
STATE_HEADERS, /* reading headers */
STATE_BODY, /* reading body */
STATE_DONE /* we've sent EOF */
} state;
/* Buffer for accumulating a line from the response. */
serf_linebuf_t linebuf;
int type; /* 0 = header, 1 = message */ /* TODO enum? */
int channel;
char *phrase;
apr_size_t length;
} incoming_context_t;
serf_bucket_t *serf_bucket_bwtp_channel_close(
int channel,
serf_bucket_alloc_t *allocator)
{
frame_context_t *ctx;
ctx = serf_bucket_mem_alloc(allocator, sizeof(*ctx));
ctx->type = 0;
ctx->open = 0;
ctx->channel = channel;
ctx->phrase = "CLOSED";
ctx->headers = serf_bucket_headers_create(allocator);
return serf_bucket_create(&serf_bucket_type_bwtp_frame, allocator, ctx);
}
serf_bucket_t *serf_bucket_bwtp_channel_open(
int channel,
const char *uri,
serf_bucket_alloc_t *allocator)
{
frame_context_t *ctx;
ctx = serf_bucket_mem_alloc(allocator, sizeof(*ctx));
ctx->type = 0;
ctx->open = 1;
ctx->channel = channel;
ctx->phrase = uri;
ctx->headers = serf_bucket_headers_create(allocator);
return serf_bucket_create(&serf_bucket_type_bwtp_frame, allocator, ctx);
}
serf_bucket_t *serf_bucket_bwtp_header_create(
int channel,
const char *phrase,
serf_bucket_alloc_t *allocator)
{
frame_context_t *ctx;
ctx = serf_bucket_mem_alloc(allocator, sizeof(*ctx));
ctx->type = 0;
ctx->open = 0;
ctx->channel = channel;
ctx->phrase = phrase;
ctx->headers = serf_bucket_headers_create(allocator);
return serf_bucket_create(&serf_bucket_type_bwtp_frame, allocator, ctx);
}
serf_bucket_t *serf_bucket_bwtp_message_create(
int channel,
serf_bucket_t *body,
serf_bucket_alloc_t *allocator)
{
frame_context_t *ctx;
ctx = serf_bucket_mem_alloc(allocator, sizeof(*ctx));
ctx->type = 1;
ctx->open = 0;
ctx->channel = channel;
ctx->phrase = "MESSAGE";
ctx->headers = serf_bucket_headers_create(allocator);
return serf_bucket_create(&serf_bucket_type_bwtp_frame, allocator, ctx);
}
int serf_bucket_bwtp_frame_get_channel(
serf_bucket_t *bucket)
{
if (SERF_BUCKET_IS_BWTP_FRAME(bucket)) {
frame_context_t *ctx = bucket->data;
return ctx->channel;
}
else if (SERF_BUCKET_IS_BWTP_INCOMING_FRAME(bucket)) {
incoming_context_t *ctx = bucket->data;
return ctx->channel;
}
return -1;
}
int serf_bucket_bwtp_frame_get_type(
serf_bucket_t *bucket)
{
if (SERF_BUCKET_IS_BWTP_FRAME(bucket)) {
frame_context_t *ctx = bucket->data;
return ctx->type;
}
else if (SERF_BUCKET_IS_BWTP_INCOMING_FRAME(bucket)) {
incoming_context_t *ctx = bucket->data;
return ctx->type;
}
return -1;
}
const char *serf_bucket_bwtp_frame_get_phrase(
serf_bucket_t *bucket)
{
if (SERF_BUCKET_IS_BWTP_FRAME(bucket)) {
frame_context_t *ctx = bucket->data;
return ctx->phrase;
}
else if (SERF_BUCKET_IS_BWTP_INCOMING_FRAME(bucket)) {
incoming_context_t *ctx = bucket->data;
return ctx->phrase;
}
return NULL;
}
serf_bucket_t *serf_bucket_bwtp_frame_get_headers(
serf_bucket_t *bucket)
{
if (SERF_BUCKET_IS_BWTP_FRAME(bucket)) {
frame_context_t *ctx = bucket->data;
return ctx->headers;
}
else if (SERF_BUCKET_IS_BWTP_INCOMING_FRAME(bucket)) {
incoming_context_t *ctx = bucket->data;
return ctx->headers;
}
return NULL;
}
static int count_size(void *baton, const char *key, const char *value)
{
apr_size_t *c = baton;
/* TODO Deal with folding. Yikes. */
/* Add in ": " and CRLF - so an extra four bytes. */
*c += strlen(key) + strlen(value) + 4;
return 0;
}
static apr_size_t calc_header_size(serf_bucket_t *hdrs)
{
apr_size_t size = 0;
serf_bucket_headers_do(hdrs, count_size, &size);
return size;
}
static void serialize_data(serf_bucket_t *bucket)
{
frame_context_t *ctx = bucket->data;
serf_bucket_t *new_bucket;
apr_size_t req_len;
/* Serialize the request-line and headers into one mother string,
* and wrap a bucket around it.
*/
req_len = apr_snprintf(ctx->req_line, sizeof(ctx->req_line),
"%s %d " "%" APR_UINT64_T_HEX_FMT " %s%s\r\n",
(ctx->type ? "BWM" : "BWH"),
ctx->channel, calc_header_size(ctx->headers),
(ctx->open ? "OPEN " : ""),
ctx->phrase);
new_bucket = serf_bucket_simple_copy_create(ctx->req_line, req_len,
bucket->allocator);
/* Build up the new bucket structure.
*
* Note that self needs to become an aggregate bucket so that a
* pointer to self still represents the "right" data.
*/
serf_bucket_aggregate_become(bucket);
/* Insert the two buckets. */
serf_bucket_aggregate_append(bucket, new_bucket);
serf_bucket_aggregate_append(bucket, ctx->headers);
/* Our private context is no longer needed, and is not referred to by
* any existing bucket. Toss it.
*/
serf_bucket_mem_free(bucket->allocator, ctx);
}
static apr_status_t serf_bwtp_frame_read(serf_bucket_t *bucket,
apr_size_t requested,
const char **data, apr_size_t *len)
{
/* Seralize our private data into a new aggregate bucket. */
serialize_data(bucket);
/* Delegate to the "new" aggregate bucket to do the read. */
return serf_bucket_read(bucket, requested, data, len);
}
static apr_status_t serf_bwtp_frame_readline(serf_bucket_t *bucket,
int acceptable, int *found,
const char **data, apr_size_t *len)
{
/* Seralize our private data into a new aggregate bucket. */
serialize_data(bucket);
/* Delegate to the "new" aggregate bucket to do the readline. */
return serf_bucket_readline(bucket, acceptable, found, data, len);
}
static apr_status_t serf_bwtp_frame_read_iovec(serf_bucket_t *bucket,
apr_size_t requested,
int vecs_size,
struct iovec *vecs,
int *vecs_used)
{
/* Seralize our private data into a new aggregate bucket. */
serialize_data(bucket);
/* Delegate to the "new" aggregate bucket to do the read. */
return serf_bucket_read_iovec(bucket, requested,
vecs_size, vecs, vecs_used);
}
static apr_status_t serf_bwtp_frame_peek(serf_bucket_t *bucket,
const char **data,
apr_size_t *len)
{
/* Seralize our private data into a new aggregate bucket. */
serialize_data(bucket);
/* Delegate to the "new" aggregate bucket to do the peek. */
return serf_bucket_peek(bucket, data, len);
}
const serf_bucket_type_t serf_bucket_type_bwtp_frame = {
"BWTP-FRAME",
serf_bwtp_frame_read,
serf_bwtp_frame_readline,
serf_bwtp_frame_read_iovec,
serf_default_read_for_sendfile,
serf_default_read_bucket,
serf_bwtp_frame_peek,
serf_default_destroy_and_data,
};
serf_bucket_t *serf_bucket_bwtp_incoming_frame_create(
serf_bucket_t *stream,
serf_bucket_alloc_t *allocator)
{
incoming_context_t *ctx;
ctx = serf_bucket_mem_alloc(allocator, sizeof(*ctx));
ctx->stream = stream;
ctx->body = NULL;
ctx->headers = serf_bucket_headers_create(allocator);
ctx->state = STATE_STATUS_LINE;
ctx->length = 0;
ctx->channel = -1;
ctx->phrase = NULL;
serf_linebuf_init(&ctx->linebuf);
return serf_bucket_create(&serf_bucket_type_bwtp_incoming_frame, allocator, ctx);
}
static void bwtp_incoming_destroy_and_data(serf_bucket_t *bucket)
{
incoming_context_t *ctx = bucket->data;
if (ctx->state != STATE_STATUS_LINE && ctx->phrase) {
serf_bucket_mem_free(bucket->allocator, (void*)ctx->phrase);
}
serf_bucket_destroy(ctx->stream);
if (ctx->body != NULL)
serf_bucket_destroy(ctx->body);
serf_bucket_destroy(ctx->headers);
serf_default_destroy_and_data(bucket);
}
static apr_status_t fetch_line(incoming_context_t *ctx, int acceptable)
{
return serf_linebuf_fetch(&ctx->linebuf, ctx->stream, acceptable);
}
static apr_status_t parse_status_line(incoming_context_t *ctx,
serf_bucket_alloc_t *allocator)
{
int res;
char *reason; /* ### stupid APR interface makes this non-const */
/* ctx->linebuf.line should be of form: BW* */
res = apr_date_checkmask(ctx->linebuf.line, "BW*");
if (!res) {
/* Not an BWTP response? Well, at least we won't understand it. */
return APR_EGENERAL;
}
if (ctx->linebuf.line[2] == 'H') {
ctx->type = 0;
}
else if (ctx->linebuf.line[2] == 'M') {
ctx->type = 1;
}
else {
ctx->type = -1;
}
ctx->channel = apr_strtoi64(ctx->linebuf.line + 3, &reason, 16);
/* Skip leading spaces for the reason string. */
if (apr_isspace(*reason)) {
reason++;
}
ctx->length = apr_strtoi64(reason, &reason, 16);
/* Skip leading spaces for the reason string. */
if (reason - ctx->linebuf.line < ctx->linebuf.used) {
if (apr_isspace(*reason)) {
reason++;
}
ctx->phrase = serf_bstrmemdup(allocator, reason,
ctx->linebuf.used
- (reason - ctx->linebuf.line));
} else {
ctx->phrase = NULL;
}
return APR_SUCCESS;
}
/* This code should be replaced with header buckets. */
static apr_status_t fetch_headers(serf_bucket_t *bkt, incoming_context_t *ctx)
{
apr_status_t status;
/* RFC 2616 says that CRLF is the only line ending, but we can easily
* accept any kind of line ending.
*/
status = fetch_line(ctx, SERF_NEWLINE_ANY);
if (SERF_BUCKET_READ_ERROR(status)) {
return status;
}
/* Something was read. Process it. */
if (ctx->linebuf.state == SERF_LINEBUF_READY && ctx->linebuf.used) {
const char *end_key;
const char *c;
end_key = c = memchr(ctx->linebuf.line, ':', ctx->linebuf.used);
if (!c) {
/* Bad headers? */
return APR_EGENERAL;
}
/* Skip over initial : and spaces. */
while (apr_isspace(*++c))
continue;
/* Always copy the headers (from the linebuf into new mem). */
/* ### we should be able to optimize some mem copies */
serf_bucket_headers_setx(
ctx->headers,
ctx->linebuf.line, end_key - ctx->linebuf.line, 1,
c, ctx->linebuf.line + ctx->linebuf.used - c, 1);
}
return status;
}
/* Perform one iteration of the state machine.
*
* Will return when one the following conditions occurred:
* 1) a state change
* 2) an error
* 3) the stream is not ready or at EOF
* 4) APR_SUCCESS, meaning the machine can be run again immediately
*/
static apr_status_t run_machine(serf_bucket_t *bkt, incoming_context_t *ctx)
{
apr_status_t status = APR_SUCCESS; /* initialize to avoid gcc warnings */
switch (ctx->state) {
case STATE_STATUS_LINE:
/* RFC 2616 says that CRLF is the only line ending, but we can easily
* accept any kind of line ending.
*/
status = fetch_line(ctx, SERF_NEWLINE_ANY);
if (SERF_BUCKET_READ_ERROR(status))
return status;
if (ctx->linebuf.state == SERF_LINEBUF_READY && ctx->linebuf.used) {
/* The Status-Line is in the line buffer. Process it. */
status = parse_status_line(ctx, bkt->allocator);
if (status)
return status;
if (ctx->length) {
ctx->body =
serf_bucket_barrier_create(ctx->stream, bkt->allocator);
ctx->body = serf_bucket_limit_create(ctx->body, ctx->length,
bkt->allocator);
if (!ctx->type) {
ctx->state = STATE_HEADERS;
} else {
ctx->state = STATE_BODY;
}
} else {
ctx->state = STATE_DONE;
}
}
else {
/* The connection closed before we could get the next
* response. Treat the request as lost so that our upper
* end knows the server never tried to give us a response.
*/
if (APR_STATUS_IS_EOF(status)) {
return SERF_ERROR_REQUEST_LOST;
}
}
break;
case STATE_HEADERS:
status = fetch_headers(ctx->body, ctx);
if (SERF_BUCKET_READ_ERROR(status))
return status;
/* If an empty line was read, then we hit the end of the headers.
* Move on to the body.
*/
if (ctx->linebuf.state == SERF_LINEBUF_READY && !ctx->linebuf.used) {
/* Advance the state. */
ctx->state = STATE_DONE;
}
break;
case STATE_BODY:
/* Don't do anything. */
break;
case STATE_DONE:
return APR_EOF;
default:
/* Not reachable */
return APR_EGENERAL;
}
return status;
}
static apr_status_t wait_for_body(serf_bucket_t *bkt, incoming_context_t *ctx)
{
apr_status_t status;
/* Keep reading and moving through states if we aren't at the BODY */
while (ctx->state != STATE_BODY) {
status = run_machine(bkt, ctx);
/* Anything other than APR_SUCCESS means that we cannot immediately
* read again (for now).
*/
if (status)
return status;
}
/* in STATE_BODY */
return APR_SUCCESS;
}
apr_status_t serf_bucket_bwtp_incoming_frame_wait_for_headers(
serf_bucket_t *bucket)
{
incoming_context_t *ctx = bucket->data;
return wait_for_body(bucket, ctx);
}
static apr_status_t bwtp_incoming_read(serf_bucket_t *bucket,
apr_size_t requested,
const char **data, apr_size_t *len)
{
incoming_context_t *ctx = bucket->data;
apr_status_t rv;
rv = wait_for_body(bucket, ctx);
if (rv) {
/* It's not possible to have read anything yet! */
if (APR_STATUS_IS_EOF(rv) || APR_STATUS_IS_EAGAIN(rv)) {
*len = 0;
}
return rv;
}
rv = serf_bucket_read(ctx->body, requested, data, len);
if (APR_STATUS_IS_EOF(rv)) {
ctx->state = STATE_DONE;
}
return rv;
}
static apr_status_t bwtp_incoming_readline(serf_bucket_t *bucket,
int acceptable, int *found,
const char **data, apr_size_t *len)
{
incoming_context_t *ctx = bucket->data;
apr_status_t rv;
rv = wait_for_body(bucket, ctx);
if (rv) {
return rv;
}
/* Delegate to the stream bucket to do the readline. */
return serf_bucket_readline(ctx->body, acceptable, found, data, len);
}
/* ### need to implement */
#define bwtp_incoming_peek NULL
const serf_bucket_type_t serf_bucket_type_bwtp_incoming_frame = {
"BWTP-INCOMING",
bwtp_incoming_read,
bwtp_incoming_readline,
serf_default_read_iovec,
serf_default_read_for_sendfile,
serf_default_read_bucket,
bwtp_incoming_peek,
bwtp_incoming_destroy_and_data,
};