Add source files

1 files changed, 562 insertions, 0 deletions

diff --git a/src/main.c b/src/main.c
new file mode 100644
index 0000000..a2430f5
--- /dev/null
+++ b/src/main.c
@@ -0,0 +1,562 @@
+/* License: LGPL-2.1
+ * vim: set sts=2 sw=2 et : */
+
+#include "lib.h"
+#include "encode.h"
+
+#ifdef DEBUG
+#include "debug/local-play.h"
+#endif
+
+#include <fcntl.h>
+#include <string.h>
+
+static SoupServer *server;
+
+static TranscodeServerCtx*
+get_server_ctx_from_msg (SoupMessage *msg,
+                         GHashTable  *ctx_table)
+{
+  SoupURI *uri;
+  TranscodeServerCtx *ctx;
+
+  g_object_get (msg, "uri", &uri, NULL);
+  ctx = g_hash_table_lookup (ctx_table, soup_uri_get_path (uri));
+  soup_uri_free (uri);
+
+  return ctx;
+}
+
+static void
+enough_data_pause_message (GstElement         *appsrc,
+                           guint               size,
+                           TranscodeServerCtx *ctx)
+{
+  /* Pausing the request can cause weirdness if
+   * the recv rate is too high. So, we don't do this. */
+  //g_print ("Enough data, pause. ");
+  //soup_server_pause_message (server, ctx->msg);
+}
+
+static void
+need_data_unpause_message (GstElement         *appsrc,
+                           guint               size,
+                           TranscodeServerCtx *ctx)
+{
+  //g_print ("Need data, unpause. \n");
+  //soup_server_unpause_message (server, ctx->msg);
+}
+
+static GstPadProbeReturn
+tee_src_pad_blocked_cb (GstPad             *srcpad,
+                        GstPadProbeInfo    *info,
+                        TranscodeClientCtx *ctx)
+{
+  GstCaps *caps;
+  GstPad *sinkpad;
+  GstBuffer *buffer;
+  GstState state;
+  GstStateChangeReturn ret;
+  TranscodeServerCtx *server_ctx = ctx->server_ctx;
+  GstElement *sinkbin = GST_ELEMENT (gst_element_get_parent (ctx->appsink));
+
+  /* Remove the probe, XXX: hence unblocking the pipeline? */
+  gst_pad_remove_probe (srcpad, GST_PAD_PROBE_INFO_ID (info));
+
+  gst_bin_add (GST_BIN (server_ctx->pipeline), sinkbin);
+
+  /* t. ! queue ! appsink */
+  sinkpad = gst_element_get_static_pad (sinkbin, "qsink");
+  gst_pad_link (srcpad, sinkpad);
+
+  /* Send the WebM stream header through the pipeline first */
+  caps = gst_pad_get_current_caps (srcpad);
+  /* TODO: cache keyframes */
+  buffer = stp_get_streamheader_from_caps (caps);
+  if (!buffer)
+    goto err;
+  gst_pad_push (srcpad, buffer);
+
+  ret = gst_element_get_state (server_ctx->pipeline, &state, NULL, 1);
+  g_print ("Linked pads, removed probe. State was %s:%s.\n",
+           gst_element_state_change_return_get_name (ret),
+           gst_element_state_get_name (state));
+ 
+  if (gst_element_set_state (server_ctx->pipeline, GST_STATE_PLAYING) ==
+      GST_STATE_CHANGE_FAILURE) {
+    g_critical ("Unable to set pipeline back to PLAYING\n");
+    goto err;
+  } else {
+    g_print ("pipeline set to PLAYING successfully\n");
+  }
+
+out:
+  gst_caps_unref (caps);
+  gst_object_unref (sinkpad);
+  gst_object_unref (sinkbin);
+  return GST_PAD_PROBE_OK;
+err:
+  soup_message_set_status (ctx->msg, SOUP_STATUS_INTERNAL_SERVER_ERROR);
+  stp_cleanup_transcode_client_ctx (ctx);
+  /* FIXME: PROBE_OK for errors too? What to do here? */
+  goto out;
+}
+
+static GstFlowReturn
+write_first_client_chunk_cb (GstElement         *appsink,
+                             TranscodeClientCtx *ctx)
+{
+  GstMapInfo info;
+  GstSample *sample;
+  GstBuffer *buffer;
+  GstMemory *memory;
+
+  g_assert (ctx->first_sample_handler_id);
+
+  g_signal_handler_disconnect (appsink, ctx->first_sample_handler_id);
+  ctx->first_sample_handler_id = 0;
+
+  /* XXX: If no samples are available, this will block till a sample is available */
+  g_signal_emit_by_name (appsink, "pull-sample", &sample);
+  buffer = gst_sample_get_buffer (sample);
+  memory = gst_buffer_get_all_memory (buffer); /* copy */
+  gst_memory_map (memory, &info, GST_MAP_READ);
+
+  soup_message_body_append (ctx->msg->response_body, SOUP_MEMORY_COPY,
+                            info.data, info.size); /* copy */
+
+  gst_memory_unmap (memory, &info);
+  gst_memory_unref (memory);
+  gst_sample_unref (sample);
+
+  soup_server_unpause_message (server, ctx->msg);
+  g_print (".");
+  return GST_FLOW_OK;
+}
+
+static void
+write_next_client_chunk_cb (SoupMessage        *msg,
+                            TranscodeClientCtx *ctx)
+{
+  GstMapInfo info;
+  GstSample *sample;
+  GstBuffer *buffer;
+  GstMemory *memory;
+  gboolean eos;
+
+  g_signal_emit_by_name (ctx->appsink, "pull-sample", &sample);
+  if (!sample) {
+    g_print ("Null sample, ending stream\n");
+    g_object_get (ctx->appsink, "eos", &eos, NULL);
+    if (eos)
+      soup_message_set_status (msg, SOUP_STATUS_OK);
+    else
+      soup_message_set_status (msg, SOUP_STATUS_INTERNAL_SERVER_ERROR);
+    soup_message_body_complete (msg->response_body);
+    return;
+  }
+
+  buffer = gst_sample_get_buffer (sample);
+  memory = gst_buffer_get_all_memory (buffer); /* copy */
+  gst_memory_map (memory, &info, GST_MAP_READ);
+
+  soup_message_body_append (msg->response_body, SOUP_MEMORY_COPY,
+                            info.data, info.size); /* copy */
+
+  gst_memory_unmap (memory, &info);
+  gst_memory_unref (memory);
+  gst_sample_unref (sample);
+
+  soup_server_unpause_message (server, msg);
+  g_print (".");
+  return;
+}
+
+static void
+client_finished_cb (SoupMessage        *msg,
+                    TranscodeClientCtx *ctx)
+{
+  stp_cleanup_transcode_client_ctx (ctx);
+}
+
+static void
+handle_request_cb (SoupServer        *server,
+                   SoupMessage       *msg,
+                   const char        *path,
+                   GHashTable        *query,
+                   SoupClientContext *client,
+                   GHashTable        *ctx_table)
+{
+  TranscodeServerCtx *server_ctx;
+  
+  if (!msg)
+    return;
+
+  g_print ("Handling %s request on path %s\n", msg->method, path);
+
+  if (msg->method == SOUP_METHOD_PUT ||
+      msg->method == SOUP_METHOD_POST)
+    goto PUT;
+  else if (msg->method == SOUP_METHOD_GET)
+    goto GET;
+  else
+    /* We return NOT_IMPLEMENTED for this
+     * when we get the request headers */
+    g_assert_not_reached ();
+
+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;
+    soup_message_set_status (server_ctx->msg, SOUP_STATUS_OK);
+    return;
+  }
+
+GET:
+  {
+    /* A GET request was received. We connect from the pipeline to the
+     * client requesting the stream and start writing the response. */
+    GstElement *bin, *tee, *q2, *appsink;
+    GstPadTemplate *template;
+    GstPad *srcpad, *sinkpadq2;
+    TranscodeClientCtx *client_ctx;
+    GstStateChangeReturn state;
+ 
+    server_ctx = get_server_ctx_from_msg (msg, ctx_table);
+    if (server_ctx == NULL ||
+        server_ctx->pipeline == NULL) {
+      /* There's no request streaming on this path (yet) */
+      soup_message_set_status (msg, SOUP_STATUS_NOT_FOUND);
+      return;
+    }
+
+    /* Check if the pipeline successfully started PLAYING */
+    state = gst_element_get_state (server_ctx->pipeline,
+                                   NULL, NULL, 100*GST_MSECOND);
+    switch (state) {
+      case GST_STATE_CHANGE_SUCCESS:
+        break;
+      case GST_STATE_CHANGE_FAILURE:
+        /* PUT stream should've  */
+        g_critical ("GET request, but state change failure?");
+      case GST_STATE_CHANGE_ASYNC:
+        soup_message_set_status (msg, SOUP_STATUS_SERVICE_UNAVAILABLE);
+        return;
+      default:
+        g_assert_not_reached ();
+    }
+ 
+    /* Connect appsink to tee, and start streaming */
+    client_ctx = g_new0 (TranscodeClientCtx, 1);
+    client_ctx->msg = msg;
+    client_ctx->server_ctx = server_ctx;
+ 
+    bin = gst_bin_new (NULL);
+
+    /* queue ! appsink */
+    q2 = gst_element_factory_make ("queue", "q2");
+    appsink = gst_element_factory_make ("appsink", NULL);
+    g_object_set (appsink, "drop", FALSE,
+                  "emit-signals", TRUE,
+                  "max-buffers", 0, NULL);
+    gst_bin_add_many (GST_BIN (bin), q2, appsink, NULL);
+    gst_element_link (q2, appsink);
+
+    sinkpadq2 = gst_element_get_static_pad (q2, "sink");
+    gst_element_add_pad (bin, gst_ghost_pad_new ("qsink", sinkpadq2));
+    gst_object_unref (sinkpadq2);
+
+    /* Set to PAUSED so the bin can pre-roll */
+    if (gst_element_set_state (bin, GST_STATE_PAUSED) ==
+        GST_STATE_CHANGE_FAILURE) {
+      g_critical ("Unable to set appsink to PAUSED\n");
+      soup_message_set_status (client_ctx->msg, SOUP_STATUS_INTERNAL_SERVER_ERROR);
+      stp_cleanup_transcode_client_ctx (client_ctx);
+    } else {
+      g_print ("sinkbin set to PAUSED successfully\n");
+    }
+
+    /* Request a new src pad from the tee in the pipeline */
+    tee = gst_bin_get_by_name (GST_BIN (server_ctx->pipeline), "tee");
+    template = gst_pad_template_new ("teesrc", GST_PAD_SRC,
+                                     GST_PAD_REQUEST, GST_CAPS_ANY);
+    srcpad = gst_element_request_pad (tee, template, NULL, GST_CAPS_ANY);
+    
+    gst_pad_add_probe (srcpad, GST_PAD_PROBE_TYPE_BLOCK,
+                       (GstPadProbeCallback) tee_src_pad_blocked_cb,
+                       client_ctx, NULL);
+    gst_object_unref (srcpad);
+
+    gst_object_ref (appsink);
+    client_ctx->appsink = appsink;
+
+    client_ctx->first_sample_handler_id =
+      g_signal_connect (appsink, "new-sample",
+                        G_CALLBACK (write_first_client_chunk_cb), client_ctx);
+    g_signal_connect (client_ctx->msg, "wrote-chunk",
+                      G_CALLBACK (write_next_client_chunk_cb), client_ctx);
+    g_signal_connect (client_ctx->msg, "finished",
+                      G_CALLBACK (client_finished_cb), client_ctx);
+
+    server_ctx->clients = g_list_prepend (server_ctx->clients, client_ctx);
+
+    /* Since we set the response as chunked, the server will 
+     * automatically pause the message for us, and the response 
+     * won't be completed till we call soup_message_body_complete() */
+    soup_message_set_status (msg, SOUP_STATUS_OK);
+  }
+}
+
+static void
+got_request_body_chunk (SoupMessage        *msg,
+                        SoupBuffer         *chunk,
+                        TranscodeServerCtx *ctx)
+{
+  GstBuffer *buffer;
+  GstFlowReturn ret;
+
+  /* XXX: Unfortunately, there doesn't seem to be a way to cancel the
+   * 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)
+    return;
+
+  /* We need to update this every chunk */
+  ctx->msg = msg;
+
+  /* Actually just a ref */
+  buffer = stp_get_gst_buffer (chunk);
+  g_signal_emit_by_name (ctx->appsrc, "push-buffer", buffer, &ret);
+  gst_buffer_unref (buffer);
+  if (ret != GST_FLOW_OK) {
+    g_critical ("Unable to push buffer\n");
+    soup_message_set_status (msg, SOUP_STATUS_INTERNAL_SERVER_ERROR);
+    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
+got_request_headers (SoupMessage *msg,
+                     GHashTable  *ctx_table)
+{
+  TranscodeServerCtx *ctx;
+  SoupURI *uri;
+
+#ifdef DEBUG
+  SoupMessageHeadersIter iter;
+  const char *name, *value;
+  soup_message_headers_iter_init (&iter, msg->request_headers);
+  while (soup_message_headers_iter_next (&iter, &name, &value))
+    g_print ("%s: %s\n", name, value);
+#endif
+
+  g_object_get (msg, "uri", &uri, NULL);
+  g_print ("%s on uri %s\n", msg->method, soup_uri_get_path (uri));
+
+  if (msg->method == SOUP_METHOD_PUT ||
+      msg->method == SOUP_METHOD_POST)
+    goto PUT;
+  else if (msg->method == SOUP_METHOD_GET)
+    goto GET;
+  else {
+    /* Nothing to do; server doesn't implement this method */
+    soup_message_set_status (msg, SOUP_STATUS_NOT_IMPLEMENTED);
+    goto out;
+  }
+
+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);
+      goto out;
+    }
+
+    ctx = g_new0 (TranscodeServerCtx, 1);
+    ctx->msg = msg;
+    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 DEBUG
+      stp_play_from_msg (ctx);
+#else
+      stp_encode_from_msg (ctx);
+#endif
+      g_signal_connect (ctx->appsrc, "need-data",
+                        G_CALLBACK (need_data_unpause_message), ctx);
+      g_signal_connect (ctx->appsrc, "enough-data",
+                        G_CALLBACK (enough_data_pause_message), ctx);
+    }
+    
+    /* 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_signal_connect (msg, "got-chunk",
+                      G_CALLBACK (got_request_body_chunk), ctx);
+    goto out;
+   }
+
+GET: {
+    if (!g_hash_table_contains (ctx_table, soup_uri_get_path (uri))) {
+      g_print ("No stream on URI: %s\n", soup_uri_get_path (uri));
+      soup_message_set_status (msg, SOUP_STATUS_NOT_FOUND);
+      goto out;
+    }
+
+    /* Our response will be chunked, and not with a fixed Content-Length */
+    soup_message_headers_set_encoding (msg->response_headers, SOUP_ENCODING_CHUNKED);
+    /* 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->response_body, FALSE);
+   }
+
+out:
+  soup_uri_free (uri);
+  return;
+}
+
+static void
+request_started_cb (SoupServer        *server,
+                    SoupMessage       *msg,
+                    SoupClientContext *client,
+                    GHashTable        *ctx_table)
+{
+  g_print ("New %s request started\n", msg->method);
+
+  /* SoupMessage is useful only once we have headers,
+   * so we do all initialization there */
+  g_signal_connect (msg, "got-headers",
+                    G_CALLBACK (got_request_headers), ctx_table);
+}
+
+static void
+request_read_cb (SoupServer        *server,
+                 SoupMessage       *msg,
+                 SoupClientContext *client,
+                 GHashTable        *ctx_table)
+{
+  if (msg->method == SOUP_METHOD_PUT ||
+      msg->method == SOUP_METHOD_POST)
+    goto PUT;
+  else if (msg->method == SOUP_METHOD_GET)
+    goto GET;
+  else
+    /* We return NOT_IMPLEMENTED for this
+     * when we get the request headers */
+    g_assert_not_reached ();
+
+PUT:
+  {
+    gboolean ret;
+    TranscodeServerCtx *ctx;
+
+    ctx = get_server_ctx_from_msg (msg, ctx_table);
+    g_return_if_fail (ctx);
+
+    /* Incoming stream has ended */
+    g_signal_emit_by_name (ctx->appsrc, "end-of-stream", &ret);
+    if (!ret)
+      g_printerr ("Unable to emit end-of-stream\n");
+  }
+
+GET:
+  {
+    /* Nothing to do */
+  }
+
+  g_print ("%s request read successfully\n", msg->method);
+}
+
+static void
+request_finished_cb (SoupServer        *server,
+                     SoupMessage       *msg,
+                     SoupClientContext *client,
+                     GHashTable        *ctx_table)
+{
+  g_print ("%s request ended\n", msg->method);
+}
+
+static void
+request_aborted_cb (SoupServer        *server,
+                    SoupMessage       *msg,
+                    SoupClientContext *client,
+                    GHashTable        *ctx_table)
+{
+  g_print ("%s request aborted!\n", msg->method);
+  if (msg->method == SOUP_METHOD_PUT ||
+      msg->method == SOUP_METHOD_POST)
+    goto PUT;
+  else if (msg->method == SOUP_METHOD_GET)
+    goto GET;
+  else
+    /* We return NOT_IMPLEMENTED for this
+     * when we get the request headers */
+    g_assert_not_reached ();
+
+PUT:
+  {
+    /* FIXME: Is there anything to do here? */
+  }
+
+GET:
+  {
+    /* We do all cleanup in the "finished" msg signal, nothing to do here */
+  }
+}
+
+int
+main (int   argc,
+      char *argv[])
+{
+  GHashTable *ctx_table;
+
+  gst_init (&argc, &argv);
+
+  /* Keys are paths, and values are TranscodeServerCtxs */
+  ctx_table = g_hash_table_new_full (g_str_hash, g_str_equal,
+                                     (GDestroyNotify)g_free,
+                                     (GDestroyNotify)stp_cleanup_transcode_server_ctx);
+  server = soup_server_new (SOUP_SERVER_SERVER_HEADER,
+                            "soup-transcode-proxy ",
+                            SOUP_SERVER_PORT, 8000, NULL);
+  soup_server_add_handler (server, NULL,
+                           (SoupServerCallback)handle_request_cb,
+                           ctx_table, NULL);
+  g_signal_connect (server, "request-started",
+                    G_CALLBACK (request_started_cb), ctx_table);
+  g_signal_connect (server, "request-read",
+                    G_CALLBACK (request_read_cb), ctx_table);
+  g_signal_connect (server, "request-finished",
+                    G_CALLBACK (request_finished_cb), ctx_table);
+  g_signal_connect (server, "request-aborted",
+                    G_CALLBACK (request_aborted_cb), ctx_table);
+
+  soup_server_run (server);
+  return 0;
+}