start flushing out new APIs

6 files changed, 83 insertions, 38 deletions

diff --git a/src/bin/geniza-drive.rs b/src/bin/geniza-drive.rs
index 42e06e2..385ec3d 100644
--- a/src/bin/geniza-drive.rs
+++ b/src/bin/geniza-drive.rs
@@ -141,7 +141,7 @@ fn run() -> Result<()> {
                 None => Path::new("/").join(path.file_name().unwrap()),
                 Some(p) => Path::new("/").join(p)
             };
-            drive.import_dir(&path, &fpath)?;
+            drive.import_dir_all(&path, &fpath)?;
 
         }
         ("export-dir", Some(subm)) => {
diff --git a/src/bitfield.rs b/src/bitfield.rs
index 11640ea..018ccbc 100644
--- a/src/bitfield.rs
+++ b/src/bitfield.rs
@@ -2,14 +2,22 @@
 use errors::*;
 use integer_encoding::VarInt;
 use bit_field::BitArray;
+use network_msgs::Have;
 
+pub struct Bitfield {
+    inner: Vec<u64>,
+}
+
+impl Bitfield {
+
+    pub fn from_have_msg(msg: &Have) -> Bitfield {
+        unimplemented!()
+    }
 
-// WrappedBitfield
-//
-// uses vec of u64 internally?
-// 
-// fn from_message()
-// fn get(u64)
+    pub fn get(&self, index: u64) -> Result<bool> {
+        unimplemented!()
+    }
+}
 
 pub fn decode_bitfield(raw_bf: &[u8]) -> Result<Vec<u8>> {
     let mut offset = 0; // byte offset that we have read up to
diff --git a/src/drive.rs b/src/drive.rs
index 7ec2288..01807a2 100644
--- a/src/drive.rs
+++ b/src/drive.rs
@@ -490,7 +490,7 @@ impl<'a> DatDrive {
     /// Copies Stat metadata and all content from a directory (recursively) from the "real"
     /// filesystem into the DatDrive.
     /// On success, returns version number including all the added files.
-    pub fn import_dir<P: AsRef<Path>, Q: AsRef<Path>>(&mut self, source: P, dest: Q) -> Result<u64> {
+    pub fn import_dir_all<P: AsRef<Path>, Q: AsRef<Path>>(&mut self, source: P, dest: Q) -> Result<u64> {
         let source = source.as_ref();
         let dest = dest.as_ref();
         // TODO: check that dest doesn't exist (or is directory)
@@ -511,7 +511,7 @@ impl<'a> DatDrive {
                     continue
                 }
                 if path.is_dir() {
-                    ret = self.import_dir(path, dest.join(fname))?;
+                    ret = self.import_dir_all(path, dest.join(fname))?;
                 } else {
                     ret = self.import_file(path, dest.join(fname))?;
                 }
@@ -522,7 +522,7 @@ impl<'a> DatDrive {
         Ok(ret)
     }
 
-    /// Copies a file from the drive to the "real" filesystem, preserving Stat metadata.
+    /// Copies a full directory from the drive to the "real" filesystem, preserving Stat metadata.
     pub fn export_dir<P: AsRef<Path>, Q: AsRef<Path>>(&mut self, source: P, dest: Q) -> Result<()> {
         let source = source.as_ref();
         let dest = dest.as_ref();
@@ -776,7 +776,7 @@ fn test_dd_export_file() {
 }
 
 #[test]
-fn test_dd_import_dir() {
+fn test_dd_import_dir_all() {
 
     use tempdir::TempDir;
     use env_logger;
@@ -784,13 +784,13 @@ fn test_dd_import_dir() {
     let tmp_dir = TempDir::new("geniza-test").unwrap();
     let mut dd = DatDrive::create(tmp_dir.path()).unwrap();
 
-    dd.import_dir("test-data/dat/tree/Animalia/", "/").unwrap();
+    dd.import_dir_all("test-data/dat/tree/Animalia/", "/").unwrap();
 
     assert_eq!(dd.read_dir("/").count(), 0);
     assert_eq!(dd.read_dir_recursive("/").count(), 2);
 
     dd.import_file("test-data/dat/alphabet/a", "/a").unwrap();
-    assert!(dd.import_dir("test-data/dat/tree/Animalia/", "/a/").is_err());
+    assert!(dd.import_dir_all("test-data/dat/tree/Animalia/", "/a/").is_err());
 
 }
 
@@ -803,7 +803,7 @@ fn test_dd_export_dir() {
     let tmp_dir = TempDir::new("geniza-test").unwrap();
     let mut dd = DatDrive::create(tmp_dir.path()).unwrap();
 
-    dd.import_dir("test-data/dat/tree/Animalia/", "/").unwrap();
+    dd.import_dir_all("test-data/dat/tree/Animalia/", "/").unwrap();
 
     dd.export_dir("/", tmp_dir.path()).unwrap();
     dd.export_dir("/Chordata/Mammalia/Carnivora/Caniformia/", tmp_dir.path()).unwrap();
diff --git a/src/peer.rs b/src/peer.rs
index 5535b09..29b90a2 100644
--- a/src/peer.rs
+++ b/src/peer.rs
@@ -1,12 +1,31 @@
 
-//use errors::*;
-
-// DatPeer
-//  connection (thread?)
-//  last_sent_ts
-//  last_received_ts
-//  has_messages
-//  registers: vec? map?
-//
-// fn has(reg, u64)
-// fn has_intersection(reg, bitfield)
+use errors::*;
+use sleep_register::{HyperRegister, SleepDirRegister};
+use protocol::DatConnection;
+use bitfield::Bitfield;
+
+pub struct DatPeer {
+    registers: Vec<SleepDirRegister>,
+    connection: DatConnection,
+    have_log: Vec<Vec<Bitfield>>,
+}
+
+impl DatPeer {
+
+    pub fn new(connection: DatConnection, registers: Vec<SleepDirRegister>) -> DatPeer {
+        DatPeer {
+            registers,
+            connection,
+            have_log: vec![],
+        }
+    }
+
+    pub fn has(self, register: u64, index: u64) -> Result<bool> {
+        for bitfield in self.have_log[register as usize].iter() {
+            if bitfield.get(index)? {
+                return Ok(true)
+            }
+        }
+        Ok(false)
+    }
+}
diff --git a/src/protocol.rs b/src/protocol.rs
index 2b23959..5940840 100644
--- a/src/protocol.rs
+++ b/src/protocol.rs
@@ -60,6 +60,7 @@ fn msg_sugar(msg: &DatNetMessage) -> &Message {
 
 /// This helper is pretty slow/inefficient; lots of copying memory
 fn bytewise_stream_xor_ic_inplace(buf: &mut [u8], byte_offset: u64, nonce: &Nonce, key: &Key) {
+    // TODO: switch to new stream_xor_ic_inplace() variant?
     let mut offset = byte_offset;
 
     // We may have a partial-64-byte-block to finish encrypting first
@@ -70,14 +71,14 @@ fn bytewise_stream_xor_ic_inplace(buf: &mut [u8], byte_offset: u64, nonce: &Nonc
         for i in 0..partial_len {
             partial[partial_offset + i] = buf[i];
         }
-        let partial_enc = stream_xor_ic(&partial, offset / 64, &nonce, &key);
+        let partial_enc = stream_xor_ic(&partial, &nonce, offset / 64, &key);
         offset += partial_len as u64;
         for i in 0..partial_len {
             buf[i] = partial_enc[partial_offset + i];
         }
     }
     if buf.len() > partial_len {
-        let main_enc = stream_xor_ic(&buf[partial_len..], offset / 64, &nonce, &key);
+        let main_enc = stream_xor_ic(&buf[partial_len..], &nonce, offset / 64, &key);
         //offset += main_enc.len() as u64;
         for i in 0..main_enc.len() {
             buf[partial_len + i] = main_enc[i];
diff --git a/src/synchronizer.rs b/src/synchronizer.rs
index 216df6e..4e803fb 100644
--- a/src/synchronizer.rs
+++ b/src/synchronizer.rs
@@ -4,18 +4,35 @@ use network_msgs::*;
 use bitfield::*;
 use protocol::{DatNetMessage, DatConnection};
 use sleep_register::HyperRegister;
+use peer::DatPeer;
+use sleep_register::SleepDirRegister;
+
+pub enum SyncMode {
+    RxMax,
+    RxEndless,
+    TxEndless,
+    RxTxEndless,
+}
+
+pub struct Synchronizer {
+    peers: Vec<DatPeer>,
+    registers: Vec<SleepDirRegister>,
+    mode: SyncMode,
+    wanted: Bitfield,
+    inflight: Vec<Vec<u64>>,
+}
+
+impl Synchronizer {
 
-// Synchronizer
-//  register_keys
-//  peers: vec
-//  registers: HyperRegisters
-//  mode: enum
-//  state: enum
-//  wanted: bitfield
-//  requested: vec
-//
-// fn next_wanted() -> Option((reg, u64))
-// fn tick()
+    pub fn next_wanted(&mut self, reg: u64) -> Option<(u64, u64)> {
+        // XXX
+        None
+    }
+
+    pub fn tick(&mut self) -> Result<()> {
+        Ok(())
+    }
+}
 
 
 fn max_index(have_msg: &Have) -> Result<u64> {