Compare commits
8 Commits
851058245d
...
8306c76a0c
Author | SHA1 | Date | |
---|---|---|---|
8306c76a0c | |||
965c021d83 | |||
a1a1d41403 | |||
b768d27f83 | |||
7cbf5439b7 | |||
8c4631e89f | |||
467ebfcc67 | |||
80dda8d821 |
30
Cargo.lock
generated
30
Cargo.lock
generated
@ -320,6 +320,24 @@ dependencies = [
|
|||||||
"zip",
|
"zip",
|
||||||
]
|
]
|
||||||
|
|
||||||
|
[[package]]
|
||||||
|
name = "chgk_ledb_async"
|
||||||
|
version = "1.1.0"
|
||||||
|
dependencies = [
|
||||||
|
"async_zip",
|
||||||
|
"bincode",
|
||||||
|
"chgk_ledb_lib",
|
||||||
|
"clap 4.3.16",
|
||||||
|
"futures",
|
||||||
|
"rand",
|
||||||
|
"serde",
|
||||||
|
"serde_derive",
|
||||||
|
"serde_json",
|
||||||
|
"tempfile",
|
||||||
|
"tokio",
|
||||||
|
"tokio-stream",
|
||||||
|
]
|
||||||
|
|
||||||
[[package]]
|
[[package]]
|
||||||
name = "chgk_ledb_lib"
|
name = "chgk_ledb_lib"
|
||||||
version = "1.1.0"
|
version = "1.1.0"
|
||||||
@ -1538,6 +1556,7 @@ dependencies = [
|
|||||||
"autocfg",
|
"autocfg",
|
||||||
"backtrace",
|
"backtrace",
|
||||||
"bytes",
|
"bytes",
|
||||||
|
"num_cpus",
|
||||||
"pin-project-lite",
|
"pin-project-lite",
|
||||||
"tokio-macros",
|
"tokio-macros",
|
||||||
]
|
]
|
||||||
@ -1553,6 +1572,17 @@ dependencies = [
|
|||||||
"syn 2.0.26",
|
"syn 2.0.26",
|
||||||
]
|
]
|
||||||
|
|
||||||
|
[[package]]
|
||||||
|
name = "tokio-stream"
|
||||||
|
version = "0.1.14"
|
||||||
|
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||||
|
checksum = "397c988d37662c7dda6d2208364a706264bf3d6138b11d436cbac0ad38832842"
|
||||||
|
dependencies = [
|
||||||
|
"futures-core",
|
||||||
|
"pin-project-lite",
|
||||||
|
"tokio",
|
||||||
|
]
|
||||||
|
|
||||||
[[package]]
|
[[package]]
|
||||||
name = "tokio-util"
|
name = "tokio-util"
|
||||||
version = "0.7.8"
|
version = "0.7.8"
|
||||||
|
@ -2,6 +2,7 @@
|
|||||||
resolver = "2"
|
resolver = "2"
|
||||||
members = [
|
members = [
|
||||||
"app",
|
"app",
|
||||||
|
"app_async",
|
||||||
"lib"
|
"lib"
|
||||||
]
|
]
|
||||||
|
|
||||||
|
33
app_async/Cargo.toml
Normal file
33
app_async/Cargo.toml
Normal file
@ -0,0 +1,33 @@
|
|||||||
|
[package]
|
||||||
|
name = "chgk_ledb_async"
|
||||||
|
version = "1.1.0"
|
||||||
|
authors = ["Dmitry <b4tm4n@mail.ru>"]
|
||||||
|
edition = "2021"
|
||||||
|
repository = "https://gitea.b4tman.ru/b4tman/chgk_ledb"
|
||||||
|
license = "MIT"
|
||||||
|
description = "Утилита загружающая базу данных ЧГК вопросов из ZIP файла в JSON формате в базу данных."
|
||||||
|
|
||||||
|
# See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
|
||||||
|
|
||||||
|
[dependencies]
|
||||||
|
chgk_ledb_lib = {path = "../lib", features = ["async", "source_async", "convert_async"]}
|
||||||
|
serde_json="1.0"
|
||||||
|
async_zip = { git = "https://github.com/Majored/rs-async-zip", rev = "ff0d985", features = [
|
||||||
|
"zstd",
|
||||||
|
"tokio",
|
||||||
|
"tokio-fs"] }
|
||||||
|
tokio = { version = "1", features = [
|
||||||
|
"io-util",
|
||||||
|
"fs",
|
||||||
|
"rt-multi-thread"
|
||||||
|
] }
|
||||||
|
tokio-stream = "0.1"
|
||||||
|
rand="0.8"
|
||||||
|
clap = { version = "4.2.7", features = ["derive"] }
|
||||||
|
futures = "0.3"
|
||||||
|
|
||||||
|
[dev-dependencies]
|
||||||
|
tempfile = "3.3"
|
||||||
|
bincode = "^2.0.0-rc.2"
|
||||||
|
serde="1.0"
|
||||||
|
serde_derive="1.0"
|
194
app_async/src/main.rs
Normal file
194
app_async/src/main.rs
Normal file
@ -0,0 +1,194 @@
|
|||||||
|
extern crate serde_json;
|
||||||
|
use clap::{Parser, Subcommand};
|
||||||
|
use futures::{pin_mut, Future};
|
||||||
|
use rand::distributions::Uniform;
|
||||||
|
use rand::seq::IteratorRandom;
|
||||||
|
use rand::{thread_rng, Rng};
|
||||||
|
|
||||||
|
use async_zip::tokio::read::seek::ZipFileReader;
|
||||||
|
use futures::stream::{self, StreamExt};
|
||||||
|
use std::time::Instant;
|
||||||
|
use tokio::sync::mpsc::{self, UnboundedReceiver, UnboundedSender};
|
||||||
|
|
||||||
|
use async_db::WriterOpts;
|
||||||
|
|
||||||
|
use tokio::{fs, io};
|
||||||
|
use tokio_stream::wrappers::UnboundedReceiverStream;
|
||||||
|
|
||||||
|
use chgk_ledb_lib::async_db;
|
||||||
|
use chgk_ledb_lib::questions::{Question, QuestionsConverterAsync};
|
||||||
|
use chgk_ledb_lib::source::ReadSourceQuestionsBatchesAsync;
|
||||||
|
|
||||||
|
const ZIP_FILENAME: &str = "json.zip";
|
||||||
|
const NEW_DB_FILENAME: &str = "db.dat";
|
||||||
|
|
||||||
|
#[derive(Subcommand, Debug)]
|
||||||
|
enum Command {
|
||||||
|
Write,
|
||||||
|
Print {
|
||||||
|
#[clap(value_parser, default_value = "0")]
|
||||||
|
id: u32,
|
||||||
|
},
|
||||||
|
ZipPrint {
|
||||||
|
#[clap(value_parser, default_value = "0")]
|
||||||
|
file_num: usize,
|
||||||
|
#[clap(value_parser, default_value = "0")]
|
||||||
|
num: usize,
|
||||||
|
},
|
||||||
|
}
|
||||||
|
|
||||||
|
#[derive(Parser, Debug)]
|
||||||
|
#[clap(author, version, about, long_about = None)]
|
||||||
|
#[clap(propagate_version = true)]
|
||||||
|
struct Cli {
|
||||||
|
#[clap(subcommand)]
|
||||||
|
command: Command,
|
||||||
|
#[clap(short, long, action)]
|
||||||
|
measure: bool,
|
||||||
|
}
|
||||||
|
|
||||||
|
async fn zip_reader_task(tx: UnboundedSender<Question>) {
|
||||||
|
let mut file = fs::File::open(ZIP_FILENAME).await.expect("open zip");
|
||||||
|
let archive = ZipFileReader::with_tokio(&mut file)
|
||||||
|
.await
|
||||||
|
.expect("open zip file reader");
|
||||||
|
let mut source_questions = archive.source_questions();
|
||||||
|
let source_questions = source_questions.stream();
|
||||||
|
pin_mut!(source_questions);
|
||||||
|
|
||||||
|
let converter = QuestionsConverterAsync::from(source_questions).convert();
|
||||||
|
pin_mut!(converter);
|
||||||
|
|
||||||
|
let mut num: u32 = 0;
|
||||||
|
while let Some(mut question) = converter.next().await {
|
||||||
|
question.num = 1 + num;
|
||||||
|
tx.send(question).expect("send");
|
||||||
|
num += 1;
|
||||||
|
}
|
||||||
|
|
||||||
|
println!("read done");
|
||||||
|
}
|
||||||
|
|
||||||
|
async fn print_question_from<F>(get_q: F)
|
||||||
|
where
|
||||||
|
F: Future<Output = Option<Question>>,
|
||||||
|
{
|
||||||
|
let q = get_q.await.expect("question not found");
|
||||||
|
println!("{:#?}", q)
|
||||||
|
}
|
||||||
|
|
||||||
|
async fn read_from_zip(file_num: usize, mut num: usize) -> Option<Question> {
|
||||||
|
let mut rng = thread_rng();
|
||||||
|
let zip_file = fs::File::open(ZIP_FILENAME).await.expect("open zip file");
|
||||||
|
let mut zip_reader = io::BufReader::new(zip_file);
|
||||||
|
let archive = ZipFileReader::with_tokio(&mut zip_reader)
|
||||||
|
.await
|
||||||
|
.expect("open zip file reader");
|
||||||
|
|
||||||
|
let mut source = archive.source_questions();
|
||||||
|
let files_count = source.len();
|
||||||
|
let file_index = if file_num == 0 {
|
||||||
|
let files = Uniform::new(0, files_count);
|
||||||
|
rng.sample(files)
|
||||||
|
} else {
|
||||||
|
file_num - 1
|
||||||
|
};
|
||||||
|
|
||||||
|
let src = source.get(file_index).await;
|
||||||
|
let src = stream::once(async { src.expect("get source file") });
|
||||||
|
pin_mut!(src);
|
||||||
|
let converter = QuestionsConverterAsync::from(src);
|
||||||
|
let questions: Vec<_> = converter.convert().collect().await;
|
||||||
|
if num == 0 {
|
||||||
|
num = (1..=questions.len()).choose(&mut rng).unwrap();
|
||||||
|
}
|
||||||
|
let mut question = questions.get(num - 1).expect("get question").clone();
|
||||||
|
question.num = num as u32;
|
||||||
|
Some(question)
|
||||||
|
}
|
||||||
|
|
||||||
|
// measure and return time elapsed in `fut` in seconds
|
||||||
|
pub async fn measure<F: Future>(fut: F) -> f64 {
|
||||||
|
let start = Instant::now();
|
||||||
|
fut.await;
|
||||||
|
let elapsed = start.elapsed();
|
||||||
|
(elapsed.as_secs() as f64) + (elapsed.subsec_nanos() as f64 / 1_000_000_000.0)
|
||||||
|
}
|
||||||
|
|
||||||
|
pub async fn measure_and_print<F: Future>(fut: F) {
|
||||||
|
let m = measure(fut).await;
|
||||||
|
eprintln!("{}", m);
|
||||||
|
}
|
||||||
|
|
||||||
|
#[tokio::main]
|
||||||
|
async fn main() {
|
||||||
|
let args = Cli::parse();
|
||||||
|
|
||||||
|
let mut action: Box<dyn Future<Output = _>> = match &args.command {
|
||||||
|
Command::Write => Box::new(write_db()),
|
||||||
|
Command::Print { id } => {
|
||||||
|
let get_question = read_from_db(*id);
|
||||||
|
Box::new(print_question_from(get_question))
|
||||||
|
}
|
||||||
|
Command::ZipPrint { file_num, num } => {
|
||||||
|
let get_question = read_from_zip(*file_num, *num);
|
||||||
|
Box::new(print_question_from(get_question))
|
||||||
|
}
|
||||||
|
};
|
||||||
|
|
||||||
|
if args.measure {
|
||||||
|
action = Box::new(measure_and_print(Box::into_pin(action)));
|
||||||
|
}
|
||||||
|
|
||||||
|
Box::into_pin(action).await;
|
||||||
|
}
|
||||||
|
|
||||||
|
async fn read_from_db(id: u32) -> Option<Question> {
|
||||||
|
let reader: async_db::Reader<Question> = async_db::Reader::new(NEW_DB_FILENAME, 2048)
|
||||||
|
.await
|
||||||
|
.expect("new db reader");
|
||||||
|
|
||||||
|
let len = reader.len();
|
||||||
|
|
||||||
|
let index = if len == 0 {
|
||||||
|
let mut rng = thread_rng();
|
||||||
|
let questions = Uniform::new(0, len);
|
||||||
|
rng.sample(questions)
|
||||||
|
} else {
|
||||||
|
id as usize - 1
|
||||||
|
};
|
||||||
|
|
||||||
|
println!("{}", &index); // DEBUG
|
||||||
|
|
||||||
|
match reader.get(index).await {
|
||||||
|
Ok(question) => Some(question),
|
||||||
|
Err(e) => {
|
||||||
|
println!("{:#?}", e); // DEBUG
|
||||||
|
None
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
async fn write_db() {
|
||||||
|
let (tx, rx) = mpsc::unbounded_channel::<Question>();
|
||||||
|
let (r, _) = tokio::join!(tokio::spawn(zip_reader_task(tx)), db_writer_task(rx),);
|
||||||
|
r.expect("tokio join");
|
||||||
|
println!("all done");
|
||||||
|
}
|
||||||
|
async fn db_writer_task(rx: UnboundedReceiver<Question>) {
|
||||||
|
let writer_opts = WriterOpts::default();
|
||||||
|
let mut writer: async_db::Writer<Question> =
|
||||||
|
async_db::Writer::new(NEW_DB_FILENAME, writer_opts)
|
||||||
|
.await
|
||||||
|
.expect("new db writer");
|
||||||
|
|
||||||
|
let mut stream: UnboundedReceiverStream<_> = rx.into();
|
||||||
|
|
||||||
|
writer
|
||||||
|
.load(&mut stream)
|
||||||
|
.await
|
||||||
|
.unwrap_or_else(|e| panic!("db writer load, {e:#?}"));
|
||||||
|
|
||||||
|
writer.finish().await.expect("db writer finish");
|
||||||
|
|
||||||
|
println!("write done");
|
||||||
|
}
|
@ -1,4 +1,4 @@
|
|||||||
use std::{marker::PhantomData, path::Path, sync::Arc};
|
use std::{path::Path, sync::Arc};
|
||||||
|
|
||||||
use async_compression::tokio::bufread::ZstdDecoder;
|
use async_compression::tokio::bufread::ZstdDecoder;
|
||||||
use async_compression::tokio::write::ZstdEncoder;
|
use async_compression::tokio::write::ZstdEncoder;
|
||||||
@ -48,7 +48,7 @@ where
|
|||||||
data_buf: Vec<u8>,
|
data_buf: Vec<u8>,
|
||||||
table: Vec<LSize>,
|
table: Vec<LSize>,
|
||||||
compress_lvl: Level,
|
compress_lvl: Level,
|
||||||
_t: PhantomData<*const T>,
|
_t: Option<Arc<T>>, // PhantomData replacement
|
||||||
}
|
}
|
||||||
|
|
||||||
impl<T> Writer<T>
|
impl<T> Writer<T>
|
||||||
@ -69,7 +69,7 @@ where
|
|||||||
data_buf,
|
data_buf,
|
||||||
table,
|
table,
|
||||||
compress_lvl,
|
compress_lvl,
|
||||||
_t: PhantomData,
|
_t: None,
|
||||||
})
|
})
|
||||||
}
|
}
|
||||||
|
|
||||||
@ -282,7 +282,6 @@ where
|
|||||||
#[cfg(test)]
|
#[cfg(test)]
|
||||||
mod test {
|
mod test {
|
||||||
use super::*;
|
use super::*;
|
||||||
use async_stream::stream;
|
|
||||||
use core::fmt::Debug;
|
use core::fmt::Debug;
|
||||||
use tempfile::tempdir;
|
use tempfile::tempdir;
|
||||||
|
|
||||||
@ -290,24 +289,21 @@ mod test {
|
|||||||
struct TestData {
|
struct TestData {
|
||||||
num: u64,
|
num: u64,
|
||||||
test: String,
|
test: String,
|
||||||
|
vnum: Vec<u64>,
|
||||||
|
vstr: Vec<String>,
|
||||||
}
|
}
|
||||||
|
|
||||||
fn gen_data(count: usize) -> impl Iterator<Item = TestData> {
|
fn gen_data(count: usize) -> impl Iterator<Item = TestData> {
|
||||||
(0..count).into_iter().map(|i| TestData {
|
(0..count).into_iter().map(|i| TestData {
|
||||||
num: i as u64,
|
num: i as u64,
|
||||||
test: "test".repeat(i),
|
test: "test".repeat(i),
|
||||||
|
vnum: (0..i * 120).map(|x| (x ^ 0x345FE34) as u64).collect(),
|
||||||
|
vstr: (0..i * 111).map(|x| "test".repeat(x).to_string()).collect(),
|
||||||
})
|
})
|
||||||
}
|
}
|
||||||
|
|
||||||
fn stream_iter<I, T>(it: I) -> impl Stream<Item = T>
|
async fn assert_data_eq((x, y): (&TestData, TestData)) {
|
||||||
where
|
assert_eq!(*x, y);
|
||||||
I: Iterator<Item = T>,
|
|
||||||
{
|
|
||||||
stream! {
|
|
||||||
for i in it {
|
|
||||||
yield i;
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
}
|
||||||
|
|
||||||
#[tokio::test]
|
#[tokio::test]
|
||||||
@ -324,7 +320,7 @@ mod test {
|
|||||||
let items_iter = gen_data(5);
|
let items_iter = gen_data(5);
|
||||||
let items: Vec<TestData> = items_iter.collect();
|
let items: Vec<TestData> = items_iter.collect();
|
||||||
|
|
||||||
let src = stream_iter(items.clone().into_iter());
|
let src = futures::stream::iter(items.clone());
|
||||||
pin_mut!(src);
|
pin_mut!(src);
|
||||||
writer.load(src).await.expect("load");
|
writer.load(src).await.expect("load");
|
||||||
writer.finish().await.expect("finish write");
|
writer.finish().await.expect("finish write");
|
||||||
@ -344,7 +340,7 @@ mod test {
|
|||||||
let items_iter = gen_data(5);
|
let items_iter = gen_data(5);
|
||||||
let items: Vec<TestData> = items_iter.collect();
|
let items: Vec<TestData> = items_iter.collect();
|
||||||
|
|
||||||
let src = stream_iter(items.clone().into_iter());
|
let src = futures::stream::iter(items.clone());
|
||||||
pin_mut!(src);
|
pin_mut!(src);
|
||||||
writer.load(src).await.expect("load");
|
writer.load(src).await.expect("load");
|
||||||
writer.finish().await.expect("finish write");
|
writer.finish().await.expect("finish write");
|
||||||
@ -372,7 +368,7 @@ mod test {
|
|||||||
let items_iter = gen_data(5);
|
let items_iter = gen_data(5);
|
||||||
let items: Vec<TestData> = items_iter.collect();
|
let items: Vec<TestData> = items_iter.collect();
|
||||||
|
|
||||||
let src = stream_iter(items.clone().into_iter());
|
let src = futures::stream::iter(items.clone());
|
||||||
pin_mut!(src);
|
pin_mut!(src);
|
||||||
writer.load(src).await.expect("load");
|
writer.load(src).await.expect("load");
|
||||||
writer.finish().await.expect("finish write");
|
writer.finish().await.expect("finish write");
|
||||||
@ -381,13 +377,18 @@ mod test {
|
|||||||
assert_eq!(items.len(), reader.len());
|
assert_eq!(items.len(), reader.len());
|
||||||
|
|
||||||
let dst_stream = reader.stream();
|
let dst_stream = reader.stream();
|
||||||
let src_stream = stream_iter(items.iter());
|
let src_stream = futures::stream::iter(items.iter());
|
||||||
|
|
||||||
async fn test_values((x, y): (&TestData, TestData)) {
|
let mut count = 0;
|
||||||
assert_eq!(*x, y);
|
src_stream
|
||||||
}
|
.zip(dst_stream)
|
||||||
|
.map(|x| {
|
||||||
src_stream.zip(dst_stream).for_each(test_values).await;
|
count += 1;
|
||||||
|
x
|
||||||
|
})
|
||||||
|
.for_each(assert_data_eq)
|
||||||
|
.await;
|
||||||
|
assert_eq!(count, items.len())
|
||||||
}
|
}
|
||||||
/// sharing Reader instance between threads
|
/// sharing Reader instance between threads
|
||||||
#[tokio::test]
|
#[tokio::test]
|
||||||
@ -404,7 +405,7 @@ mod test {
|
|||||||
let items_iter = gen_data(5);
|
let items_iter = gen_data(5);
|
||||||
let items: Vec<TestData> = items_iter.collect();
|
let items: Vec<TestData> = items_iter.collect();
|
||||||
|
|
||||||
let src = stream_iter(items.clone().into_iter());
|
let src = futures::stream::iter(items.clone());
|
||||||
pin_mut!(src);
|
pin_mut!(src);
|
||||||
writer.load(src).await.expect("load");
|
writer.load(src).await.expect("load");
|
||||||
writer.finish().await.expect("finish write");
|
writer.finish().await.expect("finish write");
|
||||||
@ -412,19 +413,15 @@ mod test {
|
|||||||
let reader: Reader<TestData> = Reader::new(&tmpfile, 2048).await.expect("new reader");
|
let reader: Reader<TestData> = Reader::new(&tmpfile, 2048).await.expect("new reader");
|
||||||
assert_eq!(items.len(), reader.len());
|
assert_eq!(items.len(), reader.len());
|
||||||
|
|
||||||
async fn test_values((x, y): (&TestData, TestData)) {
|
|
||||||
assert_eq!(*x, y);
|
|
||||||
}
|
|
||||||
|
|
||||||
let reader = Arc::new(reader);
|
let reader = Arc::new(reader);
|
||||||
for _ in 0..=3 {
|
for _ in 0..=3 {
|
||||||
let cur_items = items.clone();
|
let cur_items = items.clone();
|
||||||
let cur_reader = Arc::clone(&reader);
|
let cur_reader = Arc::clone(&reader);
|
||||||
tokio::spawn(async move {
|
tokio::spawn(async move {
|
||||||
let dst_stream = cur_reader.stream();
|
let dst_stream = cur_reader.stream();
|
||||||
let src_stream = stream_iter(cur_items.iter());
|
let src_stream = futures::stream::iter(cur_items.iter());
|
||||||
|
|
||||||
src_stream.zip(dst_stream).for_each(test_values).await;
|
src_stream.zip(dst_stream).for_each(assert_data_eq).await;
|
||||||
});
|
});
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
@ -1,8 +1,3 @@
|
|||||||
#[cfg(all(feature = "convert", feature = "convert_async"))]
|
|
||||||
compile_error!(
|
|
||||||
"feature \"convert\" and feature \"convert_async\" cannot be enabled at the same time"
|
|
||||||
);
|
|
||||||
|
|
||||||
#[cfg(feature = "async")]
|
#[cfg(feature = "async")]
|
||||||
pub mod async_db;
|
pub mod async_db;
|
||||||
#[cfg(feature = "sync")]
|
#[cfg(feature = "sync")]
|
||||||
|
@ -69,8 +69,8 @@ pub struct Question {
|
|||||||
pub batch_info: BatchInfo,
|
pub batch_info: BatchInfo,
|
||||||
}
|
}
|
||||||
|
|
||||||
#[cfg(feature = "convert")]
|
#[cfg(any(feature = "convert", feature = "convert_async"))]
|
||||||
pub mod convert {
|
pub mod convert_common {
|
||||||
use super::{BatchInfo, Question};
|
use super::{BatchInfo, Question};
|
||||||
use crate::source::{SourceQuestion, SourceQuestionsBatch};
|
use crate::source::{SourceQuestion, SourceQuestionsBatch};
|
||||||
|
|
||||||
@ -116,6 +116,12 @@ pub mod convert {
|
|||||||
result
|
result
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
#[cfg(feature = "convert")]
|
||||||
|
pub mod convert {
|
||||||
|
use super::Question;
|
||||||
|
use crate::source::SourceQuestionsBatch;
|
||||||
|
|
||||||
pub trait QuestionsConverter {
|
pub trait QuestionsConverter {
|
||||||
fn convert<'a>(&'a mut self) -> Box<dyn Iterator<Item = Question> + 'a>;
|
fn convert<'a>(&'a mut self) -> Box<dyn Iterator<Item = Question> + 'a>;
|
||||||
@ -143,59 +149,12 @@ pub use convert::QuestionsConverter;
|
|||||||
|
|
||||||
#[cfg(feature = "convert_async")]
|
#[cfg(feature = "convert_async")]
|
||||||
pub mod convert_async {
|
pub mod convert_async {
|
||||||
use async_stream::stream;
|
use futures::stream;
|
||||||
use futures_core::stream::Stream;
|
use futures_core::stream::Stream;
|
||||||
use futures_core::Future;
|
use futures_util::StreamExt;
|
||||||
use futures_util::{pin_mut, StreamExt};
|
|
||||||
|
|
||||||
use std::pin::Pin;
|
use super::Question;
|
||||||
use std::task::{Context, Poll};
|
use crate::source::SourceQuestionsBatch;
|
||||||
|
|
||||||
use super::{BatchInfo, Question};
|
|
||||||
use crate::source::{SourceQuestion, SourceQuestionsBatch};
|
|
||||||
|
|
||||||
macro_rules! make {
|
|
||||||
($Target:ident; by {$($field:ident),+}; from $src:expr) => {$Target {$(
|
|
||||||
$field: $src.$field
|
|
||||||
),+}};
|
|
||||||
($Target:ident; with defaults and by {$($field:ident),+}; from $src:expr) => {$Target {$(
|
|
||||||
$field: $src.$field
|
|
||||||
),+ ,..$Target::default()}}
|
|
||||||
}
|
|
||||||
|
|
||||||
impl From<SourceQuestion> for Question {
|
|
||||||
fn from(src: SourceQuestion) -> Self {
|
|
||||||
make! {Self; with defaults and by {
|
|
||||||
num, id, description, answer, author, comment, comment1, tour, url,
|
|
||||||
date, processed_by, redacted_by, copyright, theme, kind, source, rating
|
|
||||||
}; from src}
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
impl From<SourceQuestionsBatch> for BatchInfo {
|
|
||||||
fn from(src: SourceQuestionsBatch) -> Self {
|
|
||||||
make! {Self; by {
|
|
||||||
filename, description, author, comment, url, date,
|
|
||||||
processed_by, redacted_by, copyright, theme, kind, source, rating
|
|
||||||
}; from src}
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
impl From<SourceQuestionsBatch> for Vec<Question> {
|
|
||||||
fn from(src: SourceQuestionsBatch) -> Self {
|
|
||||||
let mut result: Vec<Question> = src
|
|
||||||
.questions
|
|
||||||
.iter()
|
|
||||||
.map(|item| item.clone().into())
|
|
||||||
.collect();
|
|
||||||
let batch_info = BatchInfo::from(src);
|
|
||||||
result.iter_mut().for_each(|question| {
|
|
||||||
question.batch_info = batch_info.clone();
|
|
||||||
});
|
|
||||||
|
|
||||||
result
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
pub struct QuestionsConverterAsync<T>
|
pub struct QuestionsConverterAsync<T>
|
||||||
where
|
where
|
||||||
@ -220,39 +179,23 @@ pub mod convert_async {
|
|||||||
T: Stream<Item = (String, Result<SourceQuestionsBatch, serde_json::Error>)>
|
T: Stream<Item = (String, Result<SourceQuestionsBatch, serde_json::Error>)>
|
||||||
+ std::marker::Unpin,
|
+ std::marker::Unpin,
|
||||||
{
|
{
|
||||||
fn convert(&mut self) -> impl Stream<Item = Question> + '_ {
|
pub fn convert(self) -> impl Stream<Item = Question> {
|
||||||
stream! {
|
self.inner
|
||||||
while let Some((filename, Ok(batch))) = self.inner.next().await {
|
.filter_map(|(name, res)| async move {
|
||||||
|
if let Ok(item) = res {
|
||||||
|
Some((name, item))
|
||||||
|
} else {
|
||||||
|
None
|
||||||
|
}
|
||||||
|
})
|
||||||
|
.flat_map(|(filename, batch)| {
|
||||||
|
stream::iter({
|
||||||
let mut batch = batch;
|
let mut batch = batch;
|
||||||
batch.filename = filename;
|
batch.filename = filename;
|
||||||
let questions: Vec<Question> = batch.into();
|
let questions: Vec<Question> = batch.into();
|
||||||
for question in questions {
|
questions
|
||||||
yield question
|
})
|
||||||
}
|
})
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
impl<T> Stream for QuestionsConverterAsync<T>
|
|
||||||
where
|
|
||||||
T: Stream<Item = (String, Result<SourceQuestionsBatch, serde_json::Error>)>
|
|
||||||
+ std::marker::Unpin,
|
|
||||||
{
|
|
||||||
type Item = Question;
|
|
||||||
|
|
||||||
fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
|
|
||||||
let convert = self.convert();
|
|
||||||
pin_mut!(convert);
|
|
||||||
match Pin::new(&mut convert.next()).poll(cx) {
|
|
||||||
Poll::Ready(Some(item)) => Poll::Ready(Some(item)),
|
|
||||||
Poll::Ready(None) => Poll::Ready(None),
|
|
||||||
Poll::Pending => Poll::Pending,
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
fn size_hint(&self) -> (usize, Option<usize>) {
|
|
||||||
self.inner.size_hint()
|
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
@ -216,15 +216,10 @@ pub use reader_sync::{ReadSourceQuestionsBatches, SourceQuestionsZipReader};
|
|||||||
|
|
||||||
#[cfg(feature = "source_async")]
|
#[cfg(feature = "source_async")]
|
||||||
pub mod reader_async {
|
pub mod reader_async {
|
||||||
use crate::util::ErrorToString;
|
use async_stream::stream;
|
||||||
|
|
||||||
use async_zip::tokio::read::seek::ZipFileReader;
|
use async_zip::tokio::read::seek::ZipFileReader;
|
||||||
use futures_core::stream::Stream;
|
use futures_core::stream::Stream;
|
||||||
use futures_core::Future;
|
use futures_util::AsyncReadExt;
|
||||||
use futures_util::{pin_mut, AsyncReadExt};
|
|
||||||
|
|
||||||
use std::pin::Pin;
|
|
||||||
use std::task::{Context, Poll};
|
|
||||||
|
|
||||||
use tokio::io::{AsyncRead, AsyncSeek};
|
use tokio::io::{AsyncRead, AsyncSeek};
|
||||||
|
|
||||||
@ -248,66 +243,70 @@ pub mod reader_async {
|
|||||||
index: None,
|
index: None,
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
async fn parse_zip_entry(
|
|
||||||
|
pub fn len(&self) -> usize {
|
||||||
|
self.zipfile.file().entries().len()
|
||||||
|
}
|
||||||
|
|
||||||
|
pub fn is_empty(&self) -> bool {
|
||||||
|
self.len() == 0
|
||||||
|
}
|
||||||
|
|
||||||
|
pub async fn get(
|
||||||
&mut self,
|
&mut self,
|
||||||
|
index: usize,
|
||||||
) -> Result<(String, Result<SourceQuestionsBatch, serde_json::Error>), String>
|
) -> Result<(String, Result<SourceQuestionsBatch, serde_json::Error>), String>
|
||||||
where
|
where
|
||||||
R: AsyncRead + AsyncSeek + Unpin,
|
R: AsyncRead + AsyncSeek + Unpin,
|
||||||
{
|
{
|
||||||
let mut reader = self
|
let len = self.len();
|
||||||
.zipfile
|
if index >= len {
|
||||||
.reader_with_entry(self.index.unwrap())
|
return Err(format!("get index={index}, when len={len}"));
|
||||||
.await
|
}
|
||||||
.str_err()?;
|
|
||||||
let filename = reader.entry().filename().clone().into_string().str_err()?;
|
|
||||||
let mut data: Vec<u8> = Vec::new();
|
|
||||||
reader.read_to_end(&mut data).await.str_err()?;
|
|
||||||
let parsed: Result<SourceQuestionsBatch, _> = serde_json::from_slice(&data);
|
|
||||||
self.index = Some(self.index.unwrap() + 1);
|
|
||||||
|
|
||||||
|
let reader = self.zipfile.reader_with_entry(index).await;
|
||||||
|
if let Err(error) = reader {
|
||||||
|
return Err(format!("reader_with_entry: {error:?}"));
|
||||||
|
}
|
||||||
|
let mut reader = reader.unwrap();
|
||||||
|
|
||||||
|
let filename = reader.entry().filename().clone().into_string().unwrap();
|
||||||
|
let mut data: Vec<u8> = Vec::new();
|
||||||
|
let readed = reader.read_to_end(&mut data).await;
|
||||||
|
if let Err(error) = readed {
|
||||||
|
return Err(format!("read_to_end: {error:?}"));
|
||||||
|
}
|
||||||
|
let parsed: Result<SourceQuestionsBatch, _> = serde_json::from_slice(&data);
|
||||||
Ok((filename, parsed))
|
Ok((filename, parsed))
|
||||||
}
|
}
|
||||||
}
|
pub async fn get_next(
|
||||||
|
&mut self,
|
||||||
impl<R> Stream for SourceQuestionsZipReaderAsync<R>
|
) -> Option<Result<(String, Result<SourceQuestionsBatch, serde_json::Error>), String>>
|
||||||
where
|
where
|
||||||
R: AsyncRead + AsyncSeek + Unpin,
|
R: AsyncRead + AsyncSeek + Unpin,
|
||||||
{
|
{
|
||||||
type Item = (String, Result<SourceQuestionsBatch, serde_json::Error>);
|
if self.index.is_none() && !self.is_empty() {
|
||||||
|
|
||||||
fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
|
|
||||||
let len = self.zipfile.file().entries().len();
|
|
||||||
if self.index.is_none() && len > 0 {
|
|
||||||
self.index = Some(0);
|
self.index = Some(0);
|
||||||
}
|
}
|
||||||
|
|
||||||
let index = &mut self.index;
|
if self.index.unwrap() >= self.len() {
|
||||||
if index.unwrap() == len {
|
return None;
|
||||||
return Poll::Ready(None);
|
|
||||||
}
|
}
|
||||||
|
|
||||||
let future = self.parse_zip_entry();
|
let item = self.get(self.index.unwrap()).await;
|
||||||
pin_mut!(future);
|
self.index = Some(self.index.unwrap() + 1);
|
||||||
match Pin::new(&mut future).poll(cx) {
|
|
||||||
Poll::Ready(Ok(item)) => Poll::Ready(Some(item)),
|
|
||||||
Poll::Ready(Err(_)) => Poll::Ready(None),
|
|
||||||
Poll::Pending => Poll::Pending,
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
fn size_hint(&self) -> (usize, Option<usize>) {
|
Some(item)
|
||||||
let len = self.zipfile.file().entries().len();
|
}
|
||||||
if self.index.is_none() {
|
pub fn stream(
|
||||||
return (len, Some(len));
|
&mut self,
|
||||||
|
) -> impl Stream<Item = (String, Result<SourceQuestionsBatch, serde_json::Error>)> + '_
|
||||||
|
{
|
||||||
|
stream! {
|
||||||
|
while let Some(Ok(item)) = self.get_next().await {
|
||||||
|
yield item
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
let index = self.index.unwrap();
|
|
||||||
let rem = if len > index + 1 {
|
|
||||||
len - (index + 1)
|
|
||||||
} else {
|
|
||||||
0
|
|
||||||
};
|
|
||||||
(rem, Some(rem))
|
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
Loading…
Reference in New Issue
Block a user