Make EbmlStreamingParser use WebmetroErrors specifically
This commit is contained in:
parent
a9932ee6a1
commit
bf56789810
3 changed files with 19 additions and 39 deletions
|
@ -1,5 +1,4 @@
|
||||||
use std::io::{
|
use std::io::{
|
||||||
Error as IoError,
|
|
||||||
StdinLock,
|
StdinLock,
|
||||||
prelude::*
|
prelude::*
|
||||||
};
|
};
|
||||||
|
@ -8,6 +7,7 @@ use futures::{
|
||||||
Async,
|
Async,
|
||||||
stream::Stream
|
stream::Stream
|
||||||
};
|
};
|
||||||
|
use webmetro::error::WebmetroError;
|
||||||
|
|
||||||
pub mod dump;
|
pub mod dump;
|
||||||
pub mod filter;
|
pub mod filter;
|
||||||
|
@ -32,7 +32,7 @@ impl<'a> StdinStream<'a> {
|
||||||
|
|
||||||
impl<'a> Stream for StdinStream<'a> {
|
impl<'a> Stream for StdinStream<'a> {
|
||||||
type Item = Vec<u8>;
|
type Item = Vec<u8>;
|
||||||
type Error = IoError;
|
type Error = WebmetroError;
|
||||||
|
|
||||||
fn poll(&mut self) -> Result<Async<Option<Self::Item>>, Self::Error> {
|
fn poll(&mut self) -> Result<Async<Option<Self::Item>>, Self::Error> {
|
||||||
self.buf_reader.consume(self.read_bytes);
|
self.buf_reader.consume(self.read_bytes);
|
||||||
|
@ -44,6 +44,6 @@ impl<'a> Stream for StdinStream<'a> {
|
||||||
} else {
|
} else {
|
||||||
Async::Ready(None)
|
Async::Ready(None)
|
||||||
}
|
}
|
||||||
})
|
}).map_err(WebmetroError::IoError)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
|
@ -34,6 +34,7 @@ use webmetro::{
|
||||||
Transmitter
|
Transmitter
|
||||||
},
|
},
|
||||||
chunk::{Chunk, WebmStream, ChunkingError},
|
chunk::{Chunk, WebmStream, ChunkingError},
|
||||||
|
error::WebmetroError,
|
||||||
fixers::ChunkStream,
|
fixers::ChunkStream,
|
||||||
stream_parser::StreamEbml
|
stream_parser::StreamEbml
|
||||||
};
|
};
|
||||||
|
@ -56,8 +57,11 @@ impl RelayServer {
|
||||||
)
|
)
|
||||||
}
|
}
|
||||||
|
|
||||||
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) -> BodyStream
|
||||||
let source = stream.parse_ebml().chunk_webm();
|
where S::Error: Error {
|
||||||
|
let source = stream
|
||||||
|
.map_err(|err| WebmetroError::Unknown(err.into()))
|
||||||
|
.parse_ebml().chunk_webm();
|
||||||
let sink = Transmitter::new(self.get_channel());
|
let sink = Transmitter::new(self.get_channel());
|
||||||
|
|
||||||
Box::new(
|
Box::new(
|
||||||
|
|
|
@ -1,35 +1,11 @@
|
||||||
use std::{
|
|
||||||
error::Error,
|
|
||||||
fmt::{Display, Formatter, Result as FmtResult}
|
|
||||||
};
|
|
||||||
|
|
||||||
use bytes::BytesMut;
|
use bytes::BytesMut;
|
||||||
use bytes::BufMut;
|
use bytes::BufMut;
|
||||||
use futures::Async;
|
use futures::Async;
|
||||||
use futures::stream::Stream;
|
use futures::stream::Stream;
|
||||||
|
|
||||||
use ebml::EbmlError;
|
|
||||||
use ebml::EbmlEventSource;
|
use ebml::EbmlEventSource;
|
||||||
use ebml::FromEbml;
|
use ebml::FromEbml;
|
||||||
|
use error::WebmetroError;
|
||||||
#[derive(Debug)]
|
|
||||||
pub enum ParsingError<E> {
|
|
||||||
EbmlError(EbmlError),
|
|
||||||
OtherError(E)
|
|
||||||
}
|
|
||||||
impl<E: Display + Error> Display for ParsingError<E> {
|
|
||||||
fn fmt(&self, f: &mut Formatter) -> FmtResult {
|
|
||||||
write!(f, "Parsing error: {}", self.description())
|
|
||||||
}
|
|
||||||
}
|
|
||||||
impl<E: Error> Error for ParsingError<E> {
|
|
||||||
fn description(&self) -> &str {
|
|
||||||
match self {
|
|
||||||
&ParsingError::EbmlError(ref err) => err.description(),
|
|
||||||
&ParsingError::OtherError(ref err) => err.description()
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
pub struct EbmlStreamingParser<S> {
|
pub struct EbmlStreamingParser<S> {
|
||||||
stream: S,
|
stream: S,
|
||||||
|
@ -47,10 +23,10 @@ pub trait StreamEbml where Self: Sized + Stream, Self::Item: AsRef<[u8]> {
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
impl<I: AsRef<[u8]>, S: Stream<Item = I>> StreamEbml for S {}
|
impl<I: AsRef<[u8]>, S: Stream<Item = I, Error = WebmetroError>> StreamEbml for S {}
|
||||||
|
|
||||||
impl<I: AsRef<[u8]>, S: Stream<Item = I>> EbmlStreamingParser<S> {
|
impl<I: AsRef<[u8]>, S: Stream<Item = I, Error = WebmetroError>> EbmlStreamingParser<S> {
|
||||||
pub fn poll_event<'a, T: FromEbml<'a>>(&'a mut self) -> Result<Async<Option<T>>, ParsingError<S::Error>> {
|
pub fn poll_event<'a, T: FromEbml<'a>>(&'a mut self) -> Result<Async<Option<T>>, WebmetroError> {
|
||||||
// release buffer from previous event
|
// release buffer from previous event
|
||||||
self.buffer.advance(self.last_read);
|
self.buffer.advance(self.last_read);
|
||||||
self.last_read = 0;
|
self.last_read = 0;
|
||||||
|
@ -60,9 +36,9 @@ impl<I: AsRef<[u8]>, S: Stream<Item = I>> EbmlStreamingParser<S> {
|
||||||
Ok(None) => {
|
Ok(None) => {
|
||||||
// need to refill buffer, below
|
// need to refill buffer, below
|
||||||
},
|
},
|
||||||
other => return other.map_err(ParsingError::EbmlError).and_then(move |_| {
|
other => return other.map_err(WebmetroError::EbmlError).and_then(move |_| {
|
||||||
match T::decode_element(&self.buffer) {
|
match T::decode_element(&self.buffer) {
|
||||||
Err(err) => Err(ParsingError::EbmlError(err)),
|
Err(err) => Err(WebmetroError::EbmlError(err)),
|
||||||
Ok(None) => panic!("Buffer was supposed to have enough data to parse element, somehow did not."),
|
Ok(None) => panic!("Buffer was supposed to have enough data to parse element, somehow did not."),
|
||||||
Ok(Some((element, element_size))) => {
|
Ok(Some((element, element_size))) => {
|
||||||
self.last_read = element_size;
|
self.last_read = element_size;
|
||||||
|
@ -72,7 +48,7 @@ impl<I: AsRef<[u8]>, S: Stream<Item = I>> EbmlStreamingParser<S> {
|
||||||
})
|
})
|
||||||
}
|
}
|
||||||
|
|
||||||
match self.stream.poll().map_err(ParsingError::OtherError) {
|
match self.stream.poll() {
|
||||||
Ok(Async::Ready(Some(chunk))) => {
|
Ok(Async::Ready(Some(chunk))) => {
|
||||||
self.buffer.reserve(chunk.as_ref().len());
|
self.buffer.reserve(chunk.as_ref().len());
|
||||||
self.buffer.put_slice(chunk.as_ref());
|
self.buffer.put_slice(chunk.as_ref());
|
||||||
|
@ -84,10 +60,10 @@ impl<I: AsRef<[u8]>, S: Stream<Item = I>> EbmlStreamingParser<S> {
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
impl<I: AsRef<[u8]>, S: Stream<Item = I>> EbmlEventSource for EbmlStreamingParser<S> {
|
impl<I: AsRef<[u8]>, S: Stream<Item = I, Error = WebmetroError>> EbmlEventSource for EbmlStreamingParser<S> {
|
||||||
type Error = ParsingError<S::Error>;
|
type Error = WebmetroError;
|
||||||
|
|
||||||
fn poll_event<'a, T: FromEbml<'a>>(&'a mut self) -> Result<Async<Option<T>>, Self::Error> {
|
fn poll_event<'a, T: FromEbml<'a>>(&'a mut self) -> Result<Async<Option<T>>, WebmetroError> {
|
||||||
return EbmlStreamingParser::poll_event(self);
|
return EbmlStreamingParser::poll_event(self);
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
Loading…
Reference in a new issue