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 -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"
45 GST_DEBUG_CATEGORY_STATIC (souphttpclientsink_dbg);
46 #define GST_CAT_DEFAULT souphttpclientsink_dbg
51 static void gst_soup_http_client_sink_set_property (GObject * object,
52 guint property_id, const GValue * value, GParamSpec * pspec);
53 static void gst_soup_http_client_sink_get_property (GObject * object,
54 guint property_id, GValue * value, GParamSpec * pspec);
55 static void gst_soup_http_client_sink_dispose (GObject * object);
56 static void gst_soup_http_client_sink_finalize (GObject * object);
58 static gboolean gst_soup_http_client_sink_set_caps (GstBaseSink * sink,
60 static void gst_soup_http_client_sink_get_times (GstBaseSink * sink,
61 GstBuffer * buffer, GstClockTime * start, GstClockTime * end);
62 static gboolean gst_soup_http_client_sink_start (GstBaseSink * sink);
63 static gboolean gst_soup_http_client_sink_stop (GstBaseSink * sink);
64 static gboolean gst_soup_http_client_sink_unlock (GstBaseSink * sink);
65 static gboolean gst_soup_http_client_sink_event (GstBaseSink * sink,
67 static GstFlowReturn gst_soup_http_client_sink_preroll (GstBaseSink * sink,
69 static GstFlowReturn gst_soup_http_client_sink_render (GstBaseSink * sink,
72 static void free_buffer_list (GList * list);
73 static void gst_soup_http_client_sink_reset (GstSoupHttpClientSink *
75 static void authenticate (SoupSession * session, SoupMessage * msg,
76 SoupAuth * auth, gboolean retrying, gpointer user_data);
77 static void callback (SoupSession * session, SoupMessage * msg,
79 static gboolean gst_soup_http_client_sink_set_proxy (GstSoupHttpClientSink *
80 souphttpsink, const gchar * uri);
87 PROP_AUTOMATIC_REDIRECT,
97 #define DEFAULT_USER_AGENT "GStreamer souphttpclientsink "
101 static GstStaticPadTemplate gst_soup_http_client_sink_sink_template =
102 GST_STATIC_PAD_TEMPLATE ("sink",
105 GST_STATIC_CAPS_ANY);
108 /* class initialization */
110 #define gst_soup_http_client_sink_parent_class parent_class
111 G_DEFINE_TYPE (GstSoupHttpClientSink, gst_soup_http_client_sink,
115 gst_soup_http_client_sink_class_init (GstSoupHttpClientSinkClass * klass)
117 GObjectClass *gobject_class = G_OBJECT_CLASS (klass);
118 GstElementClass *gstelement_class = GST_ELEMENT_CLASS (klass);
119 GstBaseSinkClass *base_sink_class = GST_BASE_SINK_CLASS (klass);
121 gobject_class->set_property = gst_soup_http_client_sink_set_property;
122 gobject_class->get_property = gst_soup_http_client_sink_get_property;
123 gobject_class->dispose = gst_soup_http_client_sink_dispose;
124 gobject_class->finalize = gst_soup_http_client_sink_finalize;
126 g_object_class_install_property (gobject_class,
128 g_param_spec_string ("location", "Location",
129 "URI to send to", "", G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
130 g_object_class_install_property (gobject_class,
132 g_param_spec_string ("user-agent", "User-Agent",
133 "Value of the User-Agent HTTP request header field",
134 DEFAULT_USER_AGENT, G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
135 g_object_class_install_property (gobject_class,
136 PROP_AUTOMATIC_REDIRECT,
137 g_param_spec_boolean ("automatic-redirect", "automatic-redirect",
138 "Automatically follow HTTP redirects (HTTP Status Code 3xx)",
139 TRUE, G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
140 g_object_class_install_property (gobject_class,
142 g_param_spec_string ("proxy", "Proxy",
143 "HTTP proxy server URI", "",
144 G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
145 g_object_class_install_property (gobject_class,
147 g_param_spec_string ("user-id", "user-id",
148 "user id for authentication", "",
149 G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
150 g_object_class_install_property (gobject_class, PROP_USER_PW,
151 g_param_spec_string ("user-pw", "user-pw",
152 "user password for authentication", "",
153 G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
154 g_object_class_install_property (gobject_class, PROP_PROXY_ID,
155 g_param_spec_string ("proxy-id", "proxy-id",
156 "user id for proxy authentication", "",
157 G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
158 g_object_class_install_property (gobject_class, PROP_PROXY_PW,
159 g_param_spec_string ("proxy-pw", "proxy-pw",
160 "user password for proxy authentication", "",
161 G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
162 g_object_class_install_property (gobject_class, PROP_SESSION,
163 g_param_spec_object ("session", "session",
164 "SoupSession object to use for communication",
165 SOUP_TYPE_SESSION, G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
166 g_object_class_install_property (gobject_class, PROP_COOKIES,
167 g_param_spec_boxed ("cookies", "Cookies", "HTTP request cookies",
168 G_TYPE_STRV, G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
170 gst_element_class_add_pad_template (gstelement_class,
171 gst_static_pad_template_get (&gst_soup_http_client_sink_sink_template));
173 gst_element_class_set_details_simple (gstelement_class, "HTTP client sink",
174 "Generic", "Sends streams to HTTP server via PUT",
175 "David Schleef <ds@entropywave.com>");
177 base_sink_class->set_caps =
178 GST_DEBUG_FUNCPTR (gst_soup_http_client_sink_set_caps);
180 base_sink_class->get_times =
181 GST_DEBUG_FUNCPTR (gst_soup_http_client_sink_get_times);
182 base_sink_class->start = GST_DEBUG_FUNCPTR (gst_soup_http_client_sink_start);
183 base_sink_class->stop = GST_DEBUG_FUNCPTR (gst_soup_http_client_sink_stop);
184 base_sink_class->unlock =
185 GST_DEBUG_FUNCPTR (gst_soup_http_client_sink_unlock);
186 base_sink_class->event = GST_DEBUG_FUNCPTR (gst_soup_http_client_sink_event);
188 base_sink_class->preroll =
189 GST_DEBUG_FUNCPTR (gst_soup_http_client_sink_preroll);
190 base_sink_class->render =
191 GST_DEBUG_FUNCPTR (gst_soup_http_client_sink_render);
193 GST_DEBUG_CATEGORY_INIT (souphttpclientsink_dbg, "souphttpclientsink", 0,
194 "souphttpclientsink element");
199 gst_soup_http_client_sink_init (GstSoupHttpClientSink * souphttpsink)
203 souphttpsink->mutex = g_mutex_new ();
204 souphttpsink->cond = g_cond_new ();
206 souphttpsink->location = NULL;
207 souphttpsink->automatic_redirect = TRUE;
208 souphttpsink->user_agent = g_strdup (DEFAULT_USER_AGENT);
209 souphttpsink->user_id = NULL;
210 souphttpsink->user_pw = NULL;
211 souphttpsink->proxy_id = NULL;
212 souphttpsink->proxy_pw = NULL;
213 souphttpsink->prop_session = NULL;
214 souphttpsink->timeout = 1;
215 proxy = g_getenv ("http_proxy");
216 if (proxy && !gst_soup_http_client_sink_set_proxy (souphttpsink, proxy)) {
217 GST_WARNING_OBJECT (souphttpsink,
218 "The proxy in the http_proxy env var (\"%s\") cannot be parsed.",
222 gst_soup_http_client_sink_reset (souphttpsink);
226 gst_soup_http_client_sink_reset (GstSoupHttpClientSink * souphttpsink)
228 g_free (souphttpsink->reason_phrase);
229 souphttpsink->reason_phrase = NULL;
230 souphttpsink->status_code = 0;
231 souphttpsink->offset = 0;
236 gst_soup_http_client_sink_set_proxy (GstSoupHttpClientSink * souphttpsink,
239 if (souphttpsink->proxy) {
240 soup_uri_free (souphttpsink->proxy);
241 souphttpsink->proxy = NULL;
243 if (g_str_has_prefix (uri, "http://")) {
244 souphttpsink->proxy = soup_uri_new (uri);
246 gchar *new_uri = g_strconcat ("http://", uri, NULL);
248 souphttpsink->proxy = soup_uri_new (new_uri);
256 gst_soup_http_client_sink_set_property (GObject * object, guint property_id,
257 const GValue * value, GParamSpec * pspec)
259 GstSoupHttpClientSink *souphttpsink = GST_SOUP_HTTP_CLIENT_SINK (object);
261 g_mutex_lock (souphttpsink->mutex);
262 switch (property_id) {
264 if (souphttpsink->prop_session) {
265 g_object_unref (souphttpsink->prop_session);
267 souphttpsink->prop_session = g_value_dup_object (value);
270 g_free (souphttpsink->location);
271 souphttpsink->location = g_value_dup_string (value);
272 souphttpsink->offset = 0;
274 case PROP_USER_AGENT:
275 g_free (souphttpsink->user_agent);
276 souphttpsink->user_agent = g_value_dup_string (value);
278 case PROP_AUTOMATIC_REDIRECT:
279 souphttpsink->automatic_redirect = g_value_get_boolean (value);
282 g_free (souphttpsink->user_id);
283 souphttpsink->user_id = g_value_dup_string (value);
286 g_free (souphttpsink->user_pw);
287 souphttpsink->user_pw = g_value_dup_string (value);
290 g_free (souphttpsink->proxy_id);
291 souphttpsink->proxy_id = g_value_dup_string (value);
294 g_free (souphttpsink->proxy_pw);
295 souphttpsink->proxy_pw = g_value_dup_string (value);
301 proxy = g_value_get_string (value);
304 GST_WARNING ("proxy property cannot be NULL");
307 if (!gst_soup_http_client_sink_set_proxy (souphttpsink, proxy)) {
308 GST_WARNING ("badly formatted proxy URI");
314 g_strfreev (souphttpsink->cookies);
315 souphttpsink->cookies = g_strdupv (g_value_get_boxed (value));
318 G_OBJECT_WARN_INVALID_PROPERTY_ID (object, property_id, pspec);
322 g_mutex_unlock (souphttpsink->mutex);
326 gst_soup_http_client_sink_get_property (GObject * object, guint property_id,
327 GValue * value, GParamSpec * pspec)
329 GstSoupHttpClientSink *souphttpsink = GST_SOUP_HTTP_CLIENT_SINK (object);
331 switch (property_id) {
333 g_value_set_object (value, souphttpsink->prop_session);
336 g_value_set_string (value, souphttpsink->location);
338 case PROP_AUTOMATIC_REDIRECT:
339 g_value_set_boolean (value, souphttpsink->automatic_redirect);
341 case PROP_USER_AGENT:
342 g_value_set_string (value, souphttpsink->user_agent);
345 g_value_set_string (value, souphttpsink->user_id);
348 g_value_set_string (value, souphttpsink->user_pw);
351 g_value_set_string (value, souphttpsink->proxy_id);
354 g_value_set_string (value, souphttpsink->proxy_pw);
357 if (souphttpsink->proxy == NULL)
358 g_value_set_static_string (value, "");
360 char *proxy = soup_uri_to_string (souphttpsink->proxy, FALSE);
362 g_value_set_string (value, proxy);
367 g_value_set_boxed (value, g_strdupv (souphttpsink->cookies));
370 G_OBJECT_WARN_INVALID_PROPERTY_ID (object, property_id, pspec);
376 gst_soup_http_client_sink_dispose (GObject * object)
378 GstSoupHttpClientSink *souphttpsink = GST_SOUP_HTTP_CLIENT_SINK (object);
380 /* clean up as possible. may be called multiple times */
381 if (souphttpsink->prop_session)
382 g_object_unref (souphttpsink->prop_session);
383 souphttpsink->prop_session = NULL;
385 G_OBJECT_CLASS (parent_class)->dispose (object);
389 gst_soup_http_client_sink_finalize (GObject * object)
391 GstSoupHttpClientSink *souphttpsink = GST_SOUP_HTTP_CLIENT_SINK (object);
393 /* clean up object here */
395 g_free (souphttpsink->user_agent);
396 g_free (souphttpsink->user_id);
397 g_free (souphttpsink->user_pw);
398 g_free (souphttpsink->proxy_id);
399 g_free (souphttpsink->proxy_pw);
400 if (souphttpsink->proxy)
401 soup_uri_free (souphttpsink->proxy);
402 g_free (souphttpsink->location);
404 g_cond_free (souphttpsink->cond);
405 g_mutex_free (souphttpsink->mutex);
407 G_OBJECT_CLASS (parent_class)->finalize (object);
413 gst_soup_http_client_sink_set_caps (GstBaseSink * sink, GstCaps * caps)
415 GstSoupHttpClientSink *souphttpsink = GST_SOUP_HTTP_CLIENT_SINK (sink);
416 GstStructure *structure;
417 const GValue *value_array;
420 structure = gst_caps_get_structure (caps, 0);
421 value_array = gst_structure_get_value (structure, "streamheader");
423 free_buffer_list (souphttpsink->streamheader_buffers);
424 souphttpsink->streamheader_buffers = NULL;
426 n = gst_value_array_get_size (value_array);
427 for (i = 0; i < n; i++) {
430 value = gst_value_array_get_value (value_array, i);
431 buffer = GST_BUFFER (gst_value_get_buffer (value));
432 souphttpsink->streamheader_buffers =
433 g_list_append (souphttpsink->streamheader_buffers,
434 gst_buffer_ref (buffer));
442 gst_soup_http_client_sink_get_times (GstBaseSink * sink, GstBuffer * buffer,
443 GstClockTime * start, GstClockTime * end)
449 thread_func (gpointer ptr)
451 GstSoupHttpClientSink *souphttpsink = GST_SOUP_HTTP_CLIENT_SINK (ptr);
453 GST_DEBUG ("thread start");
455 souphttpsink->loop = g_main_loop_new (souphttpsink->context, TRUE);
456 g_main_loop_run (souphttpsink->loop);
458 GST_DEBUG ("thread quit");
464 gst_soup_http_client_sink_start (GstBaseSink * sink)
466 GstSoupHttpClientSink *souphttpsink = GST_SOUP_HTTP_CLIENT_SINK (sink);
468 if (souphttpsink->prop_session) {
469 souphttpsink->session = souphttpsink->prop_session;
471 GError *error = NULL;
473 souphttpsink->context = g_main_context_new ();
475 souphttpsink->thread = g_thread_create (thread_func, souphttpsink,
478 souphttpsink->session =
479 soup_session_async_new_with_options (SOUP_SESSION_ASYNC_CONTEXT,
480 souphttpsink->context, SOUP_SESSION_USER_AGENT,
481 souphttpsink->user_agent, SOUP_SESSION_TIMEOUT, souphttpsink->timeout,
484 //soup_session_add_feature (souphttpsink->session,
485 // SOUP_SESSION_FEATURE (soup_logger_new (SOUP_LOGGER_LOG_BODY, 100)));
487 g_signal_connect (souphttpsink->session, "authenticate",
488 G_CALLBACK (authenticate), souphttpsink);
495 gst_soup_http_client_sink_stop (GstBaseSink * sink)
497 GstSoupHttpClientSink *souphttpsink = GST_SOUP_HTTP_CLIENT_SINK (sink);
501 if (souphttpsink->prop_session == NULL) {
502 soup_session_abort (souphttpsink->session);
503 g_object_unref (souphttpsink->session);
506 if (souphttpsink->loop) {
507 g_main_loop_quit (souphttpsink->loop);
508 g_thread_join (souphttpsink->thread);
509 g_main_loop_unref (souphttpsink->loop);
510 souphttpsink->loop = NULL;
512 if (souphttpsink->context) {
513 g_main_context_unref (souphttpsink->context);
514 souphttpsink->context = NULL;
517 gst_soup_http_client_sink_reset (souphttpsink);
523 gst_soup_http_client_sink_unlock (GstBaseSink * sink)
525 GST_DEBUG ("unlock");
531 gst_soup_http_client_sink_event (GstBaseSink * sink, GstEvent * event)
533 GstSoupHttpClientSink *souphttpsink = GST_SOUP_HTTP_CLIENT_SINK (sink);
535 GST_DEBUG_OBJECT (souphttpsink, "event");
537 if (GST_EVENT_TYPE (event) == GST_EVENT_EOS) {
538 GST_DEBUG_OBJECT (souphttpsink, "got eos");
539 g_mutex_lock (souphttpsink->mutex);
540 while (souphttpsink->message) {
541 GST_DEBUG_OBJECT (souphttpsink, "waiting");
542 g_cond_wait (souphttpsink->cond, souphttpsink->mutex);
544 g_mutex_unlock (souphttpsink->mutex);
545 GST_DEBUG_OBJECT (souphttpsink, "finished eos");
548 return GST_BASE_SINK_CLASS (parent_class)->event (sink, event);
552 gst_soup_http_client_sink_preroll (GstBaseSink * sink, GstBuffer * buffer)
554 GST_DEBUG ("preroll");
560 free_buffer_list (GList * list)
563 for (g = list; g; g = g_list_next (g)) {
564 GstBuffer *buffer = g->data;
565 gst_buffer_unref (buffer);
571 send_message_locked (GstSoupHttpClientSink * souphttpsink)
576 if (souphttpsink->queued_buffers == NULL || souphttpsink->message) {
580 /* If the URI went away, drop all these buffers */
581 if (souphttpsink->location == NULL) {
582 free_buffer_list (souphttpsink->queued_buffers);
583 souphttpsink->queued_buffers = NULL;
587 souphttpsink->message = soup_message_new ("PUT", souphttpsink->location);
590 if (souphttpsink->offset == 0) {
591 for (g = souphttpsink->streamheader_buffers; g; g = g_list_next (g)) {
592 GstBuffer *buffer = g->data;
596 /* FIXME, lifetime of the buffer? */
597 data = gst_buffer_map (buffer, &size, NULL, GST_MAP_READ);
598 soup_message_body_append (souphttpsink->message->request_body,
599 SOUP_MEMORY_STATIC, data, size);
601 gst_buffer_unmap (buffer, data, size);
605 for (g = souphttpsink->queued_buffers; g; g = g_list_next (g)) {
606 GstBuffer *buffer = g->data;
607 if (!GST_BUFFER_FLAG_IS_SET (buffer, GST_BUFFER_FLAG_IN_CAPS)) {
611 /* FIXME, lifetime of the buffer? */
612 data = gst_buffer_map (buffer, &size, NULL, GST_MAP_READ);
613 soup_message_body_append (souphttpsink->message->request_body,
614 SOUP_MEMORY_STATIC, data, size);
616 gst_buffer_unmap (buffer, data, size);
620 if (souphttpsink->offset != 0) {
622 s = g_strdup_printf ("bytes %" G_GUINT64_FORMAT "-%" G_GUINT64_FORMAT "/*",
623 souphttpsink->offset, souphttpsink->offset + n - 1);
624 soup_message_headers_append (souphttpsink->message->request_headers,
630 free_buffer_list (souphttpsink->queued_buffers);
631 souphttpsink->queued_buffers = NULL;
632 g_object_unref (souphttpsink->message);
633 souphttpsink->message = NULL;
637 souphttpsink->sent_buffers = souphttpsink->queued_buffers;
638 souphttpsink->queued_buffers = NULL;
640 GST_DEBUG_OBJECT (souphttpsink,
641 "queue message %" G_GUINT64_FORMAT " %" G_GUINT64_FORMAT,
642 souphttpsink->offset, n);
643 soup_session_queue_message (souphttpsink->session, souphttpsink->message,
644 callback, souphttpsink);
646 souphttpsink->offset += n;
650 send_message (GstSoupHttpClientSink * souphttpsink)
652 g_mutex_lock (souphttpsink->mutex);
653 send_message_locked (souphttpsink);
654 g_mutex_unlock (souphttpsink->mutex);
660 callback (SoupSession * session, SoupMessage * msg, gpointer user_data)
662 GstSoupHttpClientSink *souphttpsink = GST_SOUP_HTTP_CLIENT_SINK (user_data);
664 GST_DEBUG_OBJECT (souphttpsink, "callback status=%d %s",
665 msg->status_code, msg->reason_phrase);
667 g_mutex_lock (souphttpsink->mutex);
668 g_cond_signal (souphttpsink->cond);
669 souphttpsink->message = NULL;
671 if (!SOUP_STATUS_IS_SUCCESSFUL (msg->status_code)) {
672 souphttpsink->status_code = msg->status_code;
673 souphttpsink->reason_phrase = g_strdup (msg->reason_phrase);
674 g_mutex_unlock (souphttpsink->mutex);
678 free_buffer_list (souphttpsink->sent_buffers);
679 souphttpsink->sent_buffers = NULL;
681 send_message_locked (souphttpsink);
682 g_mutex_unlock (souphttpsink->mutex);
686 gst_soup_http_client_sink_render (GstBaseSink * sink, GstBuffer * buffer)
688 GstSoupHttpClientSink *souphttpsink = GST_SOUP_HTTP_CLIENT_SINK (sink);
692 if (souphttpsink->status_code != 0) {
693 /* FIXME we should allow a moderate amount of retries. */
694 GST_ELEMENT_ERROR (souphttpsink, RESOURCE, WRITE,
695 ("Could not write to HTTP URI"),
696 ("error: %d %s", souphttpsink->status_code,
697 souphttpsink->reason_phrase));
698 return GST_FLOW_ERROR;
701 g_mutex_lock (souphttpsink->mutex);
702 if (souphttpsink->location != NULL) {
703 wake = (souphttpsink->queued_buffers == NULL);
704 souphttpsink->queued_buffers =
705 g_list_append (souphttpsink->queued_buffers, gst_buffer_ref (buffer));
708 source = g_idle_source_new ();
709 g_source_set_callback (source, (GSourceFunc) (send_message),
711 g_source_attach (source, souphttpsink->context);
712 g_source_unref (source);
715 g_mutex_unlock (souphttpsink->mutex);
721 authenticate (SoupSession * session, SoupMessage * msg,
722 SoupAuth * auth, gboolean retrying, gpointer user_data)
724 GstSoupHttpClientSink *souphttpsink = GST_SOUP_HTTP_CLIENT_SINK (user_data);
727 if (souphttpsink->user_id && souphttpsink->user_pw) {
728 soup_auth_authenticate (auth,
729 souphttpsink->user_id, souphttpsink->user_pw);