ext/soup/gstsouphttpsink.c

   1 /* GStreamer
   2  * Copyright (C) 2011 David Schleef <ds@entropywave.com>
   3  *
   4  * This library is free software; you can redistribute it and/or
   5  * modify it under the terms of the GNU Library General Public
   6  * License as published by the Free Software Foundation; either
   7  * version 2 of the License, or (at your option) any later version.
   8  *
   9  * This library is distributed in the hope that it will be useful,
  10  * but WITHOUT ANY WARRANTY; without even the implied warranty of
  11  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  12  * Library General Public License for more details.
  13  *
  14  * You should have received a copy of the GNU Library General Public
  15  * License along with this library; if not, write to the
  16  * Free Software Foundation, Inc., 51 Franklin Street, Suite 500,
  17  * Boston, MA 02110-1335, USA.
  18  */
  19 /**
  20  * SECTION:element-gstsouphttpsink
  21  *
  22  * The souphttpsink element sends pipeline data to an HTTP server
  23  * using HTTP PUT commands.
  24  *
  25  * <refsect2>
  26  * <title>Example launch line</title>
  27  * |[
  28  * gst-launch -v videotestsrc num-buffers=300 ! theoraenc ! oggmux !
  29  *   souphttpsink location=http://server/filename.ogv
  30  * ]|
  31  *
  32  * This example encodes 10 seconds of video and sends it to the HTTP
  33  * server "server" using HTTP PUT commands.
  34  * </refsect2>
  35  */
  36
  37 #ifdef HAVE_CONFIG_H
  38 #include "config.h"
  39 #endif
  40
  41 #include <gst/gst.h>
  42 #include <gst/base/gstbasesink.h>
  43 #include "gstsouphttpsink.h"
  44
  45 GST_DEBUG_CATEGORY_STATIC (gst_soup_http_sink_debug_category);
  46 #define GST_CAT_DEFAULT gst_soup_http_sink_debug_category
  47
  48 /* prototypes */
  49
  50
  51 static void gst_soup_http_sink_set_property (GObject * object,
  52     guint property_id, const GValue * value, GParamSpec * pspec);
  53 static void gst_soup_http_sink_get_property (GObject * object,
  54     guint property_id, GValue * value, GParamSpec * pspec);
  55 static void gst_soup_http_sink_dispose (GObject * object);
  56 static void gst_soup_http_sink_finalize (GObject * object);
  57
  58 static gboolean gst_soup_http_sink_set_caps (GstBaseSink * sink,
  59     GstCaps * caps);
  60 static void gst_soup_http_sink_get_times (GstBaseSink * sink,
  61     GstBuffer * buffer, GstClockTime * start, GstClockTime * end);
  62 static gboolean gst_soup_http_sink_start (GstBaseSink * sink);
  63 static gboolean gst_soup_http_sink_stop (GstBaseSink * sink);
  64 static gboolean gst_soup_http_sink_unlock (GstBaseSink * sink);
  65 static gboolean gst_soup_http_sink_event (GstBaseSink * sink, GstEvent * event);
  66 static GstFlowReturn
  67 gst_soup_http_sink_preroll (GstBaseSink * sink, GstBuffer * buffer);
  68 static GstFlowReturn
  69 gst_soup_http_sink_render (GstBaseSink * sink, GstBuffer * buffer);
  70
  71 static void free_buffer_list (GList * list);
  72 static void gst_soup_http_sink_reset (GstSoupHttpSink * souphttpsink);
  73 static void authenticate (SoupSession * session, SoupMessage * msg,
  74     SoupAuth * auth, gboolean retrying, gpointer user_data);
  75 static void
  76 callback (SoupSession * session, SoupMessage * msg, gpointer user_data);
  77 static gboolean
  78 gst_soup_http_sink_set_proxy (GstSoupHttpSink * souphttpsink,
  79     const gchar * uri);
  80
  81 enum
  82 {
  83   PROP_0,
  84   PROP_LOCATION,
  85   PROP_USER_AGENT,
  86   PROP_AUTOMATIC_REDIRECT,
  87   PROP_PROXY,
  88   PROP_USER_ID,
  89   PROP_USER_PW,
  90   PROP_PROXY_ID,
  91   PROP_PROXY_PW,
  92   PROP_COOKIES,
  93   PROP_SESSION
  94 };
  95
  96 #define DEFAULT_USER_AGENT           "GStreamer souphttpsink "
  97
  98 /* pad templates */
  99
 100 static GstStaticPadTemplate gst_soup_http_sink_sink_template =
 101 GST_STATIC_PAD_TEMPLATE ("sink",
 102     GST_PAD_SINK,
 103     GST_PAD_ALWAYS,
 104     GST_STATIC_CAPS_ANY);
 105
 106
 107 /* class initialization */
 108
 109 #define DEBUG_INIT(bla) \
 110   GST_DEBUG_CATEGORY_INIT (gst_soup_http_sink_debug_category, "souphttpsink", 0, \
 111       "debug category for souphttpsink element");
 112
 113 GST_BOILERPLATE_FULL (GstSoupHttpSink, gst_soup_http_sink, GstBaseSink,
 114     GST_TYPE_BASE_SINK, DEBUG_INIT);
 115
 116 static void
 117 gst_soup_http_sink_base_init (gpointer g_class)
 118 {
 119   GstElementClass *element_class = GST_ELEMENT_CLASS (g_class);
 120
 121   gst_element_class_add_pad_template (element_class,
 122       gst_static_pad_template_get (&gst_soup_http_sink_sink_template));
 123
 124   gst_element_class_set_details_simple (element_class, "HTTP client sink",
 125       "Generic", "Sends streams to HTTP server via PUT",
 126       "David Schleef <ds@entropywave.com>");
 127 }
 128
 129 static void
 130 gst_soup_http_sink_class_init (GstSoupHttpSinkClass * klass)
 131 {
 132   GObjectClass *gobject_class = G_OBJECT_CLASS (klass);
 133   GstBaseSinkClass *base_sink_class = GST_BASE_SINK_CLASS (klass);
 134
 135   gobject_class->set_property = gst_soup_http_sink_set_property;
 136   gobject_class->get_property = gst_soup_http_sink_get_property;
 137   gobject_class->dispose = gst_soup_http_sink_dispose;
 138   gobject_class->finalize = gst_soup_http_sink_finalize;
 139   base_sink_class->set_caps = GST_DEBUG_FUNCPTR (gst_soup_http_sink_set_caps);
 140   if (0)
 141     base_sink_class->get_times =
 142         GST_DEBUG_FUNCPTR (gst_soup_http_sink_get_times);
 143   base_sink_class->start = GST_DEBUG_FUNCPTR (gst_soup_http_sink_start);
 144   base_sink_class->stop = GST_DEBUG_FUNCPTR (gst_soup_http_sink_stop);
 145   base_sink_class->unlock = GST_DEBUG_FUNCPTR (gst_soup_http_sink_unlock);
 146   base_sink_class->event = GST_DEBUG_FUNCPTR (gst_soup_http_sink_event);
 147   if (0)
 148     base_sink_class->preroll = GST_DEBUG_FUNCPTR (gst_soup_http_sink_preroll);
 149   base_sink_class->render = GST_DEBUG_FUNCPTR (gst_soup_http_sink_render);
 150
 151   g_object_class_install_property (gobject_class,
 152       PROP_LOCATION,
 153       g_param_spec_string ("location", "Location",
 154           "URI to send to", "", G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 155   g_object_class_install_property (gobject_class,
 156       PROP_USER_AGENT,
 157       g_param_spec_string ("user-agent", "User-Agent",
 158           "Value of the User-Agent HTTP request header field",
 159           DEFAULT_USER_AGENT, G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 160   g_object_class_install_property (gobject_class,
 161       PROP_AUTOMATIC_REDIRECT,
 162       g_param_spec_boolean ("automatic-redirect", "automatic-redirect",
 163           "Automatically follow HTTP redirects (HTTP Status Code 3xx)",
 164           TRUE, G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 165   g_object_class_install_property (gobject_class,
 166       PROP_PROXY,
 167       g_param_spec_string ("proxy", "Proxy",
 168           "HTTP proxy server URI", "",
 169           G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 170   g_object_class_install_property (gobject_class,
 171       PROP_USER_ID,
 172       g_param_spec_string ("user-id", "user-id",
 173           "user id for authentication", "",
 174           G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 175   g_object_class_install_property (gobject_class, PROP_USER_PW,
 176       g_param_spec_string ("user-pw", "user-pw",
 177           "user password for authentication", "",
 178           G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 179   g_object_class_install_property (gobject_class, PROP_PROXY_ID,
 180       g_param_spec_string ("proxy-id", "proxy-id",
 181           "user id for proxy authentication", "",
 182           G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 183   g_object_class_install_property (gobject_class, PROP_PROXY_PW,
 184       g_param_spec_string ("proxy-pw", "proxy-pw",
 185           "user password for proxy authentication", "",
 186           G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 187   g_object_class_install_property (gobject_class, PROP_SESSION,
 188       g_param_spec_object ("session", "session",
 189           "SoupSession object to use for communication",
 190           SOUP_TYPE_SESSION, G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 191   g_object_class_install_property (gobject_class, PROP_COOKIES,
 192       g_param_spec_boxed ("cookies", "Cookies", "HTTP request cookies",
 193           G_TYPE_STRV, G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 194
 195 }
 196
 197 static void
 198 gst_soup_http_sink_init (GstSoupHttpSink * souphttpsink,
 199     GstSoupHttpSinkClass * souphttpsink_class)
 200 {
 201   const char *proxy;
 202
 203   souphttpsink->sinkpad =
 204       gst_pad_new_from_static_template (&gst_soup_http_sink_sink_template,
 205       "sink");
 206
 207   souphttpsink->mutex = g_mutex_new ();
 208   souphttpsink->cond = g_cond_new ();
 209
 210   souphttpsink->location = NULL;
 211   souphttpsink->automatic_redirect = TRUE;
 212   souphttpsink->user_agent = g_strdup (DEFAULT_USER_AGENT);
 213   souphttpsink->user_id = NULL;
 214   souphttpsink->user_pw = NULL;
 215   souphttpsink->proxy_id = NULL;
 216   souphttpsink->proxy_pw = NULL;
 217   souphttpsink->prop_session = NULL;
 218   souphttpsink->timeout = 1;
 219   proxy = g_getenv ("http_proxy");
 220   if (proxy && !gst_soup_http_sink_set_proxy (souphttpsink, proxy)) {
 221     GST_WARNING_OBJECT (souphttpsink,
 222         "The proxy in the http_proxy env var (\"%s\") cannot be parsed.",
 223         proxy);
 224   }
 225
 226   gst_soup_http_sink_reset (souphttpsink);
 227 }
 228
 229 static void
 230 gst_soup_http_sink_reset (GstSoupHttpSink * souphttpsink)
 231 {
 232   g_free (souphttpsink->reason_phrase);
 233   souphttpsink->reason_phrase = NULL;
 234   souphttpsink->status_code = 0;
 235   souphttpsink->offset = 0;
 236
 237 }
 238
 239 static gboolean
 240 gst_soup_http_sink_set_proxy (GstSoupHttpSink * souphttpsink, const gchar * uri)
 241 {
 242   if (souphttpsink->proxy) {
 243     soup_uri_free (souphttpsink->proxy);
 244     souphttpsink->proxy = NULL;
 245   }
 246   if (g_str_has_prefix (uri, "http://")) {
 247     souphttpsink->proxy = soup_uri_new (uri);
 248   } else {
 249     gchar *new_uri = g_strconcat ("http://", uri, NULL);
 250
 251     souphttpsink->proxy = soup_uri_new (new_uri);
 252     g_free (new_uri);
 253   }
 254
 255   return TRUE;
 256 }
 257
 258 void
 259 gst_soup_http_sink_set_property (GObject * object, guint property_id,
 260     const GValue * value, GParamSpec * pspec)
 261 {
 262   GstSoupHttpSink *souphttpsink = GST_SOUP_HTTP_SINK (object);
 263
 264   g_mutex_lock (souphttpsink->mutex);
 265   switch (property_id) {
 266     case PROP_SESSION:
 267       if (souphttpsink->prop_session) {
 268         g_object_unref (souphttpsink->prop_session);
 269       }
 270       souphttpsink->prop_session = g_value_dup_object (value);
 271       break;
 272     case PROP_LOCATION:
 273       g_free (souphttpsink->location);
 274       souphttpsink->location = g_value_dup_string (value);
 275       souphttpsink->offset = 0;
 276       break;
 277     case PROP_USER_AGENT:
 278       g_free (souphttpsink->user_agent);
 279       souphttpsink->user_agent = g_value_dup_string (value);
 280       break;
 281     case PROP_AUTOMATIC_REDIRECT:
 282       souphttpsink->automatic_redirect = g_value_get_boolean (value);
 283       break;
 284     case PROP_USER_ID:
 285       g_free (souphttpsink->user_id);
 286       souphttpsink->user_id = g_value_dup_string (value);
 287       break;
 288     case PROP_USER_PW:
 289       g_free (souphttpsink->user_pw);
 290       souphttpsink->user_pw = g_value_dup_string (value);
 291       break;
 292     case PROP_PROXY_ID:
 293       g_free (souphttpsink->proxy_id);
 294       souphttpsink->proxy_id = g_value_dup_string (value);
 295       break;
 296     case PROP_PROXY_PW:
 297       g_free (souphttpsink->proxy_pw);
 298       souphttpsink->proxy_pw = g_value_dup_string (value);
 299       break;
 300     case PROP_PROXY:
 301     {
 302       const gchar *proxy;
 303
 304       proxy = g_value_get_string (value);
 305
 306       if (proxy == NULL) {
 307         GST_WARNING ("proxy property cannot be NULL");
 308         goto done;
 309       }
 310       if (!gst_soup_http_sink_set_proxy (souphttpsink, proxy)) {
 311         GST_WARNING ("badly formatted proxy URI");
 312         goto done;
 313       }
 314       break;
 315     }
 316     case PROP_COOKIES:
 317       g_strfreev (souphttpsink->cookies);
 318       souphttpsink->cookies = g_strdupv (g_value_get_boxed (value));
 319       break;
 320     default:
 321       G_OBJECT_WARN_INVALID_PROPERTY_ID (object, property_id, pspec);
 322       break;
 323   }
 324 done:
 325   g_mutex_unlock (souphttpsink->mutex);
 326 }
 327
 328 void
 329 gst_soup_http_sink_get_property (GObject * object, guint property_id,
 330     GValue * value, GParamSpec * pspec)
 331 {
 332   GstSoupHttpSink *souphttpsink = GST_SOUP_HTTP_SINK (object);
 333
 334   switch (property_id) {
 335     case PROP_SESSION:
 336       g_value_set_object (value, souphttpsink->prop_session);
 337       break;
 338     case PROP_LOCATION:
 339       g_value_set_string (value, souphttpsink->location);
 340       break;
 341     case PROP_AUTOMATIC_REDIRECT:
 342       g_value_set_boolean (value, souphttpsink->automatic_redirect);
 343       break;
 344     case PROP_USER_AGENT:
 345       g_value_set_string (value, souphttpsink->user_agent);
 346       break;
 347     case PROP_USER_ID:
 348       g_value_set_string (value, souphttpsink->user_id);
 349       break;
 350     case PROP_USER_PW:
 351       g_value_set_string (value, souphttpsink->user_pw);
 352       break;
 353     case PROP_PROXY_ID:
 354       g_value_set_string (value, souphttpsink->proxy_id);
 355       break;
 356     case PROP_PROXY_PW:
 357       g_value_set_string (value, souphttpsink->proxy_pw);
 358       break;
 359     case PROP_PROXY:
 360       if (souphttpsink->proxy == NULL)
 361         g_value_set_static_string (value, "");
 362       else {
 363         char *proxy = soup_uri_to_string (souphttpsink->proxy, FALSE);
 364
 365         g_value_set_string (value, proxy);
 366         g_free (proxy);
 367       }
 368       break;
 369     case PROP_COOKIES:
 370       g_value_set_boxed (value, g_strdupv (souphttpsink->cookies));
 371       break;
 372     default:
 373       G_OBJECT_WARN_INVALID_PROPERTY_ID (object, property_id, pspec);
 374       break;
 375   }
 376 }
 377
 378 void
 379 gst_soup_http_sink_dispose (GObject * object)
 380 {
 381   GstSoupHttpSink *souphttpsink = GST_SOUP_HTTP_SINK (object);
 382
 383   /* clean up as possible.  may be called multiple times */
 384   if (souphttpsink->prop_session)
 385     g_object_unref (souphttpsink->prop_session);
 386   souphttpsink->prop_session = NULL;
 387
 388   G_OBJECT_CLASS (parent_class)->dispose (object);
 389 }
 390
 391 void
 392 gst_soup_http_sink_finalize (GObject * object)
 393 {
 394   GstSoupHttpSink *souphttpsink = GST_SOUP_HTTP_SINK (object);
 395
 396   /* clean up object here */
 397
 398   g_free (souphttpsink->user_agent);
 399   g_free (souphttpsink->user_id);
 400   g_free (souphttpsink->user_pw);
 401   g_free (souphttpsink->proxy_id);
 402   g_free (souphttpsink->proxy_pw);
 403   if (souphttpsink->proxy)
 404     soup_uri_free (souphttpsink->proxy);
 405   g_free (souphttpsink->location);
 406
 407   g_cond_free (souphttpsink->cond);
 408   g_mutex_free (souphttpsink->mutex);
 409
 410   G_OBJECT_CLASS (parent_class)->finalize (object);
 411 }
 412
 413
 414
 415 static gboolean
 416 gst_soup_http_sink_set_caps (GstBaseSink * sink, GstCaps * caps)
 417 {
 418   GstSoupHttpSink *souphttpsink = GST_SOUP_HTTP_SINK (sink);
 419   GstStructure *structure;
 420   const GValue *value_array;
 421   int i, n;
 422
 423   structure = gst_caps_get_structure (caps, 0);
 424   value_array = gst_structure_get_value (structure, "streamheader");
 425   if (value_array) {
 426     free_buffer_list (souphttpsink->streamheader_buffers);
 427     souphttpsink->streamheader_buffers = NULL;
 428
 429     n = gst_value_array_get_size (value_array);
 430     for (i = 0; i < n; i++) {
 431       const GValue *value;
 432       GstBuffer *buffer;
 433       value = gst_value_array_get_value (value_array, i);
 434       buffer = GST_BUFFER (gst_value_get_buffer (value));
 435       souphttpsink->streamheader_buffers =
 436           g_list_append (souphttpsink->streamheader_buffers,
 437           gst_buffer_ref (buffer));
 438     }
 439   }
 440
 441   return TRUE;
 442 }
 443
 444 static void
 445 gst_soup_http_sink_get_times (GstBaseSink * sink, GstBuffer * buffer,
 446     GstClockTime * start, GstClockTime * end)
 447 {
 448
 449 }
 450
 451 static gpointer
 452 thread_func (gpointer ptr)
 453 {
 454   GstSoupHttpSink *souphttpsink = GST_SOUP_HTTP_SINK (ptr);
 455
 456   GST_DEBUG ("thread start");
 457
 458   souphttpsink->loop = g_main_loop_new (souphttpsink->context, TRUE);
 459   g_main_loop_run (souphttpsink->loop);
 460
 461   GST_DEBUG ("thread quit");
 462
 463   return NULL;
 464 }
 465
 466 static gboolean
 467 gst_soup_http_sink_start (GstBaseSink * sink)
 468 {
 469   GstSoupHttpSink *souphttpsink = GST_SOUP_HTTP_SINK (sink);
 470
 471   if (souphttpsink->prop_session) {
 472     souphttpsink->session = souphttpsink->prop_session;
 473   } else {
 474     GError *error = NULL;
 475
 476     souphttpsink->context = g_main_context_new ();
 477
 478     souphttpsink->thread = g_thread_create (thread_func, souphttpsink,
 479         TRUE, &error);
 480
 481     souphttpsink->session =
 482         soup_session_async_new_with_options (SOUP_SESSION_ASYNC_CONTEXT,
 483         souphttpsink->context, SOUP_SESSION_USER_AGENT,
 484         souphttpsink->user_agent, SOUP_SESSION_TIMEOUT, souphttpsink->timeout,
 485         NULL);
 486
 487     //soup_session_add_feature (souphttpsink->session,
 488     //    SOUP_SESSION_FEATURE (soup_logger_new (SOUP_LOGGER_LOG_BODY, 100)));
 489
 490     g_signal_connect (souphttpsink->session, "authenticate",
 491         G_CALLBACK (authenticate), souphttpsink);
 492   }
 493
 494   return TRUE;
 495 }
 496
 497 static gboolean
 498 gst_soup_http_sink_stop (GstBaseSink * sink)
 499 {
 500   GstSoupHttpSink *souphttpsink = GST_SOUP_HTTP_SINK (sink);
 501
 502   GST_DEBUG ("stop");
 503
 504   if (souphttpsink->prop_session == NULL) {
 505     soup_session_abort (souphttpsink->session);
 506     g_object_unref (souphttpsink->session);
 507   }
 508
 509   if (souphttpsink->loop) {
 510     g_main_loop_quit (souphttpsink->loop);
 511     g_thread_join (souphttpsink->thread);
 512     g_main_loop_unref (souphttpsink->loop);
 513     souphttpsink->loop = NULL;
 514   }
 515   if (souphttpsink->context) {
 516     g_main_context_unref (souphttpsink->context);
 517     souphttpsink->context = NULL;
 518   }
 519
 520   gst_soup_http_sink_reset (souphttpsink);
 521
 522   return TRUE;
 523 }
 524
 525 static gboolean
 526 gst_soup_http_sink_unlock (GstBaseSink * sink)
 527 {
 528   GST_DEBUG ("unlock");
 529
 530   return TRUE;
 531 }
 532
 533 static gboolean
 534 gst_soup_http_sink_event (GstBaseSink * sink, GstEvent * event)
 535 {
 536   GstSoupHttpSink *souphttpsink = GST_SOUP_HTTP_SINK (sink);
 537
 538   GST_DEBUG_OBJECT (souphttpsink, "event");
 539
 540   if (GST_EVENT_TYPE (event) == GST_EVENT_EOS) {
 541     GST_DEBUG_OBJECT (souphttpsink, "got eos");
 542     g_mutex_lock (souphttpsink->mutex);
 543     while (souphttpsink->message) {
 544       GST_DEBUG_OBJECT (souphttpsink, "waiting");
 545       g_cond_wait (souphttpsink->cond, souphttpsink->mutex);
 546     }
 547     g_mutex_unlock (souphttpsink->mutex);
 548     GST_DEBUG_OBJECT (souphttpsink, "finished eos");
 549   }
 550
 551   return TRUE;
 552 }
 553
 554 static GstFlowReturn
 555 gst_soup_http_sink_preroll (GstBaseSink * sink, GstBuffer * buffer)
 556 {
 557   GST_DEBUG ("preroll");
 558
 559   return GST_FLOW_OK;
 560 }
 561
 562 static void
 563 free_buffer_list (GList * list)
 564 {
 565   GList *g;
 566   for (g = list; g; g = g_list_next (g)) {
 567     GstBuffer *buffer = g->data;
 568     gst_buffer_unref (buffer);
 569   }
 570   g_list_free (list);
 571 }
 572
 573 static void
 574 send_message_locked (GstSoupHttpSink * souphttpsink)
 575 {
 576   GList *g;
 577   guint64 n;
 578
 579   if (souphttpsink->queued_buffers == NULL || souphttpsink->message) {
 580     return;
 581   }
 582
 583   /* If the URI went away, drop all these buffers */
 584   if (souphttpsink->location == NULL) {
 585     free_buffer_list (souphttpsink->queued_buffers);
 586     souphttpsink->queued_buffers = NULL;
 587     return;
 588   }
 589
 590   souphttpsink->message = soup_message_new ("PUT", souphttpsink->location);
 591
 592   n = 0;
 593   if (souphttpsink->offset == 0) {
 594     for (g = souphttpsink->streamheader_buffers; g; g = g_list_next (g)) {
 595       GstBuffer *buffer = g->data;
 596       soup_message_body_append (souphttpsink->message->request_body,
 597           SOUP_MEMORY_STATIC, GST_BUFFER_DATA (buffer),
 598           GST_BUFFER_SIZE (buffer));
 599       n += GST_BUFFER_SIZE (buffer);
 600     }
 601   }
 602
 603   for (g = souphttpsink->queued_buffers; g; g = g_list_next (g)) {
 604     GstBuffer *buffer = g->data;
 605     if (!GST_BUFFER_FLAG_IS_SET (buffer, GST_BUFFER_FLAG_IN_CAPS)) {
 606       soup_message_body_append (souphttpsink->message->request_body,
 607           SOUP_MEMORY_STATIC, GST_BUFFER_DATA (buffer),
 608           GST_BUFFER_SIZE (buffer));
 609       n += GST_BUFFER_SIZE (buffer);
 610     }
 611   }
 612
 613   if (souphttpsink->offset != 0) {
 614     char *s;
 615     s = g_strdup_printf ("bytes %" G_GUINT64_FORMAT "-%" G_GUINT64_FORMAT "/*",
 616         souphttpsink->offset, souphttpsink->offset + n - 1);
 617     soup_message_headers_append (souphttpsink->message->request_headers,
 618         "Content-Range", s);
 619     g_free (s);
 620   }
 621
 622   if (n == 0) {
 623     free_buffer_list (souphttpsink->queued_buffers);
 624     souphttpsink->queued_buffers = NULL;
 625     g_object_unref (souphttpsink->message);
 626     souphttpsink->message = NULL;
 627     return;
 628   }
 629
 630   souphttpsink->sent_buffers = souphttpsink->queued_buffers;
 631   souphttpsink->queued_buffers = NULL;
 632
 633   GST_DEBUG_OBJECT (souphttpsink,
 634       "queue message %" G_GUINT64_FORMAT " %" G_GUINT64_FORMAT,
 635       souphttpsink->offset, n);
 636   soup_session_queue_message (souphttpsink->session, souphttpsink->message,
 637       callback, souphttpsink);
 638
 639   souphttpsink->offset += n;
 640 }
 641
 642 static gboolean
 643 send_message (GstSoupHttpSink * souphttpsink)
 644 {
 645   g_mutex_lock (souphttpsink->mutex);
 646   send_message_locked (souphttpsink);
 647   g_mutex_unlock (souphttpsink->mutex);
 648
 649   return FALSE;
 650 }
 651
 652 static void
 653 callback (SoupSession * session, SoupMessage * msg, gpointer user_data)
 654 {
 655   GstSoupHttpSink *souphttpsink = GST_SOUP_HTTP_SINK (user_data);
 656
 657   GST_DEBUG_OBJECT (souphttpsink, "callback status=%d %s",
 658       msg->status_code, msg->reason_phrase);
 659
 660   g_mutex_lock (souphttpsink->mutex);
 661   g_cond_signal (souphttpsink->cond);
 662   souphttpsink->message = NULL;
 663
 664   if (!SOUP_STATUS_IS_SUCCESSFUL (msg->status_code)) {
 665     souphttpsink->status_code = msg->status_code;
 666     souphttpsink->reason_phrase = g_strdup (msg->reason_phrase);
 667     g_mutex_unlock (souphttpsink->mutex);
 668     return;
 669   }
 670
 671   free_buffer_list (souphttpsink->sent_buffers);
 672   souphttpsink->sent_buffers = NULL;
 673
 674   send_message_locked (souphttpsink);
 675   g_mutex_unlock (souphttpsink->mutex);
 676 }
 677
 678 static GstFlowReturn
 679 gst_soup_http_sink_render (GstBaseSink * sink, GstBuffer * buffer)
 680 {
 681   GstSoupHttpSink *souphttpsink = GST_SOUP_HTTP_SINK (sink);
 682   GSource *source;
 683   gboolean wake;
 684
 685   if (souphttpsink->status_code != 0) {
 686     /* FIXME we should allow a moderate amount of retries. */
 687     GST_ELEMENT_ERROR (souphttpsink, RESOURCE, WRITE,
 688         ("Could not write to HTTP URI"),
 689         ("error: %d %s", souphttpsink->status_code,
 690             souphttpsink->reason_phrase));
 691     return GST_FLOW_ERROR;
 692   }
 693
 694   g_mutex_lock (souphttpsink->mutex);
 695   if (souphttpsink->location != NULL) {
 696     wake = (souphttpsink->queued_buffers == NULL);
 697     souphttpsink->queued_buffers =
 698         g_list_append (souphttpsink->queued_buffers, gst_buffer_ref (buffer));
 699
 700     if (wake) {
 701       source = g_idle_source_new ();
 702       g_source_set_callback (source, (GSourceFunc) (send_message),
 703           souphttpsink, NULL);
 704       g_source_attach (source, souphttpsink->context);
 705       g_source_unref (source);
 706     }
 707   }
 708   g_mutex_unlock (souphttpsink->mutex);
 709
 710   return GST_FLOW_OK;
 711 }
 712
 713 static void
 714 authenticate (SoupSession * session, SoupMessage * msg,
 715     SoupAuth * auth, gboolean retrying, gpointer user_data)
 716 {
 717   GstSoupHttpSink *souphttpsink = GST_SOUP_HTTP_SINK (user_data);
 718
 719   if (!retrying) {
 720     if (souphttpsink->user_id && souphttpsink->user_pw) {
 721       soup_auth_authenticate (auth,
 722           souphttpsink->user_id, souphttpsink->user_pw);
 723     }
 724   }
 725 }