2019-06-12 08:07:39 +00:00
|
|
|
// Copyright (C) 2019 Amazon.com, Inc. or its affiliates <mkolny@amazon.com>
|
|
|
|
//
|
|
|
|
// Licensed under the Apache License, Version 2.0 <LICENSE-APACHE or
|
|
|
|
// http://www.apache.org/licenses/LICENSE-2.0> or the MIT license
|
|
|
|
// <LICENSE-MIT or http://opensource.org/licenses/MIT>, at your
|
|
|
|
// option. This file may not be copied, modified, or distributed
|
|
|
|
// except according to those terms.
|
|
|
|
|
2021-06-03 18:20:54 +00:00
|
|
|
use gst::glib;
|
2019-06-12 08:07:39 +00:00
|
|
|
use gst::prelude::*;
|
|
|
|
use gst::subclass::prelude::*;
|
2021-11-22 09:12:51 +00:00
|
|
|
use gst::{gst_debug, gst_error, gst_info, gst_log, gst_trace, gst_warning};
|
2019-06-12 08:07:39 +00:00
|
|
|
|
|
|
|
use gst_base::subclass::prelude::*;
|
|
|
|
|
2020-04-20 01:28:30 +00:00
|
|
|
use futures::future;
|
2021-09-27 13:49:12 +00:00
|
|
|
use rusoto_core::{region::Region, request::HttpClient};
|
|
|
|
use rusoto_credential::StaticProvider;
|
2019-06-12 08:07:39 +00:00
|
|
|
use rusoto_s3::{
|
2021-12-06 12:06:10 +00:00
|
|
|
AbortMultipartUploadRequest, CompleteMultipartUploadRequest, CompletedMultipartUpload,
|
|
|
|
CompletedPart, CreateMultipartUploadRequest, S3Client, UploadPartRequest, S3,
|
2019-06-12 08:07:39 +00:00
|
|
|
};
|
|
|
|
|
2020-11-22 15:43:59 +00:00
|
|
|
use once_cell::sync::Lazy;
|
|
|
|
|
2021-11-22 09:12:51 +00:00
|
|
|
use std::collections::HashMap;
|
2019-06-12 08:07:39 +00:00
|
|
|
use std::sync::Mutex;
|
|
|
|
|
2021-04-30 13:05:35 +00:00
|
|
|
use crate::s3url::*;
|
2020-04-20 01:28:30 +00:00
|
|
|
use crate::s3utils::{self, WaitError};
|
2019-06-12 08:07:39 +00:00
|
|
|
|
2021-12-06 12:06:10 +00:00
|
|
|
use super::OnError;
|
|
|
|
|
|
|
|
const DEFAULT_MULTIPART_UPLOAD_ON_ERROR: OnError = OnError::DoNothing;
|
|
|
|
|
2019-06-12 08:07:39 +00:00
|
|
|
struct Started {
|
2020-04-20 01:28:30 +00:00
|
|
|
client: S3Client,
|
2019-06-12 08:07:39 +00:00
|
|
|
buffer: Vec<u8>,
|
|
|
|
upload_id: String,
|
|
|
|
part_number: i64,
|
|
|
|
completed_parts: Vec<CompletedPart>,
|
|
|
|
}
|
|
|
|
|
|
|
|
impl Started {
|
2020-04-20 01:28:30 +00:00
|
|
|
pub fn new(client: S3Client, buffer: Vec<u8>, upload_id: String) -> Started {
|
2019-06-12 08:07:39 +00:00
|
|
|
Started {
|
2020-04-20 01:28:30 +00:00
|
|
|
client,
|
2019-06-12 08:07:39 +00:00
|
|
|
buffer,
|
|
|
|
upload_id,
|
|
|
|
part_number: 0,
|
|
|
|
completed_parts: Vec::new(),
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
pub fn increment_part_number(&mut self) -> Result<i64, gst::ErrorMessage> {
|
|
|
|
// https://docs.aws.amazon.com/AmazonS3/latest/dev/qfacts.html
|
|
|
|
const MAX_MULTIPART_NUMBER: i64 = 10000;
|
|
|
|
|
|
|
|
if self.part_number > MAX_MULTIPART_NUMBER {
|
2020-12-20 18:43:45 +00:00
|
|
|
return Err(gst::error_msg!(
|
2019-06-12 08:07:39 +00:00
|
|
|
gst::ResourceError::Failed,
|
|
|
|
[
|
|
|
|
"Maximum number of parts ({}) reached.",
|
|
|
|
MAX_MULTIPART_NUMBER
|
|
|
|
]
|
|
|
|
));
|
|
|
|
}
|
|
|
|
|
|
|
|
self.part_number += 1;
|
|
|
|
Ok(self.part_number)
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
enum State {
|
|
|
|
Stopped,
|
|
|
|
Started(Started),
|
|
|
|
}
|
|
|
|
|
|
|
|
impl Default for State {
|
|
|
|
fn default() -> State {
|
|
|
|
State::Stopped
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
const DEFAULT_BUFFER_SIZE: u64 = 5 * 1024 * 1024;
|
|
|
|
|
|
|
|
struct Settings {
|
2021-06-21 10:20:32 +00:00
|
|
|
region: Region,
|
|
|
|
bucket: Option<String>,
|
|
|
|
key: Option<String>,
|
2019-06-12 08:07:39 +00:00
|
|
|
content_type: Option<String>,
|
|
|
|
buffer_size: u64,
|
2021-09-27 13:49:12 +00:00
|
|
|
access_key: Option<String>,
|
|
|
|
secret_access_key: Option<String>,
|
2021-11-22 09:12:51 +00:00
|
|
|
metadata: Option<gst::Structure>,
|
2021-12-06 12:06:10 +00:00
|
|
|
multipart_upload_on_error: OnError,
|
2019-06-12 08:07:39 +00:00
|
|
|
}
|
|
|
|
|
2021-06-21 10:20:32 +00:00
|
|
|
impl Settings {
|
|
|
|
fn to_uri(&self) -> String {
|
|
|
|
format!(
|
|
|
|
"s3://{}/{}/{}",
|
2021-09-27 16:16:26 +00:00
|
|
|
match self.region {
|
|
|
|
Region::Custom {
|
|
|
|
ref name,
|
|
|
|
ref endpoint,
|
|
|
|
} => {
|
|
|
|
format!(
|
|
|
|
"{}+{}",
|
|
|
|
base32::encode(
|
|
|
|
base32::Alphabet::RFC4648 { padding: true },
|
|
|
|
name.as_bytes(),
|
|
|
|
),
|
|
|
|
base32::encode(
|
|
|
|
base32::Alphabet::RFC4648 { padding: true },
|
|
|
|
endpoint.as_bytes(),
|
|
|
|
),
|
|
|
|
)
|
|
|
|
}
|
|
|
|
_ => {
|
|
|
|
String::from(self.region.name())
|
|
|
|
}
|
|
|
|
},
|
2021-06-21 10:20:32 +00:00
|
|
|
self.bucket.as_ref().unwrap(),
|
|
|
|
self.key.as_ref().unwrap()
|
|
|
|
)
|
|
|
|
}
|
2021-11-22 09:12:51 +00:00
|
|
|
|
|
|
|
fn to_metadata(&self, element: &super::S3Sink) -> Option<HashMap<String, String>> {
|
|
|
|
self.metadata.as_ref().map(|structure| {
|
|
|
|
let mut hash = HashMap::new();
|
|
|
|
|
|
|
|
for (key, value) in structure.iter() {
|
|
|
|
if let Ok(Ok(value_str)) = value.transform::<String>().map(|v| v.get()) {
|
|
|
|
gst_log!(CAT, obj: element, "metadata '{}' -> '{}'", key, value_str);
|
|
|
|
hash.insert(key.to_string(), value_str);
|
|
|
|
} else {
|
|
|
|
gst_warning!(
|
|
|
|
CAT,
|
|
|
|
obj: element,
|
|
|
|
"Failed to convert metadata '{}' to string ('{:?}')",
|
|
|
|
key,
|
|
|
|
value
|
|
|
|
);
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
hash
|
|
|
|
})
|
|
|
|
}
|
2021-06-21 10:20:32 +00:00
|
|
|
}
|
|
|
|
|
|
|
|
impl Default for Settings {
|
|
|
|
fn default() -> Self {
|
|
|
|
Settings {
|
|
|
|
region: Region::default(),
|
|
|
|
bucket: None,
|
|
|
|
key: None,
|
|
|
|
content_type: None,
|
|
|
|
buffer_size: DEFAULT_BUFFER_SIZE,
|
2021-09-27 13:49:12 +00:00
|
|
|
access_key: None,
|
|
|
|
secret_access_key: None,
|
2021-11-22 09:12:51 +00:00
|
|
|
metadata: None,
|
2021-12-06 12:06:10 +00:00
|
|
|
multipart_upload_on_error: DEFAULT_MULTIPART_UPLOAD_ON_ERROR,
|
2021-06-21 10:20:32 +00:00
|
|
|
}
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
2021-03-07 16:22:24 +00:00
|
|
|
#[derive(Default)]
|
2019-06-12 08:07:39 +00:00
|
|
|
pub struct S3Sink {
|
2021-04-30 13:05:35 +00:00
|
|
|
url: Mutex<Option<GstS3Url>>,
|
2019-06-12 08:07:39 +00:00
|
|
|
settings: Mutex<Settings>,
|
|
|
|
state: Mutex<State>,
|
2020-04-20 01:28:30 +00:00
|
|
|
canceller: Mutex<Option<future::AbortHandle>>,
|
2021-12-06 12:06:10 +00:00
|
|
|
abort_multipart_canceller: Mutex<Option<future::AbortHandle>>,
|
2019-06-12 08:07:39 +00:00
|
|
|
}
|
|
|
|
|
2020-11-22 15:43:59 +00:00
|
|
|
static CAT: Lazy<gst::DebugCategory> = Lazy::new(|| {
|
|
|
|
gst::DebugCategory::new(
|
2019-10-31 22:34:21 +00:00
|
|
|
"rusotos3sink",
|
|
|
|
gst::DebugColorFlags::empty(),
|
|
|
|
Some("Amazon S3 Sink"),
|
2020-11-22 15:43:59 +00:00
|
|
|
)
|
|
|
|
});
|
2019-10-31 22:34:21 +00:00
|
|
|
|
2019-06-12 08:07:39 +00:00
|
|
|
impl S3Sink {
|
|
|
|
fn flush_current_buffer(
|
|
|
|
&self,
|
2020-11-14 17:24:01 +00:00
|
|
|
element: &super::S3Sink,
|
2019-06-12 08:07:39 +00:00
|
|
|
) -> Result<(), Option<gst::ErrorMessage>> {
|
|
|
|
let upload_part_req = self.create_upload_part_request()?;
|
|
|
|
let part_number = upload_part_req.part_number;
|
|
|
|
|
|
|
|
let mut state = self.state.lock().unwrap();
|
|
|
|
let state = match *state {
|
|
|
|
State::Started(ref mut started_state) => started_state,
|
|
|
|
State::Stopped => {
|
|
|
|
unreachable!("Element should be started");
|
|
|
|
}
|
|
|
|
};
|
2020-04-20 01:28:30 +00:00
|
|
|
|
|
|
|
let upload_part_req_future = state.client.upload_part(upload_part_req);
|
|
|
|
|
|
|
|
let output =
|
|
|
|
s3utils::wait(&self.canceller, upload_part_req_future).map_err(|err| match err {
|
2021-12-06 12:06:10 +00:00
|
|
|
WaitError::FutureError(err) => {
|
|
|
|
let settings = self.settings.lock().unwrap();
|
|
|
|
match settings.multipart_upload_on_error {
|
|
|
|
OnError::Abort => {
|
|
|
|
gst_log!(
|
|
|
|
CAT,
|
|
|
|
obj: element,
|
|
|
|
"Aborting multipart upload request with id: {}",
|
|
|
|
state.upload_id
|
|
|
|
);
|
|
|
|
match self.abort_multipart_upload_request(state) {
|
|
|
|
Ok(()) => {
|
|
|
|
gst_log!(
|
|
|
|
CAT,
|
|
|
|
obj: element,
|
|
|
|
"Aborting multipart upload request succeeded."
|
|
|
|
);
|
|
|
|
}
|
|
|
|
Err(err) => gst_error!(
|
|
|
|
CAT,
|
|
|
|
obj: element,
|
|
|
|
"Aborting multipart upload failed: {}",
|
|
|
|
err.to_string()
|
|
|
|
),
|
|
|
|
}
|
|
|
|
}
|
|
|
|
OnError::Complete => {
|
|
|
|
gst_log!(
|
|
|
|
CAT,
|
|
|
|
obj: element,
|
|
|
|
"Completing multipart upload request with id: {}",
|
|
|
|
state.upload_id
|
|
|
|
);
|
|
|
|
match self.complete_multipart_upload_request(state) {
|
|
|
|
Ok(()) => {
|
|
|
|
gst_log!(
|
|
|
|
CAT,
|
|
|
|
obj: element,
|
|
|
|
"Complete multipart upload request succeeded."
|
|
|
|
);
|
|
|
|
}
|
|
|
|
Err(err) => gst_error!(
|
|
|
|
CAT,
|
|
|
|
obj: element,
|
|
|
|
"Completing multipart upload failed: {}",
|
|
|
|
err.to_string()
|
|
|
|
),
|
|
|
|
}
|
|
|
|
}
|
|
|
|
OnError::DoNothing => (),
|
|
|
|
}
|
|
|
|
Some(gst::error_msg!(
|
|
|
|
gst::ResourceError::OpenWrite,
|
|
|
|
["Failed to upload part: {}", err]
|
|
|
|
))
|
|
|
|
}
|
2020-04-20 01:28:30 +00:00
|
|
|
WaitError::Cancelled => None,
|
|
|
|
})?;
|
|
|
|
|
2019-06-12 08:07:39 +00:00
|
|
|
state.completed_parts.push(CompletedPart {
|
2019-11-24 22:00:27 +00:00
|
|
|
e_tag: output.e_tag,
|
2019-06-12 08:07:39 +00:00
|
|
|
part_number: Some(part_number),
|
|
|
|
});
|
2019-10-31 22:34:21 +00:00
|
|
|
gst_info!(CAT, obj: element, "Uploaded part {}", part_number);
|
2019-06-12 08:07:39 +00:00
|
|
|
|
|
|
|
Ok(())
|
|
|
|
}
|
|
|
|
|
|
|
|
fn create_upload_part_request(&self) -> Result<UploadPartRequest, gst::ErrorMessage> {
|
2021-04-30 13:05:35 +00:00
|
|
|
let url = self.url.lock().unwrap();
|
2019-06-12 08:07:39 +00:00
|
|
|
let settings = self.settings.lock().unwrap();
|
|
|
|
let mut state = self.state.lock().unwrap();
|
|
|
|
let state = match *state {
|
|
|
|
State::Started(ref mut started_state) => started_state,
|
|
|
|
State::Stopped => {
|
|
|
|
unreachable!("Element should be started");
|
|
|
|
}
|
|
|
|
};
|
|
|
|
|
|
|
|
let part_number = state.increment_part_number()?;
|
|
|
|
Ok(UploadPartRequest {
|
|
|
|
body: Some(rusoto_core::ByteStream::from(std::mem::replace(
|
|
|
|
&mut state.buffer,
|
|
|
|
Vec::with_capacity(settings.buffer_size as usize),
|
|
|
|
))),
|
2021-04-30 13:05:35 +00:00
|
|
|
bucket: url.as_ref().unwrap().bucket.to_owned(),
|
|
|
|
key: url.as_ref().unwrap().object.to_owned(),
|
2019-06-12 08:07:39 +00:00
|
|
|
upload_id: state.upload_id.to_owned(),
|
|
|
|
part_number,
|
|
|
|
..Default::default()
|
|
|
|
})
|
|
|
|
}
|
|
|
|
|
2020-04-20 01:28:30 +00:00
|
|
|
fn create_complete_multipart_upload_request(
|
|
|
|
&self,
|
|
|
|
started_state: &mut Started,
|
|
|
|
) -> CompleteMultipartUploadRequest {
|
2019-06-12 08:07:39 +00:00
|
|
|
started_state
|
|
|
|
.completed_parts
|
|
|
|
.sort_by(|a, b| a.part_number.cmp(&b.part_number));
|
|
|
|
|
|
|
|
let completed_upload = CompletedMultipartUpload {
|
2021-06-04 22:16:05 +00:00
|
|
|
parts: Some(std::mem::take(&mut started_state.completed_parts)),
|
2019-06-12 08:07:39 +00:00
|
|
|
};
|
|
|
|
|
2021-04-30 13:05:35 +00:00
|
|
|
let url = self.url.lock().unwrap();
|
2019-06-12 08:07:39 +00:00
|
|
|
CompleteMultipartUploadRequest {
|
2021-04-30 13:05:35 +00:00
|
|
|
bucket: url.as_ref().unwrap().bucket.to_owned(),
|
|
|
|
key: url.as_ref().unwrap().object.to_owned(),
|
2019-06-12 08:07:39 +00:00
|
|
|
upload_id: started_state.upload_id.to_owned(),
|
|
|
|
multipart_upload: Some(completed_upload),
|
|
|
|
..Default::default()
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
fn create_create_multipart_upload_request(
|
|
|
|
&self,
|
2021-04-30 13:05:35 +00:00
|
|
|
url: &GstS3Url,
|
2020-04-20 01:28:30 +00:00
|
|
|
settings: &Settings,
|
2021-04-30 13:05:35 +00:00
|
|
|
) -> CreateMultipartUploadRequest {
|
|
|
|
CreateMultipartUploadRequest {
|
|
|
|
bucket: url.bucket.clone(),
|
|
|
|
key: url.object.clone(),
|
2019-06-12 08:07:39 +00:00
|
|
|
content_type: settings.content_type.clone(),
|
2021-11-22 09:12:51 +00:00
|
|
|
metadata: settings.to_metadata(&self.instance()),
|
2019-06-12 08:07:39 +00:00
|
|
|
..Default::default()
|
2021-04-30 13:05:35 +00:00
|
|
|
}
|
2019-06-12 08:07:39 +00:00
|
|
|
}
|
|
|
|
|
2021-12-06 12:06:10 +00:00
|
|
|
fn create_abort_multipart_upload_request(
|
|
|
|
&self,
|
|
|
|
url: &GstS3Url,
|
|
|
|
started_state: &Started,
|
|
|
|
) -> AbortMultipartUploadRequest {
|
|
|
|
AbortMultipartUploadRequest {
|
|
|
|
bucket: url.bucket.clone(),
|
|
|
|
expected_bucket_owner: None,
|
|
|
|
key: url.object.clone(),
|
|
|
|
request_payer: None,
|
|
|
|
upload_id: started_state.upload_id.to_owned(),
|
2019-06-12 08:07:39 +00:00
|
|
|
}
|
2021-12-06 12:06:10 +00:00
|
|
|
}
|
2019-06-12 08:07:39 +00:00
|
|
|
|
2021-12-06 12:06:10 +00:00
|
|
|
fn abort_multipart_upload_request(
|
|
|
|
&self,
|
|
|
|
started_state: &Started,
|
|
|
|
) -> Result<(), gst::ErrorMessage> {
|
|
|
|
let s3url = match *self.url.lock().unwrap() {
|
|
|
|
Some(ref url) => url.clone(),
|
|
|
|
None => unreachable!("Element should be started"),
|
2020-04-20 01:28:30 +00:00
|
|
|
};
|
2021-12-06 12:06:10 +00:00
|
|
|
let abort_req = self.create_abort_multipart_upload_request(&s3url, started_state);
|
|
|
|
let abort_req_future = started_state.client.abort_multipart_upload(abort_req);
|
2020-04-20 01:28:30 +00:00
|
|
|
|
2021-12-06 12:06:10 +00:00
|
|
|
s3utils::wait(&self.abort_multipart_canceller, abort_req_future)
|
|
|
|
.map(|_| ())
|
|
|
|
.map_err(|err| match err {
|
|
|
|
WaitError::FutureError(err) => {
|
|
|
|
gst::error_msg!(
|
|
|
|
gst::ResourceError::Write,
|
|
|
|
["Failed to abort multipart upload: {}.", err.to_string()]
|
|
|
|
)
|
|
|
|
}
|
|
|
|
WaitError::Cancelled => {
|
|
|
|
gst::error_msg!(
|
|
|
|
gst::ResourceError::Write,
|
|
|
|
["Abort multipart upload request interrupted."]
|
|
|
|
)
|
|
|
|
}
|
|
|
|
})
|
|
|
|
}
|
|
|
|
|
|
|
|
fn complete_multipart_upload_request(
|
|
|
|
&self,
|
|
|
|
started_state: &mut Started,
|
|
|
|
) -> Result<(), gst::ErrorMessage> {
|
2021-04-30 13:05:35 +00:00
|
|
|
let complete_req = self.create_complete_multipart_upload_request(started_state);
|
2020-04-20 01:28:30 +00:00
|
|
|
let complete_req_future = started_state.client.complete_multipart_upload(complete_req);
|
|
|
|
|
|
|
|
s3utils::wait(&self.canceller, complete_req_future)
|
|
|
|
.map(|_| ())
|
|
|
|
.map_err(|err| match err {
|
2020-12-20 18:43:45 +00:00
|
|
|
WaitError::FutureError(err) => gst::error_msg!(
|
2019-06-12 08:07:39 +00:00
|
|
|
gst::ResourceError::Write,
|
|
|
|
["Failed to complete multipart upload: {}.", err.to_string()]
|
2020-04-20 01:28:30 +00:00
|
|
|
),
|
|
|
|
WaitError::Cancelled => {
|
2020-12-20 18:43:45 +00:00
|
|
|
gst::error_msg!(gst::LibraryError::Failed, ["Interrupted during stop"])
|
2020-04-20 01:28:30 +00:00
|
|
|
}
|
2019-06-12 08:07:39 +00:00
|
|
|
})
|
|
|
|
}
|
|
|
|
|
2021-12-06 12:06:10 +00:00
|
|
|
fn finalize_upload(&self, element: &super::S3Sink) -> Result<(), gst::ErrorMessage> {
|
|
|
|
if self.flush_current_buffer(element).is_err() {
|
|
|
|
return Err(gst::error_msg!(
|
|
|
|
gst::ResourceError::Settings,
|
|
|
|
["Failed to flush internal buffer."]
|
|
|
|
));
|
|
|
|
}
|
|
|
|
|
|
|
|
let mut state = self.state.lock().unwrap();
|
|
|
|
let started_state = match *state {
|
|
|
|
State::Started(ref mut started_state) => started_state,
|
|
|
|
State::Stopped => {
|
|
|
|
unreachable!("Element should be started");
|
|
|
|
}
|
|
|
|
};
|
|
|
|
|
|
|
|
self.complete_multipart_upload_request(started_state)
|
|
|
|
}
|
|
|
|
|
2020-04-20 01:28:30 +00:00
|
|
|
fn start(&self) -> Result<(), gst::ErrorMessage> {
|
|
|
|
let mut state = self.state.lock().unwrap();
|
|
|
|
let settings = self.settings.lock().unwrap();
|
|
|
|
|
|
|
|
if let State::Started { .. } = *state {
|
|
|
|
unreachable!("Element should be started");
|
|
|
|
}
|
|
|
|
|
2021-04-30 13:05:35 +00:00
|
|
|
let s3url = match *self.url.lock().unwrap() {
|
|
|
|
Some(ref url) => url.clone(),
|
|
|
|
None => {
|
|
|
|
return Err(gst::error_msg!(
|
|
|
|
gst::ResourceError::Settings,
|
|
|
|
["Cannot start without a URL being set"]
|
|
|
|
));
|
|
|
|
}
|
|
|
|
};
|
|
|
|
|
2021-09-27 13:49:12 +00:00
|
|
|
let client = match (
|
|
|
|
settings.access_key.as_ref(),
|
|
|
|
settings.secret_access_key.as_ref(),
|
|
|
|
) {
|
|
|
|
(Some(access_key), Some(secret_access_key)) => {
|
|
|
|
let creds =
|
|
|
|
StaticProvider::new_minimal(access_key.clone(), secret_access_key.clone());
|
|
|
|
S3Client::new_with(
|
|
|
|
HttpClient::new().expect("failed to create request dispatcher"),
|
|
|
|
creds,
|
|
|
|
s3url.region.clone(),
|
|
|
|
)
|
|
|
|
}
|
|
|
|
_ => S3Client::new(s3url.region.clone()),
|
|
|
|
};
|
2020-04-20 01:28:30 +00:00
|
|
|
|
2021-04-30 13:05:35 +00:00
|
|
|
let create_multipart_req = self.create_create_multipart_upload_request(&s3url, &settings);
|
2020-04-20 01:28:30 +00:00
|
|
|
let create_multipart_req_future = client.create_multipart_upload(create_multipart_req);
|
|
|
|
|
|
|
|
let response = s3utils::wait(&self.canceller, create_multipart_req_future).map_err(
|
|
|
|
|err| match err {
|
2020-12-20 18:43:45 +00:00
|
|
|
WaitError::FutureError(err) => gst::error_msg!(
|
2019-06-12 08:07:39 +00:00
|
|
|
gst::ResourceError::OpenWrite,
|
|
|
|
["Failed to create multipart upload: {}", err]
|
2020-04-20 01:28:30 +00:00
|
|
|
),
|
|
|
|
WaitError::Cancelled => {
|
2020-12-20 18:43:45 +00:00
|
|
|
gst::error_msg!(gst::LibraryError::Failed, ["Interrupted during start"])
|
2020-04-20 01:28:30 +00:00
|
|
|
}
|
|
|
|
},
|
|
|
|
)?;
|
2019-06-12 08:07:39 +00:00
|
|
|
|
2019-07-04 15:30:26 +00:00
|
|
|
let upload_id = response.upload_id.ok_or_else(|| {
|
2020-12-20 18:43:45 +00:00
|
|
|
gst::error_msg!(
|
2019-07-04 15:30:26 +00:00
|
|
|
gst::ResourceError::Failed,
|
|
|
|
["Failed to get multipart upload ID"]
|
|
|
|
)
|
|
|
|
})?;
|
2019-06-12 08:07:39 +00:00
|
|
|
|
2020-04-20 01:28:30 +00:00
|
|
|
*state = State::Started(Started::new(
|
|
|
|
client,
|
|
|
|
Vec::with_capacity(settings.buffer_size as usize),
|
2019-06-12 08:07:39 +00:00
|
|
|
upload_id,
|
2020-04-20 01:28:30 +00:00
|
|
|
));
|
|
|
|
|
|
|
|
Ok(())
|
2019-06-12 08:07:39 +00:00
|
|
|
}
|
|
|
|
|
|
|
|
fn update_buffer(
|
|
|
|
&self,
|
|
|
|
src: &[u8],
|
2020-11-14 17:24:01 +00:00
|
|
|
element: &super::S3Sink,
|
2019-06-12 08:07:39 +00:00
|
|
|
) -> Result<(), Option<gst::ErrorMessage>> {
|
|
|
|
let mut state = self.state.lock().unwrap();
|
|
|
|
let started_state = match *state {
|
|
|
|
State::Started(ref mut started_state) => started_state,
|
|
|
|
State::Stopped => {
|
|
|
|
unreachable!("Element should be started already");
|
|
|
|
}
|
|
|
|
};
|
|
|
|
|
|
|
|
let to_copy = std::cmp::min(
|
|
|
|
started_state.buffer.capacity() - started_state.buffer.len(),
|
|
|
|
src.len(),
|
|
|
|
);
|
|
|
|
|
|
|
|
let (head, tail) = src.split_at(to_copy);
|
|
|
|
started_state.buffer.extend_from_slice(head);
|
|
|
|
let do_flush = started_state.buffer.capacity() == started_state.buffer.len();
|
|
|
|
drop(state);
|
|
|
|
|
|
|
|
if do_flush {
|
|
|
|
self.flush_current_buffer(element)?;
|
|
|
|
}
|
|
|
|
|
|
|
|
if to_copy < src.len() {
|
|
|
|
self.update_buffer(tail, element)?;
|
|
|
|
}
|
|
|
|
|
|
|
|
Ok(())
|
|
|
|
}
|
|
|
|
|
|
|
|
fn cancel(&self) {
|
|
|
|
let mut canceller = self.canceller.lock().unwrap();
|
2021-12-06 12:06:10 +00:00
|
|
|
let mut abort_canceller = self.abort_multipart_canceller.lock().unwrap();
|
|
|
|
|
|
|
|
if let Some(c) = abort_canceller.take() {
|
|
|
|
c.abort()
|
|
|
|
};
|
2019-06-12 08:07:39 +00:00
|
|
|
|
2020-04-20 01:28:30 +00:00
|
|
|
if let Some(c) = canceller.take() {
|
|
|
|
c.abort()
|
|
|
|
};
|
2019-06-12 08:07:39 +00:00
|
|
|
}
|
2021-04-30 13:05:35 +00:00
|
|
|
|
2021-06-21 10:20:32 +00:00
|
|
|
fn set_uri(
|
|
|
|
self: &S3Sink,
|
|
|
|
object: &super::S3Sink,
|
|
|
|
url_str: Option<&str>,
|
|
|
|
) -> Result<(), glib::Error> {
|
2021-04-30 13:05:35 +00:00
|
|
|
let state = self.state.lock().unwrap();
|
|
|
|
|
|
|
|
if let State::Started { .. } = *state {
|
|
|
|
return Err(glib::Error::new(
|
|
|
|
gst::URIError::BadState,
|
|
|
|
"Cannot set URI on a started s3sink",
|
|
|
|
));
|
|
|
|
}
|
|
|
|
|
|
|
|
let mut url = self.url.lock().unwrap();
|
|
|
|
|
|
|
|
if url_str.is_none() {
|
|
|
|
*url = None;
|
|
|
|
return Ok(());
|
|
|
|
}
|
|
|
|
|
2021-06-21 10:20:32 +00:00
|
|
|
gst_debug!(CAT, obj: object, "Setting uri to {:?}", url_str);
|
|
|
|
|
2021-04-30 13:05:35 +00:00
|
|
|
let url_str = url_str.unwrap();
|
|
|
|
match parse_s3_url(url_str) {
|
|
|
|
Ok(s3url) => {
|
|
|
|
*url = Some(s3url);
|
|
|
|
Ok(())
|
|
|
|
}
|
|
|
|
Err(_) => Err(glib::Error::new(
|
|
|
|
gst::URIError::BadUri,
|
|
|
|
"Could not parse URI",
|
|
|
|
)),
|
|
|
|
}
|
|
|
|
}
|
2019-06-12 08:07:39 +00:00
|
|
|
}
|
|
|
|
|
2021-03-07 16:22:24 +00:00
|
|
|
#[glib::object_subclass]
|
2019-06-12 08:07:39 +00:00
|
|
|
impl ObjectSubclass for S3Sink {
|
2019-10-21 07:31:03 +00:00
|
|
|
const NAME: &'static str = "RusotoS3Sink";
|
2020-11-14 17:24:01 +00:00
|
|
|
type Type = super::S3Sink;
|
2019-06-12 08:07:39 +00:00
|
|
|
type ParentType = gst_base::BaseSink;
|
2021-04-30 13:05:35 +00:00
|
|
|
type Interfaces = (gst::URIHandler,);
|
2021-01-21 18:21:29 +00:00
|
|
|
}
|
2019-06-12 08:07:39 +00:00
|
|
|
|
2021-01-21 18:21:29 +00:00
|
|
|
impl ObjectImpl for S3Sink {
|
|
|
|
fn properties() -> &'static [glib::ParamSpec] {
|
|
|
|
static PROPERTIES: Lazy<Vec<glib::ParamSpec>> = Lazy::new(|| {
|
|
|
|
vec![
|
2021-11-20 10:25:14 +00:00
|
|
|
glib::ParamSpecString::new(
|
2021-06-21 10:20:32 +00:00
|
|
|
"bucket",
|
|
|
|
"S3 Bucket",
|
|
|
|
"The bucket of the file to write",
|
|
|
|
None,
|
|
|
|
glib::ParamFlags::READWRITE | gst::PARAM_FLAG_MUTABLE_READY,
|
|
|
|
),
|
2021-11-20 10:25:14 +00:00
|
|
|
glib::ParamSpecString::new(
|
2021-06-21 10:20:32 +00:00
|
|
|
"key",
|
|
|
|
"S3 Key",
|
|
|
|
"The key of the file to write",
|
|
|
|
None,
|
|
|
|
glib::ParamFlags::READWRITE | gst::PARAM_FLAG_MUTABLE_READY,
|
|
|
|
),
|
2021-11-20 10:25:14 +00:00
|
|
|
glib::ParamSpecString::new(
|
2021-06-21 10:20:32 +00:00
|
|
|
"region",
|
|
|
|
"AWS Region",
|
|
|
|
"An AWS region (e.g. eu-west-2).",
|
|
|
|
None,
|
|
|
|
glib::ParamFlags::READWRITE | gst::PARAM_FLAG_MUTABLE_READY,
|
|
|
|
),
|
2021-11-20 10:25:14 +00:00
|
|
|
glib::ParamSpecUInt64::new(
|
2021-01-21 18:21:29 +00:00
|
|
|
"part-size",
|
|
|
|
"Part size",
|
|
|
|
"A size (in bytes) of an individual part used for multipart upload.",
|
|
|
|
5 * 1024 * 1024, // 5 MB
|
|
|
|
5 * 1024 * 1024 * 1024, // 5 GB
|
|
|
|
DEFAULT_BUFFER_SIZE,
|
2021-01-31 12:44:45 +00:00
|
|
|
glib::ParamFlags::READWRITE | gst::PARAM_FLAG_MUTABLE_READY,
|
2021-01-21 18:21:29 +00:00
|
|
|
),
|
2021-11-20 10:25:14 +00:00
|
|
|
glib::ParamSpecString::new(
|
2021-04-30 13:05:35 +00:00
|
|
|
"uri",
|
|
|
|
"URI",
|
|
|
|
"The S3 object URI",
|
|
|
|
None,
|
|
|
|
glib::ParamFlags::READWRITE | gst::PARAM_FLAG_MUTABLE_READY,
|
|
|
|
),
|
2021-11-20 10:25:14 +00:00
|
|
|
glib::ParamSpecString::new(
|
2021-09-27 13:49:12 +00:00
|
|
|
"access-key",
|
|
|
|
"Access Key",
|
|
|
|
"AWS Access Key",
|
|
|
|
None,
|
|
|
|
glib::ParamFlags::READWRITE | gst::PARAM_FLAG_MUTABLE_READY,
|
|
|
|
),
|
2021-11-20 10:25:14 +00:00
|
|
|
glib::ParamSpecString::new(
|
2021-09-27 13:49:12 +00:00
|
|
|
"secret-access-key",
|
|
|
|
"Secret Access Key",
|
|
|
|
"AWS Secret Access Key",
|
|
|
|
None,
|
|
|
|
glib::ParamFlags::READWRITE | gst::PARAM_FLAG_MUTABLE_READY,
|
|
|
|
),
|
2021-11-22 09:12:51 +00:00
|
|
|
glib::ParamSpecBoxed::new(
|
|
|
|
"metadata",
|
|
|
|
"Metadata",
|
|
|
|
"A map of metadata to store with the object in S3; field values need to be convertible to strings.",
|
|
|
|
gst::Structure::static_type(),
|
|
|
|
glib::ParamFlags::READWRITE | gst::PARAM_FLAG_MUTABLE_READY,
|
|
|
|
),
|
2021-12-06 12:06:10 +00:00
|
|
|
glib::ParamSpecEnum::new(
|
|
|
|
"on-error",
|
|
|
|
"Whether to upload or complete the multipart upload on error",
|
|
|
|
"Do nothing, abort or complete a multipart upload request on error",
|
|
|
|
OnError::static_type(),
|
|
|
|
DEFAULT_MULTIPART_UPLOAD_ON_ERROR as i32,
|
|
|
|
glib::ParamFlags::READWRITE | gst::PARAM_FLAG_MUTABLE_READY,
|
|
|
|
),
|
2021-01-21 18:21:29 +00:00
|
|
|
]
|
|
|
|
});
|
2019-06-12 08:07:39 +00:00
|
|
|
|
2021-01-21 18:21:29 +00:00
|
|
|
PROPERTIES.as_ref()
|
2019-06-12 08:07:39 +00:00
|
|
|
}
|
|
|
|
|
2021-01-21 18:21:29 +00:00
|
|
|
fn set_property(
|
|
|
|
&self,
|
2021-04-30 13:05:35 +00:00
|
|
|
obj: &Self::Type,
|
2021-01-21 18:21:29 +00:00
|
|
|
_id: usize,
|
|
|
|
value: &glib::Value,
|
|
|
|
pspec: &glib::ParamSpec,
|
|
|
|
) {
|
2019-06-12 08:07:39 +00:00
|
|
|
let mut settings = self.settings.lock().unwrap();
|
|
|
|
|
2021-11-22 10:28:46 +00:00
|
|
|
gst_debug!(
|
|
|
|
CAT,
|
|
|
|
obj: obj,
|
|
|
|
"Setting property '{}' to '{:?}'",
|
|
|
|
pspec.name(),
|
|
|
|
value
|
|
|
|
);
|
|
|
|
|
2021-04-12 12:49:54 +00:00
|
|
|
match pspec.name() {
|
2021-06-21 10:20:32 +00:00
|
|
|
"bucket" => {
|
|
|
|
settings.bucket = value
|
|
|
|
.get::<Option<String>>()
|
|
|
|
.expect("type checked upstream");
|
|
|
|
if settings.key.is_some() {
|
|
|
|
let _ = self.set_uri(obj, Some(&settings.to_uri()));
|
|
|
|
}
|
|
|
|
}
|
|
|
|
"key" => {
|
|
|
|
settings.key = value
|
|
|
|
.get::<Option<String>>()
|
|
|
|
.expect("type checked upstream");
|
|
|
|
if settings.bucket.is_some() {
|
|
|
|
let _ = self.set_uri(obj, Some(&settings.to_uri()));
|
|
|
|
}
|
|
|
|
}
|
|
|
|
"region" => {
|
2021-09-27 16:16:26 +00:00
|
|
|
let region = value.get::<String>().expect("type checked upstream");
|
2021-10-31 10:54:12 +00:00
|
|
|
settings.region = region
|
|
|
|
.parse::<Region>()
|
2021-09-27 16:16:26 +00:00
|
|
|
.or_else(|_| {
|
|
|
|
let (name, endpoint) = region.split_once('+').ok_or(())?;
|
|
|
|
Ok(Region::Custom {
|
|
|
|
name: name.into(),
|
|
|
|
endpoint: endpoint.into(),
|
|
|
|
})
|
|
|
|
})
|
|
|
|
.unwrap_or_else(|_: ()| panic!("Invalid region '{}'", region));
|
|
|
|
|
2021-06-21 10:20:32 +00:00
|
|
|
if settings.key.is_some() && settings.bucket.is_some() {
|
|
|
|
let _ = self.set_uri(obj, Some(&settings.to_uri()));
|
|
|
|
}
|
|
|
|
}
|
2021-01-21 18:21:29 +00:00
|
|
|
"part-size" => {
|
2021-04-25 12:41:22 +00:00
|
|
|
settings.buffer_size = value.get::<u64>().expect("type checked upstream");
|
2019-06-12 08:07:39 +00:00
|
|
|
}
|
2021-04-30 13:05:35 +00:00
|
|
|
"uri" => {
|
|
|
|
let _ = self.set_uri(obj, value.get().expect("type checked upstream"));
|
|
|
|
}
|
2021-09-27 13:49:12 +00:00
|
|
|
"access-key" => {
|
|
|
|
settings.access_key = value.get().expect("type checked upstream");
|
|
|
|
}
|
|
|
|
"secret-access-key" => {
|
|
|
|
settings.secret_access_key = value.get().expect("type checked upstream");
|
|
|
|
}
|
2021-11-22 09:12:51 +00:00
|
|
|
"metadata" => {
|
|
|
|
settings.metadata = value.get().expect("type checked upstream");
|
|
|
|
}
|
2021-12-06 12:06:10 +00:00
|
|
|
"on-error" => {
|
|
|
|
settings.multipart_upload_on_error =
|
|
|
|
value.get::<OnError>().expect("type checked upstream");
|
|
|
|
}
|
2019-06-12 08:07:39 +00:00
|
|
|
_ => unimplemented!(),
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
2021-04-20 12:57:40 +00:00
|
|
|
fn property(&self, _: &Self::Type, _id: usize, pspec: &glib::ParamSpec) -> glib::Value {
|
2019-06-12 08:07:39 +00:00
|
|
|
let settings = self.settings.lock().unwrap();
|
|
|
|
|
2021-04-12 12:49:54 +00:00
|
|
|
match pspec.name() {
|
2021-06-21 10:20:32 +00:00
|
|
|
"key" => settings.key.to_value(),
|
|
|
|
"bucket" => settings.bucket.to_value(),
|
|
|
|
"region" => settings.region.name().to_value(),
|
2021-01-21 18:21:29 +00:00
|
|
|
"part-size" => settings.buffer_size.to_value(),
|
2021-04-30 13:05:35 +00:00
|
|
|
"uri" => {
|
|
|
|
let url = match *self.url.lock().unwrap() {
|
|
|
|
Some(ref url) => url.to_string(),
|
|
|
|
None => "".to_string(),
|
|
|
|
};
|
|
|
|
|
|
|
|
url.to_value()
|
|
|
|
}
|
2021-09-27 13:49:12 +00:00
|
|
|
"access-key" => settings.access_key.to_value(),
|
|
|
|
"secret-access-key" => settings.secret_access_key.to_value(),
|
2021-11-22 09:12:51 +00:00
|
|
|
"metadata" => settings.metadata.to_value(),
|
2021-12-06 12:06:10 +00:00
|
|
|
"on-error" => settings.multipart_upload_on_error.to_value(),
|
2019-06-12 08:07:39 +00:00
|
|
|
_ => unimplemented!(),
|
|
|
|
}
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
2021-10-23 08:57:31 +00:00
|
|
|
impl GstObjectImpl for S3Sink {}
|
|
|
|
|
2021-01-21 18:21:29 +00:00
|
|
|
impl ElementImpl for S3Sink {
|
|
|
|
fn metadata() -> Option<&'static gst::subclass::ElementMetadata> {
|
|
|
|
static ELEMENT_METADATA: Lazy<gst::subclass::ElementMetadata> = Lazy::new(|| {
|
|
|
|
gst::subclass::ElementMetadata::new(
|
|
|
|
"Amazon S3 sink",
|
|
|
|
"Source/Network",
|
|
|
|
"Writes an object to Amazon S3",
|
|
|
|
"Marcin Kolny <mkolny@amazon.com>",
|
|
|
|
)
|
|
|
|
});
|
|
|
|
|
|
|
|
Some(&*ELEMENT_METADATA)
|
|
|
|
}
|
|
|
|
|
|
|
|
fn pad_templates() -> &'static [gst::PadTemplate] {
|
|
|
|
static PAD_TEMPLATES: Lazy<Vec<gst::PadTemplate>> = Lazy::new(|| {
|
|
|
|
let caps = gst::Caps::new_any();
|
|
|
|
let sink_pad_template = gst::PadTemplate::new(
|
|
|
|
"sink",
|
|
|
|
gst::PadDirection::Sink,
|
|
|
|
gst::PadPresence::Always,
|
|
|
|
&caps,
|
|
|
|
)
|
|
|
|
.unwrap();
|
|
|
|
|
|
|
|
vec![sink_pad_template]
|
|
|
|
});
|
|
|
|
|
|
|
|
PAD_TEMPLATES.as_ref()
|
|
|
|
}
|
|
|
|
}
|
2019-06-12 08:07:39 +00:00
|
|
|
|
2021-04-30 13:05:35 +00:00
|
|
|
impl URIHandlerImpl for S3Sink {
|
|
|
|
const URI_TYPE: gst::URIType = gst::URIType::Sink;
|
|
|
|
|
|
|
|
fn protocols() -> &'static [&'static str] {
|
|
|
|
&["s3"]
|
|
|
|
}
|
|
|
|
|
|
|
|
fn uri(&self, _: &Self::Type) -> Option<String> {
|
|
|
|
self.url.lock().unwrap().as_ref().map(|s| s.to_string())
|
|
|
|
}
|
|
|
|
|
|
|
|
fn set_uri(&self, element: &Self::Type, uri: &str) -> Result<(), glib::Error> {
|
|
|
|
self.set_uri(element, Some(uri))
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
2019-06-12 08:07:39 +00:00
|
|
|
impl BaseSinkImpl for S3Sink {
|
2020-11-14 17:24:01 +00:00
|
|
|
fn start(&self, _element: &Self::Type) -> Result<(), gst::ErrorMessage> {
|
2020-04-20 01:28:30 +00:00
|
|
|
self.start()
|
2019-06-12 08:07:39 +00:00
|
|
|
}
|
|
|
|
|
2020-11-14 17:24:01 +00:00
|
|
|
fn stop(&self, element: &Self::Type) -> Result<(), gst::ErrorMessage> {
|
2019-06-12 08:07:39 +00:00
|
|
|
let mut state = self.state.lock().unwrap();
|
|
|
|
*state = State::Stopped;
|
2019-10-31 22:34:21 +00:00
|
|
|
gst_info!(CAT, obj: element, "Stopped");
|
2019-06-12 08:07:39 +00:00
|
|
|
|
|
|
|
Ok(())
|
|
|
|
}
|
|
|
|
|
|
|
|
fn render(
|
|
|
|
&self,
|
2020-11-14 17:24:01 +00:00
|
|
|
element: &Self::Type,
|
2019-06-12 08:07:39 +00:00
|
|
|
buffer: &gst::Buffer,
|
|
|
|
) -> Result<gst::FlowSuccess, gst::FlowError> {
|
|
|
|
if let State::Stopped = *self.state.lock().unwrap() {
|
2020-12-20 18:43:45 +00:00
|
|
|
gst::element_error!(element, gst::CoreError::Failed, ["Not started yet"]);
|
2019-06-12 08:07:39 +00:00
|
|
|
return Err(gst::FlowError::Error);
|
|
|
|
}
|
|
|
|
|
2019-10-31 22:34:21 +00:00
|
|
|
gst_trace!(CAT, obj: element, "Rendering {:?}", buffer);
|
2019-12-18 05:50:10 +00:00
|
|
|
let map = buffer.map_readable().map_err(|_| {
|
2020-12-20 18:43:45 +00:00
|
|
|
gst::element_error!(element, gst::CoreError::Failed, ["Failed to map buffer"]);
|
2019-06-12 08:07:39 +00:00
|
|
|
gst::FlowError::Error
|
|
|
|
})?;
|
|
|
|
|
|
|
|
match self.update_buffer(&map, element) {
|
|
|
|
Ok(_) => Ok(gst::FlowSuccess::Ok),
|
|
|
|
Err(err) => match err {
|
|
|
|
Some(error_message) => {
|
|
|
|
gst_error!(
|
2019-10-31 22:34:21 +00:00
|
|
|
CAT,
|
2019-06-12 08:07:39 +00:00
|
|
|
obj: element,
|
|
|
|
"Multipart upload failed: {}",
|
|
|
|
error_message
|
|
|
|
);
|
2020-06-30 20:57:22 +00:00
|
|
|
element.post_error_message(error_message);
|
2019-06-12 08:07:39 +00:00
|
|
|
Err(gst::FlowError::Error)
|
|
|
|
}
|
|
|
|
_ => {
|
2019-10-31 22:34:21 +00:00
|
|
|
gst_info!(CAT, obj: element, "Upload interrupted. Flushing...");
|
2019-06-12 08:07:39 +00:00
|
|
|
Err(gst::FlowError::Flushing)
|
|
|
|
}
|
|
|
|
},
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
2020-11-14 17:24:01 +00:00
|
|
|
fn unlock(&self, _element: &Self::Type) -> Result<(), gst::ErrorMessage> {
|
2019-06-12 08:07:39 +00:00
|
|
|
self.cancel();
|
|
|
|
|
|
|
|
Ok(())
|
|
|
|
}
|
|
|
|
|
2020-11-14 17:24:01 +00:00
|
|
|
fn event(&self, element: &Self::Type, event: gst::Event) -> bool {
|
2019-11-24 22:00:27 +00:00
|
|
|
if let gst::EventView::Eos(_) = event.view() {
|
|
|
|
if let Err(error_message) = self.finalize_upload(element) {
|
|
|
|
gst_error!(
|
|
|
|
CAT,
|
|
|
|
obj: element,
|
|
|
|
"Failed to finalize the upload: {}",
|
|
|
|
error_message
|
|
|
|
);
|
|
|
|
return false;
|
2019-06-12 08:07:39 +00:00
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
BaseSinkImplExt::parent_event(self, element, event)
|
|
|
|
}
|
|
|
|
}
|