| /* |
| * GstCurlHttpSrc |
| * Copyright 2017 British Broadcasting Corporation - Research and Development |
| * |
| * Author: Sam Hurst <samuelh@rd.bbc.co.uk> |
| * |
| * Permission is hereby granted, free of charge, to any person obtaining a |
| * copy of this software and associated documentation files (the "Software"), |
| * to deal in the Software without restriction, including without limitation |
| * the rights to use, copy, modify, merge, publish, distribute, sublicense, |
| * and/or sell copies of the Software, and to permit persons to whom the |
| * Software is furnished to do so, subject to the following conditions: |
| * |
| * The above copyright notice and this permission notice shall be included in |
| * all copies or substantial portions of the Software. |
| * |
| * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR |
| * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, |
| * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE |
| * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER |
| * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING |
| * FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER |
| * DEALINGS IN THE SOFTWARE. |
| * |
| * Alternatively, the contents of this file may be used under the |
| * GNU Lesser General Public License Version 2.1 (the "LGPL"), in |
| * which case the following provisions apply instead of the ones |
| * mentioned above: |
| * |
| * This library is free software; you can redistribute it and/or |
| * modify it under the terms of the GNU Library General Public |
| * License as published by the Free Software Foundation; either |
| * version 2 of the License, or (at your option) any later version. |
| * |
| * This library is distributed in the hope that it will be useful, |
| * but WITHOUT ANY WARRANTY; without even the implied warranty of |
| * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU |
| * Library General Public License for more details. |
| * |
| * You should have received a copy of the GNU Library General Public |
| * License along with this library; if not, write to the |
| * Free Software Foundation, Inc., 59 Temple Place - Suite 330, |
| * Boston, MA 02111-1307, USA. |
| */ |
| |
| /** |
| * SECTION:element-curlhttpsrc |
| * |
| * This plugin reads data from a remote location specified by a URI, when the |
| * protocol is 'http' or 'https'. |
| * |
| * It is based on the cURL project (http://curl.haxx.se/) and is specifically |
| * designed to be also used with nghttp2 (http://nghttp2.org) to enable HTTP/2 |
| * support for GStreamer. Your libcurl library MUST be compiled against nghttp2 |
| * for HTTP/2 support for this functionality. HTTPS support is dependent on |
| * cURL being built with SSL support (OpenSSL/PolarSSL/NSS/GnuTLS). |
| * |
| * An HTTP proxy must be specified by URL. |
| * If the "http_proxy" environment variable is set, its value is used. |
| * The #GstCurlHttpSrc:proxy property can be used to override the default. |
| * |
| * <refsect2> |
| * <title>Example launch line</title> |
| * |[ |
| * gst-launch-1.0 curlhttpsrc location=http://127.0.1.1/index.html ! fakesink dump=1 |
| * ]| The above pipeline reads a web page from the local machine using HTTP and |
| * dumps it to stdout. |
| * |[ |
| * gst-launch-1.0 playbin uri=http://rdmedia.bbc.co.uk/dash/testmpds/multiperiod/bbb.php |
| * ]| The above pipeline will start up a DASH streaming session from the given |
| * MPD file. This requires GStreamer to have been built with dashdemux from |
| * gst-plugins-bad. |
| * </refsect2> |
| */ |
| |
| #ifdef HAVE_CONFIG_H |
| #include <config.h> |
| #endif |
| |
| #include <gst/gst-i18n-plugin.h> |
| |
| #include "gstcurlhttpsrc.h" |
| #include "gstcurlqueue.h" |
| #include "gstcurldefaults.h" |
| |
| GST_DEBUG_CATEGORY_STATIC (gst_curl_http_src_debug); |
| #define GST_CAT_DEFAULT gst_curl_http_src_debug |
| GST_DEBUG_CATEGORY_STATIC (gst_curl_loop_debug); |
| |
| /* |
| * Make a source pad template to be able to kick out recv'd data |
| */ |
| static GstStaticPadTemplate srcpadtemplate = GST_STATIC_PAD_TEMPLATE ("src", |
| GST_PAD_SRC, |
| GST_PAD_ALWAYS, |
| GST_STATIC_CAPS_ANY); |
| |
| /* |
| * Function Definitions |
| */ |
| /* Gstreamer generic element functions */ |
| static void gst_curl_http_src_set_property (GObject * object, guint prop_id, |
| const GValue * value, GParamSpec * pspec); |
| static void gst_curl_http_src_get_property (GObject * object, guint prop_id, |
| GValue * value, GParamSpec * pspec); |
| static void gst_curl_http_src_ref_multi (GstCurlHttpSrc * src); |
| static void gst_curl_http_src_unref_multi (GstCurlHttpSrc * src); |
| static void gst_curl_http_src_finalize (GObject * obj); |
| static GstFlowReturn gst_curl_http_src_create (GstPushSrc * psrc, |
| GstBuffer ** outbuf); |
| static GstFlowReturn gst_curl_http_src_handle_response (GstCurlHttpSrc * src); |
| static gboolean gst_curl_http_src_negotiate_caps (GstCurlHttpSrc * src); |
| static GstStateChangeReturn gst_curl_http_src_change_state (GstElement * |
| element, GstStateChange transition); |
| static void gst_curl_http_src_cleanup_instance (GstCurlHttpSrc * src); |
| static gboolean gst_curl_http_src_query (GstBaseSrc * bsrc, GstQuery * query); |
| static gboolean gst_curl_http_src_get_content_length (GstBaseSrc * bsrc, |
| guint64 * size); |
| static gboolean gst_curl_http_src_unlock (GstBaseSrc * bsrc); |
| static gboolean gst_curl_http_src_unlock_stop (GstBaseSrc * bsrc); |
| |
| /* URI Handler functions */ |
| static void gst_curl_http_src_uri_handler_init (gpointer g_iface, |
| gpointer iface_data); |
| static guint gst_curl_http_src_urihandler_get_type (GType type); |
| static const gchar *const *gst_curl_http_src_urihandler_get_protocols (GType |
| type); |
| static gchar *gst_curl_http_src_urihandler_get_uri (GstURIHandler * handler); |
| static gboolean gst_curl_http_src_urihandler_set_uri (GstURIHandler * handler, |
| const gchar * uri, GError ** error); |
| |
| /* GstTask functions */ |
| static void gst_curl_http_src_curl_multi_loop (gpointer thread_data); |
| static CURL *gst_curl_http_src_create_easy_handle (GstCurlHttpSrc * s); |
| static inline void gst_curl_http_src_destroy_easy_handle (GstCurlHttpSrc * src); |
| static size_t gst_curl_http_src_get_header (void *header, size_t size, |
| size_t nmemb, void *src); |
| static size_t gst_curl_http_src_get_chunks (void *chunk, size_t size, |
| size_t nmemb, void *src); |
| static void gst_curl_http_src_request_remove (GstCurlHttpSrc * src); |
| static char *gst_curl_http_src_strcasestr (const char *haystack, |
| const char *needle); |
| |
| curl_version_info_data *gst_curl_http_src_curl_capabilities; |
| GstCurlHttpVersion pref_http_ver; |
| gchar *gst_curl_http_src_default_useragent; |
| |
| #define GST_TYPE_CURL_HTTP_VERSION (gst_curl_http_version_get_type ()) |
| static GType |
| gst_curl_http_version_get_type (void) |
| { |
| static GType gtype = 0; |
| |
| if (!gtype) { |
| static const GEnumValue http_versions[] = { |
| {GSTCURL_HTTP_VERSION_1_0, "HTTP Version 1.0", "1.0"}, |
| {GSTCURL_HTTP_VERSION_1_1, "HTTP Version 1.1", "1.1"}, |
| #ifdef CURL_VERSION_HTTP2 |
| {GSTCURL_HTTP_VERSION_2_0, "HTTP Version 2.0", "2.0"}, |
| #endif |
| {0, NULL, NULL} |
| }; |
| gtype = g_enum_register_static ("GstCurlHttpVersionType", http_versions); |
| } |
| return gtype; |
| } |
| |
| #define gst_curl_http_src_parent_class parent_class |
| G_DEFINE_TYPE_WITH_CODE (GstCurlHttpSrc, gst_curl_http_src, GST_TYPE_PUSH_SRC, |
| G_IMPLEMENT_INTERFACE (GST_TYPE_URI_HANDLER, |
| gst_curl_http_src_uri_handler_init)); |
| |
| static void |
| gst_curl_http_src_class_init (GstCurlHttpSrcClass * klass) |
| { |
| GObjectClass *gobject_class; |
| GstElementClass *gstelement_class; |
| GstBaseSrcClass *gstbasesrc_class; |
| GstPushSrcClass *gstpushsrc_class; |
| const gchar *http_env; |
| GstCurlHttpVersion default_http_version; |
| |
| gobject_class = (GObjectClass *) klass; |
| gstelement_class = (GstElementClass *) klass; |
| gstbasesrc_class = (GstBaseSrcClass *) klass; |
| gstpushsrc_class = (GstPushSrcClass *) klass; |
| |
| GST_DEBUG_CATEGORY_INIT (gst_curl_http_src_debug, "curlhttpsrc", |
| 0, "UriHandler for libcURL"); |
| |
| GST_INFO_OBJECT (klass, "class_init started!"); |
| |
| gstelement_class->change_state = |
| GST_DEBUG_FUNCPTR (gst_curl_http_src_change_state); |
| gstpushsrc_class->create = GST_DEBUG_FUNCPTR (gst_curl_http_src_create); |
| gstbasesrc_class->query = GST_DEBUG_FUNCPTR (gst_curl_http_src_query); |
| gstbasesrc_class->get_size = |
| GST_DEBUG_FUNCPTR (gst_curl_http_src_get_content_length); |
| gstbasesrc_class->unlock = GST_DEBUG_FUNCPTR (gst_curl_http_src_unlock); |
| gstbasesrc_class->unlock_stop = |
| GST_DEBUG_FUNCPTR (gst_curl_http_src_unlock_stop); |
| |
| gst_element_class_add_pad_template (gstelement_class, |
| gst_static_pad_template_get (&srcpadtemplate)); |
| |
| gst_curl_http_src_curl_capabilities = curl_version_info (CURLVERSION_NOW); |
| #ifdef CURL_VERSION_HTTP2 |
| if (gst_curl_http_src_curl_capabilities->features & CURL_VERSION_HTTP2) { |
| default_http_version = GSTCURL_HTTP_VERSION_2_0; |
| } else |
| #endif |
| default_http_version = GSTCURL_HTTP_VERSION_1_1; |
| |
| http_env = g_getenv ("GST_CURL_HTTP_VER"); |
| if (http_env != NULL) { |
| GST_INFO_OBJECT (klass, "Seen env var GST_CURL_HTTP_VER with value %s", |
| http_env); |
| if (!strcmp (http_env, "1.0")) { |
| pref_http_ver = GSTCURL_HTTP_VERSION_1_0; |
| } else if (!strcmp (http_env, "1.1")) { |
| pref_http_ver = GSTCURL_HTTP_VERSION_1_1; |
| } else if (!strcmp (http_env, "2.0")) { |
| #ifdef CURL_VERSION_HTTP2 |
| if (gst_curl_http_src_curl_capabilities->features & CURL_VERSION_HTTP2) { |
| pref_http_ver = GSTCURL_HTTP_VERSION_2_0; |
| } else { |
| goto unsupported_http_version; |
| } |
| #endif |
| } else { |
| unsupported_http_version: |
| GST_WARNING_OBJECT (klass, |
| "Unsupported HTTP version: %s. Fallback to default", http_env); |
| pref_http_ver = default_http_version; |
| } |
| } else { |
| pref_http_ver = default_http_version; |
| } |
| |
| gst_curl_http_src_default_useragent = |
| g_strdup_printf ("GStreamer curlhttpsrc libcurl/%s", |
| gst_curl_http_src_curl_capabilities->version); |
| |
| gobject_class->set_property = gst_curl_http_src_set_property; |
| gobject_class->get_property = gst_curl_http_src_get_property; |
| gobject_class->finalize = gst_curl_http_src_finalize; |
| |
| g_object_class_install_property (gobject_class, PROP_URI, |
| g_param_spec_string ("location", "Location", "URI of resource to read", |
| GSTCURL_HANDLE_DEFAULT_CURLOPT_URL, |
| G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS)); |
| |
| g_object_class_install_property (gobject_class, PROP_USERNAME, |
| g_param_spec_string ("user-id", "user-id", |
| "HTTP location URI user id for authentication", |
| GSTCURL_HANDLE_DEFAULT_CURLOPT_USERNAME, |
| G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS)); |
| |
| g_object_class_install_property (gobject_class, PROP_PASSWORD, |
| g_param_spec_string ("user-pw", "user-pw", |
| "HTTP location URI password for authentication", |
| GSTCURL_HANDLE_DEFAULT_CURLOPT_PASSWORD, |
| G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS)); |
| |
| g_object_class_install_property (gobject_class, PROP_PROXYURI, |
| g_param_spec_string ("proxy", "Proxy", "URI of HTTP proxy server", |
| GSTCURL_HANDLE_DEFAULT_CURLOPT_PROXY, |
| G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS)); |
| |
| g_object_class_install_property (gobject_class, PROP_PROXYUSERNAME, |
| g_param_spec_string ("proxy-id", "proxy-id", |
| "HTTP proxy URI user id for authentication", |
| GSTCURL_HANDLE_DEFAULT_CURLOPT_PROXYUSERNAME, |
| G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS)); |
| |
| g_object_class_install_property (gobject_class, PROP_PROXYPASSWORD, |
| g_param_spec_string ("proxy-pw", "proxy-pw", |
| "HTTP proxy URI password for authentication", |
| GSTCURL_HANDLE_DEFAULT_CURLOPT_PROXYPASSWORD, |
| G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS)); |
| |
| g_object_class_install_property (gobject_class, PROP_COOKIES, |
| g_param_spec_boxed ("cookies", "Cookies", "List of HTTP Cookies", |
| G_TYPE_STRV, G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS)); |
| |
| g_object_class_install_property (gobject_class, PROP_USERAGENT, |
| g_param_spec_string ("user-agent", "User-Agent", |
| "URI of resource requested", GSTCURL_HANDLE_DEFAULT_CURLOPT_USERAGENT, |
| G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS)); |
| |
| g_object_class_install_property (gobject_class, PROP_COMPRESS, |
| g_param_spec_boolean ("compress", "Compress", |
| "Allow compressed content encodings", |
| GSTCURL_HANDLE_DEFAULT_CURLOPT_ACCEPT_ENCODING, |
| G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS)); |
| |
| g_object_class_install_property (gobject_class, PROP_REDIRECT, |
| g_param_spec_boolean ("automatic-redirect", "automatic-redirect", |
| "Allow HTTP Redirections (HTTP Status Code 300 series)", |
| GSTCURL_HANDLE_DEFAULT_CURLOPT_FOLLOWLOCATION, |
| G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS)); |
| |
| g_object_class_install_property (gobject_class, PROP_MAXREDIRECT, |
| g_param_spec_int ("max-redirect", "Max-Redirect", |
| "Maximum number of permitted redirections. -1 is unlimited.", |
| GSTCURL_HANDLE_MIN_CURLOPT_MAXREDIRS, |
| GSTCURL_HANDLE_MAX_CURLOPT_MAXREDIRS, |
| GSTCURL_HANDLE_DEFAULT_CURLOPT_MAXREDIRS, |
| G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS)); |
| |
| g_object_class_install_property (gobject_class, PROP_KEEPALIVE, |
| g_param_spec_boolean ("keep-alive", "Keep-Alive", |
| "Toggle keep-alive for connection reuse.", |
| GSTCURL_HANDLE_DEFAULT_CURLOPT_TCP_KEEPALIVE, |
| G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS)); |
| |
| g_object_class_install_property (gobject_class, PROP_TIMEOUT, |
| g_param_spec_int ("timeout", "Timeout", |
| "Value in seconds before timeout a blocking request (0 = no timeout)", |
| GSTCURL_HANDLE_MIN_CURLOPT_TIMEOUT, |
| GSTCURL_HANDLE_MAX_CURLOPT_TIMEOUT, |
| GSTCURL_HANDLE_DEFAULT_CURLOPT_TIMEOUT, |
| G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS)); |
| |
| g_object_class_install_property (gobject_class, PROP_HEADERS, |
| g_param_spec_boxed ("extra-headers", "Extra Headers", |
| "Extra headers to append to the HTTP request", |
| GST_TYPE_STRUCTURE, G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS)); |
| |
| g_object_class_install_property (gobject_class, PROP_STRICT_SSL, |
| g_param_spec_boolean ("ssl-strict", "SSL Strict", |
| "Strict SSL certificate checking", |
| GSTCURL_HANDLE_DEFAULT_CURLOPT_SSL_VERIFYPEER, |
| G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS)); |
| |
| g_object_class_install_property (gobject_class, PROP_SSL_CA_FILE, |
| g_param_spec_string ("ssl-ca-file", "SSL CA File", |
| "Location of an SSL CA file to use for checking SSL certificates", |
| GSTCURL_HANDLE_DEFAULT_CURLOPT_CAINFO, |
| G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS)); |
| |
| g_object_class_install_property (gobject_class, PROP_RETRIES, |
| g_param_spec_int ("retries", "Retries", |
| "Maximum number of retries until giving up (-1=infinite)", |
| GSTCURL_HANDLE_MIN_RETRIES, GSTCURL_HANDLE_MAX_RETRIES, |
| GSTCURL_HANDLE_DEFAULT_RETRIES, |
| G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS)); |
| |
| g_object_class_install_property (gobject_class, PROP_CONNECTIONMAXTIME, |
| g_param_spec_uint ("max-connection-time", "Max-Connection-Time", |
| "Maximum amount of time to keep-alive HTTP connections", |
| GSTCURL_MIN_CONNECTION_TIME, GSTCURL_MAX_CONNECTION_TIME, |
| GSTCURL_DEFAULT_CONNECTION_TIME, |
| G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS)); |
| |
| g_object_class_install_property (gobject_class, PROP_MAXCONCURRENT_SERVER, |
| g_param_spec_uint ("max-connections-per-server", |
| "Max-Connections-Per-Server", |
| "Maximum number of connections allowed per server for HTTP/1.x", |
| GSTCURL_MIN_CONNECTIONS_SERVER, GSTCURL_MAX_CONNECTIONS_SERVER, |
| GSTCURL_DEFAULT_CONNECTIONS_SERVER, |
| G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS)); |
| |
| g_object_class_install_property (gobject_class, PROP_MAXCONCURRENT_PROXY, |
| g_param_spec_uint ("max-connections-per-proxy", |
| "Max-Connections-Per-Proxy", |
| "Maximum number of concurrent connections allowed per proxy for HTTP/1.x", |
| GSTCURL_MIN_CONNECTIONS_PROXY, GSTCURL_MAX_CONNECTIONS_PROXY, |
| GSTCURL_DEFAULT_CONNECTIONS_PROXY, |
| G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS)); |
| |
| g_object_class_install_property (gobject_class, PROP_MAXCONCURRENT_GLOBAL, |
| g_param_spec_uint ("max-connections", "Max-Connections", |
| "Maximum number of concurrent connections allowed for HTTP/1.x", |
| GSTCURL_MIN_CONNECTIONS_GLOBAL, GSTCURL_MAX_CONNECTIONS_GLOBAL, |
| GSTCURL_DEFAULT_CONNECTIONS_GLOBAL, |
| G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS)); |
| |
| g_object_class_install_property (gobject_class, PROP_HTTPVERSION, |
| g_param_spec_enum ("http-version", "HTTP-Version", |
| "The preferred HTTP protocol version", |
| GST_TYPE_CURL_HTTP_VERSION, pref_http_ver, |
| G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS)); |
| |
| /* Add a debugging task so it's easier to debug in the Multi worker thread */ |
| GST_DEBUG_CATEGORY_INIT (gst_curl_loop_debug, "curl_multi_loop", 0, |
| "libcURL loop thread debugging"); |
| #ifndef GST_DISABLE_GST_DEBUG |
| gst_debug_log (gst_curl_loop_debug, GST_LEVEL_INFO, __FILE__, __func__, |
| __LINE__, NULL, "Testing the curl_multi_loop debugging prints"); |
| #endif |
| |
| g_mutex_init (&klass->multi_task_context.mutex); |
| g_cond_init (&klass->multi_task_context.signal); |
| g_rec_mutex_init (&klass->multi_task_context.task_rec_mutex); |
| |
| gst_element_class_set_static_metadata (gstelement_class, |
| "HTTP Client Source using libcURL", |
| "Source/Network", |
| "Receiver data as a client over a network via HTTP using cURL", |
| "Sam Hurst <samuelh@rd.bbc.co.uk>"); |
| } |
| |
| static void |
| gst_curl_http_src_set_property (GObject * object, guint prop_id, |
| const GValue * value, GParamSpec * pspec) |
| { |
| GstCurlHttpSrc *source = GST_CURLHTTPSRC (object); |
| GSTCURL_FUNCTION_ENTRY (source); |
| |
| switch (prop_id) { |
| case PROP_URI: |
| g_free (source->uri); |
| source->uri = g_value_dup_string (value); |
| break; |
| case PROP_USERNAME: |
| g_free (source->username); |
| source->username = g_value_dup_string (value); |
| break; |
| case PROP_PASSWORD: |
| g_free (source->password); |
| source->password = g_value_dup_string (value); |
| break; |
| case PROP_PROXYURI: |
| g_free (source->proxy_uri); |
| source->proxy_uri = g_value_dup_string (value); |
| break; |
| case PROP_PROXYUSERNAME: |
| g_free (source->proxy_user); |
| source->proxy_user = g_value_dup_string (value); |
| break; |
| case PROP_PROXYPASSWORD: |
| g_free (source->proxy_pass); |
| source->proxy_pass = g_value_dup_string (value); |
| break; |
| case PROP_COOKIES: |
| g_strfreev (source->cookies); |
| source->cookies = g_strdupv (g_value_get_boxed (value)); |
| source->number_cookies = g_strv_length (source->cookies); |
| break; |
| case PROP_USERAGENT: |
| g_free (source->user_agent); |
| source->user_agent = g_value_dup_string (value); |
| break; |
| case PROP_HEADERS: |
| { |
| const GstStructure *s = gst_value_get_structure (value); |
| if (source->request_headers) |
| gst_structure_free (source->request_headers); |
| source->request_headers = s ? gst_structure_copy (s) : NULL; |
| } |
| break; |
| case PROP_COMPRESS: |
| source->accept_compressed_encodings = g_value_get_boolean (value); |
| break; |
| case PROP_REDIRECT: |
| source->allow_3xx_redirect = g_value_get_boolean (value); |
| break; |
| case PROP_MAXREDIRECT: |
| source->max_3xx_redirects = g_value_get_int (value); |
| break; |
| case PROP_KEEPALIVE: |
| source->keep_alive = g_value_get_boolean (value); |
| break; |
| case PROP_TIMEOUT: |
| source->timeout_secs = g_value_get_int (value); |
| break; |
| case PROP_STRICT_SSL: |
| source->strict_ssl = g_value_get_boolean (value); |
| break; |
| case PROP_SSL_CA_FILE: |
| source->custom_ca_file = g_value_dup_string (value); |
| break; |
| case PROP_RETRIES: |
| source->total_retries = g_value_get_int (value); |
| break; |
| case PROP_CONNECTIONMAXTIME: |
| source->max_connection_time = g_value_get_uint (value); |
| break; |
| case PROP_MAXCONCURRENT_SERVER: |
| source->max_conns_per_server = g_value_get_uint (value); |
| break; |
| case PROP_MAXCONCURRENT_PROXY: |
| source->max_conns_per_proxy = g_value_get_uint (value); |
| break; |
| case PROP_MAXCONCURRENT_GLOBAL: |
| source->max_conns_global = g_value_get_uint (value); |
| break; |
| case PROP_HTTPVERSION: |
| source->preferred_http_version = g_value_get_enum (value); |
| break; |
| default: |
| G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec); |
| break; |
| } |
| GSTCURL_FUNCTION_EXIT (source); |
| } |
| |
| static void |
| gst_curl_http_src_get_property (GObject * object, guint prop_id, |
| GValue * value, GParamSpec * pspec) |
| { |
| GstCurlHttpSrc *source = GST_CURLHTTPSRC (object); |
| GSTCURL_FUNCTION_ENTRY (source); |
| |
| switch (prop_id) { |
| case PROP_URI: |
| g_value_set_string (value, source->uri); |
| break; |
| case PROP_USERNAME: |
| g_value_set_string (value, source->username); |
| break; |
| case PROP_PASSWORD: |
| g_value_set_string (value, source->password); |
| break; |
| case PROP_PROXYURI: |
| g_value_set_string (value, source->proxy_uri); |
| break; |
| case PROP_PROXYUSERNAME: |
| g_value_set_string (value, source->proxy_user); |
| break; |
| case PROP_PROXYPASSWORD: |
| g_value_set_string (value, source->proxy_pass); |
| break; |
| case PROP_COOKIES: |
| g_value_set_boxed (value, source->cookies); |
| break; |
| case PROP_USERAGENT: |
| g_value_set_string (value, source->user_agent); |
| break; |
| case PROP_HEADERS: |
| gst_value_set_structure (value, source->request_headers); |
| break; |
| case PROP_COMPRESS: |
| g_value_set_boolean (value, source->accept_compressed_encodings); |
| break; |
| case PROP_REDIRECT: |
| g_value_set_boolean (value, source->allow_3xx_redirect); |
| break; |
| case PROP_MAXREDIRECT: |
| g_value_set_int (value, source->max_3xx_redirects); |
| break; |
| case PROP_KEEPALIVE: |
| g_value_set_boolean (value, source->keep_alive); |
| break; |
| case PROP_TIMEOUT: |
| g_value_set_int (value, source->timeout_secs); |
| break; |
| case PROP_STRICT_SSL: |
| g_value_set_boolean (value, source->strict_ssl); |
| break; |
| case PROP_SSL_CA_FILE: |
| g_value_set_string (value, source->custom_ca_file); |
| break; |
| case PROP_RETRIES: |
| g_value_set_int (value, source->total_retries); |
| break; |
| case PROP_CONNECTIONMAXTIME: |
| g_value_set_uint (value, source->max_connection_time); |
| break; |
| case PROP_MAXCONCURRENT_SERVER: |
| g_value_set_uint (value, source->max_conns_per_server); |
| break; |
| case PROP_MAXCONCURRENT_PROXY: |
| g_value_set_uint (value, source->max_conns_per_proxy); |
| break; |
| case PROP_MAXCONCURRENT_GLOBAL: |
| g_value_set_uint (value, source->max_conns_global); |
| break; |
| case PROP_HTTPVERSION: |
| g_value_set_enum (value, source->preferred_http_version); |
| break; |
| default: |
| G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec); |
| break; |
| } |
| GSTCURL_FUNCTION_EXIT (source); |
| } |
| |
| static void |
| gst_curl_http_src_init (GstCurlHttpSrc * source) |
| { |
| GSTCURL_FUNCTION_ENTRY (source); |
| |
| /* Assume everything is already free'd */ |
| source->uri = NULL; |
| source->redirect_uri = NULL; |
| source->username = GSTCURL_HANDLE_DEFAULT_CURLOPT_USERNAME; |
| source->password = GSTCURL_HANDLE_DEFAULT_CURLOPT_PASSWORD; |
| source->proxy_uri = NULL; |
| source->proxy_user = NULL; |
| source->proxy_pass = NULL; |
| source->cookies = NULL; |
| source->user_agent = GSTCURL_HANDLE_DEFAULT_CURLOPT_USERAGENT; |
| source->number_cookies = 0; |
| source->request_headers = NULL; |
| source->allow_3xx_redirect = GSTCURL_HANDLE_DEFAULT_CURLOPT_FOLLOWLOCATION; |
| source->max_3xx_redirects = GSTCURL_HANDLE_DEFAULT_CURLOPT_MAXREDIRS; |
| source->keep_alive = GSTCURL_HANDLE_DEFAULT_CURLOPT_TCP_KEEPALIVE; |
| source->timeout_secs = GSTCURL_HANDLE_DEFAULT_CURLOPT_TIMEOUT; |
| source->max_connection_time = GSTCURL_DEFAULT_CONNECTION_TIME; |
| source->max_conns_per_server = GSTCURL_DEFAULT_CONNECTIONS_SERVER; |
| source->max_conns_per_proxy = GSTCURL_DEFAULT_CONNECTIONS_PROXY; |
| source->max_conns_global = GSTCURL_DEFAULT_CONNECTIONS_GLOBAL; |
| source->strict_ssl = GSTCURL_HANDLE_DEFAULT_CURLOPT_SSL_VERIFYPEER; |
| source->custom_ca_file = NULL; |
| source->preferred_http_version = pref_http_ver; |
| source->total_retries = GSTCURL_HANDLE_DEFAULT_RETRIES; |
| source->retries_remaining = source->total_retries; |
| source->slist = NULL; |
| |
| gst_caps_replace (&source->caps, NULL); |
| gst_base_src_set_automatic_eos (GST_BASE_SRC (source), FALSE); |
| |
| source->proxy_uri = g_strdup (g_getenv ("http_proxy")); |
| source->no_proxy_list = g_strdup (g_getenv ("no_proxy")); |
| |
| g_mutex_init (&source->uri_mutex); |
| g_mutex_init (&source->buffer_mutex); |
| g_cond_init (&source->signal); |
| |
| source->buffer = NULL; |
| source->buffer_len = 0; |
| source->state = GSTCURL_NONE; |
| source->pending_state = GSTCURL_NONE; |
| source->status_code = 0; |
| |
| source->http_headers = NULL; |
| source->hdrs_updated = FALSE; |
| |
| source->curl_result = CURLE_OK; |
| |
| GSTCURL_FUNCTION_EXIT (source); |
| } |
| |
| /* |
| * Check if the Curl multi loop has been started. If not, initialise it and |
| * start it running. If it is already running, increment the refcount. |
| */ |
| static void |
| gst_curl_http_src_ref_multi (GstCurlHttpSrc * src) |
| { |
| GstCurlHttpSrcClass *klass; |
| |
| GSTCURL_FUNCTION_ENTRY (src); |
| |
| /*klass = (GstCurlHttpSrcClass) g_type_class_peek_parent (src); */ |
| klass = G_TYPE_INSTANCE_GET_CLASS (src, GST_TYPE_CURL_HTTP_SRC, |
| GstCurlHttpSrcClass); |
| |
| g_mutex_lock (&klass->multi_task_context.mutex); |
| if (klass->multi_task_context.refcount == 0) { |
| /* Set up various in-task properties */ |
| |
| /* NULL is treated as the start of the list, no need to allocate. */ |
| klass->multi_task_context.queue = NULL; |
| |
| /* set up curl */ |
| klass->multi_task_context.multi_handle = curl_multi_init (); |
| |
| curl_multi_setopt (klass->multi_task_context.multi_handle, |
| CURLMOPT_PIPELINING, 1); |
| #ifdef CURLMOPT_MAX_HOST_CONNECTIONS |
| curl_multi_setopt (klass->multi_task_context.multi_handle, |
| CURLMOPT_MAX_HOST_CONNECTIONS, 1); |
| #endif |
| |
| /* Start the thread */ |
| klass->multi_task_context.task = gst_task_new ( |
| (GstTaskFunction) gst_curl_http_src_curl_multi_loop, |
| (gpointer) & klass->multi_task_context, NULL); |
| gst_task_set_lock (klass->multi_task_context.task, |
| &klass->multi_task_context.task_rec_mutex); |
| if (gst_task_start (klass->multi_task_context.task) == FALSE) { |
| /* |
| * This is a pretty critical failure and is not recoverable, so commit |
| * sudoku and run away. |
| */ |
| GSTCURL_ERROR_PRINT ("Couldn't start curl_multi task! Aborting."); |
| abort (); |
| } |
| GSTCURL_INFO_PRINT ("Curl multi loop has been correctly initialised!"); |
| } |
| klass->multi_task_context.refcount++; |
| g_mutex_unlock (&klass->multi_task_context.mutex); |
| |
| GSTCURL_FUNCTION_EXIT (src); |
| } |
| |
| /* |
| * Decrement the reference count on the curl multi loop. If this is called by |
| * the last instance to hold a reference, shut down the worker. (Otherwise |
| * GStreamer can't close down with a thread still running). Also offers the |
| * "force_all" boolean parameter, which if TRUE removes all references and shuts |
| * down. |
| */ |
| static void |
| gst_curl_http_src_unref_multi (GstCurlHttpSrc * src) |
| { |
| GstCurlHttpSrcClass *klass; |
| |
| GSTCURL_FUNCTION_ENTRY (src); |
| |
| klass = G_TYPE_INSTANCE_GET_CLASS (src, GST_TYPE_CURL_HTTP_SRC, |
| GstCurlHttpSrcClass); |
| |
| g_mutex_lock (&klass->multi_task_context.mutex); |
| klass->multi_task_context.refcount--; |
| GST_INFO_OBJECT (src, "Closing instance, worker thread refcount is now %u", |
| klass->multi_task_context.refcount); |
| |
| if (klass->multi_task_context.refcount <= 0) { |
| /* Everything's done! Clean up. */ |
| gst_task_pause (klass->multi_task_context.task); |
| klass->multi_task_context.state = GSTCURL_MULTI_LOOP_STATE_STOP; |
| g_cond_signal (&klass->multi_task_context.signal); |
| g_mutex_unlock (&klass->multi_task_context.mutex); |
| gst_task_join (klass->multi_task_context.task); |
| } else { |
| g_mutex_unlock (&klass->multi_task_context.mutex); |
| } |
| |
| GSTCURL_FUNCTION_EXIT (src); |
| } |
| |
| static void |
| gst_curl_http_src_finalize (GObject * obj) |
| { |
| GstCurlHttpSrc *src = GST_CURLHTTPSRC (obj); |
| |
| GSTCURL_FUNCTION_ENTRY (src); |
| |
| /* Cleanup all memory allocated */ |
| gst_curl_http_src_cleanup_instance (src); |
| |
| GSTCURL_FUNCTION_EXIT (src); |
| } |
| |
| /* |
| * Do the transfer. If the transfer hasn't begun yet, start a new curl handle |
| * and pass it to the multi queue to be operated on. Then wait for any blocks |
| * of data and push them to the source pad. |
| */ |
| static GstFlowReturn |
| gst_curl_http_src_create (GstPushSrc * psrc, GstBuffer ** outbuf) |
| { |
| GstFlowReturn ret; |
| GstCurlHttpSrc *src = GST_CURLHTTPSRC (psrc); |
| GstCurlHttpSrcClass *klass; |
| GstStructure *empty_headers; |
| |
| klass = G_TYPE_INSTANCE_GET_CLASS (src, GST_TYPE_CURL_HTTP_SRC, |
| GstCurlHttpSrcClass); |
| |
| GSTCURL_FUNCTION_ENTRY (src); |
| ret = GST_FLOW_OK; |
| |
| g_mutex_lock (&src->buffer_mutex); |
| if (src->state == GSTCURL_UNLOCK) { |
| ret = GST_FLOW_FLUSHING; |
| goto escape; |
| } |
| |
| retry: |
| if (!src->transfer_begun) { |
| GST_DEBUG_OBJECT (src, "Starting new request for URI %s", src->uri); |
| /* Create the Easy Handle and set up the session. */ |
| src->curl_handle = gst_curl_http_src_create_easy_handle (src); |
| if (src->curl_handle == NULL) { |
| ret = GST_FLOW_ERROR; |
| goto escape; |
| } |
| |
| g_mutex_lock (&klass->multi_task_context.mutex); |
| |
| if (gst_curl_http_src_add_queue_item (&klass->multi_task_context.queue, src) |
| == FALSE) { |
| GST_ERROR_OBJECT (src, "Couldn't create new queue item! Aborting..."); |
| ret = GST_FLOW_ERROR; |
| goto escape; |
| } |
| |
| /* Signal the worker thread */ |
| klass->multi_task_context.state = GSTCURL_MULTI_LOOP_STATE_QUEUE_EVENT; |
| g_cond_signal (&klass->multi_task_context.signal); |
| g_mutex_unlock (&klass->multi_task_context.mutex); |
| |
| src->state = GSTCURL_OK; |
| src->transfer_begun = TRUE; |
| src->data_received = FALSE; |
| |
| GST_DEBUG_OBJECT (src, "Submitted request for URI %s to curl", src->uri); |
| |
| empty_headers = gst_structure_new_empty (RESPONSE_HEADERS_NAME); |
| src->http_headers = gst_structure_new (HTTP_HEADERS_NAME, |
| URI_NAME, G_TYPE_STRING, src->uri, |
| REQUEST_HEADERS_NAME, GST_TYPE_STRUCTURE, src->request_headers, |
| RESPONSE_HEADERS_NAME, GST_TYPE_STRUCTURE, empty_headers, NULL); |
| gst_structure_free (empty_headers); |
| GST_INFO_OBJECT (src, "Created a new headers object"); |
| } |
| |
| /* Wait for data to become available, then punt it downstream */ |
| while ((src->buffer_len == 0) && (src->state == GSTCURL_OK)) { |
| g_cond_wait (&src->signal, &src->buffer_mutex); |
| } |
| |
| if (src->state == GSTCURL_UNLOCK) { |
| if (src->buffer_len > 0) { |
| g_free (src->buffer); |
| src->buffer = NULL; |
| src->buffer_len = 0; |
| } |
| ret = GST_FLOW_FLUSHING; |
| goto escape; |
| } |
| |
| ret = gst_curl_http_src_handle_response (src); |
| switch (ret) { |
| case GST_FLOW_ERROR: |
| goto escape; /* Don't attempt a retry, just bomb out */ |
| case GST_FLOW_CUSTOM_ERROR: |
| if (src->data_received == TRUE) { |
| /* |
| * If data has already been received, we can't recall previously sent |
| * buffers so don't attempt a retry in this case. |
| * |
| * TODO: Remember the position we got to, and make a range request for |
| * the resource without the bit we've already received? |
| */ |
| GST_WARNING_OBJECT (src, |
| "Failed mid-transfer, can't continue for URI %s", src->uri); |
| ret = GST_FLOW_ERROR; |
| goto escape; |
| } |
| src->retries_remaining--; |
| if (src->retries_remaining == 0) { |
| GST_WARNING_OBJECT (src, "Out of retries for URI %s", src->uri); |
| ret = GST_FLOW_ERROR; /* Don't attempt a retry, just bomb out */ |
| goto escape; |
| } |
| GST_INFO_OBJECT (src, "Attempting retry for URI %s", src->uri); |
| src->state = GSTCURL_NONE; |
| src->transfer_begun = FALSE; |
| src->status_code = 0; |
| src->hdrs_updated = FALSE; |
| if (src->http_headers != NULL) { |
| gst_structure_free (src->http_headers); |
| src->http_headers = NULL; |
| GST_INFO_OBJECT (src, "NULL'd the headers"); |
| } |
| gst_curl_http_src_destroy_easy_handle (src); |
| g_mutex_unlock (&src->buffer_mutex); |
| goto retry; /* Attempt a retry! */ |
| default: |
| break; |
| } |
| |
| if (((src->state == GSTCURL_OK) || (src->state == GSTCURL_DONE)) && |
| (src->buffer_len > 0)) { |
| |
| GST_DEBUG_OBJECT (src, "Pushing %u bytes of transfer for URI %s to pad", |
| src->buffer_len, src->uri); |
| *outbuf = gst_buffer_new_allocate (NULL, src->buffer_len, NULL); |
| gst_buffer_fill (*outbuf, 0, src->buffer, src->buffer_len); |
| |
| g_free (src->buffer); |
| src->buffer = NULL; |
| src->buffer_len = 0; |
| src->data_received = TRUE; |
| |
| /* ret should still be GST_FLOW_OK */ |
| } else if ((src->state == GSTCURL_DONE) && (src->buffer_len == 0)) { |
| GST_INFO_OBJECT (src, "Full body received, signalling EOS for URI %s.", |
| src->uri); |
| src->state = GSTCURL_NONE; |
| src->transfer_begun = FALSE; |
| src->status_code = 0; |
| src->hdrs_updated = FALSE; |
| gst_curl_http_src_destroy_easy_handle (src); |
| ret = GST_FLOW_EOS; |
| } else { |
| switch (src->state) { |
| case GSTCURL_NONE: |
| GST_WARNING_OBJECT (src, "Got unexpected GSTCURL_NONE state!"); |
| break; |
| case GSTCURL_REMOVED: |
| GST_WARNING_OBJECT (src, "Transfer got removed from the curl queue"); |
| ret = GST_FLOW_EOS; |
| break; |
| case GSTCURL_BAD_QUEUE_REQUEST: |
| GST_ERROR_OBJECT (src, "Bad Queue Request!"); |
| ret = GST_FLOW_ERROR; |
| break; |
| case GSTCURL_TOTAL_ERROR: |
| GST_ERROR_OBJECT (src, "Critical, unrecoverable error!"); |
| ret = GST_FLOW_ERROR; |
| break; |
| case GSTCURL_PIPELINE_NULL: |
| GST_ERROR_OBJECT (src, "Pipeline null"); |
| break; |
| default: |
| GST_ERROR_OBJECT (src, "Unknown state of %u", src->state); |
| } |
| } |
| |
| escape: |
| g_mutex_unlock (&src->buffer_mutex); |
| |
| GSTCURL_FUNCTION_EXIT (src); |
| return ret; |
| } |
| |
| /* |
| * Convert header from a GstStructure type to a curl_slist type that curl will |
| * understand. |
| */ |
| static gboolean |
| _headers_to_curl_slist (GQuark field_id, const GValue * value, gpointer ptr) |
| { |
| gchar *field; |
| struct curl_slist **p_slist = ptr; |
| |
| field = g_strdup_printf ("%s: %s", g_quark_to_string (field_id), |
| g_value_get_string (value)); |
| |
| *p_slist = curl_slist_append (*p_slist, field); |
| |
| g_free (field); |
| |
| return TRUE; |
| } |
| |
| /* |
| * From the data in the queue element s, create a CURL easy handle and populate |
| * options with the URL, proxy data, login options, cookies, |
| */ |
| static CURL * |
| gst_curl_http_src_create_easy_handle (GstCurlHttpSrc * s) |
| { |
| CURL *handle; |
| gint i; |
| GSTCURL_FUNCTION_ENTRY (s); |
| |
| handle = curl_easy_init (); |
| if (handle == NULL) { |
| GST_ERROR_OBJECT (s, "Couldn't init a curl easy handle!"); |
| return NULL; |
| } |
| GST_INFO_OBJECT (s, "Creating a new handle for URI %s", s->uri); |
| |
| /* This is mandatory and yet not default option, so if this is NULL |
| * then something very bad is going on. */ |
| if (s->uri == NULL) { |
| GST_ERROR_OBJECT (s, "No URI for curl!"); |
| return NULL; |
| } |
| gst_curl_setopt_str (s, handle, CURLOPT_URL, s->uri); |
| |
| gst_curl_setopt_str (s, handle, CURLOPT_USERNAME, s->username); |
| gst_curl_setopt_str (s, handle, CURLOPT_PASSWORD, s->password); |
| gst_curl_setopt_str (s, handle, CURLOPT_PROXY, s->proxy_uri); |
| gst_curl_setopt_str (s, handle, CURLOPT_NOPROXY, s->no_proxy_list); |
| gst_curl_setopt_str (s, handle, CURLOPT_PROXYUSERNAME, s->proxy_user); |
| gst_curl_setopt_str (s, handle, CURLOPT_PROXYPASSWORD, s->proxy_pass); |
| |
| for (i = 0; i < s->number_cookies; i++) { |
| gst_curl_setopt_str (s, handle, CURLOPT_COOKIELIST, s->cookies[i]); |
| } |
| |
| /* curl_slist_append dynamically allocates memory, but I need to free it */ |
| if (s->request_headers != NULL) { |
| gst_structure_foreach (s->request_headers, _headers_to_curl_slist, |
| &s->slist); |
| if (curl_easy_setopt (handle, CURLOPT_HTTPHEADER, s->slist) != CURLE_OK) { |
| GST_WARNING_OBJECT (s, "Failed to set HTTP headers!"); |
| } |
| } |
| |
| gst_curl_setopt_str_default (s, handle, CURLOPT_USERAGENT, s->user_agent); |
| |
| /* |
| * Unlike soup, this isn't a binary op, curl wants a string here. So if it's |
| * TRUE, simply set the value as an empty string as this allows both gzip and |
| * zlib compression methods. |
| */ |
| if (s->accept_compressed_encodings == TRUE) { |
| gst_curl_setopt_str (s, handle, CURLOPT_ACCEPT_ENCODING, ""); |
| } else { |
| gst_curl_setopt_str (s, handle, CURLOPT_ACCEPT_ENCODING, "identity"); |
| } |
| |
| gst_curl_setopt_int (s, handle, CURLOPT_FOLLOWLOCATION, |
| s->allow_3xx_redirect); |
| gst_curl_setopt_int_default (s, handle, CURLOPT_MAXREDIRS, |
| s->max_3xx_redirects); |
| gst_curl_setopt_bool (s, handle, CURLOPT_TCP_KEEPALIVE, s->keep_alive); |
| gst_curl_setopt_int (s, handle, CURLOPT_TIMEOUT, s->timeout_secs); |
| gst_curl_setopt_bool (s, handle, CURLOPT_SSL_VERIFYPEER, s->strict_ssl); |
| gst_curl_setopt_str (s, handle, CURLOPT_CAINFO, s->custom_ca_file); |
| |
| switch (s->preferred_http_version) { |
| case GSTCURL_HTTP_VERSION_1_0: |
| GST_DEBUG_OBJECT (s, "Setting version as HTTP/1.0"); |
| gst_curl_setopt_int (s, handle, CURLOPT_HTTP_VERSION, |
| CURL_HTTP_VERSION_1_0); |
| break; |
| case GSTCURL_HTTP_VERSION_1_1: |
| GST_DEBUG_OBJECT (s, "Setting version as HTTP/1.1"); |
| gst_curl_setopt_int (s, handle, CURLOPT_HTTP_VERSION, |
| CURL_HTTP_VERSION_1_1); |
| break; |
| #ifdef CURL_VERSION_HTTP2 |
| case GSTCURL_HTTP_VERSION_2_0: |
| GST_DEBUG_OBJECT (s, "Setting version as HTTP/2.0"); |
| if (curl_easy_setopt (handle, CURLOPT_HTTP_VERSION, |
| CURL_HTTP_VERSION_2_0) != CURLE_OK) { |
| if (gst_curl_http_src_curl_capabilities->features & CURL_VERSION_HTTP2) { |
| GST_WARNING_OBJECT (s, |
| "Cannot set unsupported option CURLOPT_HTTP_VERSION"); |
| } else { |
| GST_INFO_OBJECT (s, "HTTP/2 unsupported by libcurl at this time"); |
| } |
| } |
| break; |
| #endif |
| default: |
| GST_WARNING_OBJECT (s, |
| "Supplied a bogus HTTP version, using curl default!"); |
| } |
| |
| gst_curl_setopt_generic (s, handle, CURLOPT_HEADERFUNCTION, |
| gst_curl_http_src_get_header); |
| gst_curl_setopt_str (s, handle, CURLOPT_HEADERDATA, s); |
| gst_curl_setopt_generic (s, handle, CURLOPT_WRITEFUNCTION, |
| gst_curl_http_src_get_chunks); |
| gst_curl_setopt_str (s, handle, CURLOPT_WRITEDATA, s); |
| |
| gst_curl_setopt_str (s, handle, CURLOPT_ERRORBUFFER, s->curl_errbuf); |
| |
| GSTCURL_FUNCTION_EXIT (s); |
| return handle; |
| } |
| |
| /* |
| * Check the return type from the curl transfer. If it was okay, then deal with |
| * any headers that were received. Headers should only be dealt with once - but |
| * we might get a second set if there are trailing headers (RFC7230 Section 4.4) |
| */ |
| static GstFlowReturn |
| gst_curl_http_src_handle_response (GstCurlHttpSrc * src) |
| { |
| glong curl_info_long; |
| gdouble curl_info_dbl; |
| gchar *redirect_url; |
| GstBaseSrc *basesrc; |
| const GValue *response_headers; |
| GstFlowReturn ret = GST_FLOW_OK; |
| |
| GSTCURL_FUNCTION_ENTRY (src); |
| |
| GST_TRACE_OBJECT (src, "status code: %d, curl return code %d", |
| src->status_code, src->curl_result); |
| |
| /* Check the curl result code first - anything not 0 is probably a failure */ |
| if (src->curl_result != 0) { |
| GST_WARNING_OBJECT (src, "Curl failed the transfer (%d): %s", |
| src->curl_result, curl_easy_strerror (src->curl_result)); |
| GST_DEBUG_OBJECT (src, "Reason for curl failure: %s", src->curl_errbuf); |
| return GST_FLOW_ERROR; |
| } |
| |
| /* |
| * What response code do we have? |
| */ |
| if (src->status_code >= 400) { |
| GST_WARNING_OBJECT (src, "Transfer for URI %s returned error status %u", |
| src->uri, src->status_code); |
| src->retries_remaining = 0; |
| return GST_FLOW_ERROR; |
| } else if (src->status_code == 0) { |
| if (curl_easy_getinfo (src->curl_handle, CURLINFO_TOTAL_TIME, |
| &curl_info_dbl) != CURLE_OK) { |
| /* Curl cannot be relied on in this state, so return an error. */ |
| return GST_FLOW_ERROR; |
| } |
| if (curl_info_dbl > src->timeout_secs) { |
| return GST_FLOW_CUSTOM_ERROR; |
| } |
| |
| if (curl_easy_getinfo (src->curl_handle, CURLINFO_OS_ERRNO, |
| &curl_info_long) != CURLE_OK) { |
| /* Curl cannot be relied on in this state, so return an error. */ |
| return GST_FLOW_ERROR; |
| |
| } |
| |
| GST_WARNING_OBJECT (src, "Errno for CONNECT call was %ld (%s)", |
| curl_info_long, g_strerror ((gint) curl_info_long)); |
| |
| /* Some of these responses are retry-able, others not. Set the returned |
| * state to ERROR so we crash out instead of fruitlessly retrying. |
| */ |
| if (curl_info_long == ECONNREFUSED) { |
| return GST_FLOW_ERROR; |
| } |
| ret = GST_FLOW_CUSTOM_ERROR; |
| } |
| |
| |
| if (ret == GST_FLOW_CUSTOM_ERROR) { |
| src->hdrs_updated = FALSE; |
| GSTCURL_FUNCTION_EXIT (src); |
| return ret; |
| } |
| |
| /* Only do this once */ |
| if (src->hdrs_updated == FALSE) { |
| GSTCURL_FUNCTION_EXIT (src); |
| return GST_FLOW_OK; |
| } |
| |
| /* |
| * Deal with redirections... |
| */ |
| if (curl_easy_getinfo (src->curl_handle, CURLINFO_EFFECTIVE_URL, |
| &redirect_url) |
| == CURLE_OK) { |
| size_t lena, lenb; |
| lena = strlen (src->uri); |
| lenb = strlen (redirect_url); |
| if (g_ascii_strncasecmp (src->uri, redirect_url, |
| (lena > lenb) ? lenb : lena) != 0) { |
| GST_INFO_OBJECT (src, "Got a redirect to %s, setting as redirect URI", |
| redirect_url); |
| src->redirect_uri = g_strdup (redirect_url); |
| gst_structure_remove_field (src->http_headers, REDIRECT_URI_NAME); |
| gst_structure_set (src->http_headers, REDIRECT_URI_NAME, |
| G_TYPE_STRING, redirect_url, NULL); |
| } |
| } |
| |
| /* |
| * Push the content length |
| */ |
| if (curl_easy_getinfo (src->curl_handle, CURLINFO_CONTENT_LENGTH_DOWNLOAD, |
| &curl_info_dbl) == CURLE_OK) { |
| if (curl_info_dbl == -1) { |
| GST_WARNING_OBJECT (src, |
| "No Content-Length was specified in the response."); |
| } else { |
| GST_INFO_OBJECT (src, "Content-Length was given as %.0f", curl_info_dbl); |
| basesrc = GST_BASE_SRC_CAST (src); |
| basesrc->segment.duration = curl_info_dbl; |
| gst_element_post_message (GST_ELEMENT (src), |
| gst_message_new_duration_changed (GST_OBJECT (src))); |
| } |
| } |
| |
| /* |
| * Push all the received headers down via a sicky event |
| */ |
| response_headers = gst_structure_get_value (src->http_headers, |
| RESPONSE_HEADERS_NAME); |
| if (gst_structure_n_fields (gst_value_get_structure (response_headers)) > 0) { |
| GstEvent *hdrs_event; |
| GstStructure *empty_headers; |
| |
| gst_element_post_message (GST_ELEMENT_CAST (src), |
| gst_message_new_element (GST_OBJECT_CAST (src), |
| gst_structure_copy (src->http_headers))); |
| |
| /* gst_event_new_custom takes ownership of our structure */ |
| hdrs_event = gst_event_new_custom (GST_EVENT_CUSTOM_DOWNSTREAM_STICKY, |
| src->http_headers); |
| gst_pad_push_event (GST_BASE_SRC_PAD (src), hdrs_event); |
| GST_INFO_OBJECT (src, "Pushed headers downstream"); |
| empty_headers = gst_structure_new_empty (RESPONSE_HEADERS_NAME); |
| src->http_headers = gst_structure_new (HTTP_HEADERS_NAME, |
| URI_NAME, G_TYPE_STRING, src->uri, |
| REQUEST_HEADERS_NAME, GST_TYPE_STRUCTURE, src->request_headers, |
| RESPONSE_HEADERS_NAME, GST_TYPE_STRUCTURE, empty_headers, NULL); |
| gst_structure_free (empty_headers); |
| } |
| |
| src->hdrs_updated = FALSE; |
| |
| GSTCURL_FUNCTION_EXIT (src); |
| |
| return ret; |
| } |
| |
| /* |
| * "Negotiate" capabilities between us and the sink. |
| * I.e. tell the sink device what data to expect. We can't be told what to send |
| * unless we implement "only return to me if this type" property. Potential TODO |
| */ |
| static gboolean |
| gst_curl_http_src_negotiate_caps (GstCurlHttpSrc * src) |
| { |
| GST_INFO_OBJECT (src, "Negotiating caps..."); |
| if (src->caps && src->http_headers) { |
| const GValue *response_headers = gst_structure_get_value (src->http_headers, |
| RESPONSE_HEADERS_NAME); |
| |
| if (gst_structure_has_field (gst_value_get_structure (response_headers), |
| "content-type") == TRUE) { |
| const GValue *gv_content_type = |
| gst_structure_get_value (gst_value_get_structure (response_headers), |
| "content-type"); |
| if (G_VALUE_HOLDS_STRING (gv_content_type) == TRUE) { |
| const gchar *content_type = g_value_get_string (gv_content_type); |
| GST_INFO_OBJECT (src, "Setting caps as Content-Type of %s", |
| content_type); |
| src->caps = gst_caps_make_writable (src->caps); |
| gst_caps_set_simple (src->caps, "content-type", G_TYPE_STRING, |
| content_type, NULL); |
| if (gst_base_src_set_caps (GST_BASE_SRC (src), src->caps) != TRUE) { |
| GST_ERROR_OBJECT (src, "Setting caps failed!"); |
| return FALSE; |
| } |
| } else { |
| GST_ERROR_OBJECT (src, "Content Type doesn't contain expected string"); |
| return FALSE; |
| } |
| } |
| } else { |
| GST_DEBUG_OBJECT (src, "No caps have been set, continue."); |
| } |
| |
| return TRUE; |
| } |
| |
| /* |
| * Cleanup the CURL easy handle once we're done with it. |
| */ |
| static inline void |
| gst_curl_http_src_destroy_easy_handle (GstCurlHttpSrc * src) |
| { |
| /* Thank you Handles, and well done. Well done, mate. */ |
| if (src->curl_handle != NULL) { |
| curl_easy_cleanup (src->curl_handle); |
| src->curl_handle = NULL; |
| } |
| /* In addition, clean up the curl header slist if it was used. */ |
| if (src->slist != NULL) { |
| curl_slist_free_all (src->slist); |
| src->slist = NULL; |
| } |
| } |
| |
| static GstStateChangeReturn |
| gst_curl_http_src_change_state (GstElement * element, GstStateChange transition) |
| { |
| GstStateChangeReturn ret; |
| GstCurlHttpSrc *source = GST_CURLHTTPSRC (element); |
| GSTCURL_FUNCTION_ENTRY (source); |
| |
| switch (transition) { |
| case GST_STATE_CHANGE_NULL_TO_READY: |
| gst_curl_http_src_ref_multi (source); |
| break; |
| case GST_STATE_CHANGE_READY_TO_PAUSED: |
| if (source->uri == NULL) { |
| GST_ELEMENT_ERROR (element, RESOURCE, OPEN_READ, (_("No URL set.")), |
| ("Missing URL")); |
| return GST_STATE_CHANGE_FAILURE; |
| } |
| break; |
| case GST_STATE_CHANGE_READY_TO_NULL: |
| /* The pipeline has ended, so signal any running request to end. */ |
| gst_curl_http_src_request_remove (source); |
| gst_curl_http_src_unref_multi (source); |
| break; |
| default: |
| break; |
| } |
| |
| ret = GST_ELEMENT_CLASS (parent_class)->change_state (element, transition); |
| |
| GSTCURL_FUNCTION_EXIT (source); |
| return ret; |
| } |
| |
| /* |
| * Take care of any memory that may be left over from the instance that's now |
| * closing before we leak it. |
| */ |
| static void |
| gst_curl_http_src_cleanup_instance (GstCurlHttpSrc * src) |
| { |
| gint i; |
| g_mutex_lock (&src->uri_mutex); |
| g_free (src->uri); |
| src->uri = NULL; |
| g_free (src->redirect_uri); |
| src->redirect_uri = NULL; |
| g_mutex_unlock (&src->uri_mutex); |
| g_mutex_clear (&src->uri_mutex); |
| |
| g_free (src->proxy_uri); |
| src->proxy_uri = NULL; |
| g_free (src->no_proxy_list); |
| src->no_proxy_list = NULL; |
| g_free (src->proxy_user); |
| src->proxy_user = NULL; |
| g_free (src->proxy_pass); |
| src->proxy_pass = NULL; |
| |
| for (i = 0; i < src->number_cookies; i++) { |
| g_free (src->cookies[i]); |
| src->cookies[i] = NULL; |
| } |
| g_free (src->cookies); |
| src->cookies = NULL; |
| |
| g_mutex_clear (&src->buffer_mutex); |
| |
| g_cond_clear (&src->signal); |
| |
| g_free (src->buffer); |
| src->buffer = NULL; |
| |
| if (src->http_headers != NULL) { |
| gst_structure_free (src->http_headers); |
| src->http_headers = NULL; |
| } |
| |
| gst_curl_http_src_destroy_easy_handle (src); |
| } |
| |
| static gboolean |
| gst_curl_http_src_query (GstBaseSrc * bsrc, GstQuery * query) |
| { |
| GstCurlHttpSrc *src = GST_CURLHTTPSRC (bsrc); |
| gboolean ret; |
| GSTCURL_FUNCTION_ENTRY (src); |
| |
| switch (GST_QUERY_TYPE (query)) { |
| case GST_QUERY_URI: |
| gst_query_set_uri (query, src->uri); |
| if (src->redirect_uri != NULL) { |
| gst_query_set_uri_redirection (query, src->redirect_uri); |
| } |
| ret = TRUE; |
| break; |
| default: |
| ret = GST_BASE_SRC_CLASS (parent_class)->query (bsrc, query); |
| break; |
| } |
| |
| GSTCURL_FUNCTION_EXIT (src); |
| return ret; |
| } |
| |
| static gboolean |
| gst_curl_http_src_get_content_length (GstBaseSrc * bsrc, guint64 * size) |
| { |
| GstCurlHttpSrc *src = GST_CURLHTTPSRC (bsrc); |
| const GValue *response_headers; |
| gboolean ret = FALSE; |
| |
| if (src->http_headers == NULL) { |
| return FALSE; |
| } |
| |
| response_headers = gst_structure_get_value (src->http_headers, |
| RESPONSE_HEADERS_NAME); |
| if (gst_structure_has_field (gst_value_get_structure (response_headers), |
| "content-length") == TRUE) { |
| const GValue *content_length = |
| gst_structure_get_value (gst_value_get_structure (response_headers), |
| "content-length"); |
| if (G_VALUE_HOLDS_STRING (content_length) == TRUE) { |
| const gchar *len = g_value_get_string (content_length); |
| *size = (guint64) g_ascii_strtoull (len, NULL, 10); |
| ret = TRUE; |
| } else { |
| GST_ERROR_OBJECT (src, "Content Length doesn't contain expected string"); |
| } |
| } |
| |
| GST_DEBUG_OBJECT (src, |
| "No content length has yet been set, or there was an error!"); |
| return ret; |
| } |
| |
| static void |
| gst_curl_http_src_uri_handler_init (gpointer g_iface, gpointer iface_data) |
| { |
| GstURIHandlerInterface *uri_iface = (GstURIHandlerInterface *) g_iface; |
| |
| uri_iface->get_type = gst_curl_http_src_urihandler_get_type; |
| uri_iface->get_protocols = gst_curl_http_src_urihandler_get_protocols; |
| uri_iface->get_uri = gst_curl_http_src_urihandler_get_uri; |
| uri_iface->set_uri = gst_curl_http_src_urihandler_set_uri; |
| } |
| |
| static guint |
| gst_curl_http_src_urihandler_get_type (GType type) |
| { |
| return GST_URI_SRC; |
| } |
| |
| static const gchar *const * |
| gst_curl_http_src_urihandler_get_protocols (GType type) |
| { |
| static const gchar *protocols[] = { "http", "https", NULL }; |
| |
| return protocols; |
| } |
| |
| static gchar * |
| gst_curl_http_src_urihandler_get_uri (GstURIHandler * handler) |
| { |
| gchar *ret; |
| GstCurlHttpSrc *source; |
| |
| g_return_val_if_fail (GST_IS_URI_HANDLER (handler), NULL); |
| source = GST_CURLHTTPSRC (handler); |
| |
| GSTCURL_FUNCTION_ENTRY (source); |
| |
| g_mutex_lock (&source->uri_mutex); |
| ret = g_strdup (source->uri); |
| g_mutex_unlock (&source->uri_mutex); |
| |
| GSTCURL_FUNCTION_EXIT (source); |
| return ret; |
| } |
| |
| static gboolean |
| gst_curl_http_src_urihandler_set_uri (GstURIHandler * handler, |
| const gchar * uri, GError ** error) |
| { |
| GstCurlHttpSrc *source = GST_CURLHTTPSRC (handler); |
| GSTCURL_FUNCTION_ENTRY (source); |
| |
| g_return_val_if_fail (GST_IS_URI_HANDLER (handler), FALSE); |
| g_return_val_if_fail (uri != NULL, FALSE); |
| |
| g_mutex_lock (&source->uri_mutex); |
| |
| if (source->uri != NULL) { |
| GST_DEBUG_OBJECT (source, |
| "URI already present as %s, updating to new URI %s", source->uri, uri); |
| g_free (source->uri); |
| } |
| |
| source->uri = g_strdup (uri); |
| if (source->uri == NULL) { |
| return FALSE; |
| } |
| source->retries_remaining = source->total_retries; |
| |
| g_mutex_unlock (&source->uri_mutex); |
| |
| GSTCURL_FUNCTION_EXIT (source); |
| return TRUE; |
| } |
| |
| /* |
| * Cancel any currently running transfer, and then signal all the loops to drop |
| * any received buffers. The ::create() method should return GST_FLOW_FLUSHING. |
| */ |
| static gboolean |
| gst_curl_http_src_unlock (GstBaseSrc * bsrc) |
| { |
| GstCurlHttpSrc *src = GST_CURLHTTPSRC (bsrc); |
| |
| g_mutex_lock (&src->buffer_mutex); |
| if (src->state != GSTCURL_UNLOCK) { |
| if (src->state == GSTCURL_OK) { |
| /* A transfer is running, cancel it */ |
| gst_curl_http_src_request_remove (src); |
| } |
| src->pending_state = src->state; |
| src->state = GSTCURL_UNLOCK; |
| } |
| g_cond_signal (&src->signal); |
| g_mutex_unlock (&src->buffer_mutex); |
| |
| return TRUE; |
| } |
| |
| /* |
| * Finish the unlock request above and return curlhttpsrc to the normal state. |
| * This will probably be GSTCURL_DONE, and the next return from ::create() will |
| * be GST_FLOW_EOS as we don't want to deliver parts of a HTTP body. |
| */ |
| static gboolean |
| gst_curl_http_src_unlock_stop (GstBaseSrc * bsrc) |
| { |
| GstCurlHttpSrc *src = GST_CURLHTTPSRC (bsrc); |
| |
| g_mutex_lock (&src->buffer_mutex); |
| src->state = src->pending_state; |
| src->pending_state = GSTCURL_NONE; |
| g_cond_signal (&src->signal); |
| g_mutex_unlock (&src->buffer_mutex); |
| |
| return TRUE; |
| } |
| |
| /***************************************************************************** |
| * Curl loop task functions begin |
| *****************************************************************************/ |
| static void |
| gst_curl_http_src_curl_multi_loop (gpointer thread_data) |
| { |
| GstCurlHttpSrcMultiTaskContext *context; |
| GstCurlHttpSrcQueueElement *qelement; |
| int i, still_running; |
| gboolean cond = FALSE; |
| CURLMsg *curl_message; |
| |
| context = (GstCurlHttpSrcMultiTaskContext *) thread_data; |
| |
| g_mutex_lock (&context->mutex); |
| |
| /* Someone is holding a reference to us, but isn't using us so to avoid |
| * unnecessary clock cycle wasting, sit in a conditional wait until woken. |
| */ |
| while (context->state == GSTCURL_MULTI_LOOP_STATE_WAIT) { |
| GSTCURL_DEBUG_PRINT ("Entering wait state..."); |
| g_cond_wait (&context->signal, &context->mutex); |
| GSTCURL_DEBUG_PRINT ("Received wake up call!"); |
| } |
| |
| if (context->state == GSTCURL_MULTI_LOOP_STATE_QUEUE_EVENT) { |
| GSTCURL_DEBUG_PRINT ("Received a new item on the queue!"); |
| if (context->queue == NULL) { |
| GSTCURL_ERROR_PRINT ("Request Queue was empty on a Queue Event!"); |
| context->state = GSTCURL_MULTI_LOOP_STATE_WAIT; |
| return; |
| } |
| |
| /* |
| * Use the running mutex to lock access to each element, as the |
| * mutex's memory barriers stop cache optimisations from meaning |
| * flag values can't be trusted. The trylock will only let us in |
| * once and should fail immediately prior. |
| */ |
| qelement = context->queue; |
| while (qelement != NULL) { |
| if (g_mutex_trylock (&qelement->running) == TRUE) { |
| GSTCURL_DEBUG_PRINT ("Adding easy handle for URI %s", qelement->p->uri); |
| cond = TRUE; |
| curl_multi_add_handle (context->multi_handle, qelement->p->curl_handle); |
| } |
| qelement = qelement->next; |
| } |
| |
| if (cond != TRUE) { |
| GSTCURL_WARNING_PRINT ("All curl handles already added for QUEUE_EVENT!"); |
| } else { |
| GSTCURL_DEBUG_PRINT ("Finished adding all handles, continuing."); |
| context->state = GSTCURL_MULTI_LOOP_STATE_RUNNING; |
| } |
| g_mutex_unlock (&context->mutex); |
| } else if (context->state == GSTCURL_MULTI_LOOP_STATE_RUNNING) { |
| struct timeval timeout; |
| gint rc; |
| fd_set fdread, fdwrite, fdexcep; |
| int maxfd = -1; |
| long curl_timeo = -1; |
| |
| /* Because curl can possibly take some time here, be nice and let go of the |
| * mutex so other threads can perform state/queue operations as we don't |
| * care about those until the end of this. */ |
| g_mutex_unlock (&context->mutex); |
| |
| FD_ZERO (&fdread); |
| FD_ZERO (&fdwrite); |
| FD_ZERO (&fdexcep); |
| |
| timeout.tv_sec = 1; |
| timeout.tv_usec = 0; |
| |
| curl_multi_timeout (context->multi_handle, &curl_timeo); |
| if (curl_timeo >= 0) { |
| timeout.tv_sec = curl_timeo / 1000; |
| if (timeout.tv_sec > 1) { |
| timeout.tv_sec = 1; |
| } else { |
| timeout.tv_usec = (curl_timeo % 1000) * 1000; |
| } |
| } |
| |
| /* get file descriptors from the transfers */ |
| curl_multi_fdset (context->multi_handle, &fdread, &fdwrite, &fdexcep, |
| &maxfd); |
| |
| rc = select (maxfd + 1, &fdread, &fdwrite, &fdexcep, &timeout); |
| |
| switch (rc) { |
| case -1: |
| /* select error */ |
| break; |
| case 0: |
| default: |
| /* timeout or readable/writable sockets */ |
| curl_multi_perform (context->multi_handle, &still_running); |
| break; |
| } |
| |
| /* |
| * Check the CURL message buffer to find out if any transfers have |
| * completed. If they have, call the signal_finished function which |
| * will signal the g_cond_wait call in that calling instance. |
| */ |
| i = 0; |
| while (cond != TRUE) { |
| curl_message = curl_multi_info_read (context->multi_handle, &i); |
| if (curl_message == NULL) { |
| cond = TRUE; |
| } else if (curl_message->msg == CURLMSG_DONE) { |
| /* A hack, but I have seen curl_message->easy_handle being |
| * NULL randomly, so check for that. */ |
| g_mutex_lock (&context->mutex); |
| if (curl_message->easy_handle == NULL) { |
| break; |
| } |
| curl_multi_remove_handle (context->multi_handle, |
| curl_message->easy_handle); |
| gst_curl_http_src_remove_queue_handle (&context->queue, |
| curl_message->easy_handle, curl_message->data.result); |
| g_mutex_unlock (&context->mutex); |
| } |
| } |
| |
| if (still_running == 0) { |
| /* We've finished processing, so set the state to wait. |
| * |
| * This is a little more complex, as we need to catch the edge |
| * case of another thread adding a queue item while we've been |
| * working. |
| */ |
| g_mutex_lock (&context->mutex); |
| if ((context->state != GSTCURL_MULTI_LOOP_STATE_QUEUE_EVENT) && |
| (context->state != GSTCURL_MULTI_LOOP_STATE_REQUEST_REMOVAL)) { |
| context->state = GSTCURL_MULTI_LOOP_STATE_WAIT; |
| } |
| g_mutex_unlock (&context->mutex); |
| } |
| } |
| /* Is the following even necessary any more...? */ |
| else if (context->state == GSTCURL_MULTI_LOOP_STATE_STOP) { |
| g_mutex_unlock (&context->mutex); |
| /* Something wants us to shut down, so best to do a full cleanup as it |
| * might be that something's gone bang. |
| */ |
| /*gst_curl_http_src_unref_multi (NULL, GSTCURL_RETURN_PIPELINE_NULL, TRUE); */ |
| GSTCURL_INFO_PRINT ("Got instruction to shut down"); |
| } else if (context->state == GSTCURL_MULTI_LOOP_STATE_REQUEST_REMOVAL) { |
| qelement = context->queue; |
| while (qelement != NULL) { |
| if (qelement->p == context->request_removal_element) { |
| g_mutex_lock (&qelement->p->buffer_mutex); |
| curl_multi_remove_handle (context->multi_handle, |
| context->request_removal_element->curl_handle); |
| if (qelement->p->state == GSTCURL_UNLOCK) { |
| qelement->p->pending_state = GSTCURL_REMOVED; |
| } else { |
| qelement->p->state = GSTCURL_REMOVED; |
| } |
| g_cond_signal (&qelement->p->signal); |
| g_mutex_unlock (&qelement->p->buffer_mutex); |
| gst_curl_http_src_remove_queue_item (&context->queue, qelement->p); |
| } |
| } |
| context->request_removal_element = NULL; |
| g_mutex_unlock (&context->mutex); |
| } else { |
| GSTCURL_WARNING_PRINT ("Curl Loop State was invalid or unsupported"); |
| GSTCURL_WARNING_PRINT ("Signal State is %d, resetting to RUNNING.", |
| context->state); |
| /* Reset to running, so if there isn't anything to do it'll be |
| * changed the WAIT once curl_multi_perform says it has no active |
| * handles. */ |
| context->state = GSTCURL_MULTI_LOOP_STATE_RUNNING; |
| g_mutex_unlock (&context->mutex); |
| } |
| } |
| |
| /* |
| * Receive headers from the remote server and put them into the http_headers |
| * structure to be sent downstream when we've got them all and started receiving |
| * the body (see ::_handle_response()) |
| */ |
| static size_t |
| gst_curl_http_src_get_header (void *header, size_t size, size_t nmemb, |
| void *src) |
| { |
| GstCurlHttpSrc *s = src; |
| char *substr; |
| |
| GST_DEBUG_OBJECT (s, "Received header: %s", (char *) header); |
| |
| g_mutex_lock (&s->buffer_mutex); |
| |
| if (s->state == GSTCURL_UNLOCK) { |
| g_mutex_unlock (&s->buffer_mutex); |
| return size * nmemb; |
| } |
| |
| if (s->http_headers == NULL) { |
| /* Can't do anything here, so just silently swallow the header */ |
| GST_DEBUG_OBJECT (s, "HTTP Headers Structure has already been sent," |
| " ignoring header"); |
| g_mutex_unlock (&s->buffer_mutex); |
| return size * nmemb; |
| } |
| |
| substr = gst_curl_http_src_strcasestr (header, "HTTP"); |
| if (substr == header) { |
| /* We have a status line! */ |
| gchar **status_line_fields; |
| |
| /* Have we already seen a status line? If so, delete any response headers */ |
| if (s->status_code > 0) { |
| GstStructure *empty_headers = |
| gst_structure_new_empty (RESPONSE_HEADERS_NAME); |
| gst_structure_remove_field (s->http_headers, RESPONSE_HEADERS_NAME); |
| gst_structure_set (s->http_headers, RESPONSE_HEADERS_NAME, |
| GST_TYPE_STRUCTURE, empty_headers, NULL); |
| gst_structure_free (empty_headers); |
| |
| } |
| |
| /* Process the status line */ |
| status_line_fields = g_strsplit ((gchar *) header, " ", 3); |
| if (status_line_fields == NULL) { |
| GST_ERROR_OBJECT (s, "Status line processing failed!"); |
| } else { |
| s->status_code = |
| (guint) g_ascii_strtoll (status_line_fields[1], NULL, 10); |
| GST_INFO_OBJECT (s, "Received status %u for request for URI %s: %s", |
| s->status_code, s->uri, status_line_fields[2]); |
| gst_structure_set (s->http_headers, HTTP_STATUS_CODE, |
| G_TYPE_UINT, s->status_code, NULL); |
| g_strfreev (status_line_fields); |
| } |
| } else { |
| /* Normal header line */ |
| gchar **header_tpl = g_strsplit ((gchar *) header, ": ", 2); |
| if (header_tpl == NULL) { |
| GST_ERROR_OBJECT (s, "Header processing failed! (%s)", (gchar *) header); |
| } else { |
| const GValue *gv_resp_hdrs = gst_structure_get_value (s->http_headers, |
| RESPONSE_HEADERS_NAME); |
| const GstStructure *response_headers = |
| gst_value_get_structure (gv_resp_hdrs); |
| /* Store header key lower case (g_ascii_strdown), makes searching through |
| * later on easier - end applications shouldn't care, as all HTTP headers |
| * are case-insensitive */ |
| gchar *header_key = g_ascii_strdown (header_tpl[0], -1); |
| gchar *header_value; |
| |
| /* If header field already exists, append to the end */ |
| if (gst_structure_has_field (response_headers, header_key) == TRUE) { |
| header_value = g_strdup_printf ("%s, %s", |
| g_value_get_string (gst_structure_get_value (response_headers, |
| header_key)), header_tpl[1]); |
| gst_structure_set ((GstStructure *) response_headers, header_key, |
| G_TYPE_STRING, header_value, NULL); |
| g_free (header_value); |
| } else { |
| header_value = header_tpl[1]; |
| gst_structure_set ((GstStructure *) response_headers, header_key, |
| G_TYPE_STRING, header_value, NULL); |
| } |
| |
| /* We have some special cases - deal with them here */ |
| if (g_strcmp0 (header_key, "content-type") == 0) { |
| gst_curl_http_src_negotiate_caps (src); |
| } |
| |
| g_free (header_key); |
| g_strfreev (header_tpl); |
| } |
| } |
| |
| s->hdrs_updated = TRUE; |
| |
| g_mutex_unlock (&s->buffer_mutex); |
| |
| return size * nmemb; |
| } |
| |
| /* |
| * My own quick and dirty implementation of strcasestr. This is a GNU extension |
| * (i.e. not portable) and not always guaranteed to be available. |
| * |
| * I know this doesn't work if the haystack and needle are the same size. But |
| * this isn't necessarily a bad thing, as the only place we currently use this |
| * is at a point where returning nothing even if a string match occurs but the |
| * needle is the same size as the haystack actually saves us time. |
| */ |
| static char * |
| gst_curl_http_src_strcasestr (const char *haystack, const char *needle) |
| { |
| int i, j, needle_len; |
| char *location; |
| |
| needle_len = (int) strlen (needle); |
| i = 0; |
| j = 0; |
| location = NULL; |
| |
| while (haystack[i] != '\0') { |
| if (j == needle_len) { |
| location = (char *) haystack + (i - j); |
| } |
| if (tolower (haystack[i]) == tolower (needle[j])) { |
| j++; |
| } else { |
| j = 0; |
| } |
| i++; |
| } |
| |
| return location; |
| } |
| |
| /* |
| * Receive chunks of the requested body and pass these back to the ::create() |
| * loop |
| */ |
| static size_t |
| gst_curl_http_src_get_chunks (void *chunk, size_t size, size_t nmemb, void *src) |
| { |
| GstCurlHttpSrc *s = src; |
| size_t chunk_len = size * nmemb; |
| GST_TRACE_OBJECT (s, |
| "Received curl chunk for URI %s of size %d", s->uri, (int) chunk_len); |
| g_mutex_lock (&s->buffer_mutex); |
| if (s->state == GSTCURL_UNLOCK) { |
| g_mutex_unlock (&s->buffer_mutex); |
| return chunk_len; |
| } |
| s->buffer = |
| g_realloc (s->buffer, (s->buffer_len + chunk_len + 1) * sizeof (char)); |
| if (s->buffer == NULL) { |
| GST_ERROR_OBJECT (s, "Realloc for cURL response message failed!\n"); |
| return 0; |
| } |
| memcpy (s->buffer + s->buffer_len, chunk, chunk_len); |
| s->buffer_len += chunk_len; |
| g_cond_signal (&s->signal); |
| g_mutex_unlock (&s->buffer_mutex); |
| return chunk_len; |
| } |
| |
| /* |
| * Request a cancellation of a currently running curl handle. |
| */ |
| static void |
| gst_curl_http_src_request_remove (GstCurlHttpSrc * src) |
| { |
| GstCurlHttpSrcClass *klass = G_TYPE_INSTANCE_GET_CLASS (src, |
| GST_TYPE_CURL_HTTP_SRC, |
| GstCurlHttpSrcClass); |
| g_mutex_lock (&klass->multi_task_context.mutex); |
| |
| klass->multi_task_context.state = GSTCURL_MULTI_LOOP_STATE_REQUEST_REMOVAL; |
| klass->multi_task_context.request_removal_element = src; |
| g_cond_signal (&klass->multi_task_context.signal); |
| g_mutex_unlock (&klass->multi_task_context.mutex); |
| } |