Add support for open_as_secondary APIs. (#406)

src/db.rs

@@ -51,9 +51,10 @@ unsafe impl Send for DB {}
 unsafe impl Sync for DB {}
 
 // Specifies whether open DB for read only.
-enum AccessType {
+enum AccessType<'a> {
     ReadWrite,
     ReadOnly { error_if_log_file_exist: bool },
+    Secondary { secondary_path: &'a Path },
 }
 
 impl DB {
@@ -78,6 +79,15 @@ impl DB {
         DB::open_cf_for_read_only(opts, path, None::<&str>, error_if_log_file_exist)
     }
 
+    /// Opens the database as a secondary.
+    pub fn open_as_secondary<P: AsRef<Path>>(
+        opts: &Options,
+        primary_path: P,
+        secondary_path: P,
+    ) -> Result<DB, Error> {
+        DB::open_cf_as_secondary(opts, primary_path, secondary_path, None::<&str>)
+    }
+
     /// Opens a database with the given database options and column family names.
     ///
     /// Column families opened using this function will be created with default `Options`.
@@ -120,6 +130,32 @@ impl DB {
         )
     }
 
+    /// Opens the database as a secondary with the given database options and column family names.
+    pub fn open_cf_as_secondary<P, I, N>(
+        opts: &Options,
+        primary_path: P,
+        secondary_path: P,
+        cfs: I,
+    ) -> Result<DB, Error>
+    where
+        P: AsRef<Path>,
+        I: IntoIterator<Item = N>,
+        N: AsRef<str>,
+    {
+        let cfs = cfs
+            .into_iter()
+            .map(|name| ColumnFamilyDescriptor::new(name.as_ref(), Options::default()));
+
+        DB::open_cf_descriptors_internal(
+            opts,
+            primary_path,
+            cfs,
+            AccessType::Secondary {
+                secondary_path: secondary_path.as_ref(),
+            },
+        )
+    }
+
     /// Opens a database with the given database options and column family descriptors.
     pub fn open_cf_descriptors<P, I>(opts: &Options, path: P, cfs: I) -> Result<DB, Error>
     where
@@ -232,6 +268,13 @@ impl DB {
                 AccessType::ReadWrite => {
                     ffi_try!(ffi::rocksdb_open(opts.inner, cpath.as_ptr() as *const _))
                 }
+                AccessType::Secondary { secondary_path } => {
+                    ffi_try!(ffi::rocksdb_open_as_secondary(
+                        opts.inner,
+                        cpath.as_ptr() as *const _,
+                        to_cpath(secondary_path)?.as_ptr() as *const _,
+                    ))
+                }
             }
         };
         Ok(db)
@@ -267,6 +310,17 @@ impl DB {
                     cfopts.as_ptr(),
                     cfhandles.as_mut_ptr(),
                 )),
+                AccessType::Secondary { secondary_path } => {
+                    ffi_try!(ffi::rocksdb_open_as_secondary_column_families(
+                        opts.inner,
+                        cpath.as_ptr() as *const _,
+                        to_cpath(secondary_path)?.as_ptr() as *const _,
+                        cfs_v.len() as c_int,
+                        cfnames.as_ptr(),
+                        cfopts.as_ptr(),
+                        cfhandles.as_mut_ptr(),
+                    ))
+                }
             }
         };
         Ok(db)
@@ -1047,6 +1101,15 @@ impl DB {
             Ok(DBWALIterator { inner: iter })
         }
     }
+
+    /// Tries to catch up with the primary by reading as much as possible from the
+    /// log files.
+    pub fn try_catch_up_with_primary(&self) -> Result<(), Error> {
+        unsafe {
+            ffi_try!(ffi::rocksdb_try_catch_up_with_primary(self.inner));
+        }
+        Ok(())
+    }
 }
 
 impl Drop for DB {

tests/test_db.rs

@@ -337,3 +337,28 @@ fn test_get_updates_since_out_of_range() {
     let result = db.get_updates_since(1000);
     assert!(result.is_err());
 }
+
+#[test]
+fn test_open_as_secondary() {
+    let primary_path = DBPath::new("_rust_rocksdb_test_open_as_secondary_primary");
+
+    let db = DB::open_default(&primary_path).unwrap();
+    db.put(b"key1", b"value1").unwrap();
+
+    let mut opts = Options::default();
+    opts.set_max_open_files(-1);
+
+    let secondary_path = DBPath::new("_rust_rocksdb_test_open_as_secondary_secondary");
+    let secondary = DB::open_as_secondary(&opts, &primary_path, &secondary_path).unwrap();
+
+    let result = secondary.get(b"key1");
+    let vector = result.unwrap().unwrap();
+    assert!(get_byte_slice(&vector) == b"value1");
+
+    db.put(b"key1", b"value2").unwrap();
+    assert!(secondary.try_catch_up_with_primary().is_ok());
+
+    let result = secondary.get(b"key1");
+    let vector = result.unwrap().unwrap();
+    assert!(get_byte_slice(&vector) == b"value2");
+}