generalize WebmEventSource -> EbmlEventSource

src/chunk.rs

@@ -2,6 +2,7 @@ use futures::{Async, Stream};
 use std::io::Cursor;
 use std::mem;
 use std::sync::Arc;
+use ebml::EbmlEventSource;
 use webm::*;
 
 #[derive(Clone, Debug)]
@@ -87,12 +88,12 @@ pub enum ChunkingError<E> {
     OtherError(E)
 }
 
-pub struct WebmChunker<S: WebmEventSource> {
+pub struct WebmChunker<S> {
     source: S,
     state: ChunkerState
 }
 
-impl<'a, S: WebmEventSource> Stream for WebmChunker<S>
+impl<'a, S: EbmlEventSource> Stream for WebmChunker<S>
 {
     type Item = Chunk;
     type Error = ChunkingError<S::Error>;
@@ -209,11 +210,11 @@ impl<'a, S: WebmEventSource> Stream for WebmChunker<S>
     }
 }
 
-pub trait WebmStream<T: WebmEventSource> {
+pub trait WebmStream<T: EbmlEventSource> {
     fn chunk_webm(self) -> WebmChunker<T>;
 }
 
-impl<'a, T: WebmEventSource> WebmStream<T> for T {
+impl<'a, T: EbmlEventSource> WebmStream<T> for T {
     fn chunk_webm(self) -> WebmChunker<T> {
         WebmChunker {
             source: self,

src/ebml.rs

@@ -2,6 +2,7 @@ use bytes::{BigEndian, ByteOrder, BufMut};
 use std::error::Error as ErrorTrait;
 use std::fmt::{Display, Formatter, Result as FmtResult};
 use std::io::{Cursor, Error as IoError, ErrorKind, Result as IoResult, Write, Seek, SeekFrom};
+use futures::Async;
 
 pub const EBML_HEAD_ID: u64 = 0x0A45DFA3;
 pub const DOC_TYPE_ID: u64 = 0x0282;
@@ -286,6 +287,11 @@ pub trait FromEbml<'a>: Sized {
     }
 }
 
+pub trait EbmlEventSource {
+    type Error;
+    fn poll_event<'a, T: FromEbml<'a>>(&'a mut self) -> Result<Async<Option<T>>, Self::Error>;
+}
+
 #[cfg(test)]
 mod tests {
     use bytes::{BytesMut};

src/iterator.rs

@@ -1,47 +1,79 @@
 use futures::Async;
 use ebml::Error as EbmlError;
+use ebml::EbmlEventSource;
 use ebml::FromEbml;
-use webm::*;
+use webm::WebmElement;
 
-pub struct EbmlCursor<T> {
+pub struct EbmlCursor<'a> {
-    source: T,
+    source: &'a [u8]
-    position: usize
 }
 
-impl<S> EbmlCursor<S> {
+impl<'a> EbmlCursor<'a> {
-    pub fn new(source: S) -> Self {
+    pub fn new(source: &'a [u8]) -> Self {
         EbmlCursor {
-            source,
+            source
-            position: 0
         }
     }
 }
 
-impl<'a> EbmlCursor<&'a [u8]> {
+impl<'a> Iterator for EbmlCursor<'a> {
-    fn decode_element<T: FromEbml<'a>>(&mut self) -> Result<Option<T>, EbmlError> {
+    type Item = WebmElement<'a>;
-        match T::decode_element(&self.source.as_ref()[self.position..]) {
+
-            Err(err) => Err(err),
+    fn next(&mut self) -> Option<WebmElement<'a>> {
-            Ok(None) => Ok(None),
+        match WebmElement::check_space(self.source) {
-            Ok(Some((element, element_size))) => {
+            Err(err) => {
-                self.position += element_size;
+                None
-                Ok(Some(element))
+            },
+            Ok(None) => {
+                None
+            },
+            Ok(Some(element_size)) => {
+                let (element_data, rest) = self.source.split_at(element_size);
+                self.source = rest;
+                match WebmElement::decode_element(element_data) {
+                    Err(err) => {
+                        None
+                    },
+                    Ok(None) => {
+                        // buffer should have enough data
+                        panic!("Buffer was supposed to have enough data to parse element, somehow did not.")
+                    },
+                    Ok(Some((element, _))) => {
+                        Some(element)
+                    }
+                }
             }
         }
     }
 }
 
-impl<'a> Iterator for EbmlCursor<&'a [u8]> {
+impl<'b> EbmlEventSource for EbmlCursor<'b> {
-    type Item = WebmElement<'a>;
-
-    fn next(&mut self) -> Option<WebmElement<'a>> {
-        self.decode_element().unwrap_or(None)
-    }
-}
-
-impl<'b> WebmEventSource for EbmlCursor<&'b [u8]> {
     type Error = EbmlError;
 
-    fn poll_event<'a>(&'a mut self) -> Result<Async<Option<WebmElement<'a>>>, EbmlError> {
+    fn poll_event<'a, T: FromEbml<'a>>(&'a mut self) -> Result<Async<Option<T>>, EbmlError> {
-        self.decode_element().map(Async::Ready)
+        match T::check_space(self.source) {
+            Err(err) => {
+                Err(err)
+            },
+            Ok(None) => {
+                Ok(None)
+            },
+            Ok(Some(element_size)) => {
+                let (element_data, rest) = self.source.split_at(element_size);
+                self.source = rest;
+                match T::decode_element(element_data) {
+                    Err(err) => {
+                        Err(err)
+                    },
+                    Ok(None) => {
+                        // buffer should have enough data
+                        panic!("Buffer was supposed to have enough data to parse element, somehow did not.")
+                    },
+                    Ok(Some((element, _))) => {
+                        Ok(Some(element))
+                    }
+                }
+            }
+        }.map(Async::Ready)
     }
 }

src/webm.rs

@@ -1,6 +1,5 @@
 use std::io::{Cursor, Error as IoError, ErrorKind, Result as IoResult, Write, Seek};
 use bytes::{BigEndian, BufMut, ByteOrder};
-use futures::Async;
 use ebml::*;
 use iterator::EbmlCursor;
 
@@ -13,7 +12,7 @@ const CLUSTER_ID: u64 = 0x0F43B675;
 const TIMECODE_ID: u64 = 0x67;
 const SIMPLE_BLOCK_ID: u64 = 0x23;
 
-pub fn parse_webm<'a, T: AsRef<[u8]> + ?Sized>(source: &'a T) -> EbmlCursor<&'a [u8]> {
+pub fn parse_webm<'a, T: AsRef<[u8]> + ?Sized>(source: &'a T) -> EbmlCursor<'a> {
     EbmlCursor::new(source.as_ref())
 }
 
@@ -127,11 +126,6 @@ pub fn encode_webm_element<T: Write + Seek>(element: &WebmElement, output: &mut
     }
 }
 
-pub trait WebmEventSource {
-    type Error;
-    fn poll_event<'a>(&'a mut self) -> Result<Async<Option<WebmElement<'a>>>, Self::Error>;
-}
-
 #[cfg(test)]
 mod tests {
     use tests::TEST_FILE;

src/webm_stream.rs

@@ -4,7 +4,6 @@ use futures::Async;
 use futures::stream::Stream;
 
 use ebml::*;
-use webm::*;
 
 pub enum ParsingError<E> {
     EbmlError(::ebml::Error),
@@ -71,10 +70,10 @@ impl<I: AsRef<[u8]>, S: Stream<Item = I>> WebmBuffer<S> {
     }
 }
 
-impl<I: AsRef<[u8]>, S: Stream<Item = I>> WebmEventSource for WebmBuffer<S> {
+impl<I: AsRef<[u8]>, S: Stream<Item = I>> EbmlEventSource for WebmBuffer<S> {
     type Error = ParsingError<S::Error>;
 
-    fn poll_event<'a>(&'a mut self) -> Result<Async<Option<WebmElement<'a>>>, Self::Error> {
+    fn poll_event<'a, T: FromEbml<'a>>(&'a mut self) -> Result<Async<Option<T>>, Self::Error> {
         return WebmBuffer::poll_event(self);
     }
 }