2 * Copyright (C) 2011 David Schleef <ds@entropywave.com>
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.
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.
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.
20 * SECTION:element-gstsouphttpclientsink
22 * The souphttpclientsink element sends pipeline data to an HTTP server
23 * using HTTP PUT commands.
26 * <title>Example launch line</title>
28 * gst-launch-1.0 -v videotestsrc num-buffers=300 ! theoraenc ! oggmux !
29 * souphttpclientsink location=http://server/filename.ogv
32 * This example encodes 10 seconds of video and sends it to the HTTP
33 * server "server" using HTTP PUT commands.
42 #include <gst/base/gstbasesink.h>
43 #include "gstsouphttpclientsink.h"
44 #include "gstsouputils.h"
46 #include <gst/glib-compat-private.h>
48 GST_DEBUG_CATEGORY_STATIC (souphttpclientsink_dbg);
49 #define GST_CAT_DEFAULT souphttpclientsink_dbg
54 static void gst_soup_http_client_sink_set_property (GObject * object,
55 guint property_id, const GValue * value, GParamSpec * pspec);
56 static void gst_soup_http_client_sink_get_property (GObject * object,
57 guint property_id, GValue * value, GParamSpec * pspec);
58 static void gst_soup_http_client_sink_dispose (GObject * object);
59 static void gst_soup_http_client_sink_finalize (GObject * object);
61 static gboolean gst_soup_http_client_sink_set_caps (GstBaseSink * sink,
63 static void gst_soup_http_client_sink_get_times (GstBaseSink * sink,
64 GstBuffer * buffer, GstClockTime * start, GstClockTime * end);
65 static gboolean gst_soup_http_client_sink_start (GstBaseSink * sink);
66 static gboolean gst_soup_http_client_sink_stop (GstBaseSink * sink);
67 static gboolean gst_soup_http_client_sink_unlock (GstBaseSink * sink);
68 static gboolean gst_soup_http_client_sink_event (GstBaseSink * sink,
70 static GstFlowReturn gst_soup_http_client_sink_preroll (GstBaseSink * sink,
72 static GstFlowReturn gst_soup_http_client_sink_render (GstBaseSink * sink,
75 static void free_buffer_list (GList * list);
76 static void gst_soup_http_client_sink_reset (GstSoupHttpClientSink *
78 static void authenticate (SoupSession * session, SoupMessage * msg,
79 SoupAuth * auth, gboolean retrying, gpointer user_data);
80 static void callback (SoupSession * session, SoupMessage * msg,
82 static gboolean gst_soup_http_client_sink_set_proxy (GstSoupHttpClientSink *
83 souphttpsink, const gchar * uri);
90 PROP_AUTOMATIC_REDIRECT,
101 #define DEFAULT_USER_AGENT "GStreamer souphttpclientsink "
102 #define DEFAULT_SOUP_LOG_LEVEL SOUP_LOGGER_LOG_NONE
106 static GstStaticPadTemplate gst_soup_http_client_sink_sink_template =
107 GST_STATIC_PAD_TEMPLATE ("sink",
110 GST_STATIC_CAPS_ANY);
113 /* class initialization */
115 #define gst_soup_http_client_sink_parent_class parent_class
116 G_DEFINE_TYPE (GstSoupHttpClientSink, gst_soup_http_client_sink,
120 gst_soup_http_client_sink_class_init (GstSoupHttpClientSinkClass * klass)
122 GObjectClass *gobject_class = G_OBJECT_CLASS (klass);
123 GstElementClass *gstelement_class = GST_ELEMENT_CLASS (klass);
124 GstBaseSinkClass *base_sink_class = GST_BASE_SINK_CLASS (klass);
126 gobject_class->set_property = gst_soup_http_client_sink_set_property;
127 gobject_class->get_property = gst_soup_http_client_sink_get_property;
128 gobject_class->dispose = gst_soup_http_client_sink_dispose;
129 gobject_class->finalize = gst_soup_http_client_sink_finalize;
131 g_object_class_install_property (gobject_class,
133 g_param_spec_string ("location", "Location",
134 "URI to send to", "", G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
135 g_object_class_install_property (gobject_class,
137 g_param_spec_string ("user-agent", "User-Agent",
138 "Value of the User-Agent HTTP request header field",
139 DEFAULT_USER_AGENT, G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
140 g_object_class_install_property (gobject_class,
141 PROP_AUTOMATIC_REDIRECT,
142 g_param_spec_boolean ("automatic-redirect", "automatic-redirect",
143 "Automatically follow HTTP redirects (HTTP Status Code 3xx)",
144 TRUE, G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
145 g_object_class_install_property (gobject_class,
147 g_param_spec_string ("proxy", "Proxy",
148 "HTTP proxy server URI", "",
149 G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
150 g_object_class_install_property (gobject_class,
152 g_param_spec_string ("user-id", "user-id",
153 "user id for authentication", "",
154 G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
155 g_object_class_install_property (gobject_class, PROP_USER_PW,
156 g_param_spec_string ("user-pw", "user-pw",
157 "user password for authentication", "",
158 G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
159 g_object_class_install_property (gobject_class, PROP_PROXY_ID,
160 g_param_spec_string ("proxy-id", "proxy-id",
161 "user id for proxy authentication", "",
162 G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
163 g_object_class_install_property (gobject_class, PROP_PROXY_PW,
164 g_param_spec_string ("proxy-pw", "proxy-pw",
165 "user password for proxy authentication", "",
166 G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
167 g_object_class_install_property (gobject_class, PROP_SESSION,
168 g_param_spec_object ("session", "session",
169 "SoupSession object to use for communication",
170 SOUP_TYPE_SESSION, G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
171 g_object_class_install_property (gobject_class, PROP_COOKIES,
172 g_param_spec_boxed ("cookies", "Cookies", "HTTP request cookies",
173 G_TYPE_STRV, G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
175 * GstSoupHttpClientSink::http-log-level:
177 * If set and > 0, captures and dumps HTTP session data as
178 * log messages if log level >= GST_LEVEL_TRACE
182 g_object_class_install_property (gobject_class, PROP_SOUP_LOG_LEVEL,
183 g_param_spec_enum ("http-log-level", "HTTP log level",
184 "Set log level for soup's HTTP session log",
185 SOUP_TYPE_LOGGER_LOG_LEVEL, DEFAULT_SOUP_LOG_LEVEL,
186 G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
188 gst_element_class_add_pad_template (gstelement_class,
189 gst_static_pad_template_get (&gst_soup_http_client_sink_sink_template));
191 gst_element_class_set_static_metadata (gstelement_class, "HTTP client sink",
192 "Generic", "Sends streams to HTTP server via PUT",
193 "David Schleef <ds@entropywave.com>");
195 base_sink_class->set_caps =
196 GST_DEBUG_FUNCPTR (gst_soup_http_client_sink_set_caps);
198 base_sink_class->get_times =
199 GST_DEBUG_FUNCPTR (gst_soup_http_client_sink_get_times);
200 base_sink_class->start = GST_DEBUG_FUNCPTR (gst_soup_http_client_sink_start);
201 base_sink_class->stop = GST_DEBUG_FUNCPTR (gst_soup_http_client_sink_stop);
202 base_sink_class->unlock =
203 GST_DEBUG_FUNCPTR (gst_soup_http_client_sink_unlock);
204 base_sink_class->event = GST_DEBUG_FUNCPTR (gst_soup_http_client_sink_event);
206 base_sink_class->preroll =
207 GST_DEBUG_FUNCPTR (gst_soup_http_client_sink_preroll);
208 base_sink_class->render =
209 GST_DEBUG_FUNCPTR (gst_soup_http_client_sink_render);
211 GST_DEBUG_CATEGORY_INIT (souphttpclientsink_dbg, "souphttpclientsink", 0,
212 "souphttpclientsink element");
217 gst_soup_http_client_sink_init (GstSoupHttpClientSink * souphttpsink)
221 g_mutex_init (&souphttpsink->mutex);
222 g_cond_init (&souphttpsink->cond);
224 souphttpsink->location = NULL;
225 souphttpsink->automatic_redirect = TRUE;
226 souphttpsink->user_agent = g_strdup (DEFAULT_USER_AGENT);
227 souphttpsink->user_id = NULL;
228 souphttpsink->user_pw = NULL;
229 souphttpsink->proxy_id = NULL;
230 souphttpsink->proxy_pw = NULL;
231 souphttpsink->prop_session = NULL;
232 souphttpsink->timeout = 1;
233 souphttpsink->log_level = DEFAULT_SOUP_LOG_LEVEL;
234 proxy = g_getenv ("http_proxy");
235 if (proxy && !gst_soup_http_client_sink_set_proxy (souphttpsink, proxy)) {
236 GST_WARNING_OBJECT (souphttpsink,
237 "The proxy in the http_proxy env var (\"%s\") cannot be parsed.",
241 gst_soup_http_client_sink_reset (souphttpsink);
245 gst_soup_http_client_sink_reset (GstSoupHttpClientSink * souphttpsink)
247 g_list_free_full (souphttpsink->queued_buffers,
248 (GDestroyNotify) gst_buffer_unref);
249 souphttpsink->queued_buffers = NULL;
250 g_free (souphttpsink->reason_phrase);
251 souphttpsink->reason_phrase = NULL;
252 souphttpsink->status_code = 0;
253 souphttpsink->offset = 0;
258 gst_soup_http_client_sink_set_proxy (GstSoupHttpClientSink * souphttpsink,
261 if (souphttpsink->proxy) {
262 soup_uri_free (souphttpsink->proxy);
263 souphttpsink->proxy = NULL;
265 if (g_str_has_prefix (uri, "http://")) {
266 souphttpsink->proxy = soup_uri_new (uri);
268 gchar *new_uri = g_strconcat ("http://", uri, NULL);
270 souphttpsink->proxy = soup_uri_new (new_uri);
278 gst_soup_http_client_sink_set_property (GObject * object, guint property_id,
279 const GValue * value, GParamSpec * pspec)
281 GstSoupHttpClientSink *souphttpsink = GST_SOUP_HTTP_CLIENT_SINK (object);
283 g_mutex_lock (&souphttpsink->mutex);
284 switch (property_id) {
286 if (souphttpsink->prop_session) {
287 g_object_unref (souphttpsink->prop_session);
289 souphttpsink->prop_session = g_value_dup_object (value);
292 g_free (souphttpsink->location);
293 souphttpsink->location = g_value_dup_string (value);
294 souphttpsink->offset = 0;
296 case PROP_USER_AGENT:
297 g_free (souphttpsink->user_agent);
298 souphttpsink->user_agent = g_value_dup_string (value);
300 case PROP_AUTOMATIC_REDIRECT:
301 souphttpsink->automatic_redirect = g_value_get_boolean (value);
304 g_free (souphttpsink->user_id);
305 souphttpsink->user_id = g_value_dup_string (value);
308 g_free (souphttpsink->user_pw);
309 souphttpsink->user_pw = g_value_dup_string (value);
312 g_free (souphttpsink->proxy_id);
313 souphttpsink->proxy_id = g_value_dup_string (value);
316 g_free (souphttpsink->proxy_pw);
317 souphttpsink->proxy_pw = g_value_dup_string (value);
323 proxy = g_value_get_string (value);
326 GST_WARNING ("proxy property cannot be NULL");
329 if (!gst_soup_http_client_sink_set_proxy (souphttpsink, proxy)) {
330 GST_WARNING ("badly formatted proxy URI");
336 g_strfreev (souphttpsink->cookies);
337 souphttpsink->cookies = g_strdupv (g_value_get_boxed (value));
339 case PROP_SOUP_LOG_LEVEL:
340 souphttpsink->log_level = g_value_get_enum (value);
343 G_OBJECT_WARN_INVALID_PROPERTY_ID (object, property_id, pspec);
347 g_mutex_unlock (&souphttpsink->mutex);
351 gst_soup_http_client_sink_get_property (GObject * object, guint property_id,
352 GValue * value, GParamSpec * pspec)
354 GstSoupHttpClientSink *souphttpsink = GST_SOUP_HTTP_CLIENT_SINK (object);
356 switch (property_id) {
358 g_value_set_object (value, souphttpsink->prop_session);
361 g_value_set_string (value, souphttpsink->location);
363 case PROP_AUTOMATIC_REDIRECT:
364 g_value_set_boolean (value, souphttpsink->automatic_redirect);
366 case PROP_USER_AGENT:
367 g_value_set_string (value, souphttpsink->user_agent);
370 g_value_set_string (value, souphttpsink->user_id);
373 g_value_set_string (value, souphttpsink->user_pw);
376 g_value_set_string (value, souphttpsink->proxy_id);
379 g_value_set_string (value, souphttpsink->proxy_pw);
382 if (souphttpsink->proxy == NULL)
383 g_value_set_static_string (value, "");
385 char *proxy = soup_uri_to_string (souphttpsink->proxy, FALSE);
387 g_value_set_string (value, proxy);
392 g_value_set_boxed (value, g_strdupv (souphttpsink->cookies));
394 case PROP_SOUP_LOG_LEVEL:
395 g_value_set_enum (value, souphttpsink->log_level);
398 G_OBJECT_WARN_INVALID_PROPERTY_ID (object, property_id, pspec);
404 gst_soup_http_client_sink_dispose (GObject * object)
406 GstSoupHttpClientSink *souphttpsink = GST_SOUP_HTTP_CLIENT_SINK (object);
408 /* clean up as possible. may be called multiple times */
409 if (souphttpsink->prop_session)
410 g_object_unref (souphttpsink->prop_session);
411 souphttpsink->prop_session = NULL;
413 G_OBJECT_CLASS (parent_class)->dispose (object);
417 gst_soup_http_client_sink_finalize (GObject * object)
419 GstSoupHttpClientSink *souphttpsink = GST_SOUP_HTTP_CLIENT_SINK (object);
421 /* clean up object here */
423 g_free (souphttpsink->user_agent);
424 g_free (souphttpsink->user_id);
425 g_free (souphttpsink->user_pw);
426 g_free (souphttpsink->proxy_id);
427 g_free (souphttpsink->proxy_pw);
428 if (souphttpsink->proxy)
429 soup_uri_free (souphttpsink->proxy);
430 g_free (souphttpsink->location);
431 g_strfreev (souphttpsink->cookies);
433 g_cond_clear (&souphttpsink->cond);
434 g_mutex_clear (&souphttpsink->mutex);
436 G_OBJECT_CLASS (parent_class)->finalize (object);
442 gst_soup_http_client_sink_set_caps (GstBaseSink * sink, GstCaps * caps)
444 GstSoupHttpClientSink *souphttpsink = GST_SOUP_HTTP_CLIENT_SINK (sink);
445 GstStructure *structure;
446 const GValue *value_array;
449 GST_DEBUG_OBJECT (souphttpsink, "new stream headers set");
450 structure = gst_caps_get_structure (caps, 0);
451 value_array = gst_structure_get_value (structure, "streamheader");
453 free_buffer_list (souphttpsink->streamheader_buffers);
454 souphttpsink->streamheader_buffers = NULL;
456 n = gst_value_array_get_size (value_array);
457 for (i = 0; i < n; i++) {
460 value = gst_value_array_get_value (value_array, i);
461 buffer = GST_BUFFER (gst_value_get_buffer (value));
462 souphttpsink->streamheader_buffers =
463 g_list_append (souphttpsink->streamheader_buffers,
464 gst_buffer_ref (buffer));
472 gst_soup_http_client_sink_get_times (GstBaseSink * sink, GstBuffer * buffer,
473 GstClockTime * start, GstClockTime * end)
479 thread_ready_idle_cb (gpointer data)
481 GstSoupHttpClientSink *souphttpsink = GST_SOUP_HTTP_CLIENT_SINK (data);
483 GST_LOG_OBJECT (souphttpsink, "thread ready");
485 g_mutex_lock (&souphttpsink->mutex);
486 g_cond_signal (&souphttpsink->cond);
487 g_mutex_unlock (&souphttpsink->mutex);
489 return FALSE; /* only run once */
493 thread_func (gpointer ptr)
495 GstSoupHttpClientSink *souphttpsink = GST_SOUP_HTTP_CLIENT_SINK (ptr);
497 GST_DEBUG ("thread start");
499 g_main_loop_run (souphttpsink->loop);
501 GST_DEBUG ("thread quit");
507 gst_soup_http_client_sink_start (GstBaseSink * sink)
509 GstSoupHttpClientSink *souphttpsink = GST_SOUP_HTTP_CLIENT_SINK (sink);
511 if (souphttpsink->prop_session) {
512 souphttpsink->session = souphttpsink->prop_session;
515 GError *error = NULL;
517 souphttpsink->context = g_main_context_new ();
519 /* set up idle source to signal when the main loop is running and
520 * it's safe for ::stop() to call g_main_loop_quit() */
521 source = g_idle_source_new ();
522 g_source_set_callback (source, thread_ready_idle_cb, sink, NULL);
523 g_source_attach (source, souphttpsink->context);
524 g_source_unref (source);
526 souphttpsink->loop = g_main_loop_new (souphttpsink->context, TRUE);
528 g_mutex_lock (&souphttpsink->mutex);
530 /* FIXME: error handling */
531 souphttpsink->thread = g_thread_try_new ("souphttpclientsink-thread",
532 thread_func, souphttpsink, &error);
534 GST_LOG_OBJECT (souphttpsink, "waiting for main loop thread to start up");
535 g_cond_wait (&souphttpsink->cond, &souphttpsink->mutex);
536 g_mutex_unlock (&souphttpsink->mutex);
537 GST_LOG_OBJECT (souphttpsink, "main loop thread running");
539 if (souphttpsink->proxy == NULL) {
540 souphttpsink->session =
541 soup_session_async_new_with_options (SOUP_SESSION_ASYNC_CONTEXT,
542 souphttpsink->context, SOUP_SESSION_USER_AGENT,
543 souphttpsink->user_agent, SOUP_SESSION_TIMEOUT, souphttpsink->timeout,
546 souphttpsink->session =
547 soup_session_async_new_with_options (SOUP_SESSION_ASYNC_CONTEXT,
548 souphttpsink->context, SOUP_SESSION_USER_AGENT,
549 souphttpsink->user_agent, SOUP_SESSION_TIMEOUT, souphttpsink->timeout,
550 SOUP_SESSION_PROXY_URI, souphttpsink->proxy, NULL);
553 g_signal_connect (souphttpsink->session, "authenticate",
554 G_CALLBACK (authenticate), souphttpsink);
558 gst_soup_util_log_setup (souphttpsink->session, souphttpsink->log_level,
559 GST_ELEMENT (souphttpsink));
565 gst_soup_http_client_sink_stop (GstBaseSink * sink)
567 GstSoupHttpClientSink *souphttpsink = GST_SOUP_HTTP_CLIENT_SINK (sink);
571 if (souphttpsink->prop_session == NULL) {
572 soup_session_abort (souphttpsink->session);
573 g_object_unref (souphttpsink->session);
576 if (souphttpsink->loop) {
577 g_main_loop_quit (souphttpsink->loop);
578 g_thread_join (souphttpsink->thread);
579 g_main_loop_unref (souphttpsink->loop);
580 souphttpsink->loop = NULL;
582 if (souphttpsink->context) {
583 g_main_context_unref (souphttpsink->context);
584 souphttpsink->context = NULL;
587 gst_soup_http_client_sink_reset (souphttpsink);
593 gst_soup_http_client_sink_unlock (GstBaseSink * sink)
595 GST_DEBUG ("unlock");
601 gst_soup_http_client_sink_event (GstBaseSink * sink, GstEvent * event)
603 GstSoupHttpClientSink *souphttpsink = GST_SOUP_HTTP_CLIENT_SINK (sink);
605 GST_DEBUG_OBJECT (souphttpsink, "event");
607 if (GST_EVENT_TYPE (event) == GST_EVENT_EOS) {
608 GST_DEBUG_OBJECT (souphttpsink, "got eos");
609 g_mutex_lock (&souphttpsink->mutex);
610 while (souphttpsink->message) {
611 GST_DEBUG_OBJECT (souphttpsink, "waiting");
612 g_cond_wait (&souphttpsink->cond, &souphttpsink->mutex);
614 g_mutex_unlock (&souphttpsink->mutex);
615 GST_DEBUG_OBJECT (souphttpsink, "finished eos");
618 return GST_BASE_SINK_CLASS (parent_class)->event (sink, event);
622 gst_soup_http_client_sink_preroll (GstBaseSink * sink, GstBuffer * buffer)
624 GST_DEBUG ("preroll");
630 free_buffer_list (GList * list)
633 for (g = list; g; g = g_list_next (g)) {
634 GstBuffer *buffer = g->data;
635 gst_buffer_unref (buffer);
641 send_message_locked (GstSoupHttpClientSink * souphttpsink)
646 if (souphttpsink->queued_buffers == NULL || souphttpsink->message) {
650 /* If the URI went away, drop all these buffers */
651 if (souphttpsink->location == NULL) {
652 GST_DEBUG_OBJECT (souphttpsink, "URI went away, dropping queued buffers");
653 free_buffer_list (souphttpsink->queued_buffers);
654 souphttpsink->queued_buffers = NULL;
658 souphttpsink->message = soup_message_new ("PUT", souphttpsink->location);
659 soup_message_set_flags (souphttpsink->message,
660 (souphttpsink->automatic_redirect ? 0 : SOUP_MESSAGE_NO_REDIRECT));
662 if (souphttpsink->cookies) {
665 for (cookie = souphttpsink->cookies; *cookie != NULL; cookie++) {
666 soup_message_headers_append (souphttpsink->message->request_headers,
672 if (souphttpsink->offset == 0) {
673 for (g = souphttpsink->streamheader_buffers; g; g = g_list_next (g)) {
674 GstBuffer *buffer = g->data;
677 GST_DEBUG_OBJECT (souphttpsink, "queueing stream headers");
678 gst_buffer_map (buffer, &map, GST_MAP_READ);
679 /* Stream headers are updated whenever ::set_caps is called, so there's
680 * no guarantees about their lifetime and we ask libsoup to copy them
681 * into the message body with SOUP_MEMORY_COPY. */
682 soup_message_body_append (souphttpsink->message->request_body,
683 SOUP_MEMORY_COPY, map.data, map.size);
685 gst_buffer_unmap (buffer, &map);
689 for (g = souphttpsink->queued_buffers; g; g = g_list_next (g)) {
690 GstBuffer *buffer = g->data;
691 if (!GST_BUFFER_FLAG_IS_SET (buffer, GST_BUFFER_FLAG_HEADER)) {
694 gst_buffer_map (buffer, &map, GST_MAP_READ);
695 /* Queued buffers are only freed in the next iteration of the mainloop
696 * after the message body has been written out, so we don't need libsoup
697 * to copy those while appending to the body. However, if the buffer is
698 * used elsewhere, it should be copied. Hence, SOUP_MEMORY_TEMPORARY. */
699 soup_message_body_append (souphttpsink->message->request_body,
700 SOUP_MEMORY_TEMPORARY, map.data, map.size);
702 gst_buffer_unmap (buffer, &map);
706 if (souphttpsink->offset != 0) {
708 s = g_strdup_printf ("bytes %" G_GUINT64_FORMAT "-%" G_GUINT64_FORMAT "/*",
709 souphttpsink->offset, souphttpsink->offset + n - 1);
710 soup_message_headers_append (souphttpsink->message->request_headers,
716 GST_DEBUG_OBJECT (souphttpsink,
717 "total size of buffers queued is 0, freeing everything");
718 free_buffer_list (souphttpsink->queued_buffers);
719 souphttpsink->queued_buffers = NULL;
720 g_object_unref (souphttpsink->message);
721 souphttpsink->message = NULL;
725 souphttpsink->sent_buffers = souphttpsink->queued_buffers;
726 souphttpsink->queued_buffers = NULL;
728 GST_DEBUG_OBJECT (souphttpsink,
729 "queue message %" G_GUINT64_FORMAT " %" G_GUINT64_FORMAT,
730 souphttpsink->offset, n);
731 soup_session_queue_message (souphttpsink->session, souphttpsink->message,
732 callback, souphttpsink);
734 souphttpsink->offset += n;
738 send_message (GstSoupHttpClientSink * souphttpsink)
740 g_mutex_lock (&souphttpsink->mutex);
741 send_message_locked (souphttpsink);
742 g_mutex_unlock (&souphttpsink->mutex);
748 callback (SoupSession * session, SoupMessage * msg, gpointer user_data)
750 GstSoupHttpClientSink *souphttpsink = GST_SOUP_HTTP_CLIENT_SINK (user_data);
752 GST_DEBUG_OBJECT (souphttpsink, "callback status=%d %s",
753 msg->status_code, msg->reason_phrase);
755 g_mutex_lock (&souphttpsink->mutex);
756 g_cond_signal (&souphttpsink->cond);
757 souphttpsink->message = NULL;
759 if (!SOUP_STATUS_IS_SUCCESSFUL (msg->status_code)) {
760 souphttpsink->status_code = msg->status_code;
761 souphttpsink->reason_phrase = g_strdup (msg->reason_phrase);
762 g_mutex_unlock (&souphttpsink->mutex);
766 free_buffer_list (souphttpsink->sent_buffers);
767 souphttpsink->sent_buffers = NULL;
769 send_message_locked (souphttpsink);
770 g_mutex_unlock (&souphttpsink->mutex);
774 gst_soup_http_client_sink_render (GstBaseSink * sink, GstBuffer * buffer)
776 GstSoupHttpClientSink *souphttpsink = GST_SOUP_HTTP_CLIENT_SINK (sink);
780 if (souphttpsink->status_code != 0) {
781 /* FIXME we should allow a moderate amount of retries. */
782 GST_ELEMENT_ERROR (souphttpsink, RESOURCE, WRITE,
783 ("Could not write to HTTP URI"),
784 ("error: %d %s", souphttpsink->status_code,
785 souphttpsink->reason_phrase));
786 return GST_FLOW_ERROR;
789 g_mutex_lock (&souphttpsink->mutex);
790 if (souphttpsink->location != NULL) {
791 wake = (souphttpsink->queued_buffers == NULL);
792 souphttpsink->queued_buffers =
793 g_list_append (souphttpsink->queued_buffers, gst_buffer_ref (buffer));
796 GST_DEBUG_OBJECT (souphttpsink, "setting callback for new buffers");
797 source = g_idle_source_new ();
798 g_source_set_callback (source, (GSourceFunc) (send_message),
800 g_source_attach (source, souphttpsink->context);
801 g_source_unref (source);
804 g_mutex_unlock (&souphttpsink->mutex);
810 authenticate (SoupSession * session, SoupMessage * msg,
811 SoupAuth * auth, gboolean retrying, gpointer user_data)
813 GstSoupHttpClientSink *souphttpsink = GST_SOUP_HTTP_CLIENT_SINK (user_data);
816 /* First time authentication only, if we fail and are called again with retry true fall through */
817 if (msg->status_code == SOUP_STATUS_UNAUTHORIZED) {
818 if (souphttpsink->user_id && souphttpsink->user_pw)
819 soup_auth_authenticate (auth, souphttpsink->user_id,
820 souphttpsink->user_pw);
821 } else if (msg->status_code == SOUP_STATUS_PROXY_AUTHENTICATION_REQUIRED) {
822 if (souphttpsink->proxy_id && souphttpsink->proxy_pw)
823 soup_auth_authenticate (auth, souphttpsink->proxy_id,
824 souphttpsink->proxy_pw);