Split log_reader into reverse_log_reader and forward_log_reader

4 files changed, 147 insertions, 135 deletions

diff --git a/src/common.rs b/src/common.rs
index 9ea43aa..ea179ab 100644
--- a/src/common.rs
+++ b/src/common.rs
@@ -39,6 +39,22 @@ pub const SEQ_LIT_FIELD_SEP: &[u8] = &[
     ESCAPE_CHARACTER,
 ];
 
+/// There are three special sequences that need to be handled:
+/// Here: SC = escape char, FS = field separator.
+/// - SC FS FS SC  -> actual record separator
+/// - SC SC FS SC  -> literal FS
+/// - SC SC SC SC  -> literal SC
+///
+/// Returns SpecialSequence or None if not valid.
+pub fn validate_special(buf: &[u8]) -> Option<SpecialSequence> {
+    match buf {
+        SEQ_RECORD_SEP => Some(SpecialSequence::RecordSeparator),
+        SEQ_LIT_FIELD_SEP => Some(SpecialSequence::LiteralFieldSeparator),
+        SEQ_LIT_ESCAPE => Some(SpecialSequence::LiteralEscape),
+        _ => None,
+    }
+}
+
 #[derive(Debug, Eq, PartialEq)]
 pub enum SpecialSequence {
     RecordSeparator,
diff --git a/src/forward_log_reader.rs b/src/forward_log_reader.rs
new file mode 100644
index 0000000..10788a7
--- /dev/null
+++ b/src/forward_log_reader.rs
@@ -0,0 +1,124 @@
+use super::common::*;
+use std::fs::{self};
+use std::io::{self, BufRead, Read};
+
+pub struct ForwardLogReader<'a> {
+    reader: io::BufReader<&'a mut fs::File>,
+}
+
+impl<'a> ForwardLogReader<'a> {
+    pub fn new(file: &mut fs::File) -> ForwardLogReader {
+        let reader = io::BufReader::new(file);
+        ForwardLogReader { reader }
+    }
+
+    fn read_record(&mut self) -> Result<Option<Record>, io::Error> {
+        // The buffer that stores the bytes read from the file.
+        let mut read_buf: Vec<u8> = Vec::new();
+        // The buffer that stores all the bytes of the record read so far in reverse order.
+        let mut result_buf: Vec<u8> = Vec::new();
+
+        // Try reading a byte from the file.
+        // If we've reached the end of the file, return None.
+        let mut peek_buf = vec![0];
+        match self.reader.read_exact(&mut peek_buf) {
+            Ok(_) => {
+                // Go back one byte
+                self.reader.seek_relative(-1)?;
+            }
+            Err(ref e) if e.kind() == io::ErrorKind::UnexpectedEof => {
+                return Ok(None);
+            }
+            Err(e) => {
+                return Err(e);
+            }
+        }
+
+        loop {
+            read_buf.clear();
+            self.reader.read_until(ESCAPE_CHARACTER, &mut read_buf)?;
+            self.reader.seek_relative(-1)?;
+            result_buf.extend(&read_buf[..read_buf.len() - 1]);
+
+            // Otherwise, we must have encountered an escape character.
+            match self.read_special_sequence()? {
+                SpecialSequence::RecordSeparator => {
+                    // The record is complete, so we can break out of the loop.
+                    break;
+                }
+                SpecialSequence::LiteralFieldSeparator => {
+                    // The field separator is escaped, so we need to add it to the result buffer.
+                    result_buf.push(FIELD_SEPARATOR);
+                }
+                SpecialSequence::LiteralEscape => {
+                    // The escape character is escaped, so we need to add it to the result buffer.
+                    result_buf.push(ESCAPE_CHARACTER);
+                }
+            }
+        }
+
+        let record = Record::deserialize(&result_buf);
+        Ok(Some(record))
+    }
+
+    fn read_special_sequence(&mut self) -> Result<SpecialSequence, io::Error> {
+        let mut special_buf: Vec<u8> = vec![0; SEQ_RECORD_SEP.len()];
+        self.reader.read_exact(&mut special_buf)?;
+
+        match validate_special(&special_buf.as_slice()) {
+            Some(special) => Ok(special),
+            None => Err(io::Error::new(
+                io::ErrorKind::InvalidData,
+                "Not a special sequence",
+            )),
+        }
+    }
+}
+
+impl Iterator for ForwardLogReader<'_> {
+    type Item = Record;
+
+    fn next(&mut self) -> Option<Self::Item> {
+        match self.read_record() {
+            Ok(Some(record)) => Some(record),
+            Ok(None) => None,
+            Err(err) => panic!("Error reading record: {:?}", err),
+        }
+    }
+}
+
+#[cfg(test)]
+mod reverse_reader_tests {
+    use super::*;
+    use std::path::Path;
+
+    #[test]
+    fn test_forward_log_reader_fixture_db1() {
+        let db_path = Path::new(TEST_RESOURCES_DIR).join("test_db1");
+        let mut file = fs::OpenOptions::new()
+            .read(true)
+            .open(&db_path)
+            .expect("Failed to open file");
+        let mut forward_log_reader = ForwardLogReader::new(&mut file);
+
+        // There are two records in the log with "schema": Int, Null
+
+        let first_record = forward_log_reader
+            .next()
+            .expect("Failed to read the first record");
+        assert!(match first_record.values.as_slice() {
+            [RecordValue::Int(0x1D), RecordValue::Null] => true,
+            _ => false,
+        });
+
+        let last_record = forward_log_reader
+            .next()
+            .expect("Failed to read the last record");
+        assert!(match last_record.values.as_slice() {
+            [RecordValue::Int(10), RecordValue::Null] => true,
+            _ => false,
+        });
+
+        assert!(forward_log_reader.next().is_none());
+    }
+}
diff --git a/src/lib.rs b/src/lib.rs
index 7146948..0ba2cad 100644
--- a/src/lib.rs
+++ b/src/lib.rs
@@ -2,16 +2,17 @@
 extern crate log;
 
 mod common;
-mod log_reader;
+mod forward_log_reader;
 mod primary_memtable;
+mod reverse_log_reader;
 mod secondary_memtable;
 
 pub use common::*;
+pub use forward_log_reader::ForwardLogReader;
 use fs2::lock_contended_error;
 use fs2::FileExt;
-pub use log_reader::ForwardLogReader;
-pub use log_reader::ReverseLogReader;
 use primary_memtable::PrimaryMemtable;
+pub use reverse_log_reader::ReverseLogReader;
 use secondary_memtable::SecondaryMemtable;
 use std::fmt::Debug;
 use std::fs::{self};
diff --git a/src/log_reader.rs b/src/reverse_log_reader.rs
index d07326a..778d1b4 100644
--- a/src/log_reader.rs
+++ b/src/reverse_log_reader.rs
@@ -1,21 +1,6 @@
 use super::common::*;
 use std::fs::{self};
-use std::io::{self, BufRead, Read, Seek, SeekFrom};
-use std::path::Path;
-
-/// There are three special sequences that need to be handled:
-/// Here: SC = escape char, FS = field separator.
-/// - SC FS FS SC  -> actual record separator
-/// - SC SC FS SC  -> literal FS
-/// - SC SC SC SC  -> literal SC
-fn validate_special(buf: &[u8]) -> Option<SpecialSequence> {
-    match buf {
-        SEQ_RECORD_SEP => Some(SpecialSequence::RecordSeparator),
-        SEQ_LIT_FIELD_SEP => Some(SpecialSequence::LiteralFieldSeparator),
-        SEQ_LIT_ESCAPE => Some(SpecialSequence::LiteralEscape),
-        _ => None,
-    }
-}
+use std::io::{self, Read, Seek, SeekFrom};
 
 pub struct ReverseLogReader<'a> {
     file: &'a mut fs::File,
@@ -78,7 +63,7 @@ impl<'a> ReverseLogReader<'a> {
         let mut result_buf: Vec<u8> = vec![];
         loop {
             let mut read_buf = vec![];
-            let read = self.read_until(ESCAPE_CHARACTER, &mut read_buf)?;
+            let _read = self.read_until(ESCAPE_CHARACTER, &mut read_buf)?;
 
             result_buf.extend(&read_buf);
 
@@ -195,6 +180,7 @@ impl<'a> ReverseLogReader<'a> {
 mod reverse_reader_tests {
     use super::*;
     use std::io::Write;
+    use std::path::Path;
 
     #[test]
     fn test_read_until_found() {
@@ -305,36 +291,6 @@ mod reverse_reader_tests {
     }
 
     #[test]
-    fn test_forward_log_reader_fixture_db1() {
-        let db_path = Path::new(TEST_RESOURCES_DIR).join("test_db1");
-        let mut file = fs::OpenOptions::new()
-            .read(true)
-            .open(&db_path)
-            .expect("Failed to open file");
-        let mut forward_log_reader = ForwardLogReader::new(&mut file);
-
-        // There are two records in the log with "schema": Int, Null
-
-        let first_record = forward_log_reader
-            .next()
-            .expect("Failed to read the first record");
-        assert!(match first_record.values.as_slice() {
-            [RecordValue::Int(0x1D), RecordValue::Null] => true,
-            _ => false,
-        });
-
-        let last_record = forward_log_reader
-            .next()
-            .expect("Failed to read the last record");
-        assert!(match last_record.values.as_slice() {
-            [RecordValue::Int(10), RecordValue::Null] => true,
-            _ => false,
-        });
-
-        assert!(forward_log_reader.next().is_none());
-    }
-
-    #[test]
     fn test_read_exact() {
         let mut file = tempfile::tempfile().unwrap();
         file.write_all(b"hello,world").unwrap();
@@ -378,88 +334,3 @@ impl Iterator for ReverseLogReader<'_> {
         }
     }
 }
-
-pub struct ForwardLogReader<'a> {
-    reader: io::BufReader<&'a mut fs::File>,
-}
-
-impl<'a> ForwardLogReader<'a> {
-    pub fn new(file: &mut fs::File) -> ForwardLogReader {
-        let reader = io::BufReader::new(file);
-        ForwardLogReader { reader }
-    }
-
-    fn read_record(&mut self) -> Result<Option<Record>, io::Error> {
-        // The buffer that stores the bytes read from the file.
-        let mut read_buf: Vec<u8> = Vec::new();
-        // The buffer that stores all the bytes of the record read so far in reverse order.
-        let mut result_buf: Vec<u8> = Vec::new();
-
-        // Try reading a byte from the file.
-        // If we've reached the end of the file, return None.
-        let mut peek_buf = vec![0];
-        match self.reader.read_exact(&mut peek_buf) {
-            Ok(_) => {
-                // Go back one byte
-                self.reader.seek_relative(-1)?;
-            }
-            Err(ref e) if e.kind() == io::ErrorKind::UnexpectedEof => {
-                return Ok(None);
-            }
-            Err(e) => {
-                return Err(e);
-            }
-        }
-
-        loop {
-            read_buf.clear();
-            self.reader.read_until(ESCAPE_CHARACTER, &mut read_buf)?;
-            self.reader.seek_relative(-1)?;
-            result_buf.extend(&read_buf[..read_buf.len() - 1]);
-
-            // Otherwise, we must have encountered an escape character.
-            match self.read_special_sequence()? {
-                SpecialSequence::RecordSeparator => {
-                    // The record is complete, so we can break out of the loop.
-                    break;
-                }
-                SpecialSequence::LiteralFieldSeparator => {
-                    // The field separator is escaped, so we need to add it to the result buffer.
-                    result_buf.push(FIELD_SEPARATOR);
-                }
-                SpecialSequence::LiteralEscape => {
-                    // The escape character is escaped, so we need to add it to the result buffer.
-                    result_buf.push(ESCAPE_CHARACTER);
-                }
-            }
-        }
-
-        let record = Record::deserialize(&result_buf);
-        Ok(Some(record))
-    }
-
-    fn read_special_sequence(&mut self) -> Result<SpecialSequence, io::Error> {
-        let mut special_buf: Vec<u8> = vec![0; SEQ_RECORD_SEP.len()];
-        self.reader.read_exact(&mut special_buf)?;
-
-        match validate_special(&special_buf.as_slice()) {
-            Some(special) => Ok(special),
-            None => Err(io::Error::new(
-                io::ErrorKind::InvalidData,
-                "Not a special sequence",
-            )),
-        }
-    }
-}
-
-impl Iterator for ForwardLogReader<'_> {
-    type Item = Record;
-
-    fn next(&mut self) -> Option<Self::Item> {
-        match self.read_record() {
-            Ok(Some(record)) => Some(record),
-            Ok(None) => None,
-            Err(err) => panic!("Error reading record: {:?}", err),
-        }
-    }
-}