From bf56789810020f410afd7db78a633238d01f273f Mon Sep 17 00:00:00 2001
From: Tangent 128 <Tangent128@gmail.com>
Date: Sat, 14 Apr 2018 03:34:59 -0400
Subject: [PATCH] Make EbmlStreamingParser use WebmetroErrors specifically

---
 src/commands/mod.rs   |  6 +++---
 src/commands/relay.rs |  8 ++++++--
 src/stream_parser.rs  | 44 ++++++++++---------------------------------
 3 files changed, 19 insertions(+), 39 deletions(-)

diff --git a/src/commands/mod.rs b/src/commands/mod.rs
index a4ca1a1..46f739b 100644
--- a/src/commands/mod.rs
+++ b/src/commands/mod.rs
@@ -1,5 +1,4 @@
 use std::io::{
-    Error as IoError,
     StdinLock,
     prelude::*
 };
@@ -8,6 +7,7 @@ use futures::{
     Async,
     stream::Stream
 };
+use webmetro::error::WebmetroError;
 
 pub mod dump;
 pub mod filter;
@@ -32,7 +32,7 @@ impl<'a> StdinStream<'a> {
 
 impl<'a> Stream for StdinStream<'a> {
     type Item = Vec<u8>;
-    type Error = IoError;
+    type Error = WebmetroError;
 
     fn poll(&mut self) -> Result<Async<Option<Self::Item>>, Self::Error> {
         self.buf_reader.consume(self.read_bytes);
@@ -44,6 +44,6 @@ impl<'a> Stream for StdinStream<'a> {
             } else {
                 Async::Ready(None)
             }
-        })
+        }).map_err(WebmetroError::IoError)
     }
 }
diff --git a/src/commands/relay.rs b/src/commands/relay.rs
index fad253b..6fe841c 100644
--- a/src/commands/relay.rs
+++ b/src/commands/relay.rs
@@ -34,6 +34,7 @@ use webmetro::{
         Transmitter
     },
     chunk::{Chunk, WebmStream, ChunkingError},
+    error::WebmetroError,
     fixers::ChunkStream,
     stream_parser::StreamEbml
 };
@@ -56,8 +57,11 @@ impl RelayServer {
         )
     }
 
-    fn post_stream<I: AsRef<[u8]>, S: Stream<Item = I> + 'static>(&self, stream: S) -> BodyStream {
-        let source = stream.parse_ebml().chunk_webm();
+    fn post_stream<I: AsRef<[u8]>, S: Stream<Item = I> + 'static>(&self, stream: S) -> BodyStream
+    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());
 
         Box::new(
diff --git a/src/stream_parser.rs b/src/stream_parser.rs
index 645ccf3..c1e4665 100644
--- a/src/stream_parser.rs
+++ b/src/stream_parser.rs
@@ -1,35 +1,11 @@
-use std::{
-    error::Error,
-    fmt::{Display, Formatter, Result as FmtResult}
-};
-
 use bytes::BytesMut;
 use bytes::BufMut;
 use futures::Async;
 use futures::stream::Stream;
 
-use ebml::EbmlError;
 use ebml::EbmlEventSource;
 use ebml::FromEbml;
-
-#[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()
-        }
-    }
-}
+use error::WebmetroError;
 
 pub struct EbmlStreamingParser<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> {
-    pub fn poll_event<'a, T: FromEbml<'a>>(&'a mut self) -> Result<Async<Option<T>>, ParsingError<S::Error>> {
+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>>, WebmetroError> {
         // release buffer from previous event
         self.buffer.advance(self.last_read);
         self.last_read = 0;
@@ -60,9 +36,9 @@ impl<I: AsRef<[u8]>, S: Stream<Item = I>> EbmlStreamingParser<S> {
                 Ok(None) => {
                     // 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) {
-                        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(Some((element, 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))) => {
                     self.buffer.reserve(chunk.as_ref().len());
                     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> {
-    type Error = ParsingError<S::Error>;
+impl<I: AsRef<[u8]>, S: Stream<Item = I, Error = WebmetroError>> EbmlEventSource for EbmlStreamingParser<S> {
+    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);
     }
 }