replicate from storage test

Author: MarinPostma

libsql-wal/src/replication/replicator.rs

@@ -134,6 +134,8 @@ impl<IO: Io> Replicator<IO> {
 
 #[cfg(test)]
 mod test {
+    use std::time::Duration;
+
     use tempfile::NamedTempFile;
     use tokio_stream::StreamExt;
 
@@ -252,4 +254,53 @@ mod test {
             .unwrap();
         }
     }
+
+    #[tokio::test]
+    async fn stream_from_storage() {
+        let env = TestEnv::new_store(true);
+        let conn = env.open_conn("test");
+        let shared = env.shared("test");
+
+        conn.execute("create table test (x)", ()).unwrap();
+
+        conn.execute("insert into test values (randomblob(128))", ())
+            .unwrap();
+
+        tokio::task::spawn_blocking({
+            let shared = shared.clone();
+            move || seal_current_segment(&shared)
+        }).await.unwrap();
+
+        conn.execute("create table test2 (x)", ()).unwrap();
+        conn.execute("insert into test2 values (randomblob(128))", ())
+            .unwrap();
+
+        tokio::task::spawn_blocking({
+            let shared = shared.clone();
+            move || seal_current_segment(&shared)
+        }).await.unwrap();
+
+        while !shared.current.load().tail().is_empty() {
+            tokio::time::sleep(Duration::from_millis(50)).await;
+        }
+
+        let db_content = std::fs::read(&env.db_path("test").join("data")).unwrap();
+
+        let mut replicator = Replicator::new(shared, 1);
+        let stream = replicator.frame_stream().take(3);
+
+        tokio::pin!(stream);
+
+        let tmp = NamedTempFile::new().unwrap();
+        let mut replica_content = vec![0u8; db_content.len()];
+        while let Some(f) = stream.next().await { 
+            let frame = f.unwrap();
+            dbg!(frame.header().page_no());
+            let offset = (frame.header().page_no() as usize - 1) * 4096;
+            tmp.as_file().write_all_at(frame.data(), offset as u64).unwrap();
+            replica_content[offset..offset+4096].copy_from_slice(frame.data());
+        }
+
+        assert_eq!(replica_content, db_content);
+    }
 }