tangent128
/
webmetro
1
0
Fork 0
Code Issues 1 Pull Requests Packages Releases Wiki Activity

use impl Future in a few places

This commit is contained in:
Tangent 128 2018-05-13 21:05:51 -04:00
parent 4f6cc00fe6
commit 55e1f29906
2 changed files with 22 additions and 26 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
src/commands/filter.rs
View File

@ -25,7 +25,7 @@ pub fn options() -> App<'static, 'static> {
.help("Slow down output to \"real time\" speed as determined by the timestamps (useful for streaming static files)"))
}
pub fn run(args: &ArgMatches) -> Box<Future<Item=(), Error=WebmetroError> + Send> {
pub fn run(args: &ArgMatches) -> impl Future<Item=(), Error=WebmetroError> + Send {
let mut chunk_stream: Box<Stream<Item = Chunk, Error = WebmetroError> + Send> = Box::new(
stdin_stream()
.parse_ebml()
@ -37,7 +37,7 @@ pub fn run(args: &ArgMatches) -> Box<Future<Item=(), Error=WebmetroError> + Send
chunk_stream = Box::new(chunk_stream.throttle());
}
Box::new(chunk_stream.for_each(|chunk| {
chunk_stream.for_each(|chunk| {
io::stdout().write_all(chunk.as_ref()).map_err(WebmetroError::IoError)
}))
})
}

42
src/commands/relay.rs
View File

@ -48,8 +48,6 @@ header! { (XAccelBuffering, "X-Accel-Buffering") => [String] }
const BUFFER_LIMIT: usize = 2 * 1024 * 1024;
type BodyStream = Box<Stream<Item = Chunk, Error = HyperError>>;
struct RelayServer(Arc<Mutex<Channel>>);
impl RelayServer {
@ -57,16 +55,14 @@ impl RelayServer {
self.0.clone()
}
fn get_stream(&self) -> BodyStream {
Box::new(
Listener::new(self.get_channel())
.fix_timecodes()
.find_starting_point()
.map_err(|err| match err {})
)
fn get_stream(&self) -> impl Stream<Item = Chunk, Error = HyperError> {
Listener::new(self.get_channel())
.fix_timecodes()
.find_starting_point()
.map_err(|err| match err {})
}
fn post_stream<I: AsRef<[u8]>, S: Stream<Item = I> + 'static>(&self, stream: S) -> BodyStream
fn post_stream<I: AsRef<[u8]>, S: Stream<Item = I> + 'static>(&self, stream: S) -> impl Stream<Item = Chunk, Error = HyperError>
where S::Error: Error + Send {
let source = stream
.map_err(WebmetroError::from_err)
@ -74,22 +70,22 @@ impl RelayServer {
.chunk_webm().with_soft_limit(BUFFER_LIMIT);
let sink = Transmitter::new(self.get_channel());
Box::new(
source.forward(sink.sink_map_err(|err| -> WebmetroError {match err {}}))
.into_stream()
.map(|_| empty())
.map_err(|err| {
//TODO: log something somewhere
to_hyper_error(err)
})
.flatten()
)
source.forward(sink.sink_map_err(|err| -> WebmetroError {match err {}}))
.into_stream()
.map(|_| empty())
.map_err(|err| {
println!("[Warning] {}", err);
to_hyper_error(err)
})
.flatten()
}
}
type BoxedBodyStream = Box<Stream<Item = Chunk, Error = HyperError>>;
impl Service for RelayServer {
type Request = Request;
type Response = Response<BodyStream>;
type Response = Response<BoxedBodyStream>;
type Error = HyperError;
type Future = FutureResult<Self::Response, HyperError>;
@ -110,11 +106,11 @@ impl Service for RelayServer {
.with_header(ContentType("video/webm".parse().unwrap()))
.with_header(XAccelBuffering("no".to_string()))
.with_header(CacheControl(vec![CacheDirective::NoCache, CacheDirective::NoStore]))
.with_body(self.get_stream())
.with_body(Box::new(self.get_stream()) as BoxedBodyStream)
},
(Post, "/live") | (Put, "/live") => {
Response::new()
.with_body(self.post_stream(request_body))
.with_body(Box::new(self.post_stream(request_body)) as BoxedBodyStream)
},
_ => {
Response::new()