Cargo.lock

@@ -320,24 +320,6 @@ dependencies = [
  "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]]
 name = "chgk_ledb_lib"
 version = "1.1.0"
@@ -1556,7 +1538,6 @@ dependencies = [
  "autocfg",
  "backtrace",
  "bytes",
- "num_cpus",
  "pin-project-lite",
  "tokio-macros",
 ]
@@ -1572,17 +1553,6 @@ dependencies = [
  "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]]
 name = "tokio-util"
 version = "0.7.8"

Cargo.toml

@@ -2,7 +2,6 @@
 resolver = "2"
 members = [
   "app",
-  "app_async",
   "lib"
 ]
 

app_async/Cargo.toml

@@ -1,33 +0,0 @@
-[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"

app_async/src/main.rs

@@ -1,194 +0,0 @@
-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");
-}

lib/src/async_db.rs

@@ -1,4 +1,4 @@
-use std::{path::Path, sync::Arc};
+use std::{marker::PhantomData, path::Path, sync::Arc};
 
 use async_compression::tokio::bufread::ZstdDecoder;
 use async_compression::tokio::write::ZstdEncoder;
@@ -48,7 +48,7 @@ where
     data_buf: Vec<u8>,
     table: Vec<LSize>,
     compress_lvl: Level,
-    _t: Option<Arc<T>>, // PhantomData replacement
+    _t: PhantomData<*const T>,
 }
 
 impl<T> Writer<T>
@@ -69,7 +69,7 @@ where
             data_buf,
             table,
             compress_lvl,
-            _t: None,
+            _t: PhantomData,
         })
     }
 
@@ -282,6 +282,7 @@ where
 #[cfg(test)]
 mod test {
     use super::*;
+    use async_stream::stream;
     use core::fmt::Debug;
     use tempfile::tempdir;
 
@@ -289,21 +290,24 @@ mod test {
     struct TestData {
         num: u64,
         test: String,
-        vnum: Vec<u64>,
-        vstr: Vec<String>,
     }
 
     fn gen_data(count: usize) -> impl Iterator<Item = TestData> {
         (0..count).into_iter().map(|i| TestData {
             num: i as u64,
             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(),
         })
     }
 
-    async fn assert_data_eq((x, y): (&TestData, TestData)) {
+    fn stream_iter<I, T>(it: I) -> impl Stream<Item = T>
-        assert_eq!(*x, y);
+    where
+        I: Iterator<Item = T>,
+    {
+        stream! {
+            for i in it {
+                yield i;
+            }
+        }
     }
 
     #[tokio::test]
@@ -320,7 +324,7 @@ mod test {
         let items_iter = gen_data(5);
         let items: Vec<TestData> = items_iter.collect();
 
-        let src = futures::stream::iter(items.clone());
+        let src = stream_iter(items.clone().into_iter());
         pin_mut!(src);
         writer.load(src).await.expect("load");
         writer.finish().await.expect("finish write");
@@ -340,7 +344,7 @@ mod test {
         let items_iter = gen_data(5);
         let items: Vec<TestData> = items_iter.collect();
 
-        let src = futures::stream::iter(items.clone());
+        let src = stream_iter(items.clone().into_iter());
         pin_mut!(src);
         writer.load(src).await.expect("load");
         writer.finish().await.expect("finish write");
@@ -368,7 +372,7 @@ mod test {
         let items_iter = gen_data(5);
         let items: Vec<TestData> = items_iter.collect();
 
-        let src = futures::stream::iter(items.clone());
+        let src = stream_iter(items.clone().into_iter());
         pin_mut!(src);
         writer.load(src).await.expect("load");
         writer.finish().await.expect("finish write");
@@ -377,18 +381,13 @@ mod test {
         assert_eq!(items.len(), reader.len());
 
         let dst_stream = reader.stream();
-        let src_stream = futures::stream::iter(items.iter());
+        let src_stream = stream_iter(items.iter());
 
-        let mut count = 0;
+        async fn test_values((x, y): (&TestData, TestData)) {
-        src_stream
+            assert_eq!(*x, y);
-            .zip(dst_stream)
+        }
-            .map(|x| {
+
-                count += 1;
+        src_stream.zip(dst_stream).for_each(test_values).await;
-                x
-            })
-            .for_each(assert_data_eq)
-            .await;
-        assert_eq!(count, items.len())
     }
     /// sharing Reader instance between threads
     #[tokio::test]
@@ -405,7 +404,7 @@ mod test {
         let items_iter = gen_data(5);
         let items: Vec<TestData> = items_iter.collect();
 
-        let src = futures::stream::iter(items.clone());
+        let src = stream_iter(items.clone().into_iter());
         pin_mut!(src);
         writer.load(src).await.expect("load");
         writer.finish().await.expect("finish write");
@@ -413,15 +412,19 @@ mod test {
         let reader: Reader<TestData> = Reader::new(&tmpfile, 2048).await.expect("new reader");
         assert_eq!(items.len(), reader.len());
 
+        async fn test_values((x, y): (&TestData, TestData)) {
+            assert_eq!(*x, y);
+        }
+
         let reader = Arc::new(reader);
         for _ in 0..=3 {
             let cur_items = items.clone();
             let cur_reader = Arc::clone(&reader);
             tokio::spawn(async move {
                 let dst_stream = cur_reader.stream();
-                let src_stream = futures::stream::iter(cur_items.iter());
+                let src_stream = stream_iter(cur_items.iter());
 
-                src_stream.zip(dst_stream).for_each(assert_data_eq).await;
+                src_stream.zip(dst_stream).for_each(test_values).await;
             });
         }
     }

lib/src/lib.rs

@@ -1,3 +1,8 @@
+#[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")]
 pub mod async_db;
 #[cfg(feature = "sync")]

lib/src/questions.rs

@@ -69,8 +69,8 @@ pub struct Question {
     pub batch_info: BatchInfo,
 }
 
-#[cfg(any(feature = "convert", feature = "convert_async"))]
+#[cfg(feature = "convert")]
-pub mod convert_common {
+pub mod convert {
     use super::{BatchInfo, Question};
     use crate::source::{SourceQuestion, SourceQuestionsBatch};
 
@@ -116,12 +116,6 @@ pub mod convert_common {
             result
         }
     }
-}
-
-#[cfg(feature = "convert")]
-pub mod convert {
-    use super::Question;
-    use crate::source::SourceQuestionsBatch;
 
     pub trait QuestionsConverter {
         fn convert<'a>(&'a mut self) -> Box<dyn Iterator<Item = Question> + 'a>;
@@ -149,12 +143,59 @@ pub use convert::QuestionsConverter;
 
 #[cfg(feature = "convert_async")]
 pub mod convert_async {
-    use futures::stream;
+    use async_stream::stream;
     use futures_core::stream::Stream;
-    use futures_util::StreamExt;
+    use futures_core::Future;
+    use futures_util::{pin_mut, StreamExt};
 
-    use super::Question;
+    use std::pin::Pin;
-    use crate::source::SourceQuestionsBatch;
+    use std::task::{Context, Poll};
+
+    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>
     where
@@ -179,23 +220,39 @@ pub mod convert_async {
         T: Stream<Item = (String, Result<SourceQuestionsBatch, serde_json::Error>)>
             + std::marker::Unpin,
     {
-        pub fn convert(self) -> impl Stream<Item = Question> {
+        fn convert(&mut self) -> impl Stream<Item = Question> + '_ {
-            self.inner
+            stream! {
-                .filter_map(|(name, res)| async move {
+                while let Some((filename, Ok(batch))) = self.inner.next().await {
-                    if let Ok(item) = res {
+                    let mut batch = batch;
-                        Some((name, item))
+                    batch.filename = filename;
-                    } else {
+                    let questions: Vec<Question> = batch.into();
-                        None
+                    for question in questions {
+                        yield question
                     }
-                })
+                }
-                .flat_map(|(filename, batch)| {
+            }
-                    stream::iter({
+        }
-                        let mut batch = batch;
+    }
-                        batch.filename = filename;
+
-                        let questions: Vec<Question> = batch.into();
+    impl<T> Stream for QuestionsConverterAsync<T>
-                        questions
+    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()
         }
     }
 }

lib/src/source.rs

@@ -216,10 +216,15 @@ pub use reader_sync::{ReadSourceQuestionsBatches, SourceQuestionsZipReader};
 
 #[cfg(feature = "source_async")]
 pub mod reader_async {
-    use async_stream::stream;
+    use crate::util::ErrorToString;
+
     use async_zip::tokio::read::seek::ZipFileReader;
     use futures_core::stream::Stream;
-    use futures_util::AsyncReadExt;
+    use futures_core::Future;
+    use futures_util::{pin_mut, AsyncReadExt};
+
+    use std::pin::Pin;
+    use std::task::{Context, Poll};
 
     use tokio::io::{AsyncRead, AsyncSeek};
 
@@ -243,70 +248,66 @@ pub mod reader_async {
                 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,
-            index: usize,
         ) -> Result<(String, Result<SourceQuestionsBatch, serde_json::Error>), String>
         where
             R: AsyncRead + AsyncSeek + Unpin,
         {
-            let len = self.len();
+            let mut reader = self
-            if index >= len {
+                .zipfile
-                return Err(format!("get index={index}, when len={len}"));
+                .reader_with_entry(self.index.unwrap())
-            }
+                .await
-
+                .str_err()?;
-            let reader = self.zipfile.reader_with_entry(index).await;
+            let filename = reader.entry().filename().clone().into_string().str_err()?;
-            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;
+            reader.read_to_end(&mut data).await.str_err()?;
-            if let Err(error) = readed {
-                return Err(format!("read_to_end: {error:?}"));
-            }
             let parsed: Result<SourceQuestionsBatch, _> = serde_json::from_slice(&data);
+            self.index = Some(self.index.unwrap() + 1);
+
             Ok((filename, parsed))
         }
-        pub async fn get_next(
+    }
-            &mut self,
+
-        ) -> Option<Result<(String, Result<SourceQuestionsBatch, serde_json::Error>), String>>
+    impl<R> Stream for SourceQuestionsZipReaderAsync<R>
-        where
+    where
-            R: AsyncRead + AsyncSeek + Unpin,
+        R: AsyncRead + AsyncSeek + Unpin,
-        {
+    {
-            if self.index.is_none() && !self.is_empty() {
+        type Item = (String, Result<SourceQuestionsBatch, serde_json::Error>);
+
+        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);
             }
 
-            if self.index.unwrap() >= self.len() {
+            let index = &mut self.index;
-                return None;
+            if index.unwrap() == len {
+                return Poll::Ready(None);
             }
 
-            let item = self.get(self.index.unwrap()).await;
+            let future = self.parse_zip_entry();
-            self.index = Some(self.index.unwrap() + 1);
+            pin_mut!(future);
-
+            match Pin::new(&mut future).poll(cx) {
-            Some(item)
+                Poll::Ready(Ok(item)) => Poll::Ready(Some(item)),
+                Poll::Ready(Err(_)) => Poll::Ready(None),
+                Poll::Pending => Poll::Pending,
+            }
         }
-        pub fn stream(
+
-            &mut self,
+        fn size_hint(&self) -> (usize, Option<usize>) {
-        ) -> impl Stream<Item = (String, Result<SourceQuestionsBatch, serde_json::Error>)> + '_
+            let len = self.zipfile.file().entries().len();
-        {
+            if self.index.is_none() {
-            stream! {
+                return (len, Some(len));
-                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))
         }
     }