finish the dump reader API, the dump Writer API now needs to be updated

This commit is contained in:
Tamo 2022-10-10 18:57:27 +02:00 committed by Clément Renault
parent 0284764b5e
commit 7579a363ab
No known key found for this signature in database
GPG Key ID: 92ADA4E935E71FA4
9 changed files with 121 additions and 119 deletions

View File

@ -61,7 +61,10 @@ pub(crate) mod test {
use time::{macros::datetime, Duration}; use time::{macros::datetime, Duration};
use uuid::Uuid; use uuid::Uuid;
use crate::{reader, DumpWriter, IndexMetadata, Version}; use crate::{
reader::{self, Document},
DumpWriter, IndexMetadata, Version,
};
pub fn create_test_instance_uid() -> Uuid { pub fn create_test_instance_uid() -> Uuid {
Uuid::parse_str("9e15e977-f2ae-4761-943f-1eaf75fd736d").unwrap() Uuid::parse_str("9e15e977-f2ae-4761-943f-1eaf75fd736d").unwrap()
@ -111,7 +114,7 @@ pub(crate) mod test {
settings.check() settings.check()
} }
pub fn create_test_tasks() -> Vec<(TaskView, Option<&'static [u8]>)> { pub fn create_test_tasks() -> Vec<(TaskView, Option<Vec<Document>>)> {
vec![ vec![
( (
TaskView { TaskView {
@ -150,7 +153,16 @@ pub(crate) mod test {
started_at: Some(datetime!(2022-11-20 0:00 UTC)), started_at: Some(datetime!(2022-11-20 0:00 UTC)),
finished_at: Some(datetime!(2022-11-21 0:00 UTC)), finished_at: Some(datetime!(2022-11-21 0:00 UTC)),
}, },
Some(br#"{ "id": 4, "race": "leonberg" }"#), Some(vec![
json!({ "id": 4, "race": "leonberg" })
.as_object()
.unwrap()
.clone(),
json!({ "id": 5, "race": "patou" })
.as_object()
.unwrap()
.clone(),
]),
), ),
( (
TaskView { TaskView {
@ -224,10 +236,12 @@ pub(crate) mod test {
// ========== pushing the task queue // ========== pushing the task queue
let tasks = create_test_tasks(); let tasks = create_test_tasks();
/*
let mut task_queue = dump.create_tasks_queue().unwrap(); let mut task_queue = dump.create_tasks_queue().unwrap();
for (task, update_file) in &tasks { for (task, update_file) in &tasks {
task_queue.push_task(task, update_file.map(|c| c)).unwrap(); task_queue.push_task(task, update_file.map(|c| c)).unwrap();
} }
*/
// ========== pushing the api keys // ========== pushing the api keys
let api_keys = create_test_api_keys(); let api_keys = create_test_api_keys();
@ -283,9 +297,11 @@ pub(crate) mod test {
"A content file was expected for the task {}.", "A content file was expected for the task {}.",
expected.0.uid expected.0.uid
); );
let mut update = Vec::new(); let updates = content_file
content_file.unwrap().read_to_end(&mut update).unwrap(); .unwrap()
assert_eq!(update, expected_update); .collect::<Result<Vec<_>, _>>()
.unwrap();
assert_eq!(updates, expected_update);
} }
} }

View File

@ -11,6 +11,7 @@ use self::{
use super::{ use super::{
v5::V5Reader, v5::V5Reader,
v6::{self, V6IndexReader, V6Reader}, v6::{self, V6IndexReader, V6Reader},
Document, UpdateFile,
}; };
pub mod v2_to_v3; pub mod v2_to_v3;
@ -62,7 +63,7 @@ impl Compat {
pub fn tasks( pub fn tasks(
&mut self, &mut self,
) -> Box<dyn Iterator<Item = Result<(v6::Task, Option<v6::UpdateFile>)>> + '_> { ) -> Box<dyn Iterator<Item = Result<(v6::Task, Option<Box<UpdateFile>>)>> + '_> {
match self { match self {
Compat::Current(current) => current.tasks(), Compat::Current(current) => current.tasks(),
Compat::Compat(compat) => compat.tasks(), Compat::Compat(compat) => compat.tasks(),
@ -118,14 +119,14 @@ impl CompatIndex {
} }
} }
pub fn documents(&mut self) -> Result<Box<dyn Iterator<Item = Result<v6::Document>> + '_>> { pub fn documents(&mut self) -> Result<Box<dyn Iterator<Item = Result<Document>> + '_>> {
match self { match self {
CompatIndex::Current(v6) => v6 CompatIndex::Current(v6) => v6
.documents() .documents()
.map(|iter| Box::new(iter) as Box<dyn Iterator<Item = Result<v6::Document>> + '_>), .map(|iter| Box::new(iter) as Box<dyn Iterator<Item = Result<Document>> + '_>),
CompatIndex::Compat(compat) => compat CompatIndex::Compat(compat) => compat
.documents() .documents()
.map(|iter| Box::new(iter) as Box<dyn Iterator<Item = Result<v6::Document>> + '_>), .map(|iter| Box::new(iter) as Box<dyn Iterator<Item = Result<Document>> + '_>),
} }
} }

View File

@ -1,4 +1,4 @@
use crate::reader::{v4, v5}; use crate::reader::{v4, v5, Document};
use crate::Result; use crate::Result;
use super::v3_to_v4::{CompatIndexV3ToV4, CompatV3ToV4}; use super::v3_to_v4::{CompatIndexV3ToV4, CompatV3ToV4};
@ -66,8 +66,6 @@ impl CompatV4ToV5 {
}; };
Box::new(tasks.map(|task| { Box::new(tasks.map(|task| {
task.map(|(task, content_file)| { task.map(|(task, content_file)| {
// let task_view: v4::tasks::TaskView = task.into();
let task = v5::Task { let task = v5::Task {
id: task.id, id: task.id,
content: match task.content { content: match task.content {
@ -244,14 +242,14 @@ impl CompatIndexV4ToV5 {
} }
} }
pub fn documents(&mut self) -> Result<Box<dyn Iterator<Item = Result<v5::Document>> + '_>> { pub fn documents(&mut self) -> Result<Box<dyn Iterator<Item = Result<Document>> + '_>> {
match self { match self {
CompatIndexV4ToV5::V4(v4) => v4 CompatIndexV4ToV5::V4(v4) => v4
.documents() .documents()
.map(|iter| Box::new(iter) as Box<dyn Iterator<Item = Result<v5::Document>> + '_>), .map(|iter| Box::new(iter) as Box<dyn Iterator<Item = Result<Document>> + '_>),
CompatIndexV4ToV5::Compat(compat) => compat CompatIndexV4ToV5::Compat(compat) => compat
.documents() .documents()
.map(|iter| Box::new(iter) as Box<dyn Iterator<Item = Result<v5::Document>> + '_>), .map(|iter| Box::new(iter) as Box<dyn Iterator<Item = Result<Document>> + '_>),
} }
} }

View File

@ -1,4 +1,4 @@
use crate::reader::{v5, v6}; use crate::reader::{v5, v6, Document, UpdateFile};
use crate::Result; use crate::Result;
use super::v4_to_v5::{CompatIndexV4ToV5, CompatV4ToV5}; use super::v4_to_v5::{CompatIndexV4ToV5, CompatV4ToV5};
@ -54,10 +54,10 @@ impl CompatV5ToV6 {
pub fn tasks( pub fn tasks(
&mut self, &mut self,
) -> Box<dyn Iterator<Item = Result<(v6::Task, Option<v6::UpdateFile>)>> + '_> { ) -> Box<dyn Iterator<Item = Result<(v6::Task, Option<Box<UpdateFile>>)>> + '_> {
let tasks = match self { let tasks = match self {
CompatV5ToV6::V5(v5) => v5.tasks(), CompatV5ToV6::V5(v5) => v5.tasks(),
CompatV5ToV6::Compat(compat) => todo!(), // compat.tasks(), CompatV5ToV6::Compat(compat) => compat.tasks(),
}; };
Box::new(tasks.map(|task| { Box::new(tasks.map(|task| {
task.map(|(task, content_file)| { task.map(|(task, content_file)| {
@ -202,14 +202,14 @@ impl CompatIndexV5ToV6 {
} }
} }
pub fn documents(&mut self) -> Result<Box<dyn Iterator<Item = Result<v6::Document>> + '_>> { pub fn documents(&mut self) -> Result<Box<dyn Iterator<Item = Result<Document>> + '_>> {
match self { match self {
CompatIndexV5ToV6::V5(v5) => v5 CompatIndexV5ToV6::V5(v5) => v5
.documents() .documents()
.map(|iter| Box::new(iter) as Box<dyn Iterator<Item = Result<v6::Document>> + '_>), .map(|iter| Box::new(iter) as Box<dyn Iterator<Item = Result<Document>> + '_>),
CompatIndexV5ToV6::Compat(compat) => compat CompatIndexV5ToV6::Compat(compat) => compat
.documents() .documents()
.map(|iter| Box::new(iter) as Box<dyn Iterator<Item = Result<v6::Document>> + '_>), .map(|iter| Box::new(iter) as Box<dyn Iterator<Item = Result<Document>> + '_>),
} }
} }

View File

@ -6,20 +6,14 @@ use flate2::bufread::GzDecoder;
use serde::Deserialize; use serde::Deserialize;
use tempfile::TempDir; use tempfile::TempDir;
use time::OffsetDateTime;
use uuid::Uuid;
// use crate::reader::compat::Compat; use crate::{Result, Version};
use crate::{IndexMetadata, Result, Version};
use self::compat::Compat; use self::compat::Compat;
// use self::loaders::{v2, v3, v4, v5};
// pub mod error;
mod compat; mod compat;
// mod loaders;
// mod v1; // pub(self) mod v1;
pub(self) mod v2; pub(self) mod v2;
pub(self) mod v3; pub(self) mod v3;
pub(self) mod v4; pub(self) mod v4;
@ -47,38 +41,15 @@ pub fn open(dump: impl Read) -> Result<Compat> {
match dump_version { match dump_version {
// Version::V1 => Ok(Box::new(v1::Reader::open(path)?)), // Version::V1 => Ok(Box::new(v1::Reader::open(path)?)),
Version::V1 => todo!(), Version::V1 => todo!(),
Version::V2 => todo!(), Version::V2 => Ok(v2::V2Reader::open(path)?
.to_v3()
.to_v4()
.to_v5()
.to_v6()
.into()),
Version::V3 => Ok(v3::V3Reader::open(path)?.to_v4().to_v5().to_v6().into()), Version::V3 => Ok(v3::V3Reader::open(path)?.to_v4().to_v5().to_v6().into()),
Version::V4 => Ok(v4::V4Reader::open(path)?.to_v5().to_v6().into()), Version::V4 => Ok(v4::V4Reader::open(path)?.to_v5().to_v6().into()),
Version::V5 => Ok(v5::V5Reader::open(path)?.to_v6().into()), Version::V5 => Ok(v5::V5Reader::open(path)?.to_v6().into()),
Version::V6 => Ok(v6::V6Reader::open(path)?.into()), Version::V6 => Ok(v6::V6Reader::open(path)?.into()),
} }
} }
pub trait DumpReader {
/// Return the version of the dump.
fn version(&self) -> Version;
/// Return at which date the dump was created if there was one.
fn date(&self) -> Option<OffsetDateTime>;
/// Return the instance-uid if there was one.
fn instance_uid(&self) -> Result<Option<Uuid>>;
/// Return an iterator over each indexes.
fn indexes(&self) -> Result<Box<dyn Iterator<Item = Result<Box<dyn IndexReader + '_>>> + '_>>;
/// Return all the tasks in the dump with a possible update file.
fn tasks(
&mut self,
) -> Box<dyn Iterator<Item = Result<(v6::Task, Option<v6::UpdateFile>)>> + '_>;
/// Return all the keys.
fn keys(&mut self) -> Box<dyn Iterator<Item = Result<v6::Key>> + '_>;
}
pub trait IndexReader {
fn metadata(&self) -> &IndexMetadata;
fn documents(&mut self) -> Result<Box<dyn Iterator<Item = Result<v6::Document>> + '_>>;
fn settings(&mut self) -> Result<v6::Settings<v6::Checked>>;
}

View File

@ -1,5 +1,3 @@
use std::{fs::File, io::BufReader};
use serde::Deserialize; use serde::Deserialize;
use time::OffsetDateTime; use time::OffsetDateTime;
use uuid::Uuid; use uuid::Uuid;

View File

@ -43,9 +43,9 @@ use tempfile::TempDir;
use time::OffsetDateTime; use time::OffsetDateTime;
use uuid::Uuid; use uuid::Uuid;
use crate::{IndexMetadata, Result, Version}; use crate::{Error, IndexMetadata, Result, Version};
use super::compat::v5_to_v6::CompatV5ToV6; use super::{compat::v5_to_v6::CompatV5ToV6, Document};
pub mod errors; pub mod errors;
pub mod keys; pub mod keys;
@ -53,13 +53,11 @@ pub mod meta;
pub mod settings; pub mod settings;
pub mod tasks; pub mod tasks;
pub type Document = serde_json::Map<String, serde_json::Value>;
pub type Settings<T> = settings::Settings<T>; pub type Settings<T> = settings::Settings<T>;
pub type Checked = settings::Checked; pub type Checked = settings::Checked;
pub type Unchecked = settings::Unchecked; pub type Unchecked = settings::Unchecked;
pub type Task = tasks::Task; pub type Task = tasks::Task;
pub type UpdateFile = File;
pub type Key = keys::Key; pub type Key = keys::Key;
// ===== Other types to clarify the code of the compat module // ===== Other types to clarify the code of the compat module
@ -150,7 +148,9 @@ impl V5Reader {
})) }))
} }
pub fn tasks(&mut self) -> Box<dyn Iterator<Item = Result<(Task, Option<UpdateFile>)>> + '_> { pub fn tasks(
&mut self,
) -> Box<dyn Iterator<Item = Result<(Task, Option<Box<super::UpdateFile>>)>> + '_> {
Box::new((&mut self.tasks).lines().map(|line| -> Result<_> { Box::new((&mut self.tasks).lines().map(|line| -> Result<_> {
let task: Task = serde_json::from_str(&line?)?; let task: Task = serde_json::from_str(&line?)?;
if !task.is_finished() { if !task.is_finished() {
@ -161,7 +161,12 @@ impl V5Reader {
.join("updates") .join("updates")
.join("updates_files") .join("updates_files")
.join(uuid.to_string()); .join(uuid.to_string());
Ok((task, Some(File::open(update_file_path).unwrap()))) Ok((
task,
Some(
Box::new(UpdateFile::new(&update_file_path)?) as Box<super::UpdateFile>
),
))
} else { } else {
Ok((task, None)) Ok((task, None))
} }
@ -224,6 +229,32 @@ impl V5IndexReader {
} }
} }
pub struct UpdateFile {
reader: BufReader<File>,
}
impl UpdateFile {
fn new(path: &Path) -> Result<Self> {
Ok(UpdateFile {
reader: BufReader::new(File::open(path)?),
})
}
}
impl Iterator for UpdateFile {
type Item = Result<Document>;
fn next(&mut self) -> Option<Self::Item> {
(&mut self.reader)
.lines()
.map(|line| {
line.map_err(Error::from)
.and_then(|line| serde_json::from_str(&line).map_err(Error::from))
})
.next()
}
}
#[cfg(test)] #[cfg(test)]
pub(crate) mod test { pub(crate) mod test {
use std::{fs::File, io::BufReader}; use std::{fs::File, io::BufReader};

View File

@ -11,18 +11,16 @@ use uuid::Uuid;
use crate::{Error, IndexMetadata, Result, Version}; use crate::{Error, IndexMetadata, Result, Version};
use super::{DumpReader, IndexReader}; use super::Document;
pub use index; pub use index;
pub type Metadata = crate::Metadata; pub type Metadata = crate::Metadata;
pub type Document = serde_json::Map<String, serde_json::Value>;
pub type Settings<T> = index::Settings<T>; pub type Settings<T> = index::Settings<T>;
pub type Checked = index::Checked; pub type Checked = index::Checked;
pub type Unchecked = index::Unchecked; pub type Unchecked = index::Unchecked;
pub type Task = index_scheduler::TaskView; pub type Task = index_scheduler::TaskView;
pub type UpdateFile = File;
pub type Key = meilisearch_auth::Key; pub type Key = meilisearch_auth::Key;
// ===== Other types to clarify the code of the compat module // ===== Other types to clarify the code of the compat module
@ -106,7 +104,9 @@ impl V6Reader {
)) ))
} }
pub fn tasks(&mut self) -> Box<dyn Iterator<Item = Result<(Task, Option<UpdateFile>)>> + '_> { pub fn tasks(
&mut self,
) -> Box<dyn Iterator<Item = Result<(Task, Option<Box<super::UpdateFile>>)>> + '_> {
Box::new((&mut self.tasks).lines().map(|line| -> Result<_> { Box::new((&mut self.tasks).lines().map(|line| -> Result<_> {
let mut task: index_scheduler::TaskView = serde_json::from_str(&line?)?; let mut task: index_scheduler::TaskView = serde_json::from_str(&line?)?;
// TODO: this can be removed once we can `Deserialize` the duration from the `TaskView`. // TODO: this can be removed once we can `Deserialize` the duration from the `TaskView`.
@ -121,7 +121,10 @@ impl V6Reader {
.join(task.uid.to_string()); .join(task.uid.to_string());
if update_file_path.exists() { if update_file_path.exists() {
Ok((task, Some(File::open(update_file_path)?))) Ok((
task,
Some(Box::new(UpdateFile::new(&update_file_path)?) as Box<super::UpdateFile>),
))
} else { } else {
Ok((task, None)) Ok((task, None))
} }
@ -137,37 +140,29 @@ impl V6Reader {
} }
} }
impl DumpReader for V6Reader { pub struct UpdateFile {
fn version(&self) -> Version { reader: BufReader<File>,
self.version() }
}
fn date(&self) -> Option<OffsetDateTime> { impl UpdateFile {
self.date() fn new(path: &Path) -> Result<Self> {
} Ok(UpdateFile {
reader: BufReader::new(File::open(path)?),
fn instance_uid(&self) -> Result<Option<Uuid>> {
self.instance_uid()
}
fn indexes(
&self,
) -> Result<Box<dyn Iterator<Item = Result<Box<dyn super::IndexReader + '_>>> + '_>> {
self.indexes().map(|iter| {
Box::new(iter.map(|result| {
result.map(|index| Box::new(index) as Box<dyn super::IndexReader + '_>)
})) as Box<dyn Iterator<Item = _>>
}) })
} }
}
fn tasks( impl Iterator for UpdateFile {
&mut self, type Item = Result<Document>;
) -> Box<dyn Iterator<Item = Result<(self::Task, Option<self::UpdateFile>)>> + '_> {
Box::new(self.tasks())
}
fn keys(&mut self) -> Box<dyn Iterator<Item = Result<self::Key>> + '_> { fn next(&mut self) -> Option<Self::Item> {
Box::new(self.keys()) (&mut self.reader)
.lines()
.map(|line| {
line.map_err(Error::from)
.and_then(|line| serde_json::from_str(&line).map_err(Error::from))
})
.next()
} }
} }
@ -205,18 +200,3 @@ impl V6IndexReader {
Ok(settings.check()) Ok(settings.check())
} }
} }
impl IndexReader for V6IndexReader {
fn metadata(&self) -> &IndexMetadata {
self.metadata()
}
fn documents(&mut self) -> Result<Box<dyn Iterator<Item = Result<Document>> + '_>> {
self.documents()
.map(|iter| Box::new(iter) as Box<dyn Iterator<Item = Result<Document>> + '_>)
}
fn settings(&mut self) -> Result<Settings<Checked>> {
self.settings()
}
}

View File

@ -160,9 +160,12 @@ pub(crate) mod test {
use flate2::bufread::GzDecoder; use flate2::bufread::GzDecoder;
use index::Unchecked; use index::Unchecked;
use crate::test::{ use crate::{
create_test_api_keys, create_test_documents, create_test_dump, create_test_instance_uid, reader::Document,
create_test_settings, create_test_tasks, test::{
create_test_api_keys, create_test_documents, create_test_dump,
create_test_instance_uid, create_test_settings, create_test_tasks,
},
}; };
use super::*; use super::*;
@ -309,8 +312,12 @@ pub(crate) mod test {
if let Some(expected_update) = expected.1 { if let Some(expected_update) = expected.1 {
let path = dump_path.join(format!("tasks/update_files/{}", expected.0.uid)); let path = dump_path.join(format!("tasks/update_files/{}", expected.0.uid));
println!("trying to open {}", path.display()); println!("trying to open {}", path.display());
let update = fs::read(path).unwrap(); let update = fs::read_to_string(path).unwrap();
assert_eq!(update, expected_update); let documents: Vec<Document> = update
.lines()
.map(|line| serde_json::from_str(line).unwrap())
.collect();
assert_eq!(documents, expected_update);
} }
} }