2017-02-16 15:52:27 +00:00
|
|
|
// Copyright (C) 2016-2017 Sebastian Dröge <sebastian@centricular.com>
|
2016-05-15 15:54:09 +00:00
|
|
|
//
|
2017-02-16 15:52:27 +00:00
|
|
|
// 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.
|
2016-05-15 15:54:09 +00:00
|
|
|
|
2016-05-14 14:04:53 +00:00
|
|
|
use std::u64;
|
2016-05-15 14:34:13 +00:00
|
|
|
use std::io::Read;
|
2016-05-14 14:04:53 +00:00
|
|
|
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};
|
2016-05-14 14:04:53 +00:00
|
|
|
|
2016-12-23 17:04:32 +00:00
|
|
|
use gst_plugin::error::*;
|
|
|
|
use gst_plugin::source::*;
|
|
|
|
use gst_plugin::buffer::*;
|
2016-12-25 11:16:12 +00:00
|
|
|
use gst_plugin::utils::*;
|
|
|
|
use gst_plugin::log::*;
|
|
|
|
|
2017-04-12 13:44:34 +00:00
|
|
|
use slog::Logger;
|
2016-05-14 14:04:53 +00:00
|
|
|
|
2016-08-22 19:25:58 +00:00
|
|
|
#[derive(Debug)]
|
|
|
|
enum StreamingState {
|
|
|
|
Stopped,
|
|
|
|
Started {
|
2016-09-01 21:52:28 +00:00
|
|
|
uri: Url,
|
2016-08-22 19:25:58 +00:00
|
|
|
response: Response,
|
|
|
|
seekable: bool,
|
|
|
|
position: u64,
|
2016-09-01 22:01:25 +00:00
|
|
|
size: Option<u64>,
|
2016-08-22 19:25:58 +00:00
|
|
|
start: u64,
|
2016-09-01 22:01:25 +00:00
|
|
|
stop: Option<u64>,
|
2016-08-22 19:25:58 +00:00
|
|
|
},
|
|
|
|
}
|
|
|
|
|
2016-05-14 14:04:53 +00:00
|
|
|
#[derive(Debug)]
|
|
|
|
pub struct HttpSrc {
|
2016-09-01 21:52:28 +00:00
|
|
|
streaming_state: StreamingState,
|
2016-12-25 11:16:12 +00:00
|
|
|
logger: Logger,
|
2016-05-14 14:04:53 +00:00
|
|
|
client: Client,
|
|
|
|
}
|
|
|
|
|
|
|
|
impl HttpSrc {
|
2016-12-25 11:16:12 +00:00
|
|
|
pub fn new(element: Element) -> HttpSrc {
|
2016-07-20 08:28:58 +00:00
|
|
|
HttpSrc {
|
2016-09-01 21:52:28 +00:00
|
|
|
streaming_state: StreamingState::Stopped,
|
2016-12-25 11:16:12 +00:00
|
|
|
logger: Logger::root(GstDebugDrain::new(Some(&element),
|
|
|
|
"rshttpsink",
|
|
|
|
0,
|
|
|
|
"Rust http sink"),
|
2017-04-12 13:44:34 +00:00
|
|
|
o!()),
|
2016-11-14 18:57:54 +00:00
|
|
|
client: Client::new().unwrap(),
|
2016-07-20 08:28:58 +00:00
|
|
|
}
|
2016-05-14 14:04:53 +00:00
|
|
|
}
|
|
|
|
|
2016-12-25 11:16:12 +00:00
|
|
|
pub fn new_boxed(element: Element) -> Box<Source> {
|
|
|
|
Box::new(HttpSrc::new(element))
|
2016-05-14 14:04:53 +00:00
|
|
|
}
|
2016-05-15 08:48:54 +00:00
|
|
|
|
2016-09-01 22:01:25 +00:00
|
|
|
fn do_request(&self,
|
|
|
|
uri: Url,
|
|
|
|
start: u64,
|
|
|
|
stop: Option<u64>)
|
|
|
|
-> Result<StreamingState, ErrorMessage> {
|
2016-09-01 21:52:28 +00:00
|
|
|
let mut req = self.client.get(uri.clone());
|
2016-08-27 08:16:17 +00:00
|
|
|
|
2016-11-25 12:42:16 +00:00
|
|
|
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)]))
|
|
|
|
}
|
2016-08-27 08:16:17 +00:00
|
|
|
}
|
|
|
|
|
2016-12-27 15:47:39 +00:00
|
|
|
debug!(self.logger, "Doing new request {:?}", req);
|
|
|
|
|
2017-04-12 13:46:11 +00:00
|
|
|
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()]))
|
|
|
|
}));
|
2016-08-27 08:16:17 +00:00
|
|
|
|
2016-11-14 18:57:54 +00:00
|
|
|
if !response.status().is_success() {
|
2016-12-27 15:47:39 +00:00
|
|
|
error!(self.logger, "Request status failed: {:?}", response);
|
2016-08-27 08:16:17 +00:00
|
|
|
return Err(error_msg!(SourceError::ReadFailed,
|
2016-11-14 18:57:54 +00:00
|
|
|
["Failed to fetch {}: {}", uri, response.status()]));
|
2016-08-27 08:16:17 +00:00
|
|
|
}
|
|
|
|
|
2017-04-12 13:46:11 +00:00
|
|
|
let size = response
|
|
|
|
.headers()
|
|
|
|
.get()
|
|
|
|
.map(|&ContentLength(cl)| cl + start);
|
2016-08-27 08:16:17 +00:00
|
|
|
|
2016-12-04 10:58:59 +00:00
|
|
|
let accept_byte_ranges = if let Some(&AcceptRanges(ref ranges)) =
|
2017-04-10 10:30:07 +00:00
|
|
|
response.headers().get() {
|
2016-08-27 08:16:17 +00:00
|
|
|
ranges.iter().any(|u| *u == RangeUnit::Bytes)
|
|
|
|
} else {
|
|
|
|
false
|
|
|
|
};
|
|
|
|
|
2016-09-01 22:01:25 +00:00
|
|
|
let seekable = size.is_some() && accept_byte_ranges;
|
2016-08-27 08:16:17 +00:00
|
|
|
|
2017-04-12 13:46:11 +00:00
|
|
|
let position = if let Some(&ContentRange(ContentRangeSpec::Bytes {
|
|
|
|
range: Some((range_start, _)), ..
|
|
|
|
})) = response.headers().get() {
|
2016-08-27 08:16:17 +00:00
|
|
|
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
|
|
|
}
|
2016-08-27 08:16:17 +00:00
|
|
|
|
2016-12-27 15:47:39 +00:00
|
|
|
debug!(self.logger, "Request successful: {:?}", response);
|
|
|
|
|
2016-08-27 08:16:17 +00:00
|
|
|
Ok(StreamingState::Started {
|
2017-04-10 10:30:07 +00:00
|
|
|
uri: uri,
|
|
|
|
response: response,
|
|
|
|
seekable: seekable,
|
|
|
|
position: 0,
|
|
|
|
size: size,
|
|
|
|
start: start,
|
|
|
|
stop: stop,
|
|
|
|
})
|
2016-05-15 08:48:54 +00:00
|
|
|
}
|
2016-05-14 14:04:53 +00:00
|
|
|
}
|
|
|
|
|
2016-09-01 21:52:28 +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()))));
|
2016-08-24 21:08:06 +00:00
|
|
|
}
|
|
|
|
|
2016-09-01 21:52:28 +00:00
|
|
|
Ok(())
|
|
|
|
}
|
2016-05-14 14:04:53 +00:00
|
|
|
|
2016-09-01 21:52:28 +00:00
|
|
|
impl Source for HttpSrc {
|
|
|
|
fn uri_validator(&self) -> Box<UriValidator> {
|
|
|
|
Box::new(validate_uri)
|
2016-05-14 14:04:53 +00:00
|
|
|
}
|
|
|
|
|
|
|
|
fn is_seekable(&self) -> bool {
|
2016-09-01 21:52:28 +00:00
|
|
|
match self.streaming_state {
|
2016-08-22 19:25:58 +00:00
|
|
|
StreamingState::Started { seekable, .. } => seekable,
|
|
|
|
_ => false,
|
|
|
|
}
|
2016-05-14 14:04:53 +00:00
|
|
|
}
|
|
|
|
|
2016-09-01 22:01:25 +00:00
|
|
|
fn get_size(&self) -> Option<u64> {
|
2016-09-01 21:52:28 +00:00
|
|
|
match self.streaming_state {
|
2016-08-22 19:25:58 +00:00
|
|
|
StreamingState::Started { size, .. } => size,
|
2016-09-01 22:01:25 +00:00
|
|
|
_ => None,
|
2016-08-22 19:25:58 +00:00
|
|
|
}
|
2016-05-14 14:04:53 +00:00
|
|
|
}
|
|
|
|
|
2016-09-03 14:37:51 +00:00
|
|
|
fn start(&mut self, uri: Url) -> Result<(), ErrorMessage> {
|
2016-09-01 21:52:28 +00:00
|
|
|
self.streaming_state = StreamingState::Stopped;
|
2016-09-03 14:37:51 +00:00
|
|
|
self.streaming_state = try!(self.do_request(uri, 0, None));
|
2016-08-27 08:16:17 +00:00
|
|
|
|
|
|
|
Ok(())
|
2016-08-22 19:25:58 +00:00
|
|
|
}
|
|
|
|
|
2016-09-01 21:52:28 +00:00
|
|
|
fn stop(&mut self) -> Result<(), ErrorMessage> {
|
|
|
|
self.streaming_state = StreamingState::Stopped;
|
2016-05-14 14:04:53 +00:00
|
|
|
|
2016-08-27 08:16:17 +00:00
|
|
|
Ok(())
|
2016-05-14 14:04:53 +00:00
|
|
|
}
|
|
|
|
|
2016-09-01 22:01:25 +00:00
|
|
|
fn seek(&mut self, start: u64, stop: Option<u64>) -> Result<(), ErrorMessage> {
|
2016-09-01 21:52:28 +00:00
|
|
|
let (position, old_stop, uri) = match self.streaming_state {
|
2017-04-12 13:46:11 +00:00
|
|
|
StreamingState::Started {
|
|
|
|
position,
|
|
|
|
stop,
|
|
|
|
ref uri,
|
|
|
|
..
|
|
|
|
} => (position, stop, uri.clone()),
|
2016-09-01 21:52:28 +00:00
|
|
|
StreamingState::Stopped => {
|
|
|
|
return Err(error_msg!(SourceError::Failure, ["Not started yet"]));
|
|
|
|
}
|
|
|
|
};
|
|
|
|
|
|
|
|
if position == start && old_stop == stop {
|
|
|
|
return Ok(());
|
|
|
|
}
|
2016-08-22 19:25:58 +00:00
|
|
|
|
2016-09-01 21:52:28 +00:00
|
|
|
self.streaming_state = StreamingState::Stopped;
|
|
|
|
self.streaming_state = try!(self.do_request(uri, start, stop));
|
2016-08-27 08:16:17 +00:00
|
|
|
|
|
|
|
Ok(())
|
2016-05-15 08:48:54 +00:00
|
|
|
}
|
|
|
|
|
2016-09-12 13:00:28 +00:00
|
|
|
fn fill(&mut self, offset: u64, _: u32, buffer: &mut Buffer) -> Result<(), FlowError> {
|
2016-12-27 15:47:39 +00:00
|
|
|
let logger = self.logger.clone();
|
|
|
|
|
2016-09-01 21:52:28 +00:00
|
|
|
let (response, position) = match self.streaming_state {
|
2017-04-12 13:46:11 +00:00
|
|
|
StreamingState::Started {
|
|
|
|
ref mut response,
|
|
|
|
ref mut position,
|
|
|
|
..
|
|
|
|
} => (response, position),
|
2016-08-27 08:16:17 +00:00
|
|
|
StreamingState::Stopped => {
|
|
|
|
return Err(FlowError::Error(error_msg!(SourceError::Failure, ["Not started yet"])));
|
|
|
|
}
|
|
|
|
};
|
|
|
|
|
2016-09-01 21:52:28 +00:00
|
|
|
if *position != offset {
|
|
|
|
return Err(FlowError::Error(error_msg!(SourceError::SeekFailed,
|
|
|
|
["Got unexpected offset {}, expected {}",
|
|
|
|
offset,
|
|
|
|
position])));
|
|
|
|
}
|
|
|
|
|
2016-09-12 13:00:28 +00:00
|
|
|
let size = {
|
2016-09-17 12:31:10 +00:00
|
|
|
let mut map = match buffer.map_readwrite() {
|
2016-09-12 13:00:28 +00:00
|
|
|
None => {
|
|
|
|
return Err(FlowError::Error(error_msg!(SourceError::Failure,
|
|
|
|
["Failed to map buffer"])));
|
|
|
|
}
|
|
|
|
Some(map) => map,
|
|
|
|
};
|
|
|
|
|
|
|
|
let data = map.as_mut_slice();
|
|
|
|
|
2017-04-12 13:46:11 +00:00
|
|
|
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()])))
|
|
|
|
}))
|
2016-09-12 13:00:28 +00:00
|
|
|
};
|
2016-08-27 08:16:17 +00:00
|
|
|
|
|
|
|
if size == 0 {
|
|
|
|
return Err(FlowError::Eos);
|
2016-05-14 14:04:53 +00:00
|
|
|
}
|
2016-08-27 08:16:17 +00:00
|
|
|
|
|
|
|
*position += size as u64;
|
2016-09-01 21:52:28 +00:00
|
|
|
|
2017-02-03 14:34:17 +00:00
|
|
|
buffer.set_size(size);
|
2016-09-12 13:00:28 +00:00
|
|
|
|
|
|
|
Ok(())
|
2016-05-14 14:04:53 +00:00
|
|
|
}
|
|
|
|
}
|