fix(wado-rs): return HTTP 404 for empty streams by peeking

CHANGELOG.md

@@ -11,6 +11,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 - Add `secret-key-env` and `access-key-env` options to load secrets from environment variables.
 
+### Fixed
+
+- Return HTTP 404 "Not Found" for empty DICOM streams
+
 ## [0.2.0] - 2024-06-27
 
 ### Added

src/api/wado/routes.rs

@@ -1,5 +1,5 @@
-use std::error::Error;
 use crate::api::wado::RetrieveInstanceRequest;
+use crate::backend::dimse::wado::DicomMultipartStream;
 use crate::backend::ServiceProvider;
 use crate::types::UI;
 use crate::AppState;
@@ -9,6 +9,8 @@ use axum::http::{Response, StatusCode};
 use axum::response::IntoResponse;
 use axum::routing::get;
 use axum::Router;
+use futures::{StreamExt, TryStreamExt};
+use std::pin::Pin;
 use tracing::{error, instrument};
 
 /// HTTP Router for the Retrieve Transaction
@@ -59,24 +61,38 @@ async fn instance_resource(
 		let response = wado.retrieve(request).await;
 
 		match response {
-			Ok(response) => Response::builder()
-				.header(
-					CONTENT_DISPOSITION,
-					format!(r#"attachment; filename="{study_instance_uid}""#,),
-				)
-				.header(
-					CONTENT_TYPE,
-					r#"multipart/related; type="application/dicom"; boundary=boundary"#,
-				)
-				.body(Body::from_stream(response.stream))
-				.unwrap(),
+			Ok(response) => {
+				let mut stream = response.stream.peekable();
+				let pinned_stream = Pin::new(&mut stream);
+				if pinned_stream.peek().await.is_none() {
+					return StatusCode::NOT_FOUND.into_response();
+				}
+
+				Response::builder()
+					.header(
+						CONTENT_DISPOSITION,
+						format!(r#"attachment; filename="{study_instance_uid}""#,),
+					)
+					.header(
+						CONTENT_TYPE,
+						r#"multipart/related; type="application/dicom"; boundary=boundary"#,
+					)
+					.body(Body::from_stream(DicomMultipartStream::new(
+						stream.into_stream(),
+					)))
+					.unwrap()
+			}
 			Err(err) => {
 				error!("{err:?}");
 				(StatusCode::INTERNAL_SERVER_ERROR, err.to_string()).into_response()
-			},
+			}
 		}
 	} else {
-		(StatusCode::SERVICE_UNAVAILABLE, "WADO-RS endpoint is disabled").into_response()
+		(
+			StatusCode::SERVICE_UNAVAILABLE,
+			"WADO-RS endpoint is disabled",
+		)
+			.into_response()
 	}
 }
 

src/api/wado/service.rs

@@ -1,18 +1,20 @@
+use crate::backend::dimse::cmove::movescu::MoveError;
+use crate::types::{AE, UI};
+use crate::AppState;
 use async_trait::async_trait;
 use axum::extract::rejection::{PathRejection, QueryRejection};
 use axum::extract::{FromRef, FromRequestParts, Path, Query};
 use axum::http::request::Parts;
 use axum::response::{IntoResponse, Response};
+use dicom::object::{FileDicomObject, InMemDicomObject};
+use futures::stream::BoxStream;
 use serde::{Deserialize, Serialize};
 use std::fmt::Debug;
 use std::num::ParseIntError;
 use std::str::FromStr;
+use std::sync::Arc;
 use thiserror::Error;
 
-use crate::backend::dimse::wado::DicomMultipartStream;
-use crate::types::{AE, UI};
-use crate::AppState;
-
 #[async_trait]
 pub trait WadoService: Send + Sync {
 	async fn retrieve(
@@ -63,7 +65,7 @@ where
 }
 
 pub struct InstanceResponse {
-	pub stream: DicomMultipartStream<'static>,
+	pub stream: BoxStream<'static, Result<Arc<FileDicomObject<InMemDicomObject>>, MoveError>>,
 }
 
 #[derive(Debug, Deserialize)]

src/backend/dimse/wado.rs

@@ -61,7 +61,9 @@ impl WadoService for DimseWadoService {
 			)
 			.await;
 
-		Ok(InstanceResponse { stream })
+		Ok(InstanceResponse {
+			stream: stream.boxed(),
+		})
 	}
 }
 
@@ -122,7 +124,7 @@ impl DimseWadoService {
 		aet: &str,
 		storescp_aet: &str,
 		identifier: InMemDicomObject,
-	) -> DicomMultipartStream<'static> {
+	) -> BoxStream<'static, Result<Arc<FileDicomObject<InMemDicomObject>>, MoveError>> {
 		let message_id = next_message_id();
 		let (tx, mut rx) = mpsc::channel::<Result<MoveSubOperation, MoveError>>(1);
 
@@ -174,7 +176,7 @@ impl DimseWadoService {
 			}
 		};
 
-		DicomMultipartStream::new(DropStream::new(rx_stream, subscription))
+		DropStream::new(rx_stream, subscription).boxed()
 	}
 }
 

src/backend/s3/wado.rs

@@ -1,6 +1,5 @@
 use crate::api::wado::{InstanceResponse, RetrieveError, RetrieveInstanceRequest, WadoService};
 use crate::backend::dimse::cmove::movescu::MoveError;
-use crate::backend::dimse::wado::DicomMultipartStream;
 use crate::config::{S3Config, S3EndpointStyle};
 use async_trait::async_trait;
 use aws_config::retry::RetryConfig;
@@ -99,6 +98,7 @@ impl WadoService for S3WadoService {
 						.await
 						.unwrap();
 
+					// TODO: Do not unwrap
 					let bytes = object.body.collect().await.unwrap().reader();
 					Result::<_, MoveError>::Ok(Arc::new(
 						FileDicomObject::from_reader(bytes).unwrap(),
@@ -112,7 +112,7 @@ impl WadoService for S3WadoService {
 			});
 
 		Ok(InstanceResponse {
-			stream: DicomMultipartStream::new(stream),
+			stream: stream.boxed(),
 		})
 	}
 }