souphttpsrc: refactor to use Soup's sync API

Replace the async API with the sync API to remove all the extra mainloop
and context handling. Currently it blocks reading until 'blocksize'
bytes are available but that can be improved by using:

https://developer.gnome.org/gio/unstable/GPollableInputStream.html#g-pollable-input-stream-read-nonblocking

https://bugzilla.gnome.org/show_bug.cgi?id=693911

2 files changed, 211 insertions, 485 deletions

diff --git a/ext/soup/gstsouphttpsrc.c b/ext/soup/gstsouphttpsrc.c
index f1475ae39..b707d0899 100644
--- a/ext/soup/gstsouphttpsrc.c
+++ b/ext/soup/gstsouphttpsrc.c
@@ -81,15 +81,6 @@
 #include "gstsouphttpsrc.h"
 #include "gstsouputils.h"
 
-/* libsoup before 2.47.0 was stealing our main context from us,
- * so we can't reliable use it to clean up all pending resources
- * once we're done... let's just continue leaking on old versions.
- * https://bugzilla.gnome.org/show_bug.cgi?id=663944
- */
-#if defined(SOUP_MINOR_VERSION) && SOUP_MINOR_VERSION >= 47
-#define LIBSOUP_DOES_NOT_STEAL_OUR_CONTEXT 1
-#endif
-
 #include <gst/tag/tag.h>
 
 GST_DEBUG_CATEGORY_STATIC (souphttpsrc_debug);
@@ -173,28 +164,14 @@ static char *gst_soup_http_src_unicodify (const char *str);
 static gboolean gst_soup_http_src_build_message (GstSoupHTTPSrc * src,
     const gchar * method);
 static void gst_soup_http_src_cancel_message (GstSoupHTTPSrc * src);
-static void gst_soup_http_src_queue_message (GstSoupHTTPSrc * src);
 static gboolean gst_soup_http_src_add_range_header (GstSoupHTTPSrc * src,
     guint64 offset, guint64 stop_offset);
-static void gst_soup_http_src_session_unpause_message (GstSoupHTTPSrc * src);
-static void gst_soup_http_src_session_pause_message (GstSoupHTTPSrc * src);
 static gboolean gst_soup_http_src_session_open (GstSoupHTTPSrc * src);
 static void gst_soup_http_src_session_close (GstSoupHTTPSrc * src);
 static void gst_soup_http_src_parse_status (SoupMessage * msg,
     GstSoupHTTPSrc * src);
-static void gst_soup_http_src_chunk_free (gpointer gstbuf);
-static SoupBuffer *gst_soup_http_src_chunk_allocator (SoupMessage * msg,
-    gsize max_len, gpointer user_data);
-static void gst_soup_http_src_got_chunk_cb (SoupMessage * msg,
-    SoupBuffer * chunk, GstSoupHTTPSrc * src);
-static void gst_soup_http_src_response_cb (SoupSession * session,
-    SoupMessage * msg, GstSoupHTTPSrc * src);
-static void gst_soup_http_src_got_headers_cb (SoupMessage * msg,
-    GstSoupHTTPSrc * src);
-static void gst_soup_http_src_got_body_cb (SoupMessage * msg,
-    GstSoupHTTPSrc * src);
-static void gst_soup_http_src_finished_cb (SoupMessage * msg,
-    GstSoupHTTPSrc * src);
+static void gst_soup_http_src_got_headers (GstSoupHTTPSrc * src,
+    SoupMessage * msg);
 static void gst_soup_http_src_authenticate_cb (SoupSession * session,
     SoupMessage * msg, SoupAuth * auth, gboolean retrying,
     GstSoupHTTPSrc * src);
@@ -450,8 +427,6 @@ gst_soup_http_src_class_init (GstSoupHTTPSrcClass * klass)
 static void
 gst_soup_http_src_reset (GstSoupHTTPSrc * src)
 {
-  src->interrupted = FALSE;
-  src->retry = FALSE;
   src->retry_count = 0;
   src->have_size = FALSE;
   src->got_headers = FALSE;
@@ -463,6 +438,11 @@ gst_soup_http_src_reset (GstSoupHTTPSrc * src)
   src->have_body = FALSE;
 
   src->ret = GST_FLOW_OK;
+  g_cancellable_reset (src->cancellable);
+  if (src->input_stream) {
+    g_object_unref (src->input_stream);
+    src->input_stream = NULL;
+  }
 
   gst_caps_replace (&src->src_caps, NULL);
   g_free (src->iradio_name);
@@ -479,7 +459,8 @@ gst_soup_http_src_init (GstSoupHTTPSrc * src)
   const gchar *proxy;
 
   g_mutex_init (&src->mutex);
-  g_cond_init (&src->request_finished_cond);
+  g_cond_init (&src->have_headers_cond);
+  src->cancellable = g_cancellable_new ();
   src->location = NULL;
   src->redirection_uri = NULL;
   src->automatic_redirect = TRUE;
@@ -490,8 +471,6 @@ gst_soup_http_src_init (GstSoupHTTPSrc * src)
   src->proxy_pw = NULL;
   src->cookies = NULL;
   src->iradio_mode = DEFAULT_IRADIO_MODE;
-  src->loop = NULL;
-  src->context = NULL;
   src->session = NULL;
   src->msg = NULL;
   src->timeout = DEFAULT_TIMEOUT;
@@ -534,7 +513,8 @@ gst_soup_http_src_finalize (GObject * gobject)
   GST_DEBUG_OBJECT (src, "finalize");
 
   g_mutex_clear (&src->mutex);
-  g_cond_clear (&src->request_finished_cond);
+  g_cond_clear (&src->have_headers_cond);
+  g_object_unref (src->cancellable);
   g_free (src->location);
   g_free (src->redirection_uri);
   g_free (src->user_agent);
@@ -787,21 +767,8 @@ gst_soup_http_src_unicodify (const gchar * str)
 static void
 gst_soup_http_src_cancel_message (GstSoupHTTPSrc * src)
 {
-  if (src->msg != NULL) {
-    GST_INFO_OBJECT (src, "Cancelling message");
-    src->session_io_status = GST_SOUP_HTTP_SRC_SESSION_IO_STATUS_CANCELLED;
-    soup_session_cancel_message (src->session, src->msg, SOUP_STATUS_CANCELLED);
-  }
-  src->session_io_status = GST_SOUP_HTTP_SRC_SESSION_IO_STATUS_IDLE;
-  src->msg = NULL;
-}
-
-static void
-gst_soup_http_src_queue_message (GstSoupHTTPSrc * src)
-{
-  soup_session_queue_message (src->session, src->msg,
-      (SoupSessionCallback) gst_soup_http_src_response_cb, src);
-  src->session_io_status = GST_SOUP_HTTP_SRC_SESSION_IO_STATUS_QUEUED;
+  g_cancellable_cancel (src->cancellable);
+  g_cond_signal (&src->have_headers_cond);
 }
 
 static gboolean
@@ -906,19 +873,6 @@ gst_soup_http_src_add_extra_headers (GstSoupHTTPSrc * src)
   return gst_structure_foreach (src->extra_headers, _append_extra_headers, src);
 }
 
-
-static void
-gst_soup_http_src_session_unpause_message (GstSoupHTTPSrc * src)
-{
-  soup_session_unpause_message (src->session, src->msg);
-}
-
-static void
-gst_soup_http_src_session_pause_message (GstSoupHTTPSrc * src)
-{
-  soup_session_pause_message (src->session, src->msg);
-}
-
 static gboolean
 gst_soup_http_src_session_open (GstSoupHTTPSrc * src)
 {
@@ -933,32 +887,18 @@ gst_soup_http_src_session_open (GstSoupHTTPSrc * src)
     return FALSE;
   }
 
-  if (!src->context)
-    src->context = g_main_context_new ();
-
-  if (!src->loop)
-    src->loop = g_main_loop_new (src->context, TRUE);
-  if (!src->loop) {
-    GST_ELEMENT_ERROR (src, LIBRARY, INIT,
-        (NULL), ("Failed to start GMainLoop"));
-    g_main_context_unref (src->context);
-    return FALSE;
-  }
-
   if (!src->session) {
     GST_DEBUG_OBJECT (src, "Creating session");
     if (src->proxy == NULL) {
       src->session =
-          soup_session_async_new_with_options (SOUP_SESSION_ASYNC_CONTEXT,
-          src->context, SOUP_SESSION_USER_AGENT, src->user_agent,
-          SOUP_SESSION_TIMEOUT, src->timeout,
+          soup_session_new_with_options (SOUP_SESSION_USER_AGENT,
+          src->user_agent, SOUP_SESSION_TIMEOUT, src->timeout,
           SOUP_SESSION_SSL_STRICT, src->ssl_strict,
           SOUP_SESSION_ADD_FEATURE_BY_TYPE, SOUP_TYPE_PROXY_RESOLVER_DEFAULT,
           SOUP_SESSION_TLS_INTERACTION, src->tls_interaction, NULL);
     } else {
       src->session =
-          soup_session_async_new_with_options (SOUP_SESSION_ASYNC_CONTEXT,
-          src->context, SOUP_SESSION_PROXY_URI, src->proxy,
+          soup_session_new_with_options (SOUP_SESSION_PROXY_URI, src->proxy,
           SOUP_SESSION_TIMEOUT, src->timeout,
           SOUP_SESSION_SSL_STRICT, src->ssl_strict,
           SOUP_SESSION_USER_AGENT, src->user_agent,
@@ -996,22 +936,11 @@ gst_soup_http_src_session_open (GstSoupHTTPSrc * src)
   return TRUE;
 }
 
-#ifdef LIBSOUP_DOES_NOT_STEAL_OUR_CONTEXT
-static gboolean
-dummy_idle_cb (gpointer data)
-{
-  return FALSE /* Idle source is removed */ ;
-}
-#endif
-
 static void
 gst_soup_http_src_session_close (GstSoupHTTPSrc * src)
 {
   GST_DEBUG_OBJECT (src, "Closing session");
 
-  if (src->loop)
-    g_main_loop_quit (src->loop);
-
   g_mutex_lock (&src->mutex);
   if (src->session) {
     soup_session_abort (src->session);  /* This unrefs the message. */
@@ -1019,33 +948,6 @@ gst_soup_http_src_session_close (GstSoupHTTPSrc * src)
     src->session = NULL;
     src->msg = NULL;
   }
-  if (src->loop) {
-#ifdef LIBSOUP_DOES_NOT_STEAL_OUR_CONTEXT
-    GSource *idle_source;
-
-    /* Iterating the main context to give GIO cancellables a chance
-     * to initiate cleanups. Wihout this, resources allocated by
-     * libsoup for the connection are not released and socket fd is
-     * leaked. */
-    idle_source = g_idle_source_new ();
-    /* Suppressing "idle souce without callback" warning */
-    g_source_set_callback (idle_source, dummy_idle_cb, NULL, NULL);
-    g_source_set_priority (idle_source, G_PRIORITY_LOW);
-    g_source_attach (idle_source, src->context);
-    /* Acquiring the context. Idle source guarantees that we'll not block. */
-    g_main_context_push_thread_default (src->context);
-    g_main_context_iteration (src->context, TRUE);
-    /* Ensuring that there's no unhandled pending events left. */
-    while (g_main_context_iteration (src->context, FALSE));
-    g_main_context_pop_thread_default (src->context);
-    g_source_unref (idle_source);
-#endif
-
-    g_main_loop_unref (src->loop);
-    g_main_context_unref (src->context);
-    src->loop = NULL;
-    src->context = NULL;
-  }
   g_mutex_unlock (&src->mutex);
 }
 
@@ -1100,7 +1002,7 @@ insert_http_header (const gchar * name, const gchar * value, gpointer user_data)
 }
 
 static void
-gst_soup_http_src_got_headers_cb (SoupMessage * msg, GstSoupHTTPSrc * src)
+gst_soup_http_src_got_headers (GstSoupHTTPSrc * src, SoupMessage * msg)
 {
   const char *value;
   GstTagList *tag_list;
@@ -1127,11 +1029,14 @@ gst_soup_http_src_got_headers_cb (SoupMessage * msg, GstSoupHTTPSrc * src)
     return;
   }
 
-  if (msg->status_code == SOUP_STATUS_UNAUTHORIZED)
+  if (msg->status_code == SOUP_STATUS_UNAUTHORIZED) {
+    /* force an error */
+    gst_soup_http_src_parse_status (msg, src);
     return;
+  }
 
-  src->session_io_status = GST_SOUP_HTTP_SRC_SESSION_IO_STATUS_RUNNING;
   src->got_headers = TRUE;
+  g_cond_broadcast (&src->have_headers_cond);
 
   http_headers = gst_structure_new_empty ("http-headers");
   gst_structure_set (http_headers, "uri", G_TYPE_STRING, src->location, NULL);
@@ -1299,251 +1204,23 @@ gst_soup_http_src_got_headers_cb (SoupMessage * msg, GstSoupHTTPSrc * src)
    * GST_FLOW_ERROR from the create function instead of having
    * got_chunk_cb overwrite src->ret with FLOW_OK again. */
   if (src->ret == GST_FLOW_ERROR || src->ret == GST_FLOW_EOS) {
-    gst_soup_http_src_session_pause_message (src);
-
-    if (src->loop)
-      g_main_loop_quit (src->loop);
-  }
-  g_cond_signal (&src->request_finished_cond);
-}
-
-/* Have body. Signal EOS. */
-static void
-gst_soup_http_src_got_body_cb (SoupMessage * msg, GstSoupHTTPSrc * src)
-{
-  if (G_UNLIKELY (msg != src->msg)) {
-    GST_DEBUG_OBJECT (src, "got body, but not for current message");
-    return;
-  }
-  if (G_UNLIKELY (src->session_io_status !=
-          GST_SOUP_HTTP_SRC_SESSION_IO_STATUS_RUNNING)) {
-    /* Probably a redirect. */
-    return;
-  }
-  GST_DEBUG_OBJECT (src, "got body");
-  src->ret = GST_FLOW_EOS;
-  src->have_body = TRUE;
-
-  /* no need to interrupt the message here, we do it on the
-   * finished_cb anyway if needed. And getting the body might mean
-   * that the connection was hang up before finished. This happens when
-   * the pipeline is stalled for too long (long pauses during playback).
-   * Best to let it continue from here and pause because it reached the
-   * final bytes based on content_size or received an out of range error */
-}
-
-/* Finished. Signal EOS. */
-static void
-gst_soup_http_src_finished_cb (SoupMessage * msg, GstSoupHTTPSrc * src)
-{
-  if (G_UNLIKELY (msg != src->msg)) {
-    GST_DEBUG_OBJECT (src, "finished, but not for current message");
-    return;
-  }
-  GST_INFO_OBJECT (src, "finished, io status: %d", src->session_io_status);
-  src->ret = GST_FLOW_EOS;
-  if (src->session_io_status == GST_SOUP_HTTP_SRC_SESSION_IO_STATUS_CANCELLED) {
-    /* gst_soup_http_src_cancel_message() triggered this; probably a seek
-     * that occurred in the QUEUEING state; i.e. before the connection setup
-     * was complete. Do nothing */
-    GST_DEBUG_OBJECT (src, "cancelled");
-  } else if (src->session_io_status ==
-      GST_SOUP_HTTP_SRC_SESSION_IO_STATUS_RUNNING && src->read_position > 0 &&
-      (src->have_size && src->read_position < src->content_size) &&
-      (src->max_retries == -1 || src->retry_count < src->max_retries)) {
-    /* The server disconnected while streaming. Reconnect and seeking to the
-     * last location. */
-    src->retry = TRUE;
-    src->retry_count++;
-    src->ret = GST_FLOW_CUSTOM_ERROR;
-  } else if (G_UNLIKELY (src->session_io_status !=
-          GST_SOUP_HTTP_SRC_SESSION_IO_STATUS_RUNNING)) {
-    if (msg->method == SOUP_METHOD_HEAD) {
-      GST_DEBUG_OBJECT (src, "Ignoring error %d:%s during HEAD request",
-          msg->status_code, msg->reason_phrase);
-    } else {
-      gst_soup_http_src_parse_status (msg, src);
-    }
   }
-  if (src->loop)
-    g_main_loop_quit (src->loop);
-  g_cond_signal (&src->request_finished_cond);
-}
-
-/* Buffer lifecycle management.
- *
- * gst_soup_http_src_create() runs the GMainLoop for this element, to let
- * Soup take control.
- * A GstBuffer is allocated in gst_soup_http_src_chunk_allocator() and
- * associated with a SoupBuffer.
- * Soup reads HTTP data in the GstBuffer's data buffer.
- * The gst_soup_http_src_got_chunk_cb() is then called with the SoupBuffer.
- * That sets gst_soup_http_src_create()'s return argument to the GstBuffer,
- * increments its refcount (to 2), pauses the flow of data from the HTTP
- * source to prevent gst_soup_http_src_got_chunk_cb() from being called
- * again and breaks out of the GMainLoop.
- * Because the SOUP_MESSAGE_OVERWRITE_CHUNKS flag is set, Soup frees the
- * SoupBuffer and calls gst_soup_http_src_chunk_free(), which decrements the
- * refcount (to 1).
- * gst_soup_http_src_create() returns the GstBuffer. It will be freed by a
- * downstream element.
- * If Soup fails to read HTTP data, it does not call
- * gst_soup_http_src_got_chunk_cb(), but still frees the SoupBuffer and
- * calls gst_soup_http_src_chunk_free(), which decrements the GstBuffer's
- * refcount to 0, freeing it.
- */
-
-typedef struct
-{
-  GstBuffer *buffer;
-  GstMapInfo map;
-} SoupGstChunk;
-
-static void
-gst_soup_http_src_chunk_free (gpointer user_data)
-{
-  SoupGstChunk *chunk = (SoupGstChunk *) user_data;
-
-  gst_buffer_unmap (chunk->buffer, &chunk->map);
-  gst_buffer_unref (chunk->buffer);
-  g_slice_free (SoupGstChunk, chunk);
 }
 
-static SoupBuffer *
-gst_soup_http_src_chunk_allocator (SoupMessage * msg, gsize max_len,
-    gpointer user_data)
+static GstBuffer *
+gst_soup_http_src_alloc_buffer (GstSoupHTTPSrc * src)
 {
-  GstSoupHTTPSrc *src = (GstSoupHTTPSrc *) user_data;
   GstBaseSrc *basesrc = GST_BASE_SRC_CAST (src);
-  GstBuffer *gstbuf;
-  SoupBuffer *soupbuf;
-  gsize length;
   GstFlowReturn rc;
-  SoupGstChunk *chunk;
-
-  if (max_len)
-    length = MIN (basesrc->blocksize, max_len);
-  else
-    length = basesrc->blocksize;
-  GST_DEBUG_OBJECT (src, "alloc %" G_GSIZE_FORMAT " bytes <= %" G_GSIZE_FORMAT,
-      length, max_len);
+  GstBuffer *gstbuf;
 
-  rc = GST_BASE_SRC_CLASS (parent_class)->alloc (basesrc, -1, length, &gstbuf);
+  rc = GST_BASE_SRC_CLASS (parent_class)->alloc (basesrc, -1,
+      basesrc->blocksize, &gstbuf);
   if (G_UNLIKELY (rc != GST_FLOW_OK)) {
-    /* Failed to allocate buffer. Stall SoupSession and return error code
-     * to create(). */
-    src->ret = rc;
-    g_main_loop_quit (src->loop);
     return NULL;
   }
 
-  chunk = g_slice_new0 (SoupGstChunk);
-  chunk->buffer = gstbuf;
-  gst_buffer_map (gstbuf, &chunk->map, GST_MAP_READWRITE);
-
-  soupbuf = soup_buffer_new_with_owner (chunk->map.data, chunk->map.size,
-      chunk, gst_soup_http_src_chunk_free);
-
-  return soupbuf;
-}
-
-static void
-gst_soup_http_src_got_chunk_cb (SoupMessage * msg, SoupBuffer * chunk,
-    GstSoupHTTPSrc * src)
-{
-  GstBaseSrc *basesrc;
-  guint64 new_position;
-  SoupGstChunk *gchunk;
-
-  if (G_UNLIKELY (msg != src->msg)) {
-    GST_DEBUG_OBJECT (src, "got chunk, but not for current message");
-    return;
-  }
-  if (G_UNLIKELY (!src->outbuf)) {
-    GST_DEBUG_OBJECT (src, "got chunk but we're not expecting one");
-    src->ret = GST_FLOW_OK;
-    gst_soup_http_src_cancel_message (src);
-    g_main_loop_quit (src->loop);
-    return;
-  }
-
-  /* We got data, reset the retry counter */
-  src->retry_count = 0;
-
-  src->have_body = FALSE;
-  if (G_UNLIKELY (src->session_io_status !=
-          GST_SOUP_HTTP_SRC_SESSION_IO_STATUS_RUNNING)) {
-    /* Probably a redirect. */
-    return;
-  }
-  basesrc = GST_BASE_SRC_CAST (src);
-  GST_DEBUG_OBJECT (src, "got chunk of %" G_GSIZE_FORMAT " bytes",
-      chunk->length);
-
-  /* Extract the GstBuffer from the SoupBuffer and set its fields. */
-  gchunk = (SoupGstChunk *) soup_buffer_get_owner (chunk);
-  *src->outbuf = gchunk->buffer;
-
-  gst_buffer_resize (*src->outbuf, 0, chunk->length);
-  GST_BUFFER_OFFSET (*src->outbuf) = basesrc->segment.position;
-
-  gst_buffer_ref (*src->outbuf);
-
-  new_position = src->read_position + chunk->length;
-  if (G_LIKELY (src->request_position == src->read_position))
-    src->request_position = new_position;
-  src->read_position = new_position;
-
-  if (src->have_size) {
-    if (new_position > src->content_size) {
-      GST_DEBUG_OBJECT (src, "Got position previous estimated content size "
-          "(%" G_GINT64_FORMAT " > %" G_GINT64_FORMAT ")", new_position,
-          src->content_size);
-      src->content_size = new_position;
-      basesrc->segment.duration = src->content_size;
-      gst_element_post_message (GST_ELEMENT (src),
-          gst_message_new_duration_changed (GST_OBJECT (src)));
-    } else if (new_position == src->content_size) {
-      GST_DEBUG_OBJECT (src, "We're EOS now");
-    }
-  }
-
-  src->ret = GST_FLOW_OK;
-  g_main_loop_quit (src->loop);
-  gst_soup_http_src_session_pause_message (src);
-}
-
-static void
-gst_soup_http_src_response_cb (SoupSession * session, SoupMessage * msg,
-    GstSoupHTTPSrc * src)
-{
-  if (G_UNLIKELY (msg != src->msg)) {
-    GST_DEBUG_OBJECT (src, "got response %d: %s, but not for current message",
-        msg->status_code, msg->reason_phrase);
-    return;
-  }
-  if (G_UNLIKELY (src->session_io_status !=
-          GST_SOUP_HTTP_SRC_SESSION_IO_STATUS_RUNNING)
-      && SOUP_STATUS_IS_REDIRECTION (msg->status_code)) {
-    /* Ignore redirections. */
-    return;
-  }
-  GST_INFO_OBJECT (src, "got response %d: %s", msg->status_code,
-      msg->reason_phrase);
-  if (src->session_io_status == GST_SOUP_HTTP_SRC_SESSION_IO_STATUS_RUNNING &&
-      src->read_position > 0 && (src->have_size
-          && src->read_position < src->content_size) &&
-      (src->max_retries == -1 || src->retry_count < src->max_retries)) {
-    /* The server disconnected while streaming. Reconnect and seeking to the
-     * last location. */
-    src->retry = TRUE;
-    src->retry_count++;
-  } else {
-    gst_soup_http_src_parse_status (msg, src);
-  }
-  /* The session's SoupMessage object expires after this callback returns. */
-  src->msg = NULL;
-  g_main_loop_quit (src->loop);
+  return gstbuf;
 }
 
 #define SOUP_HTTP_SRC_ERROR(src,soup_msg,cat,code,error_message)     \
@@ -1579,8 +1256,6 @@ gst_soup_http_src_parse_status (SoupMessage * msg, GstSoupHTTPSrc * src)
         break;
       case SOUP_STATUS_IO_ERROR:
         if (src->max_retries == -1 || src->retry_count < src->max_retries) {
-          src->retry = TRUE;
-          src->retry_count++;
           src->ret = GST_FLOW_CUSTOM_ERROR;
         } else {
           SOUP_HTTP_SRC_ERROR (src, msg, RESOURCE, READ,
@@ -1653,7 +1328,6 @@ gst_soup_http_src_build_message (GstSoupHTTPSrc * src, const gchar * method)
         ("Error parsing URL."), ("URL: %s", src->location));
     return FALSE;
   }
-  src->session_io_status = GST_SOUP_HTTP_SRC_SESSION_IO_STATUS_IDLE;
   if (!src->keep_alive) {
     soup_message_headers_append (src->msg->request_headers, "Connection",
         "close");
@@ -1670,20 +1344,9 @@ gst_soup_http_src_build_message (GstSoupHTTPSrc * src, const gchar * method)
           *cookie);
     }
   }
-  src->retry = FALSE;
 
-  g_signal_connect (src->msg, "got_headers",
-      G_CALLBACK (gst_soup_http_src_got_headers_cb), src);
-  g_signal_connect (src->msg, "got_body",
-      G_CALLBACK (gst_soup_http_src_got_body_cb), src);
-  g_signal_connect (src->msg, "finished",
-      G_CALLBACK (gst_soup_http_src_finished_cb), src);
-  g_signal_connect (src->msg, "got_chunk",
-      G_CALLBACK (gst_soup_http_src_got_chunk_cb), src);
   soup_message_set_flags (src->msg, SOUP_MESSAGE_OVERWRITE_CHUNKS |
       (src->automatic_redirect ? 0 : SOUP_MESSAGE_NO_REDIRECT));
-  soup_message_set_chunk_allocator (src->msg,
-      gst_soup_http_src_chunk_allocator, src, NULL);
   gst_soup_http_src_add_range_header (src, src->request_position,
       src->stop_position);
 
@@ -1693,145 +1356,217 @@ gst_soup_http_src_build_message (GstSoupHTTPSrc * src, const gchar * method)
 }
 
 static GstFlowReturn
-gst_soup_http_src_do_request (GstSoupHTTPSrc * src, const gchar * method,
-    GstBuffer ** outbuf)
+gst_soup_http_src_send_message (GstSoupHTTPSrc * src)
 {
-  /* If we're not OK, just go out of here */
+  g_return_val_if_fail (src->msg != NULL, GST_FLOW_ERROR);
+
+  /* FIXME We are ignoring the GError here, might be useful to debug */
+  src->input_stream =
+      soup_session_send (src->session, src->msg, src->cancellable, NULL);
+  gst_soup_http_src_got_headers (src, src->msg);
   if (src->ret != GST_FLOW_OK) {
-    GST_DEBUG_OBJECT (src, "Previous flow return not OK: %s",
-        gst_flow_get_name (src->ret));
     return src->ret;
   }
 
+  if (!src->input_stream) {
+    GST_DEBUG_OBJECT (src, "Didn't get an input stream");
+    return GST_FLOW_ERROR;
+  }
+
+  if (SOUP_STATUS_IS_SUCCESSFUL (src->msg->status_code)) {
+    GST_DEBUG_OBJECT (src, "Successfully got a reply");
+  } else {
+    /* FIXME - be more helpful to people debugging */
+    return GST_FLOW_ERROR;
+  }
+
+  return GST_FLOW_OK;
+}
+
+static GstFlowReturn
+gst_soup_http_src_do_request (GstSoupHTTPSrc * src, const gchar * method)
+{
+  if (src->max_retries != -1 && src->retry_count > src->max_retries) {
+    GST_DEBUG_OBJECT (src, "Max retries reached");
+    src->ret = GST_FLOW_ERROR;
+    return src->ret;
+  }
+
+  src->retry_count++;
+  /* EOS immediately if we have an empty segment */
+  if (src->request_position == src->stop_position)
+    return GST_FLOW_EOS;
+
   GST_LOG_OBJECT (src, "Running request for method: %s", method);
+
+  /* Update the position if we are retrying */
   if (src->msg && (src->request_position != src->read_position)) {
-    if (src->session_io_status == GST_SOUP_HTTP_SRC_SESSION_IO_STATUS_IDLE) {
-      /* EOS immediately if we have an empty segment */
-      if (src->request_position == src->stop_position)
-        return GST_FLOW_EOS;
+    gst_soup_http_src_add_range_header (src, src->request_position,
+        src->stop_position);
+  }
 
-      gst_soup_http_src_add_range_header (src, src->request_position,
-          src->stop_position);
-    } else {
-      GST_DEBUG_OBJECT (src, "Seek from position %" G_GUINT64_FORMAT
-          " to %" G_GUINT64_FORMAT ": requeueing connection request",
-          src->read_position, src->request_position);
-      gst_soup_http_src_cancel_message (src);
+  if (!src->msg) {
+    if (!gst_soup_http_src_build_message (src, method)) {
+      return GST_FLOW_ERROR;
     }
   }
-  if (!src->msg) {
-    /* EOS immediately if we have an empty segment */
-    if (src->request_position == src->stop_position)
-      return GST_FLOW_EOS;
 
-    if (!gst_soup_http_src_build_message (src, method))
-      return GST_FLOW_ERROR;
+  if (g_cancellable_is_cancelled (src->cancellable)) {
+    GST_INFO_OBJECT (src, "interrupted");
+    src->ret = GST_FLOW_FLUSHING;
+    goto done;
   }
+  src->ret = gst_soup_http_src_send_message (src);
 
-  src->ret = GST_FLOW_CUSTOM_ERROR;
-  src->outbuf = outbuf;
-  do {
-    if (src->interrupted) {
-      GST_INFO_OBJECT (src, "interrupted");
-      src->ret = GST_FLOW_FLUSHING;
-      break;
-    }
-    if (src->retry) {
-      GST_INFO_OBJECT (src, "Reconnecting");
+done:
+  return src->ret;
+}
 
-      /* EOS immediately if we have an empty segment */
-      if (src->request_position == src->stop_position)
-        return GST_FLOW_EOS;
+static void
+gst_soup_http_src_update_position (GstSoupHTTPSrc * src, gint64 bytes_read)
+{
+  GstBaseSrc *basesrc = GST_BASE_SRC_CAST (src);
+  guint64 new_position;
 
-      if (!gst_soup_http_src_build_message (src, method))
-        return GST_FLOW_ERROR;
-      src->retry = FALSE;
-      continue;
-    }
-    if (!src->msg) {
-      GST_DEBUG_OBJECT (src, "EOS reached");
-      break;
-    }
+  new_position = src->read_position + bytes_read;
+  if (G_LIKELY (src->request_position == src->read_position))
+    src->request_position = new_position;
+  src->read_position = new_position;
 
-    switch (src->session_io_status) {
-      case GST_SOUP_HTTP_SRC_SESSION_IO_STATUS_IDLE:
-        GST_INFO_OBJECT (src, "Queueing connection request");
-        gst_soup_http_src_queue_message (src);
-        break;
-      case GST_SOUP_HTTP_SRC_SESSION_IO_STATUS_QUEUED:
-        break;
-      case GST_SOUP_HTTP_SRC_SESSION_IO_STATUS_RUNNING:
-        gst_soup_http_src_session_unpause_message (src);
-        break;
-      case GST_SOUP_HTTP_SRC_SESSION_IO_STATUS_CANCELLED:
-        /* Impossible. */
-        break;
+  if (src->have_size) {
+    if (new_position > src->content_size) {
+      GST_DEBUG_OBJECT (src, "Got position previous estimated content size "
+          "(%" G_GINT64_FORMAT " > %" G_GINT64_FORMAT ")", new_position,
+          src->content_size);
+      src->content_size = new_position;
+      basesrc->segment.duration = src->content_size;
+      gst_element_post_message (GST_ELEMENT (src),
+          gst_message_new_duration_changed (GST_OBJECT (src)));
+    } else if (new_position == src->content_size) {
+      GST_DEBUG_OBJECT (src, "We're EOS now");
     }
+  }
+}
 
-    if (src->ret == GST_FLOW_CUSTOM_ERROR) {
-      g_main_context_push_thread_default (src->context);
-      g_main_loop_run (src->loop);
-      g_main_context_pop_thread_default (src->context);
-    }
+static GstFlowReturn
+gst_soup_http_src_read_buffer (GstSoupHTTPSrc * src, GstBuffer ** outbuf)
+{
+  gssize read_bytes;
+  GstMapInfo mapinfo;
+  GstBaseSrc *bsrc;
+  GstFlowReturn ret;
 
-  } while (src->ret == GST_FLOW_CUSTOM_ERROR);
+  bsrc = GST_BASE_SRC_CAST (src);
 
-  /* Let the request finish if we had a stop position and are there */
-  if (src->ret == GST_FLOW_OK && src->stop_position != -1
-      && src->read_position >= src->stop_position) {
-    src->outbuf = NULL;
-    gst_soup_http_src_session_unpause_message (src);
-    g_main_context_push_thread_default (src->context);
-    g_main_loop_run (src->loop);
-    g_main_context_pop_thread_default (src->context);
+  *outbuf = gst_soup_http_src_alloc_buffer (src);
+  if (!*outbuf) {
+    GST_WARNING_OBJECT (src, "Failed to allocate buffer");
+    return GST_FLOW_ERROR;
+  }
 
-    g_cond_signal (&src->request_finished_cond);
-    /* Return OK unconditionally here, src->ret will
-     * be most likely be EOS now but we want to
-     * consume the buffer we got above */
-    return GST_FLOW_OK;
+  if (!gst_buffer_map (*outbuf, &mapinfo, GST_MAP_WRITE)) {
+    GST_WARNING_OBJECT (src, "Failed to map buffer");
+    return GST_FLOW_ERROR;
   }
 
-  if (src->ret == GST_FLOW_CUSTOM_ERROR)
-    src->ret = GST_FLOW_EOS;
-  g_cond_signal (&src->request_finished_cond);
+  read_bytes =
+      g_input_stream_read (src->input_stream, mapinfo.data, mapinfo.size,
+      src->cancellable, NULL);
+  GST_DEBUG_OBJECT (src, "Read %" G_GSSIZE_FORMAT " bytes from http input",
+      read_bytes);
 
-  /* basesrc assumes that we don't return a buffer if
-   * something else than OK is returned. It will just
-   * leak any buffer we might accidentially provide
-   * here.
-   *
-   * This can potentially happen during flushing.
-   */
-  if (src->ret != GST_FLOW_OK && outbuf && *outbuf) {
-    gst_buffer_unref (*outbuf);
-    *outbuf = NULL;
+  g_mutex_lock (&src->mutex);
+  if (g_cancellable_is_cancelled (src->cancellable)) {
+    g_mutex_unlock (&src->mutex);
+    return GST_FLOW_FLUSHING;
   }
+  g_mutex_unlock (&src->mutex);
 
-  return src->ret;
+  gst_buffer_unmap (*outbuf, &mapinfo);
+  if (read_bytes > 0) {
+    gst_buffer_set_size (*outbuf, read_bytes);
+    GST_BUFFER_OFFSET (*outbuf) = bsrc->segment.position;
+    ret = GST_FLOW_OK;
+    gst_soup_http_src_update_position (src, read_bytes);
+
+    /* Got some data, reset retry counter */
+    src->retry_count = 0;
+  } else {
+    gst_buffer_unref (*outbuf);
+    if (read_bytes < 0) {
+      /* Maybe the server disconnected, retry */
+      ret = GST_FLOW_CUSTOM_ERROR;
+    } else {
+      ret = GST_FLOW_EOS;
+      src->have_body = TRUE;
+    }
+  }
+  return ret;
 }
 
 static GstFlowReturn
 gst_soup_http_src_create (GstPushSrc * psrc, GstBuffer ** outbuf)
 {
   GstSoupHTTPSrc *src;
-  GstFlowReturn ret;
-  GstEvent *http_headers_event;
+  GstFlowReturn ret = GST_FLOW_OK;
+  GstEvent *http_headers_event = NULL;
 
   src = GST_SOUP_HTTP_SRC (psrc);
 
+retry:
   g_mutex_lock (&src->mutex);
-  *outbuf = NULL;
-  ret =
-      gst_soup_http_src_do_request (src,
-      src->method ? src->method : SOUP_METHOD_GET, outbuf);
-  http_headers_event = src->http_headers_event;
-  src->http_headers_event = NULL;
+
+  /* Check for pending position change */
+  if (src->request_position != src->read_position) {
+    if (src->input_stream) {
+      g_input_stream_close (src->input_stream, src->cancellable, NULL);
+      g_object_unref (src->input_stream);
+      src->input_stream = NULL;
+    }
+  }
+
+  if (g_cancellable_is_cancelled (src->cancellable)) {
+    ret = GST_FLOW_FLUSHING;
+    g_mutex_unlock (&src->mutex);
+    goto done;
+  }
+
+  /* If we have no open connection to the server, start one */
+  if (!src->input_stream) {
+    *outbuf = NULL;
+    ret =
+        gst_soup_http_src_do_request (src,
+        src->method ? src->method : SOUP_METHOD_GET);
+    http_headers_event = src->http_headers_event;
+    src->http_headers_event = NULL;
+  }
   g_mutex_unlock (&src->mutex);
 
-  if (http_headers_event)
-    gst_pad_push_event (GST_BASE_SRC_PAD (src), http_headers_event);
+  if (ret == GST_FLOW_OK || ret == GST_FLOW_CUSTOM_ERROR) {
+    if (http_headers_event) {
+      gst_pad_push_event (GST_BASE_SRC_PAD (src), http_headers_event);
+      http_headers_event = NULL;
+    }
+  }
+
+  if (ret == GST_FLOW_OK)
+    ret = gst_soup_http_src_read_buffer (src, outbuf);
 
+done:
+  GST_DEBUG_OBJECT (src, "Returning %d %s", ret, gst_flow_get_name (ret));
+  if (ret != GST_FLOW_OK) {
+    if (http_headers_event)
+      gst_event_unref (http_headers_event);
+
+    g_mutex_lock (&src->mutex);
+    if (src->input_stream) {
+      g_object_unref (src->input_stream);
+      src->input_stream = NULL;
+    }
+    g_mutex_unlock (&src->mutex);
+    if (ret == GST_FLOW_CUSTOM_ERROR)
+      goto retry;
+  }
   return ret;
 }
 
@@ -1891,11 +1626,8 @@ gst_soup_http_src_unlock (GstBaseSrc * bsrc)
   src = GST_SOUP_HTTP_SRC (bsrc);
   GST_DEBUG_OBJECT (src, "unlock()");
 
-  src->interrupted = TRUE;
   src->ret = GST_FLOW_FLUSHING;
-  if (src->loop)
-    g_main_loop_quit (src->loop);
-  g_cond_signal (&src->request_finished_cond);
+  gst_soup_http_src_cancel_message (src);
   return TRUE;
 }
 
@@ -1908,8 +1640,8 @@ gst_soup_http_src_unlock_stop (GstBaseSrc * bsrc)
   src = GST_SOUP_HTTP_SRC (bsrc);
   GST_DEBUG_OBJECT (src, "unlock_stop()");
 
-  src->interrupted = FALSE;
   src->ret = GST_FLOW_OK;
+  g_cancellable_reset (src->cancellable);
   return TRUE;
 }
 
@@ -1941,14 +1673,14 @@ gst_soup_http_src_check_seekable (GstSoupHTTPSrc * src)
    */
   if (!src->got_headers && GST_STATE (src) >= GST_STATE_PAUSED) {
     g_mutex_lock (&src->mutex);
-    while (!src->got_headers && !src->interrupted && ret == GST_FLOW_OK) {
-      if ((src->msg && src->msg->method != SOUP_METHOD_HEAD) &&
-          src->session_io_status != GST_SOUP_HTTP_SRC_SESSION_IO_STATUS_IDLE) {
+    while (!src->got_headers && !g_cancellable_is_cancelled (src->cancellable)
+        && ret == GST_FLOW_OK) {
+      if ((src->msg && src->msg->method != SOUP_METHOD_HEAD)) {
         /* wait for the current request to finish */
-        g_cond_wait (&src->request_finished_cond, &src->mutex);
+        g_cond_wait (&src->have_headers_cond, &src->mutex);
       } else {
         if (gst_soup_http_src_session_open (src)) {
-          ret = gst_soup_http_src_do_request (src, SOUP_METHOD_HEAD, NULL);
+          ret = gst_soup_http_src_do_request (src, SOUP_METHOD_HEAD);
         }
       }
     }
@@ -1956,8 +1688,6 @@ gst_soup_http_src_check_seekable (GstSoupHTTPSrc * src)
       /* A HEAD request shouldn't lead to EOS */
       src->ret = GST_FLOW_OK;
     }
-    /* resets status to idle */
-    gst_soup_http_src_cancel_message (src);
     g_mutex_unlock (&src->mutex);
   }
 }
diff --git a/ext/soup/gstsouphttpsrc.h b/ext/soup/gstsouphttpsrc.h
index 71725817e..2d5b3ee84 100644
--- a/ext/soup/gstsouphttpsrc.h
+++ b/ext/soup/gstsouphttpsrc.h
@@ -59,16 +59,9 @@ struct _GstSoupHTTPSrc {
   gchar *proxy_id;             /* Authentication user id for proxy URI. */
   gchar *proxy_pw;             /* Authentication user password for proxy URI. */
   gchar **cookies;             /* HTTP request cookies. */
-  GMainContext *context;       /* I/O context. */
-  GMainLoop *loop;             /* Event loop. */
   SoupSession *session;        /* Async context. */
-  GstSoupHTTPSrcSessionIOStatus session_io_status;
-                               /* Async I/O status. */
   SoupMessage *msg;            /* Request message. */
   GstFlowReturn ret;           /* Return code from callback. */
-  GstBuffer **outbuf;          /* Return buffer allocated by callback. */
-  gboolean interrupted;        /* Signal unlock(). */
-  gboolean retry;              /* Should attempt to reconnect. */
   gint retry_count;            /* Number of retries since we received data */
   gint max_retries;            /* Maximum number of retries */
   gchar *method;               /* HTTP method */
@@ -94,6 +87,9 @@ struct _GstSoupHTTPSrc {
   GTlsDatabase *tls_database;
   GTlsInteraction *tls_interaction;
 
+  GCancellable *cancellable;
+  GInputStream *input_stream;
+
   /* Shoutcast/icecast metadata extraction handling. */
   gboolean iradio_mode;
   GstCaps *src_caps;
@@ -110,7 +106,7 @@ struct _GstSoupHTTPSrc {
   guint timeout;
 
   GMutex mutex;
-  GCond request_finished_cond;
+  GCond have_headers_cond;
 
   GstEvent *http_headers_event;
 };