mirrors/gst-plugins-rs
mirrors/gst-plugins-rs
1
1
Fork 0
mirror of https://gitlab.freedesktop.org/gstreamer/gst-plugins-rs.git synced 2024-12-22 10:06:29 +00:00
Code Issues Projects Releases Wiki Activity

Run everything through rustfmt-nightly

Browse source
This commit is contained in:
Sebastian Dröge 2017-07-31 14:36:35 +01:00
parent 3c27685e38
commit 6f04ddf797
22 changed files with 1175 additions and 909 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

63
gst-plugin-file/src/filesink.rs
View file

@ -37,11 +37,10 @@ impl FileSink {
pub fn new(element: Element) -> FileSink {
FileSink {
streaming_state: StreamingState::Stopped,
logger: Logger::root(GstDebugDrain::new(Some(&element),
"rsfilesink",
0,
"Rust file sink"),
o!()),
logger: Logger::root(
GstDebugDrain::new(Some(&element), "rsfilesink", 0, "Rust file sink"),
o!(),
),
}
}
@ -51,11 +50,11 @@ impl FileSink {
}
fn validate_uri(uri: &Url) -> Result<(), UriError> {
let _ = try!(uri.to_file_path()
.or_else(|_| {
Err(UriError::new(UriErrorKind::UnsupportedProtocol,
Some(format!("Unsupported file URI '{}'",
uri.as_str()))))
let _ = try!(uri.to_file_path().or_else(|_| {
Err(UriError::new(
UriErrorKind::UnsupportedProtocol,
Some(format!("Unsupported file URI '{}'", uri.as_str())),
))
}));
Ok(())
}
@ -70,23 +69,29 @@ impl Sink for FileSink {
return Err(error_msg!(SinkError::Failure, ["Sink already started"]));
}
let location =
try!(uri.to_file_path()
.or_else(|_| {
let location = try!(uri.to_file_path().or_else(|_| {
error!(self.logger, "Unsupported file URI '{}'", uri.as_str());
Err(error_msg!(SinkError::Failure,
["Unsupported file URI '{}'", uri.as_str()]))
Err(error_msg!(
SinkError::Failure,
["Unsupported file URI '{}'", uri.as_str()]
))
}));
let file = try!(File::create(location.as_path()).or_else(|err| {
error!(self.logger,
error!(
self.logger,
"Could not open file for writing: {}",
err.to_string());
Err(error_msg!(SinkError::OpenFailed,
["Could not open file for writing '{}': {}",
err.to_string()
);
Err(error_msg!(
SinkError::OpenFailed,
[
"Could not open file for writing '{}': {}",
location.to_str().unwrap_or("Non-UTF8 path"),
err.to_string()]))
err.to_string()
]
))
}));
debug!(self.logger, "Opened file {:?}", file);
@ -117,24 +122,28 @@ impl Sink for FileSink {
ref mut position,
} => (file, position),
StreamingState::Stopped => {
return Err(FlowError::Error(error_msg!(SinkError::Failure, ["Not started yet"])));
return Err(FlowError::Error(
error_msg!(SinkError::Failure, ["Not started yet"]),
));
}
};
let map = match buffer.map_read() {
None => {
return Err(FlowError::Error(error_msg!(SinkError::Failure,
["Failed to map buffer"])));
return Err(FlowError::Error(
error_msg!(SinkError::Failure, ["Failed to map buffer"]),
));
}
Some(map) => map,
};
let data = map.as_slice();
try!(file.write_all(data)
.or_else(|err| {
try!(file.write_all(data).or_else(|err| {
error!(logger, "Failed to write: {}", err);
Err(FlowError::Error(error_msg!(SinkError::WriteFailed,
["Failed to write: {}", err])))
Err(FlowError::Error(error_msg!(
SinkError::WriteFailed,
["Failed to write: {}", err]
)))
}));
*position += data.len() as u64;

76
gst-plugin-file/src/filesrc.rs
View file

@ -35,11 +35,10 @@ impl FileSrc {
pub fn new(element: Element) -> FileSrc {
FileSrc {
streaming_state: StreamingState::Stopped,
logger: Logger::root(GstDebugDrain::new(Some(&element),
"rsfilesrc",
0,
"Rust file source"),
o!()),
logger: Logger::root(
GstDebugDrain::new(Some(&element), "rsfilesrc", 0, "Rust file source"),
o!(),
),
}
}
@ -49,11 +48,11 @@ impl FileSrc {
}
fn validate_uri(uri: &Url) -> Result<(), UriError> {
let _ = try!(uri.to_file_path()
.or_else(|_| {
Err(UriError::new(UriErrorKind::UnsupportedProtocol,
Some(format!("Unsupported file URI '{}'",
uri.as_str()))))
let _ = try!(uri.to_file_path().or_else(|_| {
Err(UriError::new(
UriErrorKind::UnsupportedProtocol,
Some(format!("Unsupported file URI '{}'", uri.as_str())),
))
}));
Ok(())
}
@ -80,22 +79,28 @@ impl Source for FileSrc {
return Err(error_msg!(SourceError::Failure, ["Source already started"]));
}
let location =
try!(uri.to_file_path()
.or_else(|_| {
let location = try!(uri.to_file_path().or_else(|_| {
error!(self.logger, "Unsupported file URI '{}'", uri.as_str());
Err(error_msg!(SourceError::Failure,
["Unsupported file URI '{}'", uri.as_str()]))
Err(error_msg!(
SourceError::Failure,
["Unsupported file URI '{}'", uri.as_str()]
))
}));
let file = try!(File::open(location.as_path()).or_else(|err| {
error!(self.logger,
error!(
self.logger,
"Could not open file for reading: {}",
err.to_string());
Err(error_msg!(SourceError::OpenFailed,
["Could not open file for reading '{}': {}",
err.to_string()
);
Err(error_msg!(
SourceError::OpenFailed,
[
"Could not open file for reading '{}': {}",
location.to_str().unwrap_or("Non-UTF8 path"),
err.to_string()]))
err.to_string()
]
))
}));
debug!(self.logger, "Opened file {:?}", file);
@ -124,18 +129,19 @@ impl Source for FileSrc {
ref mut position,
} => (file, position),
StreamingState::Stopped => {
return Err(FlowError::Error(error_msg!(SourceError::Failure, ["Not started yet"])));
return Err(FlowError::Error(
error_msg!(SourceError::Failure, ["Not started yet"]),
));
}
};
if *position != offset {
try!(file.seek(SeekFrom::Start(offset))
.or_else(|err| {
try!(file.seek(SeekFrom::Start(offset)).or_else(|err| {
error!(logger, "Failed to seek to {}: {:?}", offset, err);
Err(FlowError::Error(error_msg!(SourceError::SeekFailed,
["Failed to seek to {}: {}",
offset,
err.to_string()])))
Err(FlowError::Error(error_msg!(
SourceError::SeekFailed,
["Failed to seek to {}: {}", offset, err.to_string()]
)))
}));
*position = offset;
}
@ -143,21 +149,21 @@ impl Source for FileSrc {
let size = {
let mut map = match buffer.map_readwrite() {
None => {
return Err(FlowError::Error(error_msg!(SourceError::Failure,
["Failed to map buffer"])));
return Err(FlowError::Error(
error_msg!(SourceError::Failure, ["Failed to map buffer"]),
));
}
Some(map) => map,
};
let data = map.as_mut_slice();
try!(file.read(data)
.or_else(|err| {
try!(file.read(data).or_else(|err| {
error!(logger, "Failed to read: {:?}", err);
Err(FlowError::Error(error_msg!(SourceError::ReadFailed,
["Failed to read at {}: {}",
offset,
err.to_string()])))
Err(FlowError::Error(error_msg!(
SourceError::ReadFailed,
["Failed to read at {}: {}", offset, err.to_string()]
)))
}))
};

18
gst-plugin-file/src/lib.rs
View file

@ -25,7 +25,8 @@ use filesrc::FileSrc;
use filesink::FileSink;
fn plugin_init(plugin: &Plugin) -> bool {
source_register(plugin,
source_register(
plugin,
SourceInfo {
name: "rsfilesrc".into(),
long_name: "File Source".into(),
@ -36,9 +37,11 @@ fn plugin_init(plugin: &Plugin) -> bool {
create_instance: FileSrc::new_boxed,
protocols: vec!["file".into()],
push_only: false,
});
},
);
sink_register(plugin,
sink_register(
plugin,
SinkInfo {
name: "rsfilesink".into(),
long_name: "File Sink".into(),
@ -48,12 +51,14 @@ fn plugin_init(plugin: &Plugin) -> bool {
rank: 256 + 100,
create_instance: FileSink::new_boxed,
protocols: vec!["file".into()],
});
},
);
true
}
plugin_define!(b"rsfile\0",
plugin_define!(
b"rsfile\0",
b"Rust File Plugin\0",
plugin_init,
b"1.0\0",
@ -61,4 +66,5 @@ plugin_define!(b"rsfile\0",
b"rsfile\0",
b"rsfile\0",
b"https://github.com/sdroege/rsplugin\0",
b"2016-12-08\0");
b"2016-12-08\0"
);

297
gst-plugin-flv/src/flvdemux.rs
View file

@ -7,7 +7,7 @@
// except according to those terms.
use std::cmp;
use std::io::{Write, Cursor};
use std::io::{Cursor, Write};
use nom;
use nom::IResult;
@ -87,17 +87,18 @@ struct AudioFormat {
// Ignores bitrate
impl PartialEq for AudioFormat {
fn eq(&self, other: &Self) -> bool {
self.format.eq(&other.format) && self.rate.eq(&other.rate) &&
self.width.eq(&other.width) && self.channels.eq(&other.channels) &&
self.format.eq(&other.format) && self.rate.eq(&other.rate) && self.width.eq(&other.width) &&
self.channels.eq(&other.channels) &&
self.aac_sequence_header.eq(&other.aac_sequence_header)
}
}
impl AudioFormat {
fn new(data_header: &flavors::AudioDataHeader,
fn new(
data_header: &flavors::AudioDataHeader,
metadata: &Option<Metadata>,
aac_sequence_header: &Option<GstRc<Buffer>>)
-> AudioFormat {
aac_sequence_header: &Option<GstRc<Buffer>>,
) -> AudioFormat {
let numeric_rate = match (data_header.sound_format, data_header.sound_rate) {
(flavors::SoundFormat::NELLYMOSER_16KHZ_MONO, _) => 16000,
(flavors::SoundFormat::NELLYMOSER_8KHZ_MONO, _) => 8000,
@ -146,47 +147,52 @@ impl AudioFormat {
fn to_caps(&self) -> Option<GstRc<Caps>> {
let mut caps = match self.format {
flavors::SoundFormat::MP3 |
flavors::SoundFormat::MP3_8KHZ => {
Some(Caps::new_simple("audio/mpeg",
&[("mpegversion", 1i32.into()), ("layer", 3i32.into())]))
}
flavors::SoundFormat::PCM_NE |
flavors::SoundFormat::PCM_LE => {
flavors::SoundFormat::MP3 | flavors::SoundFormat::MP3_8KHZ => Some(Caps::new_simple(
"audio/mpeg",
&[("mpegversion", 1i32.into()), ("layer", 3i32.into())],
)),
flavors::SoundFormat::PCM_NE | flavors::SoundFormat::PCM_LE => {
if self.rate != 0 && self.channels != 0 {
// Assume little-endian for "PCM_NE", it's probably more common and we have no
// way to know what the endianness of the system creating the stream was
Some(Caps::new_simple("audio/x-raw",
&[("layout", "interleaved".into()),
("format",
Some(Caps::new_simple(
"audio/x-raw",
&[
("layout", "interleaved".into()),
(
"format",
if self.width == 8 {
"U8".into()
} else {
"S16LE".into()
})]))
},
),
],
))
} else {
None
}
}
flavors::SoundFormat::ADPCM => {
Some(Caps::new_simple("audio/x-adpcm", &[("layout", "swf".into())]))
}
flavors::SoundFormat::ADPCM => Some(Caps::new_simple(
"audio/x-adpcm",
&[("layout", "swf".into())],
)),
flavors::SoundFormat::NELLYMOSER_16KHZ_MONO |
flavors::SoundFormat::NELLYMOSER_8KHZ_MONO |
flavors::SoundFormat::NELLYMOSER => Some(Caps::new_simple("audio/x-nellymoser", &[])),
flavors::SoundFormat::PCM_ALAW => Some(Caps::new_simple("audio/x-alaw", &[])),
flavors::SoundFormat::PCM_ULAW => Some(Caps::new_simple("audio/x-mulaw", &[])),
flavors::SoundFormat::AAC => {
self.aac_sequence_header
.as_ref()
.map(|header| {
Caps::new_simple("audio/mpeg",
&[("mpegversion", 4i32.into()),
flavors::SoundFormat::AAC => self.aac_sequence_header.as_ref().map(|header| {
Caps::new_simple(
"audio/mpeg",
&[
("mpegversion", 4i32.into()),
("framed", true.into()),
("stream-format", "raw".into()),
("codec_data", header.as_ref().into())])
})
}
("codec_data", header.as_ref().into()),
],
)
}),
flavors::SoundFormat::SPEEX => {
let header = {
let header_size = 80;
@ -227,9 +233,10 @@ impl AudioFormat {
};
let comment = Buffer::from_vec(comment).unwrap();
Some(Caps::new_simple("audio/x-speex",
&[("streamheader",
vec![header.into(), comment.into()].into())]))
Some(Caps::new_simple(
"audio/x-speex",
&[("streamheader", vec![header.into(), comment.into()].into())],
))
}
flavors::SoundFormat::DEVICE_SPECIFIC => {
// Nobody knows
@ -238,16 +245,14 @@ impl AudioFormat {
};
if self.rate != 0 {
caps.as_mut()
.map(|c| {
caps.as_mut().map(|c| {
c.get_mut()
.unwrap()
.set_simple(&[("rate", (self.rate as i32).into())])
});
}
if self.channels != 0 {
caps.as_mut()
.map(|c| {
caps.as_mut().map(|c| {
c.get_mut()
.unwrap()
.set_simple(&[("channels", (self.channels as i32).into())])
@ -270,17 +275,16 @@ struct VideoFormat {
}
impl VideoFormat {
fn new(data_header: &flavors::VideoDataHeader,
fn new(
data_header: &flavors::VideoDataHeader,
metadata: &Option<Metadata>,
avc_sequence_header: &Option<GstRc<Buffer>>)
-> VideoFormat {
avc_sequence_header: &Option<GstRc<Buffer>>,
) -> VideoFormat {
VideoFormat {
format: data_header.codec_id,
width: metadata.as_ref().and_then(|m| m.video_width),
height: metadata.as_ref().and_then(|m| m.video_height),
pixel_aspect_ratio: metadata
.as_ref()
.and_then(|m| m.video_pixel_aspect_ratio),
pixel_aspect_ratio: metadata.as_ref().and_then(|m| m.video_pixel_aspect_ratio),
framerate: metadata.as_ref().and_then(|m| m.video_framerate),
bitrate: metadata.as_ref().and_then(|m| m.video_bitrate),
avc_sequence_header: avc_sequence_header.clone(),
@ -324,28 +328,28 @@ impl VideoFormat {
fn to_caps(&self) -> Option<GstRc<Caps>> {
let mut caps = match self.format {
flavors::CodecId::SORENSON_H263 => {
Some(Caps::new_simple("video/x-flash-video", &[("flvversion", 1i32.into())]))
}
flavors::CodecId::SORENSON_H263 => Some(Caps::new_simple(
"video/x-flash-video",
&[("flvversion", 1i32.into())],
)),
flavors::CodecId::SCREEN => Some(Caps::new_simple("video/x-flash-screen", &[])),
flavors::CodecId::VP6 => Some(Caps::new_simple("video/x-vp6-flash", &[])),
flavors::CodecId::VP6A => Some(Caps::new_simple("video/x-vp6-flash-alpha", &[])),
flavors::CodecId::SCREEN2 => Some(Caps::new_simple("video/x-flash-screen2", &[])),
flavors::CodecId::H264 => {
self.avc_sequence_header
.as_ref()
.map(|header| {
Caps::new_simple("video/x-h264",
&[("stream-format", "avc".into()),
("codec_data", header.as_ref().into())])
})
}
flavors::CodecId::H264 => self.avc_sequence_header.as_ref().map(|header| {
Caps::new_simple(
"video/x-h264",
&[
("stream-format", "avc".into()),
("codec_data", header.as_ref().into()),
],
)
}),
flavors::CodecId::H263 => Some(Caps::new_simple("video/x-h263", &[])),
flavors::CodecId::MPEG4Part2 => {
Some(Caps::new_simple("video/x-h263",
&[("mpegversion", 4i32.into()),
("systemstream", false.into())]))
}
flavors::CodecId::MPEG4Part2 => Some(Caps::new_simple(
"video/x-h263",
&[("mpegversion", 4i32.into()), ("systemstream", false.into())],
)),
flavors::CodecId::JPEG => {
// Unused according to spec
None
@ -353,19 +357,17 @@ impl VideoFormat {
};
if let (Some(width), Some(height)) = (self.width, self.height) {
caps.as_mut()
.map(|c| {
c.get_mut()
.unwrap()
.set_simple(&[("width", (width as i32).into()),
("height", (height as i32).into())])
caps.as_mut().map(|c| {
c.get_mut().unwrap().set_simple(&[
("width", (width as i32).into()),
("height", (height as i32).into()),
])
});
}
if let Some(par) = self.pixel_aspect_ratio {
if *par.numer() != 0 && par.numer() != par.denom() {
caps.as_mut()
.map(|c| {
caps.as_mut().map(|c| {
c.get_mut()
.unwrap()
.set_simple(&[("pixel-aspect-ratio", par.into())])
@ -375,8 +377,7 @@ impl VideoFormat {
if let Some(fps) = self.framerate {
if *fps.numer() != 0 {
caps.as_mut()
.map(|c| {
caps.as_mut().map(|c| {
c.get_mut()
.unwrap()
.set_simple(&[("framerate", fps.into())])
@ -509,11 +510,10 @@ pub struct FlvDemux {
impl FlvDemux {
pub fn new(element: Element) -> FlvDemux {
FlvDemux {
logger: Logger::root(GstDebugDrain::new(Some(&element),
"rsflvdemux",
0,
"Rust FLV demuxer"),
o!()),
logger: Logger::root(
GstDebugDrain::new(Some(&element), "rsflvdemux", 0, "Rust FLV demuxer"),
o!(),
),
state: State::Stopped,
adapter: Adapter::new(),
streaming_state: None,
@ -524,9 +524,10 @@ impl FlvDemux {
Box::new(FlvDemux::new(element))
}
fn handle_script_tag(&mut self,
tag_header: &flavors::TagHeader)
-> Result<HandleBufferResult, FlowError> {
fn handle_script_tag(
&mut self,
tag_header: &flavors::TagHeader,
) -> Result<HandleBufferResult, FlowError> {
if self.adapter.get_available() < (15 + tag_header.data_size) as usize {
return Ok(HandleBufferResult::NeedMoreData);
}
@ -564,14 +565,14 @@ impl FlvDemux {
let mut streams = Vec::new();
if audio_changed {
if let Some(caps) =
streaming_state.audio.as_ref().and_then(|a| a.to_caps()) {
if let Some(caps) = streaming_state.audio.as_ref().and_then(|a| a.to_caps())
{
streams.push(Stream::new(AUDIO_STREAM_ID, caps, String::from("audio")));
}
}
if video_changed {
if let Some(caps) =
streaming_state.video.as_ref().and_then(|v| v.to_caps()) {
if let Some(caps) = streaming_state.video.as_ref().and_then(|v| v.to_caps())
{
streams.push(Stream::new(VIDEO_STREAM_ID, caps, String::from("video")));
}
}
@ -582,8 +583,7 @@ impl FlvDemux {
IResult::Done(_, ref script_data) => {
trace!(self.logger, "Got script tag: {:?}", script_data);
}
IResult::Error(_) |
IResult::Incomplete(_) => {
IResult::Error(_) | IResult::Incomplete(_) => {
// ignore
}
}
@ -591,17 +591,20 @@ impl FlvDemux {
Ok(HandleBufferResult::Again)
}
fn update_audio_stream(&mut self,
data_header: &flavors::AudioDataHeader)
-> Result<HandleBufferResult, FlowError> {
fn update_audio_stream(
&mut self,
data_header: &flavors::AudioDataHeader,
) -> Result<HandleBufferResult, FlowError> {
let logger = self.logger.clone();
trace!(logger, "Got audio data header: {:?}", data_header);
let streaming_state = self.streaming_state.as_mut().unwrap();
let new_audio_format = AudioFormat::new(data_header,
let new_audio_format = AudioFormat::new(
data_header,
&streaming_state.metadata,
&streaming_state.aac_sequence_header);
&streaming_state.aac_sequence_header,
);
if streaming_state.audio.as_ref() != Some(&new_audio_format) {
debug!(logger, "Got new audio format: {:?}", new_audio_format);
@ -623,7 +626,8 @@ impl FlvDemux {
if !streaming_state.got_all_streams && streaming_state.audio != None &&
(streaming_state.expect_video && streaming_state.video != None ||
!streaming_state.expect_video) {
!streaming_state.expect_video)
{
streaming_state.got_all_streams = true;
return Ok(HandleBufferResult::HaveAllStreams);
}
@ -631,10 +635,11 @@ impl FlvDemux {
Ok(HandleBufferResult::Again)
}
fn handle_audio_tag(&mut self,
fn handle_audio_tag(
&mut self,
tag_header: &flavors::TagHeader,
data_header: &flavors::AudioDataHeader)
-> Result<HandleBufferResult, FlowError> {
data_header: &flavors::AudioDataHeader,
) -> Result<HandleBufferResult, FlowError> {
let res = self.update_audio_stream(data_header);
match res {
Ok(HandleBufferResult::Again) => (),
@ -652,17 +657,18 @@ impl FlvDemux {
self.adapter
.flush(15 + tag_header.data_size as usize)
.unwrap();
warn!(self.logger,
warn!(
self.logger,
"Too small packet for AAC packet header {}",
15 + tag_header.data_size);
15 + tag_header.data_size
);
return Ok(HandleBufferResult::Again);
}
let mut data = [0u8; 17];
self.adapter.peek_into(&mut data).unwrap();
match flavors::aac_audio_packet_header(&data[16..]) {
IResult::Error(_) |
IResult::Incomplete(_) => {
IResult::Error(_) | IResult::Incomplete(_) => {
unimplemented!();
}
IResult::Done(_, header) => {
@ -673,10 +679,12 @@ impl FlvDemux {
let buffer = self.adapter
.get_buffer((tag_header.data_size - 1 - 1) as usize)
.unwrap();
debug!(self.logger,
debug!(
self.logger,
"Got AAC sequence header {:?} of size {}",
buffer,
tag_header.data_size - 1 - 1);
tag_header.data_size - 1 - 1
);
let streaming_state = self.streaming_state.as_mut().unwrap();
streaming_state.aac_sequence_header = Some(buffer);
@ -731,26 +739,31 @@ impl FlvDemux {
buffer.set_pts(Some((tag_header.timestamp as u64) * 1000 * 1000));
}
trace!(self.logger,
trace!(
self.logger,
"Outputting audio buffer {:?} for tag {:?} of size {}",
buffer,
tag_header,
tag_header.data_size - 1);
tag_header.data_size - 1
);
Ok(HandleBufferResult::BufferForStream(AUDIO_STREAM_ID, buffer))
}
fn update_video_stream(&mut self,
data_header: &flavors::VideoDataHeader)
-> Result<HandleBufferResult, FlowError> {
fn update_video_stream(
&mut self,
data_header: &flavors::VideoDataHeader,
) -> Result<HandleBufferResult, FlowError> {
let logger = self.logger.clone();
trace!(logger, "Got video data header: {:?}", data_header);
let streaming_state = self.streaming_state.as_mut().unwrap();
let new_video_format = VideoFormat::new(data_header,
let new_video_format = VideoFormat::new(
data_header,
&streaming_state.metadata,
&streaming_state.avc_sequence_header);
&streaming_state.avc_sequence_header,
);
if streaming_state.video.as_ref() != Some(&new_video_format) {
debug!(logger, "Got new video format: {:?}", new_video_format);
@ -773,7 +786,8 @@ impl FlvDemux {
if !streaming_state.got_all_streams && streaming_state.video != None &&
(streaming_state.expect_audio && streaming_state.audio != None ||
!streaming_state.expect_audio) {
!streaming_state.expect_audio)
{
streaming_state.got_all_streams = true;
return Ok(HandleBufferResult::HaveAllStreams);
}
@ -781,10 +795,11 @@ impl FlvDemux {
Ok(HandleBufferResult::Again)
}
fn handle_video_tag(&mut self,
fn handle_video_tag(
&mut self,
tag_header: &flavors::TagHeader,
data_header: &flavors::VideoDataHeader)
-> Result<HandleBufferResult, FlowError> {
data_header: &flavors::VideoDataHeader,
) -> Result<HandleBufferResult, FlowError> {
let res = self.update_video_stream(data_header);
match res {
Ok(HandleBufferResult::Again) => (),
@ -804,17 +819,18 @@ impl FlvDemux {
self.adapter
.flush(15 + tag_header.data_size as usize)
.unwrap();
warn!(self.logger,
warn!(
self.logger,
"Too small packet for AVC packet header {}",
15 + tag_header.data_size);
15 + tag_header.data_size
);
return Ok(HandleBufferResult::Again);
}
let mut data = [0u8; 20];
self.adapter.peek_into(&mut data).unwrap();
match flavors::avc_video_packet_header(&data[16..]) {
IResult::Error(_) |
IResult::Incomplete(_) => {
IResult::Error(_) | IResult::Incomplete(_) => {
unimplemented!();
}
IResult::Done(_, header) => {
@ -825,10 +841,12 @@ impl FlvDemux {
let buffer = self.adapter
.get_buffer((tag_header.data_size - 1 - 4) as usize)
.unwrap();
debug!(self.logger,
debug!(
self.logger,
"Got AVC sequence header {:?} of size {}",
buffer,
tag_header.data_size - 1 - 4);
tag_header.data_size - 1 - 4
);
let streaming_state = self.streaming_state.as_mut().unwrap();
streaming_state.avc_sequence_header = Some(buffer);
@ -864,8 +882,7 @@ impl FlvDemux {
self.adapter.flush(16).unwrap();
let offset = match video.format {
flavors::CodecId::VP6 |
flavors::CodecId::VP6A => 1,
flavors::CodecId::VP6 | flavors::CodecId::VP6A => 1,
flavors::CodecId::H264 => 4,
_ => 0,
};
@ -905,12 +922,14 @@ impl FlvDemux {
buffer.set_pts(Some(pts * 1000 * 1000));
}
trace!(self.logger,
trace!(
self.logger,
"Outputting video buffer {:?} for tag {:?} of size {}, keyframe: {}",
buffer,
tag_header,
tag_header.data_size - 1 - offset,
is_keyframe);
is_keyframe
);
Ok(HandleBufferResult::BufferForStream(VIDEO_STREAM_ID, buffer))
}
@ -924,8 +943,7 @@ impl FlvDemux {
self.adapter.peek_into(&mut data).unwrap();
match flavors::header(&data) {
IResult::Error(_) |
IResult::Incomplete(_) => {
IResult::Error(_) | IResult::Incomplete(_) => {
// fall through
}
IResult::Done(_, ref header) => {
@ -964,7 +982,9 @@ impl FlvDemux {
Ok(HandleBufferResult::Again)
}
State::Skipping { ref mut skip_left, .. } => {
State::Skipping {
ref mut skip_left, ..
} => {
let skip = cmp::min(self.adapter.get_available(), *skip_left as usize);
self.adapter.flush(skip).unwrap();
*skip_left -= skip as u32;
@ -980,8 +1000,7 @@ impl FlvDemux {
self.adapter.peek_into(&mut data).unwrap();
match nom::be_u32(&data[0..4]) {
IResult::Error(_) |
IResult::Incomplete(_) => {
IResult::Error(_) | IResult::Incomplete(_) => {
unimplemented!();
}
IResult::Done(_, previous_size) => {
@ -991,8 +1010,7 @@ impl FlvDemux {
}
let tag_header = match flavors::tag_header(&data[4..]) {
IResult::Error(_) |
IResult::Incomplete(_) => {
IResult::Error(_) | IResult::Incomplete(_) => {
unimplemented!();
}
IResult::Done(_, tag_header) => tag_header,
@ -1008,8 +1026,7 @@ impl FlvDemux {
trace!(self.logger, "Found audio tag");
let data_header = match flavors::audio_data_header(&data[15..]) {
IResult::Error(_) |
IResult::Incomplete(_) => {
IResult::Error(_) | IResult::Incomplete(_) => {
unimplemented!();
}
IResult::Done(_, data_header) => data_header,
@ -1021,8 +1038,7 @@ impl FlvDemux {
trace!(self.logger, "Found video tag");
let data_header = match flavors::video_data_header(&data[15..]) {
IResult::Error(_) |
IResult::Incomplete(_) => {
IResult::Error(_) | IResult::Incomplete(_) => {
unimplemented!();
}
IResult::Done(_, data_header) => data_header,
@ -1056,10 +1072,11 @@ impl FlvDemux {
}
impl Demuxer for FlvDemux {
fn start(&mut self,
fn start(
&mut self,
_upstream_size: Option<u64>,
_random_access: bool)
-> Result<(), ErrorMessage> {
_random_access: bool,
) -> Result<(), ErrorMessage> {
self.state = State::NeedHeader;
Ok(())
@ -1077,9 +1094,10 @@ impl Demuxer for FlvDemux {
unimplemented!();
}
fn handle_buffer(&mut self,
buffer: Option<GstRc<Buffer>>)
-> Result<HandleBufferResult, FlowError> {
fn handle_buffer(
&mut self,
buffer: Option<GstRc<Buffer>>,
) -> Result<HandleBufferResult, FlowError> {
if let Some(buffer) = buffer {
self.adapter.push(buffer);
}
@ -1105,8 +1123,11 @@ impl Demuxer for FlvDemux {
}
fn get_duration(&self) -> Option<u64> {
if let Some(StreamingState { metadata: Some(Metadata { duration, .. }), .. }) =
self.streaming_state {
if let Some(StreamingState {
metadata: Some(Metadata { duration, .. }),
..
}) = self.streaming_state
{
return duration;
}

12
gst-plugin-flv/src/lib.rs
View file

@ -26,7 +26,8 @@ mod flvdemux;
use flvdemux::FlvDemux;
fn plugin_init(plugin: &Plugin) -> bool {
demuxer_register(plugin,
demuxer_register(
plugin,
&DemuxerInfo {
name: "rsflvdemux",
long_name: "FLV Demuxer",
@ -37,12 +38,14 @@ fn plugin_init(plugin: &Plugin) -> bool {
create_instance: FlvDemux::new_boxed,
input_caps: &Caps::new_simple("video/x-flv", &[]),
output_caps: &Caps::new_any(),
});
},
);
true
}
plugin_define!(b"rsflv\0",
plugin_define!(
b"rsflv\0",
b"Rust FLV Plugin\0",
plugin_init,
b"1.0\0",
@ -50,4 +53,5 @@ plugin_define!(b"rsflv\0",
b"rsflv\0",
b"rsflv\0",
b"https://github.com/sdroege/rsplugin\0",
b"2016-12-08\0");
b"2016-12-08\0"
);

90
gst-plugin-http/src/httpsrc.rs
View file

@ -10,8 +10,8 @@ use std::u64;
use std::io::Read;
use url::Url;
use reqwest::{Client, Response};
use reqwest::header::{ContentLength, ContentRange, ContentRangeSpec, Range, ByteRangeSpec,
AcceptRanges, RangeUnit};
use reqwest::header::{AcceptRanges, ByteRangeSpec, ContentLength, ContentRange, ContentRangeSpec,
Range, RangeUnit};
use gst_plugin::error::*;
use gst_plugin::source::*;
@ -46,11 +46,10 @@ impl HttpSrc {
pub fn new(element: Element) -> HttpSrc {
HttpSrc {
streaming_state: StreamingState::Stopped,
logger: Logger::root(GstDebugDrain::new(Some(&element),
"rshttpsink",
0,
"Rust http sink"),
o!()),
logger: Logger::root(
GstDebugDrain::new(Some(&element), "rshttpsink", 0, "Rust http sink"),
o!(),
),
client: Client::new().unwrap(),
}
}
@ -59,11 +58,12 @@ impl HttpSrc {
Box::new(HttpSrc::new(element))
}
fn do_request(&self,
fn do_request(
&self,
uri: Url,
start: u64,
stop: Option<u64>)
-> Result<StreamingState, ErrorMessage> {
stop: Option<u64>,
) -> Result<StreamingState, ErrorMessage> {
let mut req = self.client.get(uri.clone()).unwrap();
match (start != 0, stop) {
@ -78,18 +78,20 @@ impl HttpSrc {
debug!(self.logger, "Doing new request {:?}", req);
let response =
try!(req.send()
.or_else(|err| {
let response = try!(req.send().or_else(|err| {
error!(self.logger, "Request failed: {:?}", err);
Err(error_msg!(SourceError::ReadFailed,
["Failed to fetch {}: {}", uri, err.to_string()]))
Err(error_msg!(
SourceError::ReadFailed,
["Failed to fetch {}: {}", uri, err.to_string()]
))
}));
if !response.status().is_success() {
error!(self.logger, "Request status failed: {:?}", response);
return Err(error_msg!(SourceError::ReadFailed,
["Failed to fetch {}: {}", uri, response.status()]));
return Err(error_msg!(
SourceError::ReadFailed,
["Failed to fetch {}: {}", uri, response.status()]
));
}
let size = response
@ -97,8 +99,7 @@ impl HttpSrc {
.get()
.map(|&ContentLength(cl)| cl + start);
let accept_byte_ranges = if let Some(&AcceptRanges(ref ranges)) =
response.headers().get() {
let accept_byte_ranges = if let Some(&AcceptRanges(ref ranges)) = response.headers().get() {
ranges.iter().any(|u| *u == RangeUnit::Bytes)
} else {
false
@ -106,17 +107,23 @@ impl HttpSrc {
let seekable = size.is_some() && accept_byte_ranges;
let position = if let Some(&ContentRange(ContentRangeSpec::Bytes {
range: Some((range_start, _)), ..
})) = response.headers().get() {
let position = if let Some(
&ContentRange(ContentRangeSpec::Bytes {
range: Some((range_start, _)),
..
}),
) = response.headers().get()
{
range_start
} else {
start
};
if position != start {
return Err(error_msg!(SourceError::SeekFailed,
["Failed to seek to {}: Got {}", start, position]));
return Err(error_msg!(
SourceError::SeekFailed,
["Failed to seek to {}: Got {}", start, position]
));
}
debug!(self.logger, "Request successful: {:?}", response);
@ -135,8 +142,10 @@ impl HttpSrc {
fn validate_uri(uri: &Url) -> Result<(), UriError> {
if uri.scheme() != "http" && uri.scheme() != "https" {
return Err(UriError::new(UriErrorKind::UnsupportedProtocol,
Some(format!("Unsupported URI '{}'", uri.as_str()))));
return Err(UriError::new(
UriErrorKind::UnsupportedProtocol,
Some(format!("Unsupported URI '{}'", uri.as_str())),
));
}
Ok(())
@ -207,36 +216,37 @@ impl Source for HttpSrc {
..
} => (response, position),
StreamingState::Stopped => {
return Err(FlowError::Error(error_msg!(SourceError::Failure, ["Not started yet"])));
return Err(FlowError::Error(
error_msg!(SourceError::Failure, ["Not started yet"]),
));
}
};
if *position != offset {
return Err(FlowError::Error(error_msg!(SourceError::SeekFailed,
["Got unexpected offset {}, expected {}",
offset,
position])));
return Err(FlowError::Error(error_msg!(
SourceError::SeekFailed,
["Got unexpected offset {}, expected {}", offset, position]
)));
}
let size = {
let mut map = match buffer.map_readwrite() {
None => {
return Err(FlowError::Error(error_msg!(SourceError::Failure,
["Failed to map buffer"])));
return Err(FlowError::Error(
error_msg!(SourceError::Failure, ["Failed to map buffer"]),
));
}
Some(map) => map,
};
let data = map.as_mut_slice();
try!(response
.read(data)
.or_else(|err| {
try!(response.read(data).or_else(|err| {
error!(logger, "Failed to read: {:?}", err);
Err(FlowError::Error(error_msg!(SourceError::ReadFailed,
["Failed to read at {}: {}",
offset,
err.to_string()])))
Err(FlowError::Error(error_msg!(
SourceError::ReadFailed,
["Failed to read at {}: {}", offset, err.to_string()]
)))
}))
};

12
gst-plugin-http/src/lib.rs
View file

@ -23,7 +23,8 @@ mod httpsrc;
use httpsrc::HttpSrc;
fn plugin_init(plugin: &Plugin) -> bool {
source_register(plugin,
source_register(
plugin,
SourceInfo {
name: "rshttpsrc".into(),
long_name: "HTTP/HTTPS Source".into(),
@ -34,12 +35,14 @@ fn plugin_init(plugin: &Plugin) -> bool {
create_instance: HttpSrc::new_boxed,
protocols: vec!["http".into(), "https".into()],
push_only: true,
});
},
);
true
}
plugin_define!(b"rshttp\0",
plugin_define!(
b"rshttp\0",
b"Rust HTTP Plugin\0",
plugin_init,
b"1.0\0",
@ -47,4 +50,5 @@ plugin_define!(b"rshttp\0",
b"rshttp\0",
b"rshttp\0",
b"https://github.com/sdroege/rsplugin\0",
b"2016-12-08\0");
b"2016-12-08\0"
);

56
gst-plugin/src/adapter.rs
View file

@ -50,11 +50,13 @@ impl Adapter {
let size = buffer.get_size();
self.size += size;
trace!(LOGGER,
trace!(
LOGGER,
"Storing {:?} of size {}, now have size {}",
buffer,
size,
self.size);
self.size
);
self.deque
.push_back(Buffer::into_read_mapped_buffer(buffer).unwrap());
}
@ -82,11 +84,13 @@ impl Adapter {
let data_item = item.as_slice();
let to_copy = cmp::min(left, data_item.len() - skip);
trace!(LOGGER,
trace!(
LOGGER,
"Copying {} bytes from {:?}, {} more to go",
to_copy,
item,
left - to_copy);
left - to_copy
);
data[idx..idx + to_copy].copy_from_slice(&data_item[skip..skip + to_copy]);
skip = 0;
@ -103,10 +107,12 @@ impl Adapter {
let size = data.len();
if self.size < size {
debug!(LOGGER,
debug!(
LOGGER,
"Peeking {} bytes into, not enough data: have {}",
size,
self.size);
self.size
);
return Err(AdapterError::NotEnoughData);
}
@ -121,10 +127,12 @@ impl Adapter {
pub fn peek(&mut self, size: usize) -> Result<&[u8], AdapterError> {
if self.size < size {
debug!(LOGGER,
debug!(
LOGGER,
"Peeking {} bytes, not enough data: have {}",
size,
self.size);
self.size
);
return Err(AdapterError::NotEnoughData);
}
@ -153,10 +161,12 @@ impl Adapter {
pub fn get_buffer(&mut self, size: usize) -> Result<GstRc<Buffer>, AdapterError> {
if self.size < size {
debug!(LOGGER,
debug!(
LOGGER,
"Get buffer of {} bytes, not enough data: have {}",
size,
self.size);
self.size
);
return Err(AdapterError::NotEnoughData);
}
@ -164,9 +174,8 @@ impl Adapter {
return Ok(Buffer::new());
}
let sub = self.deque
.front()
.and_then(|front| if front.get_size() - self.skip >= size {
let sub = self.deque.front().and_then(
|front| if front.get_size() - self.skip >= size {
trace!(LOGGER, "Get buffer of {} bytes, subbuffer of first", size);
let new = front
.get_buffer()
@ -175,7 +184,8 @@ impl Adapter {
Some(new)
} else {
None
});
},
);
if let Some(s) = sub {
self.flush(size).unwrap();
@ -195,10 +205,12 @@ impl Adapter {
pub fn flush(&mut self, size: usize) -> Result<(), AdapterError> {
if self.size < size {
debug!(LOGGER,
debug!(
LOGGER,
"Flush {} bytes, not enough data: have {}",
size,
self.size);
self.size
);
return Err(AdapterError::NotEnoughData);
}
@ -213,19 +225,23 @@ impl Adapter {
let front_size = self.deque.front().unwrap().get_size() - self.skip;
if front_size <= left {
trace!(LOGGER,
trace!(
LOGGER,
"Flushing whole {:?}, {} more to go",
self.deque.front(),
left - front_size);
left - front_size
);
self.deque.pop_front();
self.size -= front_size;
self.skip = 0;
left -= front_size;
} else {
trace!(LOGGER,
trace!(
LOGGER,
"Flushing partial {:?}, {} more left",
self.deque.front(),
front_size - left);
front_size - left
);
self.skip += left;
self.size -= left;
left = 0;

54
gst-plugin/src/buffer.rs
View file

@ -82,13 +82,15 @@ impl Buffer {
let size = vec.len();
let data = vec.as_mut_ptr();
let user_data = Box::into_raw(vec);
gst::gst_buffer_new_wrapped_full(gst::GstMemoryFlags::empty(),
gst::gst_buffer_new_wrapped_full(
gst::GstMemoryFlags::empty(),
data as glib::gpointer,
maxsize,
0,
size,
user_data as glib::gpointer,
Some(Buffer::vec_drop))
Some(Buffer::vec_drop),
)
};
if raw.is_null() {
@ -101,9 +103,11 @@ impl Buffer {
pub fn map_read(&self) -> Option<ReadBufferMap> {
let mut map_info: gst::GstMapInfo = unsafe { mem::zeroed() };
let res = unsafe {
gst::gst_buffer_map(self.as_mut_ptr() as *mut gst::GstBuffer,
gst::gst_buffer_map(
self.as_mut_ptr() as *mut gst::GstBuffer,
&mut map_info,
gst::GST_MAP_READ)
gst::GST_MAP_READ,
)
};
if res == glib::GTRUE {
Some(ReadBufferMap {
@ -161,8 +165,10 @@ impl Buffer {
pub fn append(buffer: GstRc<Buffer>, other: GstRc<Buffer>) -> GstRc<Buffer> {
unsafe {
GstRc::from_owned_ptr(gst::gst_buffer_append(buffer.into_ptr() as *mut gst::GstBuffer,
other.into_ptr() as *mut gst::GstBuffer))
GstRc::from_owned_ptr(gst::gst_buffer_append(
buffer.into_ptr() as *mut gst::GstBuffer,
other.into_ptr() as *mut gst::GstBuffer,
))
}
}
@ -170,10 +176,12 @@ impl Buffer {
let size_real = size.unwrap_or(usize::MAX);
let raw = unsafe {
gst::gst_buffer_copy_region(self.as_mut_ptr(),
gst::gst_buffer_copy_region(
self.as_mut_ptr(),
gst::GST_BUFFER_COPY_ALL,
offset,
size_real)
size_real,
)
};
if raw.is_null() {
@ -194,7 +202,11 @@ impl Buffer {
gst::gst_buffer_fill(self.as_mut_ptr(), offset, src as glib::gconstpointer, size)
};
if copied == size { Ok(()) } else { Err(copied) }
if copied == size {
Ok(())
} else {
Err(copied)
}
}
pub fn copy_to_slice(&self, offset: usize, slice: &mut [u8]) -> Result<(), usize> {
@ -208,7 +220,11 @@ impl Buffer {
gst::gst_buffer_extract(self.as_mut_ptr(), offset, dest as glib::gpointer, size)
};
if copied == size { Ok(()) } else { Err(copied) }
if copied == size {
Ok(())
} else {
Err(copied)
}
}
pub fn get_size(&self) -> usize {
@ -219,11 +235,13 @@ impl Buffer {
let mut maxsize: usize = 0;
unsafe {
gst::gst_buffer_get_sizes_range(self.as_mut_ptr(),
gst::gst_buffer_get_sizes_range(
self.as_mut_ptr(),
0,
-1,
ptr::null_mut(),
&mut maxsize as *mut usize);
&mut maxsize as *mut usize,
);
};
maxsize
@ -270,7 +288,11 @@ impl Buffer {
pub fn get_pts(&self) -> Option<u64> {
let pts = self.0.pts;
if pts == u64::MAX { None } else { Some(pts) }
if pts == u64::MAX {
None
} else {
Some(pts)
}
}
pub fn set_pts(&mut self, pts: Option<u64>) {
@ -281,7 +303,11 @@ impl Buffer {
pub fn get_dts(&self) -> Option<u64> {
let dts = self.0.dts;
if dts == u64::MAX { None } else { Some(dts) }
if dts == u64::MAX {
None
} else {
Some(dts)
}
}
pub fn set_dts(&mut self, dts: Option<u64>) {

14
gst-plugin/src/bytes.rs
View file

@ -6,7 +6,7 @@
// option. This file may not be copied, modified, or distributed
// except according to those terms.
pub use byteorder::{ReadBytesExt, WriteBytesExt, LittleEndian, BigEndian};
pub use byteorder::{BigEndian, LittleEndian, ReadBytesExt, WriteBytesExt};
use std::io;
pub trait ReadBytesExtShort: io::Read {
@ -72,7 +72,11 @@ pub trait ReadBytesExtShort: io::Read {
}
}
impl<T> ReadBytesExtShort for T where T: ReadBytesExt {}
impl<T> ReadBytesExtShort for T
where
T: ReadBytesExt,
{
}
pub trait WriteBytesExtShort: WriteBytesExt {
fn write_u16le(&mut self, n: u16) -> io::Result<()> {
@ -137,4 +141,8 @@ pub trait WriteBytesExtShort: WriteBytesExt {
}
}
impl<T> WriteBytesExtShort for T where T: WriteBytesExt {}
impl<T> WriteBytesExtShort for T
where
T: WriteBytesExt,
{
}

39
gst-plugin/src/caps.rs
View file

@ -88,7 +88,9 @@ impl Caps {
return None;
}
Some(Structure::from_borrowed_ptr(structure as *const gst::GstStructure))
Some(Structure::from_borrowed_ptr(
structure as *const gst::GstStructure,
))
}
}
@ -99,7 +101,9 @@ impl Caps {
return None;
}
Some(Structure::from_borrowed_mut_ptr(structure as *mut gst::GstStructure))
Some(Structure::from_borrowed_mut_ptr(
structure as *mut gst::GstStructure,
))
}
}
@ -146,24 +150,35 @@ mod tests {
fn test_simple() {
init();
let caps = Caps::new_simple("foo/bar",
&[("int", 12.into()),
let caps = Caps::new_simple(
"foo/bar",
&[
("int", 12.into()),
("bool", true.into()),
("string", "bla".into()),
("fraction", (1, 2).into()),
("array", vec![1.into(), 2.into()].into())]);
assert_eq!(caps.to_string(),
("array", vec![1.into(), 2.into()].into()),
],
);
assert_eq!(
caps.to_string(),
"foo/bar, int=(int)12, bool=(boolean)true, string=(string)bla, \
fraction=(fraction)1/2, array=(int)< 1, 2 >");
fraction=(fraction)1/2, array=(int)< 1, 2 >"
);
let s = caps.get_structure(0).unwrap();
assert_eq!(s,
OwnedStructure::new("foo/bar",
&[("int", 12.into()),
assert_eq!(
s,
OwnedStructure::new(
"foo/bar",
&[
("int", 12.into()),
("bool", true.into()),
("string", "bla".into()),
("fraction", (1, 2).into()),
("array", vec![1.into(), 2.into()].into())])
.as_ref());
("array", vec![1.into(), 2.into()].into())
]
).as_ref()
);
}
}

144
gst-plugin/src/demuxer.rs
View file

@ -57,16 +57,18 @@ pub enum HandleBufferResult {
}
pub trait Demuxer {
fn start(&mut self,
fn start(
&mut self,
upstream_size: Option<u64>,
random_access: bool)
-> Result<(), ErrorMessage>;
random_access: bool,
) -> Result<(), ErrorMessage>;
fn stop(&mut self) -> Result<(), ErrorMessage>;
fn seek(&mut self, start: u64, stop: Option<u64>) -> Result<SeekResult, ErrorMessage>;
fn handle_buffer(&mut self,
buffer: Option<GstRc<Buffer>>)
-> Result<HandleBufferResult, FlowError>;
fn handle_buffer(
&mut self,
buffer: Option<GstRc<Buffer>>,
) -> Result<HandleBufferResult, FlowError>;
fn end_of_stream(&mut self) -> Result<(), ErrorMessage>;
fn is_seekable(&self) -> bool;
@ -102,11 +104,15 @@ impl DemuxerWrapper {
fn new(raw: *mut gst::GstElement, demuxer: Box<Demuxer>) -> DemuxerWrapper {
DemuxerWrapper {
raw: raw,
logger: Logger::root(GstDebugDrain::new(Some(unsafe { &Element::new(raw) }),
logger: Logger::root(
GstDebugDrain::new(
Some(unsafe { &Element::new(raw) }),
"rsdemux",
0,
"Rust demuxer base class"),
o!()),
"Rust demuxer base class",
),
o!(),
),
demuxer: Mutex::new(demuxer),
panicked: AtomicBool::new(false),
}
@ -115,10 +121,12 @@ impl DemuxerWrapper {
fn start(&self, upstream_size: u64, random_access: bool) -> bool {
let demuxer = &mut self.demuxer.lock().unwrap();
debug!(self.logger,
debug!(
self.logger,
"Starting with upstream size {} and random access {}",
upstream_size,
random_access);
random_access
);
let upstream_size = if upstream_size == u64::MAX {
None
@ -250,19 +258,24 @@ impl DemuxerWrapper {
fn handle_buffer(&self, buffer: GstRc<Buffer>) -> gst::GstFlowReturn {
extern "C" {
fn gst_rs_demuxer_stream_eos(raw: *mut gst::GstElement, index: u32);
fn gst_rs_demuxer_add_stream(raw: *mut gst::GstElement,
fn gst_rs_demuxer_add_stream(
raw: *mut gst::GstElement,
index: u32,
caps: *const gst::GstCaps,
stream_id: *const c_char);
stream_id: *const c_char,
);
fn gst_rs_demuxer_added_all_streams(raw: *mut gst::GstElement);
// fn gst_rs_demuxer_remove_all_streams(raw: *mut gst::GstElement);
fn gst_rs_demuxer_stream_format_changed(raw: *mut gst::GstElement,
fn gst_rs_demuxer_stream_format_changed(
raw: *mut gst::GstElement,
index: u32,
caps: *const gst::GstCaps);
fn gst_rs_demuxer_stream_push_buffer(raw: *mut gst::GstElement,
caps: *const gst::GstCaps,
);
fn gst_rs_demuxer_stream_push_buffer(
raw: *mut gst::GstElement,
index: u32,
buffer: *mut gst::GstBuffer)
-> gst::GstFlowReturn;
buffer: *mut gst::GstBuffer,
) -> gst::GstFlowReturn;
};
let mut res = {
@ -275,8 +288,9 @@ impl DemuxerWrapper {
Err(flow_error) => {
error!(self.logger, "Failed handling buffer: {:?}", flow_error);
match flow_error {
FlowError::NotNegotiated(ref msg) |
FlowError::Error(ref msg) => self.post_message(msg),
FlowError::NotNegotiated(ref msg) | FlowError::Error(ref msg) => {
self.post_message(msg)
}
_ => (),
}
return flow_error.to_native();
@ -296,34 +310,40 @@ impl DemuxerWrapper {
let stream_id_cstr = CString::new(stream.stream_id.as_bytes()).unwrap();
unsafe {
gst_rs_demuxer_add_stream(self.raw,
gst_rs_demuxer_add_stream(
self.raw,
stream.index,
stream.caps.as_ptr(),
stream_id_cstr.as_ptr());
stream_id_cstr.as_ptr(),
);
}
}
HandleBufferResult::HaveAllStreams => unsafe {
gst_rs_demuxer_added_all_streams(self.raw);
},
HandleBufferResult::StreamChanged(stream) => unsafe {
gst_rs_demuxer_stream_format_changed(self.raw,
gst_rs_demuxer_stream_format_changed(
self.raw,
stream.index,
stream.caps.as_ptr());
stream.caps.as_ptr(),
);
},
HandleBufferResult::StreamsChanged(streams) => {
for stream in streams {
HandleBufferResult::StreamsChanged(streams) => for stream in streams {
unsafe {
gst_rs_demuxer_stream_format_changed(self.raw,
gst_rs_demuxer_stream_format_changed(
self.raw,
stream.index,
stream.caps.as_ptr());
}
}
stream.caps.as_ptr(),
);
}
},
HandleBufferResult::BufferForStream(index, buffer) => {
let flow_ret = unsafe {
gst_rs_demuxer_stream_push_buffer(self.raw,
gst_rs_demuxer_stream_push_buffer(
self.raw,
index,
buffer.into_ptr() as *mut gst::GstBuffer)
buffer.into_ptr() as *mut gst::GstBuffer,
)
};
if flow_ret != gst::GST_FLOW_OK {
return flow_ret;
@ -352,8 +372,9 @@ impl DemuxerWrapper {
Err(flow_error) => {
error!(self.logger, "Failed calling again: {:?}", flow_error);
match flow_error {
FlowError::NotNegotiated(ref msg) |
FlowError::Error(ref msg) => self.post_message(msg),
FlowError::NotNegotiated(ref msg) | FlowError::Error(ref msg) => {
self.post_message(msg)
}
_ => (),
}
return flow_error.to_native();
@ -384,9 +405,10 @@ impl DemuxerWrapper {
}
#[no_mangle]
pub unsafe extern "C" fn demuxer_new(demuxer: *mut gst::GstElement,
create_instance: fn(Element) -> Box<Demuxer>)
-> *mut DemuxerWrapper {
pub unsafe extern "C" fn demuxer_new(
demuxer: *mut gst::GstElement,
create_instance: fn(Element) -> Box<Demuxer>,
) -> *mut DemuxerWrapper {
let instance = create_instance(Element::new(demuxer));
Box::into_raw(Box::new(DemuxerWrapper::new(demuxer, instance)))
}
@ -397,10 +419,11 @@ pub unsafe extern "C" fn demuxer_drop(ptr: *mut DemuxerWrapper) {
}
#[no_mangle]
pub unsafe extern "C" fn demuxer_start(ptr: *const DemuxerWrapper,
pub unsafe extern "C" fn demuxer_start(
ptr: *const DemuxerWrapper,
upstream_size: u64,
random_access: glib::gboolean)
-> glib::gboolean {
random_access: glib::gboolean,
) -> glib::gboolean {
let wrap: &DemuxerWrapper = &*ptr;
panic_to_error!(wrap, glib::GFALSE, {
@ -439,9 +462,10 @@ pub unsafe extern "C" fn demuxer_is_seekable(ptr: *const DemuxerWrapper) -> glib
}
#[no_mangle]
pub unsafe extern "C" fn demuxer_get_position(ptr: *const DemuxerWrapper,
position: *mut u64)
-> glib::gboolean {
pub unsafe extern "C" fn demuxer_get_position(
ptr: *const DemuxerWrapper,
position: *mut u64,
) -> glib::gboolean {
let wrap: &DemuxerWrapper = &*ptr;
panic_to_error!(wrap, glib::GFALSE, {
@ -451,9 +475,10 @@ pub unsafe extern "C" fn demuxer_get_position(ptr: *const DemuxerWrapper,
}
#[no_mangle]
pub unsafe extern "C" fn demuxer_get_duration(ptr: *const DemuxerWrapper,
duration: *mut u64)
-> glib::gboolean {
pub unsafe extern "C" fn demuxer_get_duration(
ptr: *const DemuxerWrapper,
duration: *mut u64,
) -> glib::gboolean {
let wrap: &DemuxerWrapper = &*ptr;
panic_to_error!(wrap, glib::GFALSE, {
@ -463,11 +488,12 @@ pub unsafe extern "C" fn demuxer_get_duration(ptr: *const DemuxerWrapper,
}
#[no_mangle]
pub unsafe extern "C" fn demuxer_seek(ptr: *mut DemuxerWrapper,
pub unsafe extern "C" fn demuxer_seek(
ptr: *mut DemuxerWrapper,
start: u64,
stop: u64,
offset: *mut u64)
-> glib::gboolean {
offset: *mut u64,
) -> glib::gboolean {
let wrap: &mut DemuxerWrapper = &mut *ptr;
@ -483,9 +509,10 @@ pub unsafe extern "C" fn demuxer_seek(ptr: *mut DemuxerWrapper,
}
#[no_mangle]
pub unsafe extern "C" fn demuxer_handle_buffer(ptr: *mut DemuxerWrapper,
buffer: *mut gst::GstBuffer)
-> gst::GstFlowReturn {
pub unsafe extern "C" fn demuxer_handle_buffer(
ptr: *mut DemuxerWrapper,
buffer: *mut gst::GstBuffer,
) -> gst::GstFlowReturn {
let wrap: &mut DemuxerWrapper = &mut *ptr;
panic_to_error!(wrap, gst::GST_FLOW_ERROR, {
@ -517,7 +544,8 @@ pub struct DemuxerInfo<'a> {
pub fn demuxer_register(plugin: &Plugin, demuxer_info: &DemuxerInfo) {
extern "C" {
fn gst_rs_demuxer_register(plugin: *const gst::GstPlugin,
fn gst_rs_demuxer_register(
plugin: *const gst::GstPlugin,
name: *const c_char,
long_name: *const c_char,
description: *const c_char,
@ -526,8 +554,8 @@ pub fn demuxer_register(plugin: &Plugin, demuxer_info: &DemuxerInfo) {
rank: i32,
create_instance: *const c_void,
input_caps: *const gst::GstCaps,
output_caps: *const gst::GstCaps)
-> glib::gboolean;
output_caps: *const gst::GstCaps,
) -> glib::gboolean;
}
let cname = CString::new(demuxer_info.name).unwrap();
@ -537,7 +565,8 @@ pub fn demuxer_register(plugin: &Plugin, demuxer_info: &DemuxerInfo) {
let cauthor = CString::new(demuxer_info.author).unwrap();
unsafe {
gst_rs_demuxer_register(plugin.as_ptr(),
gst_rs_demuxer_register(
plugin.as_ptr(),
cname.as_ptr(),
clong_name.as_ptr(),
cdescription.as_ptr(),
@ -546,6 +575,7 @@ pub fn demuxer_register(plugin: &Plugin, demuxer_info: &DemuxerInfo) {
demuxer_info.rank,
demuxer_info.create_instance as *const c_void,
demuxer_info.input_caps.as_ptr(),
demuxer_info.output_caps.as_ptr());
demuxer_info.output_caps.as_ptr(),
);
}
}

41
gst-plugin/src/error.rs
View file

@ -81,13 +81,14 @@ pub struct ErrorMessage {
}
impl ErrorMessage {
pub fn new<T: ToGError>(error: &T,
pub fn new<T: ToGError>(
error: &T,
message: Option<Cow<str>>,
debug: Option<Cow<str>>,
filename: &'static str,
function: &'static str,
line: u32)
-> ErrorMessage {
line: u32,
) -> ErrorMessage {
let (gdomain, gcode) = error.to_gerror();
ErrorMessage {
@ -125,7 +126,8 @@ impl ErrorMessage {
let function_cstr = CString::new(function.as_bytes()).unwrap();
let function_ptr = function_cstr.as_ptr();
gst::gst_element_message_full(element,
gst::gst_element_message_full(
element,
gst::GST_MESSAGE_ERROR,
error_domain,
error_code,
@ -133,7 +135,8 @@ impl ErrorMessage {
glib::g_strndup(debug_ptr, debug_len),
file_ptr,
function_ptr,
line as i32);
line as i32,
);
}
}
@ -160,18 +163,18 @@ impl Display for FlowError {
fn fmt(&self, f: &mut Formatter) -> Result<(), FmtError> {
match *self {
FlowError::Flushing | FlowError::Eos => f.write_str(self.description()),
FlowError::NotNegotiated(ref m) => {
f.write_fmt(format_args!("{}: {} ({})",
FlowError::NotNegotiated(ref m) => f.write_fmt(format_args!(
"{}: {} ({})",
self.description(),
m.message.as_ref().map_or("None", |s| s.as_str()),
m.debug.as_ref().map_or("None", |s| s.as_str())))
}
FlowError::Error(ref m) => {
f.write_fmt(format_args!("{}: {} ({})",
m.debug.as_ref().map_or("None", |s| s.as_str())
)),
FlowError::Error(ref m) => f.write_fmt(format_args!(
"{}: {} ({})",
self.description(),
m.message.as_ref().map_or("None", |s| s.as_str()),
m.debug.as_ref().map_or("None", |s| s.as_str())))
}
m.debug.as_ref().map_or("None", |s| s.as_str())
)),
}
}
}
@ -220,15 +223,19 @@ impl UriError {
pub unsafe fn into_gerror(self, err: *mut *mut glib::GError) {
if let Some(msg) = self.message {
let cmsg = CString::new(msg.as_str()).unwrap();
glib::g_set_error_literal(err,
glib::g_set_error_literal(
err,
gst::gst_uri_error_quark(),
self.error_kind as i32,
cmsg.as_ptr());
cmsg.as_ptr(),
);
} else {
glib::g_set_error_literal(err,
glib::g_set_error_literal(
err,
gst::gst_uri_error_quark(),
self.error_kind as i32,
ptr::null());
ptr::null(),
);
}
}
}

27
gst-plugin/src/log.rs
View file

@ -8,7 +8,7 @@
use libc::c_char;
use std::ffi::CString;
use slog::{Drain, Record, OwnedKVList, Never, Level};
use slog::{Drain, Level, Never, OwnedKVList, Record};
use std::fmt;
use std::ptr;
use std::mem;
@ -24,16 +24,18 @@ pub struct GstDebugDrain {
}
impl GstDebugDrain {
pub fn new(element: Option<&Element>,
pub fn new(
element: Option<&Element>,
name: &str,
color: u32,
description: &str)
-> GstDebugDrain {
description: &str,
) -> GstDebugDrain {
extern "C" {
fn _gst_debug_category_new(name: *const c_char,
fn _gst_debug_category_new(
name: *const c_char,
color: u32,
description: *const c_char)
-> *mut gst::GstDebugCategory;
description: *const c_char,
) -> *mut gst::GstDebugCategory;
}
let name_cstr = CString::new(name.as_bytes()).unwrap();
@ -99,16 +101,19 @@ impl Drain for GstDebugDrain {
let message_cstr = CString::new(fmt::format(*record.msg()).as_bytes()).unwrap();
unsafe {
let element = gobject::g_weak_ref_get(&*self.element as *const gobject::GWeakRef as
*mut gobject::GWeakRef);
let element = gobject::g_weak_ref_get(
&*self.element as *const gobject::GWeakRef as *mut gobject::GWeakRef,
);
gst::gst_debug_log(self.category,
gst::gst_debug_log(
self.category,
level,
file_cstr.as_ptr(),
function_cstr.as_ptr(),
record.line() as i32,
element as *mut gobject::GObject,
message_cstr.as_ptr());
message_cstr.as_ptr(),
);
if !element.is_null() {
gst::gst_object_unref(element as *mut gst::GstObject);

24
gst-plugin/src/miniobject.rs
View file

@ -6,7 +6,7 @@
// option. This file may not be copied, modified, or distributed
// except according to those terms.
use std::{fmt, ops, borrow};
use std::{borrow, fmt, ops};
use std::mem;
use std::marker::PhantomData;
@ -41,9 +41,10 @@ impl<T: MiniObject> GstRc<T> {
return &mut *self.0;
}
self.0 = T::from_mut_ptr(gst::gst_mini_object_make_writable(self.as_mut_ptr() as
*mut gst::GstMiniObject) as
*mut T::PtrType);
self.0 = T::from_mut_ptr(
gst::gst_mini_object_make_writable(self.as_mut_ptr() as *mut gst::GstMiniObject) as
*mut T::PtrType,
);
assert!(self.is_writable());
&mut *self.0
@ -60,16 +61,16 @@ impl<T: MiniObject> GstRc<T> {
pub fn copy(&self) -> Self {
unsafe {
GstRc::from_owned_ptr(gst::gst_mini_object_copy(self.as_ptr() as
*const gst::GstMiniObject) as
*const T::PtrType)
GstRc::from_owned_ptr(
gst::gst_mini_object_copy(self.as_ptr() as *const gst::GstMiniObject) as
*const T::PtrType,
)
}
}
fn is_writable(&self) -> bool {
(unsafe {
gst::gst_mini_object_is_writable(self.as_ptr() as *const gst::GstMiniObject)
} == glib::GTRUE)
(unsafe { gst::gst_mini_object_is_writable(self.as_ptr() as *const gst::GstMiniObject) } ==
glib::GTRUE)
}
pub unsafe fn into_ptr(self) -> *mut T::PtrType {
@ -132,7 +133,8 @@ impl<T: MiniObject + fmt::Display> fmt::Display for GstRc<T> {
}
pub unsafe trait MiniObject
where Self: Sized
where
Self: Sized,
{
type PtrType;

105
gst-plugin/src/sink.rs
View file

@ -77,11 +77,15 @@ impl SinkWrapper {
fn new(raw: *mut gst::GstElement, sink: Box<Sink>) -> SinkWrapper {
SinkWrapper {
raw: raw,
logger: Logger::root(GstDebugDrain::new(Some(unsafe { &Element::new(raw) }),
logger: Logger::root(
GstDebugDrain::new(
Some(unsafe { &Element::new(raw) }),
"rssink",
0,
"Rust sink base class"),
o!()),
"Rust sink base class",
),
o!(),
),
uri: Mutex::new((None, false)),
uri_validator: sink.uri_validator(),
sink: Mutex::new(sink),
@ -95,7 +99,10 @@ impl SinkWrapper {
debug!(self.logger, "Setting URI {:?}", uri_str);
if uri_storage.1 {
return Err(UriError::new(UriErrorKind::BadState, Some("Already started".to_string())));
return Err(UriError::new(
UriErrorKind::BadState,
Some("Already started".to_string()),
));
}
uri_storage.0 = None;
@ -107,10 +114,10 @@ impl SinkWrapper {
uri_storage.0 = Some(uri);
Ok(())
}
Err(err) => {
Err(UriError::new(UriErrorKind::BadUri,
Some(format!("Failed to parse URI '{}': {}", uri_str, err))))
}
Err(err) => Err(UriError::new(
UriErrorKind::BadUri,
Some(format!("Failed to parse URI '{}': {}", uri_str, err)),
)),
}
} else {
Ok(())
@ -119,10 +126,7 @@ impl SinkWrapper {
fn get_uri(&self) -> Option<String> {
let uri_storage = &self.uri.lock().unwrap();
uri_storage
.0
.as_ref()
.map(|uri| String::from(uri.as_str()))
uri_storage.0.as_ref().map(|uri| String::from(uri.as_str()))
}
fn start(&self) -> bool {
@ -187,8 +191,9 @@ impl SinkWrapper {
Err(flow_error) => {
error!(self.logger, "Failed to render: {:?}", flow_error);
match flow_error {
FlowError::NotNegotiated(ref msg) |
FlowError::Error(ref msg) => self.post_message(msg),
FlowError::NotNegotiated(ref msg) | FlowError::Error(ref msg) => {
self.post_message(msg)
}
_ => (),
}
flow_error.to_native()
@ -203,10 +208,11 @@ impl SinkWrapper {
}
}
unsafe fn sink_set_uri(ptr: *const RsSink,
unsafe fn sink_set_uri(
ptr: *const RsSink,
uri_ptr: *const c_char,
cerr: *mut *mut glib::GError)
-> glib::gboolean {
cerr: *mut *mut glib::GError,
) -> glib::gboolean {
let sink = &*(ptr as *const RsSink);
let wrap: &SinkWrapper = &*sink.wrap;
@ -266,16 +272,15 @@ unsafe extern "C" fn sink_stop(ptr: *mut gst_base::GstBaseSink) -> glib::gboolea
})
}
unsafe extern "C" fn sink_render(ptr: *mut gst_base::GstBaseSink,
buffer: *mut gst::GstBuffer)
-> gst::GstFlowReturn {
unsafe extern "C" fn sink_render(
ptr: *mut gst_base::GstBaseSink,
buffer: *mut gst::GstBuffer,
) -> gst::GstFlowReturn {
let sink = &*(ptr as *const RsSink);
let wrap: &SinkWrapper = &*sink.wrap;
let buffer: &Buffer = Buffer::from_ptr(buffer);
panic_to_error!(wrap, gst::GST_FLOW_ERROR, {
wrap.render(buffer)
})
panic_to_error!(wrap, gst::GST_FLOW_ERROR, { wrap.render(buffer) })
}
pub struct SinkInfo {
@ -314,10 +319,12 @@ unsafe extern "C" fn sink_finalize(obj: *mut gobject::GObject) {
parent_klass.finalize.map(|f| f(obj));
}
unsafe extern "C" fn sink_set_property(obj: *mut gobject::GObject,
unsafe extern "C" fn sink_set_property(
obj: *mut gobject::GObject,
id: u32,
value: *mut gobject::GValue,
_pspec: *mut gobject::GParamSpec) {
_pspec: *mut gobject::GParamSpec,
) {
let sink = &*(obj as *const RsSink);
match id {
@ -329,10 +336,12 @@ unsafe extern "C" fn sink_set_property(obj: *mut gobject::GObject,
}
}
unsafe extern "C" fn sink_get_property(obj: *mut gobject::GObject,
unsafe extern "C" fn sink_get_property(
obj: *mut gobject::GObject,
id: u32,
value: *mut gobject::GValue,
_pspec: *mut gobject::GParamSpec) {
_pspec: *mut gobject::GParamSpec,
) {
let sink = &*(obj as *const RsSink);
match id {
@ -362,12 +371,17 @@ unsafe extern "C" fn sink_class_init(klass: glib::gpointer, klass_data: glib::gp
let nick_cstr = CString::new("URI").unwrap();
let blurb_cstr = CString::new("URI to read from").unwrap();
gobject::g_object_class_install_property(klass as *mut gobject::GObjectClass, 1,
gobject::g_param_spec_string(name_cstr.as_ptr(),
gobject::g_object_class_install_property(
klass as *mut gobject::GObjectClass,
1,
gobject::g_param_spec_string(
name_cstr.as_ptr(),
nick_cstr.as_ptr(),
blurb_cstr.as_ptr(),
ptr::null_mut(),
gobject::G_PARAM_READWRITE));
gobject::G_PARAM_READWRITE,
),
);
}
{
@ -378,18 +392,22 @@ unsafe extern "C" fn sink_class_init(klass: glib::gpointer, klass_data: glib::gp
let description_cstr = CString::new(sink_info.classification.clone()).unwrap();
let author_cstr = CString::new(sink_info.author.clone()).unwrap();
gst::gst_element_class_set_static_metadata(element_klass,
gst::gst_element_class_set_static_metadata(
element_klass,
longname_cstr.into_raw(),
classification_cstr.into_raw(),
description_cstr.into_raw(),
author_cstr.into_raw());
author_cstr.into_raw(),
);
let caps = Caps::new_any();
let templ_name = CString::new("sink").unwrap();
let pad_template = gst::gst_pad_template_new(templ_name.into_raw(),
let pad_template = gst::gst_pad_template_new(
templ_name.into_raw(),
gst::GST_PAD_SINK,
gst::GST_PAD_ALWAYS,
caps.as_ptr() as *mut gst::GstCaps);
caps.as_ptr() as *mut gst::GstCaps,
);
gst::gst_element_class_add_pad_template(element_klass, pad_template);
}
@ -418,8 +436,10 @@ unsafe extern "C" fn sink_init(instance: *mut gobject::GTypeInstance, klass: gli
sink.sink_info = sink_info;
let wrap = Box::new(SinkWrapper::new(&mut sink.parent.element,
(sink_info.create_instance)(Element::new(&mut sink.parent.element))));
let wrap = Box::new(SinkWrapper::new(
&mut sink.parent.element,
(sink_info.create_instance)(Element::new(&mut sink.parent.element)),
));
sink.wrap = Box::into_raw(wrap);
gst_base::gst_base_sink_set_sync(&mut sink.parent, glib::GFALSE);
@ -439,10 +459,11 @@ unsafe extern "C" fn sink_uri_handler_get_uri(uri_handler: *mut gst::GstURIHandl
sink_get_uri(uri_handler as *const RsSink)
}
unsafe extern "C" fn sink_uri_handler_set_uri(uri_handler: *mut gst::GstURIHandler,
unsafe extern "C" fn sink_uri_handler_set_uri(
uri_handler: *mut gst::GstURIHandler,
uri: *const c_char,
err: *mut *mut glib::GError)
-> glib::gboolean {
err: *mut *mut glib::GError,
) -> glib::gboolean {
sink_set_uri(uri_handler as *const RsSink, uri, err)
}
@ -481,10 +502,12 @@ pub fn sink_register(plugin: &Plugin, sink_info: SinkInfo) {
value_table: ptr::null(),
};
let type_ = gobject::g_type_register_static(parent_type,
let type_ = gobject::g_type_register_static(
parent_type,
type_name_cstr.as_ptr(),
&type_info,
gobject::GTypeFlags::empty());
gobject::GTypeFlags::empty(),
);
let iface_info = gobject::GInterfaceInfo {
interface_init: Some(sink_uri_handler_init),

134
gst-plugin/src/source.rs
View file

@ -80,11 +80,15 @@ impl SourceWrapper {
fn new(raw: *mut gst::GstElement, source: Box<Source>) -> SourceWrapper {
SourceWrapper {
raw: raw,
logger: Logger::root(GstDebugDrain::new(Some(unsafe { &Element::new(raw) }),
logger: Logger::root(
GstDebugDrain::new(
Some(unsafe { &Element::new(raw) }),
"rssrc",
0,
"Rust source base class"),
o!()),
"Rust source base class",
),
o!(),
),
uri: Mutex::new((None, false)),
uri_validator: source.uri_validator(),
source: Mutex::new(source),
@ -98,7 +102,10 @@ impl SourceWrapper {
debug!(self.logger, "Setting URI {:?}", uri_str);
if uri_storage.1 {
return Err(UriError::new(UriErrorKind::BadState, Some("Already started".to_string())));
return Err(UriError::new(
UriErrorKind::BadState,
Some("Already started".to_string()),
));
}
uri_storage.0 = None;
@ -110,10 +117,10 @@ impl SourceWrapper {
uri_storage.0 = Some(uri);
Ok(())
}
Err(err) => {
Err(UriError::new(UriErrorKind::BadUri,
Some(format!("Failed to parse URI '{}': {}", uri_str, err))))
}
Err(err) => Err(UriError::new(
UriErrorKind::BadUri,
Some(format!("Failed to parse URI '{}': {}", uri_str, err)),
)),
}
} else {
Ok(())
@ -122,10 +129,7 @@ impl SourceWrapper {
fn get_uri(&self) -> Option<String> {
let uri_storage = &self.uri.lock().unwrap();
uri_storage
.0
.as_ref()
.map(|uri| String::from(uri.as_str()))
uri_storage.0.as_ref().map(|uri| String::from(uri.as_str()))
}
fn is_seekable(&self) -> bool {
@ -193,19 +197,22 @@ impl SourceWrapper {
fn fill(&self, offset: u64, length: u32, buffer: &mut Buffer) -> gst::GstFlowReturn {
let source = &mut self.source.lock().unwrap();
trace!(self.logger,
trace!(
self.logger,
"Filling buffer {:?} with offset {} and length {}",
buffer,
offset,
length);
length
);
match source.fill(offset, length, buffer) {
Ok(()) => gst::GST_FLOW_OK,
Err(flow_error) => {
error!(self.logger, "Failed to fill: {:?}", flow_error);
match flow_error {
FlowError::NotNegotiated(ref msg) |
FlowError::Error(ref msg) => self.post_message(msg),
FlowError::NotNegotiated(ref msg) | FlowError::Error(ref msg) => {
self.post_message(msg)
}
_ => (),
}
flow_error.to_native()
@ -235,10 +242,11 @@ impl SourceWrapper {
}
}
unsafe fn source_set_uri(ptr: *const RsSrc,
unsafe fn source_set_uri(
ptr: *const RsSrc,
uri_ptr: *const c_char,
cerr: *mut *mut glib::GError)
-> glib::gboolean {
cerr: *mut *mut glib::GError,
) -> glib::gboolean {
let src = &*(ptr as *const RsSrc);
let wrap: &SourceWrapper = &*src.wrap;
@ -285,9 +293,10 @@ unsafe extern "C" fn source_is_seekable(ptr: *mut gst_base::GstBaseSrc) -> glib:
})
}
unsafe extern "C" fn source_get_size(ptr: *mut gst_base::GstBaseSrc,
size: *mut u64)
-> glib::gboolean {
unsafe extern "C" fn source_get_size(
ptr: *mut gst_base::GstBaseSrc,
size: *mut u64,
) -> glib::gboolean {
let src = &*(ptr as *const RsSrc);
let wrap: &SourceWrapper = &*src.wrap;
@ -323,23 +332,27 @@ unsafe extern "C" fn source_stop(ptr: *mut gst_base::GstBaseSrc) -> glib::gboole
})
}
unsafe extern "C" fn source_fill(ptr: *mut gst_base::GstBaseSrc,
unsafe extern "C" fn source_fill(
ptr: *mut gst_base::GstBaseSrc,
offset: u64,
length: u32,
buffer: *mut gst::GstBuffer)
-> gst::GstFlowReturn {
buffer: *mut gst::GstBuffer,
) -> gst::GstFlowReturn {
let src = &*(ptr as *const RsSrc);
let wrap: &SourceWrapper = &*src.wrap;
let buffer: &mut Buffer = <Buffer as MiniObject>::from_mut_ptr(buffer);
panic_to_error!(wrap, gst::GST_FLOW_ERROR, {
wrap.fill(offset, length, buffer)
})
panic_to_error!(
wrap,
gst::GST_FLOW_ERROR,
{ wrap.fill(offset, length, buffer) }
)
}
unsafe extern "C" fn source_seek(ptr: *mut gst_base::GstBaseSrc,
segment: *mut gst::GstSegment)
-> glib::gboolean {
unsafe extern "C" fn source_seek(
ptr: *mut gst_base::GstBaseSrc,
segment: *mut gst::GstSegment,
) -> glib::gboolean {
let src = &*(ptr as *const RsSrc);
let wrap: &SourceWrapper = &*src.wrap;
@ -392,10 +405,12 @@ unsafe extern "C" fn source_finalize(obj: *mut gobject::GObject) {
parent_klass.finalize.map(|f| f(obj));
}
unsafe extern "C" fn source_set_property(obj: *mut gobject::GObject,
unsafe extern "C" fn source_set_property(
obj: *mut gobject::GObject,
id: u32,
value: *mut gobject::GValue,
_pspec: *mut gobject::GParamSpec) {
_pspec: *mut gobject::GParamSpec,
) {
let src = &*(obj as *const RsSrc);
match id {
@ -407,10 +422,12 @@ unsafe extern "C" fn source_set_property(obj: *mut gobject::GObject,
}
}
unsafe extern "C" fn source_get_property(obj: *mut gobject::GObject,
unsafe extern "C" fn source_get_property(
obj: *mut gobject::GObject,
id: u32,
value: *mut gobject::GValue,
_pspec: *mut gobject::GParamSpec) {
_pspec: *mut gobject::GParamSpec,
) {
let src = &*(obj as *const RsSrc);
match id {
@ -441,12 +458,17 @@ unsafe extern "C" fn source_class_init(klass: glib::gpointer, klass_data: glib::
let nick_cstr = CString::new("URI").unwrap();
let blurb_cstr = CString::new("URI to read from").unwrap();
gobject::g_object_class_install_property(klass as *mut gobject::GObjectClass, 1,
gobject::g_param_spec_string(name_cstr.as_ptr(),
gobject::g_object_class_install_property(
klass as *mut gobject::GObjectClass,
1,
gobject::g_param_spec_string(
name_cstr.as_ptr(),
nick_cstr.as_ptr(),
blurb_cstr.as_ptr(),
ptr::null_mut(),
gobject::G_PARAM_READWRITE));
gobject::G_PARAM_READWRITE,
),
);
}
{
@ -457,18 +479,22 @@ unsafe extern "C" fn source_class_init(klass: glib::gpointer, klass_data: glib::
let description_cstr = CString::new(source_info.classification.clone()).unwrap();
let author_cstr = CString::new(source_info.author.clone()).unwrap();
gst::gst_element_class_set_static_metadata(element_klass,
gst::gst_element_class_set_static_metadata(
element_klass,
longname_cstr.into_raw(),
classification_cstr.into_raw(),
description_cstr.into_raw(),
author_cstr.into_raw());
author_cstr.into_raw(),
);
let caps = Caps::new_any();
let templ_name = CString::new("src").unwrap();
let pad_template = gst::gst_pad_template_new(templ_name.into_raw(),
let pad_template = gst::gst_pad_template_new(
templ_name.into_raw(),
gst::GST_PAD_SRC,
gst::GST_PAD_ALWAYS,
caps.as_ptr() as *mut gst::GstCaps);
caps.as_ptr() as *mut gst::GstCaps,
);
gst::gst_element_class_add_pad_template(element_klass, pad_template);
}
@ -500,8 +526,10 @@ unsafe extern "C" fn source_init(instance: *mut gobject::GTypeInstance, klass: g
src.source_info = source_info;
let wrap = Box::new(SourceWrapper::new(&mut src.parent.parent.element,
(source_info.create_instance)(Element::new(&mut src.parent.parent.element))));
let wrap = Box::new(SourceWrapper::new(
&mut src.parent.parent.element,
(source_info.create_instance)(Element::new(&mut src.parent.parent.element)),
));
src.wrap = Box::into_raw(wrap);
gst_base::gst_base_src_set_blocksize(&mut src.parent.parent, 4096);
@ -517,15 +545,17 @@ unsafe extern "C" fn source_uri_handler_get_protocols(type_: glib::GType) -> *co
(*src_klass.protocols).as_ptr()
}
unsafe extern "C" fn source_uri_handler_get_uri(uri_handler: *mut gst::GstURIHandler)
-> *mut c_char {
unsafe extern "C" fn source_uri_handler_get_uri(
uri_handler: *mut gst::GstURIHandler,
) -> *mut c_char {
source_get_uri(uri_handler as *const RsSrc)
}
unsafe extern "C" fn source_uri_handler_set_uri(uri_handler: *mut gst::GstURIHandler,
unsafe extern "C" fn source_uri_handler_set_uri(
uri_handler: *mut gst::GstURIHandler,
uri: *const c_char,
err: *mut *mut glib::GError)
-> glib::gboolean {
err: *mut *mut glib::GError,
) -> glib::gboolean {
source_set_uri(uri_handler as *const RsSrc, uri, err)
}
@ -568,10 +598,12 @@ pub fn source_register(plugin: &Plugin, source_info: SourceInfo) {
value_table: ptr::null(),
};
let type_ = gobject::g_type_register_static(parent_type,
let type_ = gobject::g_type_register_static(
parent_type,
type_name_cstr.as_ptr(),
&type_info,
gobject::GTypeFlags::empty());
gobject::GTypeFlags::empty(),
);
let iface_info = gobject::GInterfaceInfo {
interface_init: Some(source_uri_handler_init),

29
gst-plugin/src/streams.rs
View file

@ -39,20 +39,23 @@ bitflags! {
}
impl Stream {
pub fn new(stream_id: &str,
pub fn new(
stream_id: &str,
caps: Option<GstRc<Caps>>,
t: StreamType,
flags: StreamFlags)
-> Self {
flags: StreamFlags,
) -> Self {
let stream_id_cstr = CString::new(stream_id).unwrap();
let caps = caps.map(|caps| unsafe { caps.as_mut_ptr() })
.unwrap_or(ptr::null_mut());
Stream(unsafe {
gst::gst_stream_new(stream_id_cstr.as_ptr(),
gst::gst_stream_new(
stream_id_cstr.as_ptr(),
caps,
mem::transmute(t.bits()),
mem::transmute(flags.bits()))
mem::transmute(flags.bits()),
)
})
}
@ -118,7 +121,10 @@ impl Stream {
impl Clone for Stream {
fn clone(&self) -> Self {
unsafe { Stream(gst::gst_object_ref(self.0 as *mut gst::GstObject) as *mut gst::GstStream) }
unsafe {
Stream(gst::gst_object_ref(self.0 as *mut gst::GstObject) as
*mut gst::GstStream)
}
}
}
@ -131,8 +137,9 @@ impl Drop for Stream {
impl StreamCollection {
pub fn new(upstream_id: &str, streams: &[Stream]) -> Self {
let upstream_id_cstr = CString::new(upstream_id).unwrap();
let collection =
StreamCollection(unsafe { gst::gst_stream_collection_new(upstream_id_cstr.as_ptr()) });
let collection = StreamCollection(unsafe {
gst::gst_stream_collection_new(upstream_id_cstr.as_ptr())
});
for stream in streams {
unsafe { gst::gst_stream_collection_add_stream(collection.0, stream.clone().0) };
@ -196,7 +203,8 @@ impl<'a> Iterator for StreamCollectionIterator<'a> {
self.position += 1;
Some(unsafe {
Stream(gst::gst_object_ref(stream as *mut gst::GstObject) as *mut gst::GstStream)
Stream(gst::gst_object_ref(stream as *mut gst::GstObject) as
*mut gst::GstStream)
})
}
@ -227,7 +235,8 @@ impl<'a> DoubleEndedIterator for StreamCollectionIterator<'a> {
}
Some(unsafe {
Stream(gst::gst_object_ref(stream as *mut gst::GstObject) as *mut gst::GstStream)
Stream(gst::gst_object_ref(stream as *mut gst::GstObject) as
*mut gst::GstStream)
})
}
}

49
gst-plugin/src/structure.rs
View file

@ -11,7 +11,7 @@ use std::ptr;
use std::mem;
use std::ffi::{CStr, CString};
use std::ops::{Deref, DerefMut};
use std::borrow::{Borrow, ToOwned, BorrowMut};
use std::borrow::{Borrow, BorrowMut, ToOwned};
use std::marker::PhantomData;
use value::*;
@ -24,10 +24,10 @@ pub struct OwnedStructure(*mut Structure, PhantomData<Structure>);
impl OwnedStructure {
pub fn new_empty(name: &str) -> OwnedStructure {
let name_cstr = CString::new(name).unwrap();
OwnedStructure(unsafe {
gst::gst_structure_new_empty(name_cstr.as_ptr()) as *mut Structure
},
PhantomData)
OwnedStructure(
unsafe { gst::gst_structure_new_empty(name_cstr.as_ptr()) as *mut Structure },
PhantomData,
)
}
pub fn new(name: &str, values: &[(&str, Value)]) -> OwnedStructure {
@ -88,8 +88,10 @@ impl AsMut<Structure> for OwnedStructure {
impl Clone for OwnedStructure {
fn clone(&self) -> Self {
OwnedStructure(unsafe { gst::gst_structure_copy(&(*self.0).0) as *mut Structure },
PhantomData)
OwnedStructure(
unsafe { gst::gst_structure_copy(&(*self.0).0) as *mut Structure },
PhantomData,
)
}
}
@ -135,8 +137,10 @@ impl ToOwned for Structure {
type Owned = OwnedStructure;
fn to_owned(&self) -> OwnedStructure {
OwnedStructure(unsafe { gst::gst_structure_copy(&self.0) as *mut Structure },
PhantomData)
OwnedStructure(
unsafe { gst::gst_structure_copy(&self.0) as *mut Structure },
PhantomData,
)
}
}
@ -167,8 +171,7 @@ impl Structure {
}
pub fn get<'a, T: ValueType<'a>>(&'a self, name: &str) -> Option<TypedValueRef<'a, T>> {
self.get_value(name)
.and_then(TypedValueRef::from_value_ref)
self.get_value(name).and_then(TypedValueRef::from_value_ref)
}
pub fn get_value<'a>(&'a self, name: &str) -> Option<ValueRef<'a>> {
@ -331,7 +334,9 @@ pub struct Iter<'a> {
impl<'a> Iter<'a> {
pub fn new(structure: &'a Structure) -> Iter<'a> {
Iter { iter: FieldIterator::new(structure) }
Iter {
iter: FieldIterator::new(structure),
}
}
}
@ -385,17 +390,25 @@ mod tests {
assert_eq!(s.get::<&str>("f2").unwrap().get(), "bcd");
assert_eq!(s.get::<i32>("f3").unwrap().get(), 123i32);
assert_eq!(s.fields().collect::<Vec<_>>(), vec!["f1", "f2", "f3"]);
assert_eq!(s.iter()
assert_eq!(
s.iter()
.map(|(f, v)| (f, Value::from_value_ref(&v)))
.collect::<Vec<_>>(),
vec![("f1", Value::new("abc")),
vec![
("f1", Value::new("abc")),
("f2", Value::new("bcd")),
("f3", Value::new(123i32))]);
("f3", Value::new(123i32)),
]
);
let s2 = OwnedStructure::new("test",
&[("f1", "abc".into()),
let s2 = OwnedStructure::new(
"test",
&[
("f1", "abc".into()),
("f2", "bcd".into()),
("f3", 123i32.into())]);
("f3", 123i32.into()),
],
);
assert_eq!(s, s2);
}
}

27
gst-plugin/src/tags.rs
View file

@ -81,17 +81,20 @@ impl TagList {
}
pub fn add<'a, T: Tag<'a>>(&mut self, value: T::TagType, mode: MergeMode)
where T::TagType: Into<Value>
where
T::TagType: Into<Value>,
{
unsafe {
let v = value.into();
let mut gvalue = v.into_raw();
let tag_name = CString::new(T::tag_name()).unwrap();
gst::gst_tag_list_add_value(self.as_mut_ptr(),
gst::gst_tag_list_add_value(
self.as_mut_ptr(),
mode.to_ffi(),
tag_name.as_ptr(),
&gvalue);
&gvalue,
);
gobject::g_value_unset(&mut gvalue);
}
@ -252,8 +255,10 @@ mod tests {
tags.add::<Title>("some title".into(), MergeMode::Append);
tags.add::<Duration>((1000u64 * 1000 * 1000 * 120).into(), MergeMode::Append);
}
assert_eq!(tags.to_string(),
"taglist, title=(string)\"some\\ title\", duration=(guint64)120000000000;");
assert_eq!(
tags.to_string(),
"taglist, title=(string)\"some\\ title\", duration=(guint64)120000000000;"
);
}
#[test]
@ -269,10 +274,14 @@ mod tests {
}
assert_eq!(tags.get::<Title>().unwrap().get(), "some title");
assert_eq!(tags.get::<Duration>().unwrap().get(),
(1000u64 * 1000 * 1000 * 120));
assert_eq!(
tags.get::<Duration>().unwrap().get(),
(1000u64 * 1000 * 1000 * 120)
);
assert_eq!(tags.get_index::<Title>(0).unwrap().get(), "some title");
assert_eq!(tags.get_index::<Duration>(0).unwrap().get(),
(1000u64 * 1000 * 1000 * 120));
assert_eq!(
tags.get_index::<Duration>(0).unwrap().get(),
(1000u64 * 1000 * 1000 * 120)
);
}
}

9
gst-plugin/src/utils.rs
View file

@ -79,7 +79,8 @@ pub fn f64_to_fraction(val: f64) -> Option<Rational32> {
// Prevent overflow
if a != 0 &&
(n1 > (i32::MAX as u32) / a || d1 > (i32::MAX as u32) / a ||
a * n1 > (i32::MAX as u32) - n0 || a * d1 > (i32::MAX as u32) - d0) {
a * n1 > (i32::MAX as u32) - n0 || a * d1 > (i32::MAX as u32) - d0)
{
break;
}
@ -156,8 +157,10 @@ mod tests {
fn test_f64_to_fraction() {
assert_eq!(f64_to_fraction(2.0), Some(Rational32::new(2, 1)));
assert_eq!(f64_to_fraction(2.5), Some(Rational32::new(5, 2)));
assert_eq!(f64_to_fraction(0.127659574),
Some(Rational32::new(29013539, 227272723)));
assert_eq!(
f64_to_fraction(0.127659574),
Some(Rational32::new(29013539, 227272723))
);
assert_eq!(f64_to_fraction(29.97), Some(Rational32::new(2997, 100)));
}
}

80
gst-plugin/src/value.rs
View file

@ -46,7 +46,8 @@ impl<'a> ValueView<'a> {
}
pub trait ValueType<'a>
where Self: Sized
where
Self: Sized,
{
fn g_type() -> glib::GType;
@ -138,13 +139,12 @@ impl Value {
typ if typ == *TYPE_FRACTION => {
ValueView::Fraction(Rational32::from_value(&self.0).unwrap())
}
gobject::G_TYPE_STRING => {
ValueView::String(Cow::Borrowed(<&str as ValueType>::from_value(&self.0).unwrap()))
}
typ if typ == *TYPE_GST_VALUE_ARRAY => {
ValueView::Array(Cow::Borrowed(<&[Value] as ValueType>::from_value(&self.0)
.unwrap()))
}
gobject::G_TYPE_STRING => ValueView::String(Cow::Borrowed(
<&str as ValueType>::from_value(&self.0).unwrap(),
)),
typ if typ == *TYPE_GST_VALUE_ARRAY => ValueView::Array(Cow::Borrowed(
<&[Value] as ValueType>::from_value(&self.0).unwrap(),
)),
typ if typ == *TYPE_BUFFER => {
ValueView::Buffer(<GstRc<Buffer> as ValueType>::from_value(&self.0).unwrap())
}
@ -228,13 +228,12 @@ impl<'a> ValueRef<'a> {
typ if typ == *TYPE_FRACTION => {
ValueView::Fraction(Rational32::from_value(self.0).unwrap())
}
gobject::G_TYPE_STRING => {
ValueView::String(Cow::Borrowed(<&str as ValueType>::from_value(self.0).unwrap()))
}
typ if typ == *TYPE_GST_VALUE_ARRAY => {
ValueView::Array(Cow::Borrowed(<&[Value] as ValueType>::from_value(self.0)
.unwrap()))
}
gobject::G_TYPE_STRING => ValueView::String(Cow::Borrowed(
<&str as ValueType>::from_value(self.0).unwrap(),
)),
typ if typ == *TYPE_GST_VALUE_ARRAY => ValueView::Array(Cow::Borrowed(
<&[Value] as ValueType>::from_value(self.0).unwrap(),
)),
typ if typ == *TYPE_BUFFER => {
ValueView::Buffer(<GstRc<Buffer> as ValueType>::from_value(self.0).unwrap())
}
@ -335,16 +334,20 @@ impl_value_type_simple!(u64,
gobject::G_TYPE_UINT64,
|value: &gobject::GValue| gobject::g_value_get_uint64(value),
|value: &mut gobject::GValue, v| gobject::g_value_set_uint64(value, v));
impl_value_type_simple!(Rational32,
impl_value_type_simple!(
Rational32,
Fraction,
*TYPE_FRACTION,
|value: &gobject::GValue| {
Rational32::new(gst::gst_value_get_fraction_numerator(value),
gst::gst_value_get_fraction_denominator(value))
Rational32::new(
gst::gst_value_get_fraction_numerator(value),
gst::gst_value_get_fraction_denominator(value),
)
},
|value: &mut gobject::GValue, v: Rational32| {
gst::gst_value_set_fraction(value, *v.numer(), *v.denom())
});
}
);
impl<'a> ValueType<'a> for &'a str {
fn g_type() -> glib::GType {
@ -469,7 +472,10 @@ impl<'a> ValueType<'a> for &'a [Value] {
Some(&[])
} else {
let arr = &*arr;
Some(slice::from_raw_parts(arr.data as *const Value, arr.len as usize))
Some(slice::from_raw_parts(
arr.data as *const Value,
arr.len as usize,
))
}
}
}
@ -491,20 +497,19 @@ impl<'a> From<Cow<'a, [Value]>> for Value {
gobject::g_value_init(&mut value.0, <&[Value] as ValueType>::g_type());
match v {
Cow::Borrowed(array) => {
for e in array {
gst::gst_value_array_append_value(&mut value.0,
e.as_ptr() as *mut gobject::GValue);
}
}
Cow::Owned(array) => {
for e in array {
gst::gst_value_array_append_and_take_value(&mut value.0,
e.as_ptr() as
*mut gobject::GValue);
Cow::Borrowed(array) => for e in array {
gst::gst_value_array_append_value(
&mut value.0,
e.as_ptr() as *mut gobject::GValue,
);
},
Cow::Owned(array) => for e in array {
gst::gst_value_array_append_and_take_value(
&mut value.0,
e.as_ptr() as *mut gobject::GValue,
);
mem::forget(e);
}
}
},
}
value
@ -555,7 +560,8 @@ pub struct TypedValue<T> {
}
impl<'a, T> TypedValue<T>
where T: ValueType<'a>
where
T: ValueType<'a>,
{
pub fn new<VT: Into<TypedValue<T>>>(v: VT) -> TypedValue<T> {
v.into()
@ -611,7 +617,8 @@ impl<'a, T> TypedValue<T>
}
impl<'a, T> From<T> for TypedValue<T>
where T: ValueType<'a> + Into<Value>
where
T: ValueType<'a> + Into<Value>,
{
fn from(v: T) -> Self {
TypedValue::from_value(Value::new(v)).unwrap()
@ -667,7 +674,8 @@ pub struct TypedValueRef<'a, T> {
}
impl<'a, T> TypedValueRef<'a, T>
where T: ValueType<'a>
where
T: ValueType<'a>,
{
pub fn from_typed_value(v: &'a TypedValue<T>) -> TypedValueRef<'a, T> {
TypedValueRef {