"User-Agent: %v\r\n"
"Content-Length: %llu\r\n";
+static const char *put_request_template = "PUT %s HTTP/1.1\r\n"
+ "Host: %v\r\n"
+ "User-Agent: %v\r\n"
+ "Content-Length: %llu\r\n";
+
+static const char *put_chunked_request_template =
+ "PUT %s HTTP/1.1\r\n"
+ "Host: %v\r\n"
+ "User-Agent: %v\r\n"
+ "Transfer-Encoding: chunked\r\n";
+
always_inline http_req_t *
http1_conn_alloc_req (http_conn_t *hc)
{
req->method = HTTP_REQ_POST;
req->target_path_offset = method_offset + 5;
}
+ else if (!memcmp (rx_buf + method_offset, "PUT ", 4))
+ {
+ HTTP_DBG (0, "PUT method");
+ req->method = HTTP_REQ_PUT;
+ req->target_path_offset = method_offset + 4;
+ }
else if (!memcmp (rx_buf + method_offset, "CONNECT ", 8))
{
HTTP_DBG (0, "CONNECT method");
http_get_app_msg (req, &msg);
- if (msg.data.type > HTTP_MSG_DATA_PTR)
+ if (msg.data.type >= HTTP_MSG_DATA_N_TYPES)
{
clib_warning ("no data");
sc = HTTP_STATUS_INTERNAL_ERROR;
http_get_app_msg (req, &msg);
- if (msg.data.type > HTTP_MSG_DATA_PTR)
+ if (msg.data.type >= HTTP_MSG_DATA_N_TYPES)
{
clib_warning ("no data");
goto error;
next_state = HTTP_REQ_STATE_APP_IO_MORE_DATA;
sm_result = HTTP_SM_CONTINUE;
}
+ else if (msg.method_type == HTTP_REQ_PUT)
+ {
+ /* Check if this is a streaming PUT */
+ if (msg.data.type == HTTP_MSG_DATA_STREAMING)
+ {
+ /*
+ * Streaming PUT with chunked transfer encoding
+ */
+ request = format (request, put_chunked_request_template,
+ /* target */
+ target,
+ /* Host */
+ hc->host,
+ /* User-Agent */
+ hc->app_name);
+
+ http_req_tx_buffer_init (req, &msg);
+
+ /* For streaming, we need a different state */
+ next_state = HTTP_REQ_STATE_APP_IO_MORE_STREAMING_DATA;
+ sm_result = HTTP_SM_CONTINUE;
+ }
+ else
+ {
+ if (!msg.data.body_len)
+ {
+ clib_warning ("PUT request should include data");
+ goto error;
+ }
+ /*
+ * Regular PUT with Content-Length
+ */
+ request = format (request, put_request_template,
+ /* target */
+ target,
+ /* Host */
+ hc->host,
+ /* User-Agent */
+ hc->app_name,
+ /* Content-Length */
+ msg.data.body_len);
+
+ http_req_tx_buffer_init (req, &msg);
+
+ next_state = HTTP_REQ_STATE_APP_IO_MORE_DATA;
+ sm_result = HTTP_SM_CONTINUE;
+ }
+ }
+
else
{
clib_warning ("unsupported method %d", msg.method_type);
return HTTP_SM_STOP;
}
+static http_sm_result_t
+http1_req_state_app_io_more_streaming_data (http_conn_t *hc, http_req_t *req,
+ transport_send_params_t *sp)
+{
+ u32 max_write, chunk_size, n_segs, n_written = 0;
+ http_buffer_t *hb = &req->tx_buf;
+ svm_fifo_seg_t *seg;
+ int finished = 0;
+ int chunk_sz_value_headroom = 20;
+ u8 chunk_hdr[32];
+ int hdr_len;
+
+ ASSERT (hb->type == HTTP_BUFFER_STREAMING);
+
+ /* For streaming, check if we have data available */
+ max_write = http_io_ts_max_write (hc, sp);
+ /*
+ * do not drain more than we are going to write at a max - which
+ * is max_write minus chunk_sz_value_headroom (overhead for the chunk
+ * size value) bytes to leave the room for chunk headers.
+ */
+ if (max_write < chunk_sz_value_headroom)
+ {
+ HTTP_DBG (1, "ts tx fifo full - before write");
+ goto check_fifo;
+ }
+ chunk_size = http_buffer_get_segs (hb, max_write - chunk_sz_value_headroom,
+ &seg, &n_segs);
+ if (chunk_size == 0)
+ {
+ /* No data available right now, wait for more */
+ HTTP_DBG (1, "streaming: no data available");
+ return HTTP_SM_STOP;
+ }
+
+ /* Write chunk size in hex */
+ hdr_len =
+ snprintf ((char *) chunk_hdr, sizeof (chunk_hdr), "%x\r\n", chunk_size);
+ http_io_ts_write (hc, chunk_hdr, hdr_len, sp);
+
+ /* Write chunk data */
+ n_written = http_io_ts_write_segs (hc, seg, n_segs, sp);
+
+ /* Write chunk trailer */
+ http_io_ts_write (hc, (u8 *) "\r\n", 2, sp);
+
+ http_buffer_drain (hb, n_written);
+
+ finished = http_buffer_bytes_left (hb) == 0;
+ if (finished)
+ {
+ /* Send final chunk (0-sized) */
+ http_io_ts_write (hc, (u8 *) "0\r\n\r\n", 5, sp);
+
+ /* Finished transaction:
+ * server back to HTTP_REQ_STATE_WAIT_TRANSPORT_METHOD
+ * client to HTTP_REQ_STATE_WAIT_TRANSPORT_REPLY */
+ http_req_state_change (req, (hc->flags & HTTP_CONN_F_IS_SERVER) ?
+ HTTP_REQ_STATE_WAIT_TRANSPORT_METHOD :
+ HTTP_REQ_STATE_WAIT_TRANSPORT_REPLY);
+ http_buffer_free (hb);
+ }
+ http_io_ts_after_write (hc, finished);
+
+check_fifo:
+ http1_check_and_deschedule (hc, req, sp);
+ return HTTP_SM_STOP;
+}
+
static http_sm_result_t
http1_req_state_tunnel_tx (http_conn_t *hc, http_req_t *req,
transport_send_params_t *sp)
http1_req_state_app_io_more_data,
http1_req_state_tunnel_tx,
http1_req_state_udp_tunnel_tx,
+ http1_req_state_app_io_more_streaming_data,
};
static http_sm_handler rx_state_funcs[HTTP_REQ_N_STATES] = {
0, /* app io more data */
http1_req_state_tunnel_rx,
http1_req_state_udp_tunnel_rx,
+ 0, /* app io more streaming data */
};
static_always_inline int
#include <http/http_buffer.h>
#include <http/http.h>
-static http_buffer_vft_t buf_vfts[HTTP_BUFFER_PTR + 1];
+static http_buffer_vft_t buf_vfts[HTTP_BUFFER_N_TYPES];
#define HTTP_BUFFER_REGISTER_VFT(type, vft) \
static void __attribute__ ((constructor)) http_buf_init_##type (void) \
HTTP_BUFFER_REGISTER_VFT (HTTP_BUFFER_PTR, buf_ptr_vft);
+typedef struct http_buffer_streaming_
+{
+ svm_fifo_t *src;
+ svm_fifo_seg_t *segs;
+ u64 total_len; /* total expected length (can be ~0 for unknown) */
+ u64 sent; /* bytes sent so far */
+} http_buffer_streaming_t;
+
+STATIC_ASSERT (sizeof (http_buffer_streaming_t) <= HTTP_BUFFER_DATA_SZ,
+ "buf data");
+
+static void
+buf_streaming_init (http_buffer_t *hb, void *data, u64 len)
+{
+ svm_fifo_t *f = (svm_fifo_t *) data;
+ http_buffer_streaming_t *bs;
+
+ bs = (http_buffer_streaming_t *) &hb->data;
+
+ bs->total_len = len;
+ bs->sent = 0;
+ bs->src = f;
+ bs->segs = 0;
+}
+
+static void
+buf_streaming_free (http_buffer_t *hb)
+{
+ http_buffer_streaming_t *bs = (http_buffer_streaming_t *) &hb->data;
+
+ bs->src = 0;
+ vec_free (bs->segs);
+}
+
+static u32
+buf_streaming_get_segs (http_buffer_t *hb, u32 max_len, svm_fifo_seg_t **fs,
+ u32 *n_segs)
+{
+ http_buffer_streaming_t *bs = (http_buffer_streaming_t *) &hb->data;
+
+ u32 _n_segs = 5;
+ int len;
+
+ /* For streaming, we send whatever is available */
+ u32 available = svm_fifo_max_dequeue (bs->src);
+ if (available == 0)
+ return 0;
+
+ max_len = clib_min (available, max_len);
+
+ vec_validate (bs->segs, _n_segs - 1);
+
+ len = svm_fifo_segments (bs->src, 0, bs->segs, &_n_segs, max_len);
+ if (len < 0)
+ return 0;
+
+ *n_segs = _n_segs;
+
+ HTTP_DBG (1, "streaming: available to send %u n_segs %u", len, *n_segs);
+
+ *fs = bs->segs;
+ return len;
+}
+
+static u32
+buf_streaming_drain (http_buffer_t *hb, u32 len)
+{
+ http_buffer_streaming_t *bs = (http_buffer_streaming_t *) &hb->data;
+
+ bs->sent += len;
+ svm_fifo_dequeue_drop (bs->src, len);
+ HTTP_DBG (1, "streaming: drained %u total sent %lu", len, bs->sent);
+
+ return len;
+}
+
+static u64
+buf_streaming_bytes_left (http_buffer_t *hb)
+{
+ http_buffer_streaming_t *bs = (http_buffer_streaming_t *) &hb->data;
+ if (bs->total_len == ~0)
+ {
+ return ~0;
+ }
+
+ return (bs->total_len > bs->sent ? (bs->total_len - bs->sent) : 0);
+}
+
+const static http_buffer_vft_t buf_streaming_vft = {
+ .init = buf_streaming_init,
+ .free = buf_streaming_free,
+ .get_segs = buf_streaming_get_segs,
+ .drain = buf_streaming_drain,
+ .bytes_left = buf_streaming_bytes_left,
+};
+
+HTTP_BUFFER_REGISTER_VFT (HTTP_BUFFER_STREAMING, buf_streaming_vft);
+
void
http_buffer_init (http_buffer_t *hb, http_buffer_type_t type, svm_fifo_t *f,
u64 data_len)
Code Review / vpp.git / commitdiff