#![deny(missing_docs)]

use varint::consume_single_varint;
use stream::*;
use std::io::Read;

/// A consumer for a byte stream
pub struct ByteConsumer<T: Read> {
    read: T,
    type_: StreamType,
}

impl <T: Read> ByteConsumer<T> {
    /// Return a ByteConsumer from for single messages, varint or leb128-delimited
    pub fn new(read: T, type_: StreamType) -> ByteConsumer<T> {
        ByteConsumer { read, type_ }
    }
}

impl <T: Read> Iterator for ByteConsumer<T> {
    type Item = Vec<u8>;

    fn next(&mut self) -> Option<Vec<u8>> {
        match self.type_ {
            StreamType::Leb128 | StreamType::Varint => consume_single_varint(&mut self.read),
            StreamType::Single => {
                let ret: Option<Vec<u8>>;
                let mut buf = Vec::new();
                match self.read.read_to_end(&mut buf) {
                    Ok(x) => {
                        if x > 0 {
                            ret = Some(buf);
                        } else {
                            ret = None
                        }
                    }
                    Err(_) => ret = None,
                }
                ret
            }
        }
    }
}

#![deny(missing_docs)]

use varint::consume_single_varint;
use stream::*;
use std::io::Read;

/// A consumer for a byte stream
pub struct ByteConsumer<'a> {
    read: &'a mut Read,
    type_: StreamType,
}

impl<'a> ByteConsumer<'a> {
    /// Return a ByteConsumer from for single messages, varint or leb128-delimited
    pub fn new(read: &'a mut Read, type_: StreamType) -> ByteConsumer {
        ByteConsumer { read, type_ }
    }
}

impl<'a> Iterator for ByteConsumer<'a> {
    type Item = Vec<u8>;

    fn next(&mut self) -> Option<Vec<u8>> {
        match self.type_ {
            StreamType::Leb128 | StreamType::Varint => consume_single_varint(self.read),
            StreamType::Single => {
                let ret: Option<Vec<u8>>;
                let mut buf = Vec::new();
                match self.read.read_to_end(&mut buf) {
                    Ok(x) => {
                        if x > 0 {
                            ret = Some(buf);
                        } else {
                            ret = None
                        }
                    }
                    Err(_) => ret = None,
                }
                ret
            }
        }
    }
}

--- 
+++ 
@@ -5,24 +5,24 @@
 use std::io::Read;
 
 /// A consumer for a byte stream
-pub struct ByteConsumer<'a> {
-    read: &'a mut Read,
+pub struct ByteConsumer<T: Read> {
+    read: T,
     type_: StreamType,
 }
 
-impl<'a> ByteConsumer<'a> {
+impl <T: Read> ByteConsumer<T> {
     /// Return a ByteConsumer from for single messages, varint or leb128-delimited
-    pub fn new(read: &'a mut Read, type_: StreamType) -> ByteConsumer {
+    pub fn new(read: T, type_: StreamType) -> ByteConsumer<T> {
         ByteConsumer { read, type_ }
     }
 }
 
-impl<'a> Iterator for ByteConsumer<'a> {
+impl <T: Read> Iterator for ByteConsumer<T> {
     type Item = Vec<u8>;
 
     fn next(&mut self) -> Option<Vec<u8>> {
         match self.type_ {
-            StreamType::Leb128 | StreamType::Varint => consume_single_varint(self.read),
+            StreamType::Leb128 | StreamType::Varint => consume_single_varint(&mut self.read),
             StreamType::Single => {
                 let ret: Option<Vec<u8>>;
                 let mut buf = Vec::new();
Make ByteConsumer own its reader

This is in preparation for making pq multi-threaded; the trait `Read`
is not `Send` and so with the previous declaration (read: &'a mut
Read) ByteConsumer was _not_ Send and could not be moved into a
closure, i.e., with thread::spawn.

This commit modifies the API of ByteConsumer by making it take
ownership of a value of type T where T implements the Read trait.
