Change rendering to use a pipeline approach

src/pipeline.rs

@@ -34,12 +34,6 @@ impl<S1: ProcessingStep, S2: ProcessingStep<Input = S1::Output>> ProcessingStep
     }
 }
 
-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>;
@@ -58,6 +52,14 @@ pub trait ProcessingStepChain: Sized + ProcessingStep {
 
 impl<S: ProcessingStep> ProcessingStepChain for S {}
 
+pub trait ProcessingStepParallel: Sized + ProcessingStep {
+    fn parallel(self) -> ParallelPipeline<Self> {
+        ParallelPipeline(self)
+    }
+}
+
+impl<S: ProcessingStep> ProcessingStepParallel for S {}
+
 pub trait IntoPipeline: Sized + ProcessingStep + 'static {
     fn into_pipeline(self) -> ProcessingPipeline<Self::Input, Self::Output> {
         ProcessingPipeline(Box::new(self))

src/rendering/mod.rs

@@ -1,6 +1,6 @@
 use std::{path::PathBuf, sync::Arc};
 
-use futures::future;
+use async_trait::async_trait;
 use miette::{IntoDiagnostic, Result};
 use tera::{Context as TeraContext, Tera};
 use tokio::{fs, sync::Mutex};
@@ -8,6 +8,7 @@ use tokio::{fs, sync::Mutex};
 use crate::{
     context::Context,
     data::{load_page, FolderData},
+    pipeline::{ProcessingStep, ProcessingStepChain, ProcessingStepParallel},
 };
 
 use self::style::{load_stylesheets, Stylesheets};
@@ -21,66 +22,49 @@ pub struct ContentRenderer {
     styles: Arc<Mutex<Stylesheets>>,
 }
 
-impl ContentRenderer {
-    pub async fn new(ctx: Arc<Context>) -> Result<Self> {
-        let template_glob = format!("{}/**/*", ctx.dirs.template_dir.to_string_lossy());
-        let styles = load_stylesheets(&ctx.dirs.stylesheet_dir).await?;
-
-        Ok(Self {
-            template_glob,
-            ctx,
-            styles: Arc::new(Mutex::new(styles)),
-        })
-    }
-
-    #[tracing::instrument(level = "trace", skip_all)]
-    pub async fn render_all(&self, dirs: Vec<FolderData>) -> Result<()> {
-        if self.ctx.dirs.output_dir.exists() {
-            fs::remove_dir_all(&self.ctx.dirs.output_dir)
-                .await
-                .into_diagnostic()?;
-        }
-        let mut tera = Tera::new(&self.template_glob).into_diagnostic()?;
-        super::processors::register_all(&mut tera);
-        future::try_join_all(dirs.into_iter().map(|data| self.render_folder(&tera, data))).await?;
+pub struct LoadDir;
 
-        Ok(())
-    }
+#[async_trait]
+impl ProcessingStep for LoadDir {
+    type Input = FolderData;
+    type Output = Vec<(PathBuf, String)>;
 
-    #[tracing::instrument(level = "trace", skip_all)]
-    async fn render_folder(&self, tera: &Tera, data: FolderData) -> Result<()> {
-        let dir_name = data
+    #[tracing::instrument(name = "load dir", level = "trace", skip_all)]
+    async fn process(&self, input: Self::Input) -> Result<Self::Output> {
+        let dir_name = input
             .path
             .components()
             .last()
             .unwrap()
             .as_os_str()
             .to_string_lossy();
-        let default_template = data
+        let default_template = input
             .index
             .default_template
             .to_owned()
             .unwrap_or(dir_name.into());
 
-        future::try_join_all(
-            data.pages
-                .into_iter()
-                .map(|page| self.render_page(tera, default_template.clone(), page)),
-        )
-        .await?;
-
-        Ok(())
+        Ok(input
+            .pages
+            .into_iter()
+            .map(|p| (p, default_template.clone()))
+            .collect())
     }
+}
 
-    #[tracing::instrument(level = "trace", skip_all)]
-    async fn render_page(
-        &self,
-        tera: &Tera,
-        default_template: String,
-        page_path: PathBuf,
-    ) -> Result<()> {
-        tracing::debug!("Rendering {page_path:?}");
+struct RenderPage {
+    tera: Tera,
+    styles: Arc<Mutex<Stylesheets>>,
+    ctx: Arc<Context>,
+}
+
+#[async_trait]
+impl ProcessingStep for RenderPage {
+    type Input = (PathBuf, String);
+    type Output = (PathBuf, String);
 
+    #[tracing::instrument(name = "render page", level = "trace", skip_all)]
+    async fn process(&self, (page_path, default_template): Self::Input) -> Result<Self::Output> {
         let page = load_page(&page_path).await?;
         let mut context = TeraContext::new();
         let mut template_name = default_template;
@@ -106,20 +90,81 @@ impl ContentRenderer {
 
         tracing::debug!("context = {context:?}");
 
-        let html = tera
+        let html = self
+            .tera
             .render(&format!("{template_name}.html"), &context)
             .into_diagnostic()?;
         let rel_path = page_path
             .strip_prefix(&self.ctx.dirs.content_dir)
             .into_diagnostic()?;
-        let mut out_path = self.ctx.dirs.output_dir.join(rel_path);
-        out_path.set_extension("html");
+
+        Ok((rel_path.to_owned(), html))
+    }
+}
+
+pub struct SaveOutput {
+    out_dir: PathBuf,
+    extension: &'static str,
+}
+
+#[async_trait]
+impl ProcessingStep for SaveOutput {
+    type Input = (PathBuf, String);
+    type Output = ();
+
+    #[tracing::instrument(name = "save output", level = "trace", skip_all)]
+    async fn process(&self, (rel_path, content): Self::Input) -> Result<Self::Output> {
+        let mut out_path = self.out_dir.join(rel_path);
+        out_path.set_extension(self.extension);
         let parent = out_path.parent().unwrap();
 
         if !parent.exists() {
             fs::create_dir_all(parent).await.into_diagnostic()?;
         }
-        fs::write(out_path, html).await.into_diagnostic()?;
+        fs::write(out_path, content).await.into_diagnostic()?;
+
+        Ok(())
+    }
+}
+
+impl ContentRenderer {
+    pub async fn new(ctx: Arc<Context>) -> Result<Self> {
+        let template_glob = format!("{}/**/*", ctx.dirs.template_dir.to_string_lossy());
+        let styles = load_stylesheets(&ctx.dirs.stylesheet_dir).await?;
+
+        Ok(Self {
+            template_glob,
+            ctx,
+            styles: Arc::new(Mutex::new(styles)),
+        })
+    }
+
+    #[tracing::instrument(level = "trace", skip_all)]
+    pub async fn render_all(&self, dirs: Vec<FolderData>) -> Result<()> {
+        if self.ctx.dirs.output_dir.exists() {
+            fs::remove_dir_all(&self.ctx.dirs.output_dir)
+                .await
+                .into_diagnostic()?;
+        }
+        let mut tera = Tera::new(&self.template_glob).into_diagnostic()?;
+        super::processors::register_all(&mut tera);
+
+        LoadDir
+            .chain(
+                RenderPage {
+                    tera,
+                    styles: self.styles.clone(),
+                    ctx: self.ctx.clone(),
+                }
+                .chain(SaveOutput {
+                    out_dir: self.ctx.dirs.output_dir.to_owned(),
+                    extension: "html",
+                })
+                .parallel(),
+            )
+            .parallel()
+            .process(dirs)
+            .await?;
 
         Ok(())
     }