diff options
Diffstat (limited to 'src/main.c')
| -rw-r--r-- | src/main.c | 199 |
1 files changed, 159 insertions, 40 deletions
@@ -29,6 +29,7 @@ static int port = 8000; static int client_timeout = 10; +static int server_timeout = 5; static GOptionEntry entries[] = { @@ -39,6 +40,9 @@ static GOptionEntry entries[] = static SoupServer *server; +static void stream_finished_cb (SoupMessage *msg, + TranscodeServerCtx *ctx); + static TranscodeServerCtx* get_server_ctx_from_msg (SoupMessage *msg, GHashTable *ctx_table) @@ -109,6 +113,20 @@ err: goto out; } +/* If it's been more than 10 seconds since the last time we got + * a chunk for a PUT request, we timeout and drop the connection */ +static gboolean +increment_read_timer (TranscodeServerCtx *ctx) +{ + ctx->seconds_since_read += 2; + if (ctx->seconds_since_read < server_timeout) + return G_SOURCE_CONTINUE; + + g_printerr ("Stream timed out, sending EOS\n"); + stream_finished_cb (ctx->msg, ctx); + return G_SOURCE_REMOVE; +} + /* If it's been more than 10 seconds since the last time we wrote * a chunk for a GET response, we timeout and drop the connection */ static gboolean @@ -209,7 +227,7 @@ stream_finished_cb (SoupMessage *msg, g_print ("Stream finished/aborted, queueing EOS... "); - ctx->request_finished = TRUE; + ctx->stream_finished = TRUE; /* Incoming stream has ended */ g_signal_emit_by_name (ctx->appsrc, "end-of-stream", &ret); if (ret != GST_FLOW_OK) @@ -249,7 +267,8 @@ PUT: /* The PUT request has finished streaming. Cleanup * will be done on EOS/ERROR in the gst pipeline. */ server_ctx = get_server_ctx_from_msg (msg, ctx_table); - server_ctx->request_finished = TRUE; + if (server_ctx->encoding != SOUP_ENCODING_CONTENT_LENGTH) + server_ctx->stream_finished = TRUE; soup_message_set_status (server_ctx->msg, SOUP_STATUS_OK); return; } @@ -371,6 +390,7 @@ got_request_body_chunk (SoupMessage *msg, /* We need to update this every chunk */ ctx->msg = msg; + ctx->seconds_since_read = 0; /* Actually just a ref */ buffer = stp_get_gst_buffer (chunk); @@ -382,27 +402,76 @@ got_request_body_chunk (SoupMessage *msg, stp_cleanup_transcode_server_ctx (ctx); return; } +} - /* We can only set the pipeline to playing when - * the first chunk has been received */ - if (!ctx->pipeline_is_playing) { - if (gst_element_set_state (ctx->pipeline, GST_STATE_PLAYING) == - GST_STATE_CHANGE_FAILURE) { - g_critical ("Unable to set pipeline to PLAYING\n"); - soup_message_set_status (msg, SOUP_STATUS_INTERNAL_SERVER_ERROR); - stp_cleanup_transcode_server_ctx (ctx); - } else { - g_print ("Set pipeline to PLAYING\n"); - ctx->pipeline_is_playing = TRUE; - } +static void +request_ended_no_body_cb (SoupMessage *msg, + TranscodeServerCtx *ctx) +{ + g_print ("Request ended without a body!\n"); + g_hash_table_remove (ctx->parent_ctx_table, ctx->path); + g_free (ctx); +} + +static void +got_first_request_body_chunk (SoupMessage *msg, + SoupBuffer *chunk, + TranscodeServerCtx *ctx) +{ + int num; + + /* Disconnect the ctx cleanup function */ + num = g_signal_handlers_disconnect_by_func (msg, + G_CALLBACK (request_ended_no_body_cb), ctx); + if (num != 1) + g_critical ("Unable to remove signal handler for cleanup function!\n"); + + /* Disconnect us from got-chunk */ + num = g_signal_handlers_disconnect_by_func (msg, + G_CALLBACK (got_first_request_body_chunk), ctx); + if (num != 1) + g_critical ("Unable to remove signal handler for first request chunk!\n"); + + ctx->msg = msg; + ctx->pipeline = gst_pipeline_new ("pipe"); + + /* The chunked request is copied into this stream + * for consumption by the gst pipeline */ +#ifdef PLAY_DEBUG + stp_play_from_msg (ctx); +#else + stp_encode_from_msg (ctx); +#endif + + got_request_body_chunk (msg, chunk, ctx); + + if (gst_element_set_state (ctx->pipeline, GST_STATE_PLAYING) == + GST_STATE_CHANGE_FAILURE) { + g_critical ("Unable to set pipeline to PLAYING\n"); + soup_message_set_status (msg, SOUP_STATUS_INTERNAL_SERVER_ERROR); + stp_cleanup_transcode_server_ctx (ctx); + } else { + g_print ("Set pipeline to PLAYING\n"); } + + /* Connect a different method for all further chunks */ + g_signal_connect (msg, "got-chunk", + G_CALLBACK (got_request_body_chunk), ctx); + + if (ctx->encoding != SOUP_ENCODING_CONTENT_LENGTH) + /* Our incoming stream ends if the stream ends only if + * we're using a chunked request encoding. + * + * This will also be called when the client sending + * the PUT request disconnects prematurely */ + g_signal_connect (msg, "finished", + G_CALLBACK (stream_finished_cb), ctx); } static void got_request_headers (SoupMessage *msg, GHashTable *ctx_table) { - TranscodeServerCtx *ctx; SoupURI *uri; #ifdef HEADERS_DEBUG @@ -428,42 +497,92 @@ got_request_headers (SoupMessage *msg, } PUT: { - if (g_hash_table_contains (ctx_table, soup_uri_get_path (uri))) { - /* There's already a request streaming on this path */ - g_print ("Recv duplicate request on the same URI: %s\n", - soup_uri_get_path (uri)); - soup_message_set_status (msg, SOUP_STATUS_CONFLICT); + const char *content_range; + SoupEncoding encoding; + gboolean connection_exists; + TranscodeServerCtx *ctx; + + connection_exists = g_hash_table_contains (ctx_table, soup_uri_get_path (uri)); + + /* Data might be sent to us in one of three ways: + * 1. Chunked encoding with the stream (and request body) sent in chunks + * 2. Content-Length encoding with the entire stream in one chunk + * 3. A persistent Content-Length encoding connection, and subsequent + * requests can send more stream data, forever, subject to a timeout */ + encoding = soup_message_headers_get_encoding (msg->request_headers); + switch (encoding) { + case SOUP_ENCODING_CHUNKED: + g_print ("Chunked encoding detected!\n"); + if (connection_exists) { + /* There's already a chunked request streaming on this path */ + g_printerr ("Recv duplicate request on the same URI: %s\n", + soup_uri_get_path (uri)); + soup_message_set_status (msg, SOUP_STATUS_CONFLICT); + goto out; + } + break; + case SOUP_ENCODING_CONTENT_LENGTH: + g_print ("Content-Length encoding detected!\n"); + content_range = soup_message_headers_get_one (msg->request_headers, + "Content-Range"); + /* TODO: Right now, we don't check if the Content-Range is valid, + * and just pass the data to the pipeline. */ + if (connection_exists && !content_range) { + /* A connection already exists, and this one isn't a continuation + * of the previous one, so it's a conflict */ + g_printerr ("Recv Content-Length PUT on '%s' without Content-Range\n", + soup_uri_get_path (uri)); + soup_message_set_status (msg, SOUP_STATUS_CONFLICT); + goto out; + } + break; + case SOUP_ENCODING_EOF: + case SOUP_ENCODING_UNRECOGNIZED: + case SOUP_ENCODING_NONE: + case SOUP_ENCODING_BYTERANGES: + g_critical ("Unknown encoding!\n"); + goto out; + default: + g_assert_not_reached (); + } + + /* Whether the incoming stream is chunked or fixed length, we want to + * handle it chunked, so tell libsoup to not collect chunks for + * forming a complete request body */ + soup_message_body_set_accumulate (msg->request_body, FALSE); + + if (connection_exists) { + g_print ("Stream already exists, connecting everything to that...\n"); + ctx = get_server_ctx_from_msg (msg, ctx_table); + if (ctx->stream_finished) { + g_printerr ("Recv more data on '%s' after timeout\n", + soup_uri_get_path (uri)); + soup_message_set_status (msg, SOUP_STATUS_REQUEST_TIMEOUT); + goto out; + } + /* The chunks we'll get are a continuation of the previous one */ + g_signal_connect (msg, "got-chunk", + G_CALLBACK (got_request_body_chunk), ctx); goto out; } + /* This is a new connection; treat it as such */ ctx = g_new0 (TranscodeServerCtx, 1); ctx->msg = msg; + ctx->encoding = encoding; ctx->parent_ctx_table = ctx_table; ctx->path = g_strdup (soup_uri_get_path (uri)); g_hash_table_insert (ctx_table, ctx->path, ctx); - - /* We only do this when we receive the first chunk */ - if (!ctx->pipeline) { - ctx->pipeline = gst_pipeline_new ("pipe"); - /* The chunked request is copied into this stream - * for consumption by the gst pipeline */ -#ifdef PLAY_DEBUG - stp_play_from_msg (ctx); -#else - stp_encode_from_msg (ctx); -#endif - } - - /* The request body isn't fixed length, so tell libsoup to - * not collect chunks for forming a complete request body */ - soup_message_body_set_accumulate (msg->request_body, FALSE); + g_print ("New stream on %s!\n", ctx->path); g_signal_connect (msg, "got-chunk", - G_CALLBACK (got_request_body_chunk), ctx); - /* This will also be called when the client sending - * the PUT request disconnects prematurely */ + G_CALLBACK (got_first_request_body_chunk), ctx); + /* If we don't get any data after this, we have to cleanup ctx + * otherwise we'll leak it */ g_signal_connect (msg, "finished", - G_CALLBACK (stream_finished_cb), ctx); + G_CALLBACK (request_ended_no_body_cb), ctx); + if (encoding == SOUP_ENCODING_CONTENT_LENGTH) + g_timeout_add_seconds (2, (GSourceFunc)increment_read_timer, ctx); goto out; } |