Avoid writing blurhashed images to tmp

src/blurhash.rs

@@ -1,6 +1,8 @@
-use std::ffi::OsStr;
+use std::ffi::{OsStr, OsString};
 
+use futures_core::Stream;
 use tokio::io::AsyncReadExt;
+use tokio_util::bytes::Bytes;
 
 use crate::{
     details::Details,
@@ -47,13 +49,7 @@ where
             .internal_format()
             .processable_format()
             .expect("not a video"),
-        |mut tmp_file| async move {
+        stream,
-            tmp_file
-                .write_from_stream(stream)
-                .await
-                .map_err(MagickError::Write)?;
-            Ok(tmp_file)
-        },
     )
     .await?;
 
@@ -83,37 +79,19 @@ where
     Ok(blurhash)
 }
 
-async fn read_rgba<S, F, Fut>(
+async fn read_rgba<S>(
     state: &State<S>,
     input_format: ProcessableFormat,
-    write_file: F,
+    stream: impl Stream<Item = std::io::Result<Bytes>> + 'static,
-) -> Result<ProcessRead, MagickError>
+) -> Result<ProcessRead, MagickError> {
-where
-    F: FnOnce(crate::file::File) -> Fut,
-    Fut: std::future::Future<Output = Result<crate::file::File, MagickError>>,
-{
     let temporary_path = state
         .tmp_dir
         .tmp_folder()
         .await
         .map_err(MagickError::CreateTemporaryDirectory)?;
 
-    let input_file = state.tmp_dir.tmp_file(None);
+    let mut input_arg = OsString::from(input_format.magick_format());
-    crate::store::file_store::safe_create_parent(&input_file)
+    input_arg.push(":-");
-        .await
-        .map_err(MagickError::CreateDir)?;
-
-    let tmp_one = crate::file::File::create(&input_file)
-        .await
-        .map_err(MagickError::CreateFile)?;
-    let tmp_one = (write_file)(tmp_one).await?;
-    tmp_one.close().await.map_err(MagickError::CloseFile)?;
-
-    let mut input_arg = [
-        input_format.magick_format().as_ref(),
-        input_file.as_os_str(),
-    ]
-    .join(":".as_ref());
     if input_format.coalesce() {
         input_arg.push("[0]");
     }
@@ -126,8 +104,7 @@ where
     ];
 
     let process = Process::run("magick", &args, &envs, state.config.media.process_timeout)?
-        .read()
+        .stream_read(stream)
-        .add_extras(input_file)
         .add_extras(temporary_path);
 
     Ok(process)

src/process.rs

@@ -6,11 +6,13 @@ use std::{
     time::{Duration, Instant},
 };
 
+use futures_core::Stream;
+use streem::IntoStreamer;
 use tokio::{
-    io::AsyncReadExt,
+    io::{AsyncReadExt, AsyncWriteExt},
     process::{Child, ChildStdin, Command},
 };
-use tokio_util::io::ReaderStream;
+use tokio_util::{bytes::Bytes, io::ReaderStream};
 use tracing::Instrument;
 use uuid::Uuid;
 
@@ -249,6 +251,26 @@ impl Process {
         })
     }
 
+    pub(crate) fn stream_read<S>(self, input: S) -> ProcessRead
+    where
+        S: Stream<Item = std::io::Result<Bytes>> + 'static,
+    {
+        self.spawn_fn(move |mut stdin| async move {
+            let stream = std::pin::pin!(input);
+            let mut stream = stream.into_streamer();
+
+            while let Some(mut bytes) = stream.try_next().await? {
+                match stdin.write_all_buf(&mut bytes).await {
+                    Ok(()) => {}
+                    Err(e) if e.kind() == std::io::ErrorKind::BrokenPipe => break,
+                    Err(e) => return Err(e),
+                }
+            }
+
+            Ok(())
+        })
+    }
+
     pub(crate) fn read(self) -> ProcessRead {
         self.spawn_fn(|_| async { Ok(()) })
     }