Add parallel processing implementation

Cargo.lock

@@ -154,6 +154,17 @@ version = "4.4.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "ecc7ab41815b3c653ccd2978ec3255c81349336702dfdf62ee6f7069b12a3aae"
 
+[[package]]
+name = "async-trait"
+version = "0.1.68"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "b9ccdd8f2a161be9bd5c023df56f1b2a0bd1d83872ae53b71a84a12c9bf6e842"
+dependencies = [
+ "proc-macro2",
+ "quote",
+ "syn",
+]
+
 [[package]]
 name = "async-walkdir"
 version = "0.2.0"
@@ -2006,6 +2017,7 @@ checksum = "49874b5167b65d7193b8aba1567f5c7d93d001cafc34600cee003eda787e483f"
 name = "viki"
 version = "0.1.0"
 dependencies = [
+ "async-trait",
  "async-walkdir",
  "clap",
  "comrak",

Cargo.toml

@@ -6,6 +6,7 @@ edition = "2021"
 # See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
 
 [dependencies]
+async-trait = "0.1.68"
 async-walkdir = "0.2.0"
 clap = { version = "4.3.8", features = ["derive"] }
 comrak = { version = "0.18.0", features = ["emojis"] }

src/context.rs

@@ -1,6 +1,12 @@
 use std::path::PathBuf;
 
+#[derive(Clone, Debug)]
 pub struct Context {
+    pub dirs: Dirs,
+}
+
+#[derive(Clone, Debug)]
+pub struct Dirs {
     pub content_dir: PathBuf,
     pub template_dir: PathBuf,
     pub stylesheet_dir: PathBuf,

src/main.rs

@@ -3,7 +3,7 @@ use std::{path::Path, sync::Arc};
 use args::BuildArgs;
 use clap::Parser;
 use config::{read_config, Config};
-use context::Context;
+use context::{Context, Dirs};
 use data::DirLoader;
 use miette::Result;
 use rendering::ContentRenderer;
@@ -16,6 +16,7 @@ mod args;
 mod config;
 mod context;
 pub mod data;
+mod pipeline;
 mod processors;
 mod rendering;
 
@@ -36,7 +37,7 @@ async fn build(args: &Args, _build_args: &BuildArgs, cfg: Config) -> Result<()>
     let base_path = &args.directory;
     let ctx = Arc::new(build_context(&base_path, &cfg));
 
-    let dirs = DirLoader::new(ctx.content_dir.to_owned())
+    let dirs = DirLoader::new(ctx.dirs.content_dir.to_owned())
         .read_content()
         .await?;
 
@@ -53,10 +54,12 @@ fn build_context(base_path: &Path, config: &Config) -> Context {
     let stylesheet_dir = base_path.join(folders.stylesheets.unwrap_or("styles".into()));
 
     Context {
-        content_dir,
-        template_dir,
-        stylesheet_dir,
-        output_dir,
+        dirs: Dirs {
+            content_dir,
+            template_dir,
+            stylesheet_dir,
+            output_dir,
+        },
     }
 }
 

src/pipeline.rs

@@ -0,0 +1,77 @@
+use async_trait::async_trait;
+use futures::future;
+use miette::Result;
+
+/// The result of combining two processing steps
+pub struct ProcessingChain<S1: ProcessingStep, S2: ProcessingStep<Input = S1::Output>>(S1, S2);
+
+/// An adapter to execute a step with multiple inputs in parallel
+pub struct ParallelPipeline<S: ProcessingStep>(S);
+
+/// A generic wrapper for processing pipelines
+pub struct ProcessingPipeline<I: Send + Sync, O: Send + Sync>(
+    Box<dyn ProcessingStep<Input = I, Output = O>>,
+);
+
+#[async_trait]
+pub trait ProcessingStep: Send + Sync {
+    type Input: Send + Sync;
+    type Output: Send + Sync;
+
+    async fn process(&self, input: Self::Input) -> Result<Self::Output>;
+}
+
+#[async_trait]
+impl<S1: ProcessingStep, S2: ProcessingStep<Input = S1::Output>> ProcessingStep
+    for ProcessingChain<S1, S2>
+{
+    type Input = S1::Input;
+    type Output = S2::Output;
+
+    async fn process(&self, input: Self::Input) -> Result<Self::Output> {
+        let first = self.0.process(input).await?;
+        self.1.process(first).await
+    }
+}
+
+impl<S: ProcessingStep> ParallelPipeline<S> {
+    pub fn new(step: S) -> Self {
+        Self(step)
+    }
+}
+
+#[async_trait]
+impl<S: ProcessingStep> ProcessingStep for ParallelPipeline<S> {
+    type Input = Vec<S::Input>;
+    type Output = Vec<S::Output>;
+
+    async fn process(&self, input: Self::Input) -> Result<Self::Output> {
+        future::try_join_all(input.into_iter().map(|i| self.0.process(i))).await
+    }
+}
+
+pub trait ProcessingStepChain: Sized + ProcessingStep {
+    fn chain<S: ProcessingStep<Input = Self::Output>>(self, other: S) -> ProcessingChain<Self, S> {
+        ProcessingChain(self, other)
+    }
+}
+
+impl<S: ProcessingStep> ProcessingStepChain for S {}
+
+pub trait IntoPipeline: Sized + ProcessingStep + 'static {
+    fn into_pipeline(self) -> ProcessingPipeline<Self::Input, Self::Output> {
+        ProcessingPipeline(Box::new(self))
+    }
+}
+
+impl<S: ProcessingStep + 'static> IntoPipeline for S {}
+
+#[async_trait]
+impl<I: Send + Sync, O: Send + Sync> ProcessingStep for ProcessingPipeline<I, O> {
+    type Input = I;
+    type Output = O;
+
+    async fn process(&self, input: Self::Input) -> Result<Self::Output> {
+        self.0.process(input).await
+    }
+}

src/rendering/mod.rs

@@ -23,8 +23,8 @@ pub struct ContentRenderer {
 
 impl ContentRenderer {
     pub async fn new(ctx: Arc<Context>) -> Result<Self> {
-        let template_glob = format!("{}/**/*", ctx.template_dir.to_string_lossy());
-        let styles = load_stylesheets(&ctx.stylesheet_dir).await?;
+        let template_glob = format!("{}/**/*", ctx.dirs.template_dir.to_string_lossy());
+        let styles = load_stylesheets(&ctx.dirs.stylesheet_dir).await?;
 
         Ok(Self {
             template_glob,
@@ -35,8 +35,8 @@ impl ContentRenderer {
 
     #[tracing::instrument(level = "trace", skip_all)]
     pub async fn render_all(&self, dirs: Vec<FolderData>) -> Result<()> {
-        if self.ctx.output_dir.exists() {
-            fs::remove_dir_all(&self.ctx.output_dir)
+        if self.ctx.dirs.output_dir.exists() {
+            fs::remove_dir_all(&self.ctx.dirs.output_dir)
                 .await
                 .into_diagnostic()?;
         }
@@ -99,7 +99,7 @@ impl ContentRenderer {
         {
             let mut styles = self.styles.lock().await;
             let style_embed = styles
-                .get_style_embed(&style_name, &self.ctx.output_dir)
+                .get_style_embed(&style_name, &self.ctx.dirs.output_dir)
                 .await?;
             context.insert("style", &style_embed);
         };
@@ -110,9 +110,9 @@ impl ContentRenderer {
             .render(&format!("{template_name}.html"), &context)
             .into_diagnostic()?;
         let rel_path = page_path
-            .strip_prefix(&self.ctx.content_dir)
+            .strip_prefix(&self.ctx.dirs.content_dir)
             .into_diagnostic()?;
-        let mut out_path = self.ctx.output_dir.join(rel_path);
+        let mut out_path = self.ctx.dirs.output_dir.join(rel_path);
         out_path.set_extension("html");
         let parent = out_path.parent().unwrap();