net/aws: Add support for specifying endpoint

Allow specifying an endpoint to be used for S3 requests. This makes
it possible to use integrations providing object storage based on S3
API like MinIO.

When the endpoint-uri property is specified, the endpoint resolver to
use will be overridden when making S3 requests.

net/aws/src/s3hlssink/imp.rs

@@ -23,8 +23,10 @@ use gst::{element_error, glib, prelude::ObjectExt, prelude::*, subclass::prelude
 use aws_sdk_s3::config;
 use aws_sdk_s3::model::ObjectCannedAcl;
 use aws_sdk_s3::types::ByteStream;
+use aws_sdk_s3::Endpoint;
 use aws_sdk_s3::{Client, Credentials, Region, RetryConfig};
 use aws_types::sdk_config::SdkConfig;
+use http::Uri;
 
 use crate::s3utils;
 
@@ -55,6 +57,7 @@ struct Settings {
     audio_sink: bool,
     video_sink: bool,
     config: Option<SdkConfig>,
+    endpoint_uri: Option<String>,
 }
 
 impl Default for Settings {
@@ -76,6 +79,7 @@ impl Default for Settings {
             audio_sink: false,
             video_sink: false,
             config: None,
+            endpoint_uri: None,
         }
     }
 }
@@ -346,11 +350,33 @@ fn s3client_from_settings(element: &super::S3HlsSink) -> Client {
         settings.config = Some(sdk_config);
     }
 
-    let sdk_config = settings.config.as_ref().unwrap();
-    let config = config::Builder::from(sdk_config)
+    let sdk_config = settings.config.as_ref().expect("SDK config must be set");
+    let endpoint_uri = match &settings.endpoint_uri {
+        Some(endpoint) => match endpoint.parse::<Uri>() {
+            Ok(uri) => Some(uri),
+            Err(e) => {
+                element_error!(
+                    element,
+                    gst::ResourceError::Settings,
+                    ["Invalid S3 endpoint uri. Error: {}", e]
+                );
+                None
+            }
+        },
+        None => None,
+    };
+
+    let config_builder = config::Builder::from(sdk_config)
         .region(settings.s3_region.clone())
-        .retry_config(RetryConfig::new().with_max_attempts(settings.retry_attempts))
-        .build();
+        .retry_config(RetryConfig::new().with_max_attempts(settings.retry_attempts));
+
+    let config = if let Some(uri) = endpoint_uri {
+        config_builder
+            .endpoint_resolver(Endpoint::mutable(uri))
+            .build()
+    } else {
+        config_builder.build()
+    };
 
     Client::from_conf(config)
 }
@@ -481,6 +507,13 @@ impl ObjectImpl for S3HlsSink {
                     DEFAULT_TIMEOUT_IN_MSECS,
                     glib::ParamFlags::READWRITE,
                 ),
+                glib::ParamSpecString::new(
+                    "endpoint-uri",
+                    "S3 endpoint URI",
+                    "The S3 endpoint URI to use",
+                    None,
+                    glib::ParamFlags::READWRITE | gst::PARAM_FLAG_MUTABLE_READY,
+                ),
             ]
         });
 
@@ -539,6 +572,11 @@ impl ObjectImpl for S3HlsSink {
                 settings.request_timeout =
                     Duration::from_millis(value.get::<u64>().expect("type checked upstream"));
             }
+            "endpoint-uri" => {
+                settings.endpoint_uri = value
+                    .get::<Option<String>>()
+                    .expect("type checked upstream");
+            }
             _ => unimplemented!(),
         }
     }
@@ -557,6 +595,7 @@ impl ObjectImpl for S3HlsSink {
             "acl" => settings.s3_acl.as_str().to_value(),
             "retry-attempts" => settings.retry_attempts.to_value(),
             "request-timeout" => (settings.request_timeout.as_millis() as u64).to_value(),
+            "endpoint-uri" => settings.endpoint_uri.to_value(),
             _ => unimplemented!(),
         }
     }
@@ -657,16 +696,9 @@ impl ObjectImpl for S3HlsSink {
                 };
 
                 let s3hlssink = element.imp();
+                let s3_client = s3client_from_settings(&element);
+
                 let settings = s3hlssink.settings.lock().unwrap();
-
-                let sdk_config = settings.config.as_ref().unwrap();
-                let config = config::Builder::from(sdk_config)
-                    .region(settings.s3_region.clone())
-                    .retry_config(RetryConfig::new().with_max_attempts(settings.retry_attempts))
-                    .build();
-
-                let s3_client = Client::from_conf(config);
-
                 let s3_bucket = settings.s3_bucket.as_ref().unwrap().clone();
                 let s3_location = args[1].get::<String>().unwrap();
 

net/aws/src/s3sink/imp.rs

@@ -17,7 +17,9 @@ use aws_sdk_s3::client::fluent_builders::{
 use aws_sdk_s3::config;
 use aws_sdk_s3::model::{CompletedMultipartUpload, CompletedPart};
 use aws_sdk_s3::types::ByteStream;
+use aws_sdk_s3::Endpoint;
 use aws_sdk_s3::{Client, Credentials, Region, RetryConfig};
+use http::Uri;
 
 use futures::future;
 use once_cell::sync::Lazy;
@@ -108,6 +110,7 @@ struct Settings {
     retry_attempts: u32,
     multipart_upload_on_error: OnError,
     request_timeout: Duration,
+    endpoint_uri: Option<String>,
 }
 
 impl Settings {
@@ -159,6 +162,7 @@ impl Default for Settings {
             retry_attempts: DEFAULT_RETRY_ATTEMPTS,
             multipart_upload_on_error: DEFAULT_MULTIPART_UPLOAD_ON_ERROR,
             request_timeout: Duration::from_millis(DEFAULT_REQUEST_TIMEOUT_MSEC),
+            endpoint_uri: None,
         }
     }
 }
@@ -496,9 +500,30 @@ impl S3Sink {
                     }
                 })?;
 
-        let config = config::Builder::from(&sdk_config)
-            .retry_config(RetryConfig::new().with_max_attempts(settings.retry_attempts))
-            .build();
+        let endpoint_uri = match &settings.endpoint_uri {
+            Some(endpoint) => match endpoint.parse::<Uri>() {
+                Ok(uri) => Some(uri),
+                Err(e) => {
+                    return Err(gst::error_msg!(
+                        gst::ResourceError::Settings,
+                        ["Invalid S3 endpoint uri. Error: {}", e]
+                    ));
+                }
+            },
+            None => None,
+        };
+
+        let config_builder = config::Builder::from(&sdk_config)
+            .retry_config(RetryConfig::new().with_max_attempts(settings.retry_attempts));
+
+        let config = if let Some(uri) = endpoint_uri {
+            config_builder
+                .endpoint_resolver(Endpoint::mutable(uri))
+                .build()
+        } else {
+            config_builder.build()
+        };
+
         let client = Client::from_conf(config);
 
         let create_multipart_req =
@@ -768,6 +793,13 @@ impl ObjectImpl for S3Sink {
                     DEFAULT_COMPLETE_RETRY_DURATION_MSEC as i64,
                     glib::ParamFlags::READWRITE,
                 ),
+                glib::ParamSpecString::new(
+                    "endpoint-uri",
+                    "S3 endpoint URI",
+                    "The S3 endpoint URI to use",
+                    None,
+                    glib::ParamFlags::READWRITE,
+                ),
             ]
         });
 
@@ -869,6 +901,14 @@ impl ObjectImpl for S3Sink {
             "upload-part-retry-duration" | "complete-upload-retry-duration" => {
                 gst::warning!(CAT, "Use retry-attempts. retry/upload-part/complete-upload-retry duration are deprecated.");
             }
+            "endpoint-uri" => {
+                settings.endpoint_uri = value
+                    .get::<Option<String>>()
+                    .expect("type checked upstream");
+                if settings.key.is_some() && settings.bucket.is_some() {
+                    let _ = self.set_uri(obj, Some(&settings.to_uri()));
+                }
+            }
             _ => unimplemented!(),
         }
     }
@@ -907,6 +947,7 @@ impl ObjectImpl for S3Sink {
                 let request_timeout = duration_to_millis(Some(settings.request_timeout));
                 (settings.retry_attempts as i64 * request_timeout).to_value()
             }
+            "endpoint-uri" => settings.endpoint_uri.to_value(),
             _ => unimplemented!(),
         }
     }

net/aws/src/s3src/imp.rs

@@ -13,7 +13,9 @@ use std::sync::Mutex;
 use std::time::Duration;
 
 use aws_sdk_s3::config;
+use aws_sdk_s3::Endpoint;
 use aws_sdk_s3::{Client, Credentials, RetryConfig};
+use http::Uri;
 
 use gst::glib;
 use gst::prelude::*;
@@ -53,6 +55,7 @@ struct Settings {
     session_token: Option<String>,
     retry_attempts: u32,
     request_timeout: Duration,
+    endpoint_uri: Option<String>,
 }
 
 impl Default for Settings {
@@ -65,6 +68,7 @@ impl Default for Settings {
             session_token: None,
             retry_attempts: DEFAULT_RETRY_ATTEMPTS,
             request_timeout: duration,
+            endpoint_uri: None,
         }
     }
 }
@@ -126,9 +130,29 @@ impl S3Src {
                     }
                 })?;
 
-        let config = config::Builder::from(&sdk_config)
-            .retry_config(RetryConfig::new().with_max_attempts(settings.retry_attempts))
-            .build();
+        let endpoint_uri = match &settings.endpoint_uri {
+            Some(endpoint) => match endpoint.parse::<Uri>() {
+                Ok(uri) => Some(uri),
+                Err(e) => {
+                    return Err(gst::error_msg!(
+                        gst::ResourceError::Settings,
+                        ["Invalid S3 endpoint uri. Error: {}", e]
+                    ));
+                }
+            },
+            None => None,
+        };
+
+        let config_builder = config::Builder::from(&sdk_config)
+            .retry_config(RetryConfig::new().with_max_attempts(settings.retry_attempts));
+
+        let config = if let Some(uri) = endpoint_uri {
+            config_builder
+                .endpoint_resolver(Endpoint::mutable(uri))
+                .build()
+        } else {
+            config_builder.build()
+        };
 
         Ok(Client::from_conf(config))
     }
@@ -328,6 +352,13 @@ impl ObjectImpl for S3Src {
                     DEFAULT_RETRY_ATTEMPTS,
                     glib::ParamFlags::READWRITE,
                 ),
+                glib::ParamSpecString::new(
+                    "endpoint-uri",
+                    "S3 endpoint URI",
+                    "The S3 endpoint URI to use",
+                    None,
+                    glib::ParamFlags::READWRITE,
+                ),
             ]
         });
 
@@ -378,6 +409,11 @@ impl ObjectImpl for S3Src {
             "retry-attempts" => {
                 settings.retry_attempts = value.get::<u32>().expect("type checked upstream");
             }
+            "endpoint-uri" => {
+                settings.endpoint_uri = value
+                    .get::<Option<String>>()
+                    .expect("type checked upstream");
+            }
             _ => unimplemented!(),
         }
     }
@@ -403,6 +439,7 @@ impl ObjectImpl for S3Src {
                 (settings.retry_attempts as i64 * request_timeout).to_value()
             }
             "retry-attempts" => settings.retry_attempts.to_value(),
+            "endpoint-uri" => settings.endpoint_uri.to_value(),
             _ => unimplemented!(),
         }
     }