appsink: Pull and write samples in the libsoup main context

1 files changed, 17 insertions, 20 deletions

diff --git a/src/main.c b/src/main.c
index 21843db..c0dbd7a 100644
--- a/src/main.c
+++ b/src/main.c
@@ -105,33 +105,19 @@ static void
 can_write_next_client_chunk_cb (SoupMessage        *msg,
                                 TranscodeClientCtx *ctx)
 {
-  g_mutex_unlock (&ctx->can_write_chunk);
   ctx->seconds_since_write = 0;
   ctx->msg = msg;
   stp_print_status ("_");
 }
 
-static GstFlowReturn
-write_client_chunk_cb (GstElement         *appsink,
-                       TranscodeClientCtx *ctx)
+static gboolean
+invoke_write_client_chunk (TranscodeClientCtx *ctx)
 {
   GstMapInfo info;
   GstSample *sample;
   GstBuffer *buffer;
   gboolean eos;
 
-  /* FIXME: Maybe dispatch to libsoup's main loop and do
-   * everything from there instead?
-   */
-  if (!g_mutex_trylock (&ctx->can_write_chunk)) {
-    stp_print_status ("!");
-    /* We cannot safely append to the message body till
-     * the previous chunk is written out, otherwise we get
-     * a segfault. This is likely because SoupMessageBody
-     * uses a GSList for the chunks, which isn't MT-safe. */
-    return GST_FLOW_OK;
-  }
-
   stp_print_status ("*");
   g_signal_emit_by_name (ctx->appsink, "pull-sample", &sample);
   if (!sample) {
@@ -142,7 +128,7 @@ write_client_chunk_cb (GstElement         *appsink,
     else
       soup_message_set_status (ctx->msg, SOUP_STATUS_INTERNAL_SERVER_ERROR);
     soup_message_body_complete (ctx->msg->response_body);
-    return GST_FLOW_OK;
+    return FALSE;
   }
 
   buffer = gst_sample_get_buffer (sample);
@@ -156,7 +142,16 @@ write_client_chunk_cb (GstElement         *appsink,
 
   soup_server_unpause_message (server, ctx->msg);
   stp_print_status (".");
+  return FALSE;
+}
 
+static GstFlowReturn
+write_client_chunk_cb (GstElement         *appsink,
+                       TranscodeClientCtx *ctx)
+{
+  g_main_context_invoke_full (NULL, G_PRIORITY_HIGH,
+                              (GSourceFunc)invoke_write_client_chunk,
+                              ctx, NULL);
   return GST_FLOW_OK;
 }
 
@@ -164,8 +159,11 @@ static void
 client_finished_cb (SoupMessage        *msg,
                     TranscodeClientCtx *ctx)
 {
-  g_debug ("Client finished/aborted, doing cleanup...\n");
-  stp_cleanup_transcode_client_ctx (ctx);
+  stp_print_status ("Client finished/aborted, doing cleanup...\n");
+  /* Make sure not to double-free. If the server 
+   * has shut down, this will be 0. */
+  if (ctx->timeout_handler_id)
+    stp_cleanup_transcode_client_ctx (ctx);
 }
 
 static void
@@ -268,7 +266,6 @@ GET:
     client_ctx = g_new0 (TranscodeClientCtx, 1);
     client_ctx->msg = msg;
     client_ctx->server_ctx = server_ctx;
-    g_mutex_init (&client_ctx->can_write_chunk);
 
     bin = gst_bin_new (NULL);