server: Track stream state better, and abort better

We now differentiate between STREAMING, FLUSHING, and FINISHED. Server aborting
is now consolidated into a single function that's called from everywhere.

1 files changed, 30 insertions, 15 deletions

diff --git a/src/main.c b/src/main.c
index 7cc2d25..fb62d53 100644
--- a/src/main.c
+++ b/src/main.c
@@ -63,6 +63,20 @@ get_server_ctx_from_msg (SoupMessage *msg,
   return ctx;
 }
 
+static void
+stp_abort_server_ctx (TranscodeServerCtx *ctx)
+{
+  /* Abort all processing for this stream */
+  if (ctx->status == STP_STATUS_STREAMING &&
+      ctx->encoding != SOUP_ENCODING_CONTENT_LENGTH) {
+    /* Close the PUT stream if necessary */
+    ctx->status = STP_STATUS_FINISHED;
+    soup_message_set_status (ctx->msg, SOUP_STATUS_GONE);
+  }
+  /* Disconnect all clients, and shut down the stream */
+  invoke_g_hash_table_remove (ctx);
+}
+
 /* 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
@@ -190,7 +204,13 @@ stream_finished_cb (SoupMessage        *msg,
 
   stp_print_status ("Stream finished/aborted, queueing EOS... ");
 
-  ctx->stream_finished = TRUE;
+  if (!ctx->appsrc) {
+    ctx->status = STP_STATUS_FINISHED;
+    stp_print_status ("No need.\n");
+    return;
+  }
+
+  ctx->status = STP_STATUS_FLUSHING;
   /* Incoming stream has ended */
   g_signal_emit_by_name (ctx->appsrc, "end-of-stream", &ret);
   if (ret != GST_FLOW_OK)
@@ -231,7 +251,7 @@ PUT:
      * will be done on EOS/ERROR in the gst pipeline. */
     server_ctx = get_server_ctx_from_msg (msg, ctx_table);
     if (server_ctx->encoding != SOUP_ENCODING_CONTENT_LENGTH)
-      server_ctx->stream_finished = TRUE;
+      server_ctx->status = STP_STATUS_FLUSHING;
     soup_message_set_status (server_ctx->msg, SOUP_STATUS_OK);
     return;
   }
@@ -407,7 +427,8 @@ got_request_body_chunk (SoupMessage        *msg,
    * reception of further chunks when we've prematurely set the response
    * (say due to an error), so we check if the status code is set and skip
    * all further chunks */
-  if (msg->status_code != 0)
+  if (msg->status_code != 0 ||
+      ctx->status != STP_STATUS_STREAMING)
     return;
 
   /* We need to update this every chunk */
@@ -421,7 +442,7 @@ got_request_body_chunk (SoupMessage        *msg,
   if (ret != GST_FLOW_OK) {
     g_critical ("Unable to push buffer\n");
     soup_message_set_status (msg, SOUP_STATUS_INTERNAL_SERVER_ERROR);
-    g_hash_table_remove (ctx->parent_ctx_table, ctx->path);
+    invoke_g_hash_table_remove (ctx);
     return;
   }
 }
@@ -431,7 +452,7 @@ request_ended_no_body_cb (SoupMessage        *msg,
                           TranscodeServerCtx *ctx)
 {
   g_critical ("Request ended without a body!");
-  g_hash_table_remove (ctx->parent_ctx_table, ctx->path);
+  invoke_g_hash_table_remove (ctx);
   g_free (ctx);
 }
 
@@ -471,7 +492,7 @@ got_first_request_body_chunk (SoupMessage        *msg,
       GST_STATE_CHANGE_FAILURE) {
     g_critical ("Unable to set pipeline to PLAYING\n");
     soup_message_set_status (msg, SOUP_STATUS_INTERNAL_SERVER_ERROR);
-    g_hash_table_remove (ctx->parent_ctx_table, ctx->path);
+    invoke_g_hash_table_remove (ctx);
   } else {
     g_debug ("Set pipeline to PLAYING");
   }
@@ -576,7 +597,7 @@ PUT: {
     if (connection_exists) {
       g_debug ("Stream already exists, connecting everything to that...");
       ctx = get_server_ctx_from_msg (msg, ctx_table);
-      if (ctx->stream_finished) {
+      if (ctx->status != STP_STATUS_STREAMING) {
         g_critical ("Recv more data on '%s' after timeout",
                     soup_uri_get_path (uri));
         soup_message_set_status (msg, SOUP_STATUS_REQUEST_TIMEOUT);
@@ -592,6 +613,7 @@ PUT: {
     ctx = g_new0 (TranscodeServerCtx, 1);
     ctx->msg = msg;
     ctx->encoding = encoding;
+    ctx->status = STP_STATUS_STREAMING;
     ctx->parent_ctx_table = ctx_table;
     ctx->path = g_strdup (soup_uri_get_path (uri));
     g_hash_table_insert (ctx_table, ctx->path, ctx);
@@ -623,14 +645,7 @@ GET: {
     if (query && g_strcmp0 (query, "abort") == 0) {
       /* We just use the "uri" field for this, so this is OK */
       server_ctx = get_server_ctx_from_msg (msg, ctx_table);
-      /* Abort all processing for this stream */
-      if (!server_ctx->stream_finished) {
-        /* Close the PUT stream if necessary */
-        server_ctx->stream_finished = TRUE;
-        soup_message_set_status (server_ctx->msg, SOUP_STATUS_GONE);
-      }
-      /* Disconnect all clients, and shut down the stream */
-      g_hash_table_remove (ctx_table, server_ctx->path);
+      stp_abort_server_ctx (server_ctx);
       soup_message_set_status (msg, SOUP_STATUS_OK);
       goto out;
     }