gst-plugins-rs/src/rshttpsrc.rs

236 lines
7.3 KiB
Rust
Raw Normal View History

2016-05-15 15:54:09 +00:00
// Copyright (C) 2016 Sebastian Dröge <sebastian@centricular.com>
//
// This library is free software; you can redistribute it and/or
// modify it under the terms of the GNU Library General Public
// License as published by the Free Software Foundation; either
// version 2 of the License, or (at your option) any later version.
//
// This library is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
// Library General Public License for more details.
//
// You should have received a copy of the GNU Library General Public
// License along with this library; if not, write to the
// Free Software Foundation, Inc., 51 Franklin St, Fifth Floor,
// Boston, MA 02110-1301, USA.
use std::u64;
2016-05-15 14:34:13 +00:00
use std::io::Read;
use url::Url;
2016-11-14 18:57:54 +00:00
use reqwest::{Client, Response};
use reqwest::header::{ContentLength, ContentRange, ContentRangeSpec, Range, ByteRangeSpec,
AcceptRanges, RangeUnit};
use error::*;
use rssource::*;
use buffer::*;
#[derive(Debug)]
enum StreamingState {
Stopped,
Started {
uri: Url,
response: Response,
seekable: bool,
position: u64,
size: Option<u64>,
start: u64,
stop: Option<u64>,
},
}
#[derive(Debug)]
pub struct HttpSrc {
streaming_state: StreamingState,
client: Client,
}
impl HttpSrc {
pub fn new() -> HttpSrc {
2016-07-20 08:28:58 +00:00
HttpSrc {
streaming_state: StreamingState::Stopped,
2016-11-14 18:57:54 +00:00
client: Client::new().unwrap(),
2016-07-20 08:28:58 +00:00
}
}
pub fn new_boxed() -> Box<Source> {
Box::new(HttpSrc::new())
}
2016-05-15 08:48:54 +00:00
fn do_request(&self,
uri: Url,
start: u64,
stop: Option<u64>)
-> Result<StreamingState, ErrorMessage> {
let mut req = self.client.get(uri.clone());
match (start != 0, stop) {
(false, None) => (),
(true, None) => req = req.header(Range::Bytes(vec![ByteRangeSpec::AllFrom(start)])),
(_, Some(stop)) => {
req = req.header(Range::Bytes(vec![ByteRangeSpec::FromTo(start, stop - 1)]))
}
}
let response = try!(req.send().or_else(|err| {
Err(error_msg!(SourceError::ReadFailed,
["Failed to fetch {}: {}", uri, err.to_string()]))
}));
2016-11-14 18:57:54 +00:00
if !response.status().is_success() {
return Err(error_msg!(SourceError::ReadFailed,
2016-11-14 18:57:54 +00:00
["Failed to fetch {}: {}", uri, response.status()]));
}
let size = response.headers().get().map(|&ContentLength(cl)| cl + start);
let accept_byte_ranges = if let Some(&AcceptRanges(ref ranges)) =
response.headers()
.get() {
ranges.iter().any(|u| *u == RangeUnit::Bytes)
} else {
false
};
let seekable = size.is_some() && accept_byte_ranges;
let position = if let Some(&ContentRange(ContentRangeSpec::Bytes { range: Some((range_start,
_)),
2016-11-14 18:57:54 +00:00
.. })) = response.headers()
.get() {
range_start
} else {
start
};
if position != start {
return Err(error_msg!(SourceError::SeekFailed,
["Failed to seek to {}: Got {}", start, position]));
2016-05-15 08:48:54 +00:00
}
Ok(StreamingState::Started {
uri: uri,
response: response,
seekable: seekable,
position: 0,
size: size,
start: start,
stop: stop,
})
2016-05-15 08:48:54 +00:00
}
}
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()))));
}
Ok(())
}
impl Source for HttpSrc {
fn uri_validator(&self) -> Box<UriValidator> {
Box::new(validate_uri)
}
fn is_seekable(&self) -> bool {
match self.streaming_state {
StreamingState::Started { seekable, .. } => seekable,
_ => false,
}
}
fn get_size(&self) -> Option<u64> {
match self.streaming_state {
StreamingState::Started { size, .. } => size,
_ => None,
}
}
fn start(&mut self, uri: Url) -> Result<(), ErrorMessage> {
self.streaming_state = StreamingState::Stopped;
self.streaming_state = try!(self.do_request(uri, 0, None));
Ok(())
}
fn stop(&mut self) -> Result<(), ErrorMessage> {
self.streaming_state = StreamingState::Stopped;
Ok(())
}
fn seek(&mut self, start: u64, stop: Option<u64>) -> Result<(), ErrorMessage> {
let (position, old_stop, uri) = match self.streaming_state {
StreamingState::Started { position, stop, ref uri, .. } => {
(position, stop, uri.clone())
}
StreamingState::Stopped => {
return Err(error_msg!(SourceError::Failure, ["Not started yet"]));
}
};
if position == start && old_stop == stop {
return Ok(());
}
self.streaming_state = StreamingState::Stopped;
self.streaming_state = try!(self.do_request(uri, start, stop));
Ok(())
2016-05-15 08:48:54 +00:00
}
fn fill(&mut self, offset: u64, _: u32, buffer: &mut Buffer) -> Result<(), FlowError> {
let (response, position) = match self.streaming_state {
StreamingState::Started { ref mut response, ref mut position, .. } => {
(response, position)
2016-07-20 08:28:58 +00:00
}
StreamingState::Stopped => {
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])));
}
let size = {
let mut map = match buffer.map_readwrite() {
None => {
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| {
Err(FlowError::Error(error_msg!(SourceError::ReadFailed,
["Failed to read at {}: {}",
offset,
err.to_string()])))
}))
};
if size == 0 {
return Err(FlowError::Eos);
}
*position += size as u64;
if let Err(err) = buffer.set_size(size) {
return Err(FlowError::Error(error_msg!(SourceError::Failure,
["Failed to resize buffer: {}", err])));
}
Ok(())
}
}