use bytes::Bytes;
use camel_api::{
Body, CamelError, Exchange, StreamMetadata, StreamSplitConfig, StreamSplitFormat,
fragment_exchange,
};
use futures::Stream;
use std::pin::Pin;
pub mod chunks;
pub mod lines;
pub mod ndjson;
pub const CAMEL_STREAM_ORIGIN: &str = "CamelStreamOrigin";
pub const CAMEL_STREAM_SOURCE_CONTENT_TYPE: &str = "CamelStreamSourceContentType";
pub const CAMEL_STREAM_OFFSET: &str = "CamelStreamOffset";
pub const CAMEL_STREAM_BATCH_SIZE: &str = "CamelStreamBatchSize";
pub struct StreamSplitInput {
pub parent: Exchange,
pub stream: Pin<Box<dyn Stream<Item = Result<Bytes, CamelError>> + Send>>,
pub metadata: StreamMetadata,
}
pub trait StreamSplitCodec: Send + Sync {
fn split(
&self,
input: StreamSplitInput,
config: StreamSplitConfig,
) -> Pin<Box<dyn Stream<Item = Result<Exchange, CamelError>> + Send>>;
}
pub fn resolve_format(
format: &StreamSplitFormat,
metadata: &StreamMetadata,
) -> Result<StreamSplitFormat, CamelError> {
match format {
StreamSplitFormat::Auto => {
let ct = metadata
.content_type
.as_deref()
.unwrap_or("")
.to_lowercase();
let ct = ct.split(';').next().unwrap_or("").trim();
match ct {
"application/x-ndjson" => Ok(StreamSplitFormat::Ndjson),
"text/plain" => Ok(StreamSplitFormat::Lines),
"application/octet-stream" => Ok(StreamSplitFormat::Chunks),
"application/zip" | "application/x-zip-compressed" => Err(CamelError::Config(
"stream split format=Auto: ZIP archives require explicit stream.format: zip"
.into(),
)),
"application/x-tar" => Err(CamelError::Config(
"stream split format=Auto: TAR archives require explicit stream.format: tar"
.into(),
)),
"application/gzip" | "application/x-gzip" => Err(CamelError::Config(
"stream split format=Auto: GZIP archives require explicit stream.format: tar.gz"
.into(),
)),
"" => Err(CamelError::Config(
"stream split format=Auto but stream has no content_type".into(),
)),
other => Err(CamelError::Config(format!(
"stream split format=Auto: unknown content type '{}'",
other
))),
}
}
other => Ok(other.clone()),
}
}
#[derive(Debug)]
pub enum ArchiveSplitKind {
Zip,
Tar,
TarGz,
}
pub enum ResolvedStreamSplit {
Incremental(Box<dyn StreamSplitCodec>),
MaterializedArchive(ArchiveSplitKind),
}
pub fn resolve_incremental_codec(
format: &StreamSplitFormat,
) -> Result<Box<dyn StreamSplitCodec>, CamelError> {
match format {
StreamSplitFormat::Ndjson => Ok(Box::new(ndjson::NdjsonCodec)),
StreamSplitFormat::Lines => Ok(Box::new(lines::LinesCodec)),
StreamSplitFormat::Chunks => Ok(Box::new(chunks::ChunksCodec)),
StreamSplitFormat::Zip => Err(CamelError::Config(
"Zip is a materialized archive format, not an incremental codec".into(),
)),
StreamSplitFormat::Tar | StreamSplitFormat::TarGz => Err(CamelError::Config(
"Tar and TarGz are materialized archive formats, not incremental codecs".into(),
)),
StreamSplitFormat::Auto => Err(CamelError::Config(
"resolve_incremental_codec requires a resolved format, not Auto".into(),
)),
_ => Err(CamelError::Config("unsupported stream split format".into())),
}
}
pub fn resolve_split(
format: &StreamSplitFormat,
metadata: &StreamMetadata,
) -> Result<ResolvedStreamSplit, CamelError> {
let resolved = resolve_format(format, metadata)?;
match resolved {
StreamSplitFormat::Zip => Ok(ResolvedStreamSplit::MaterializedArchive(
ArchiveSplitKind::Zip,
)),
StreamSplitFormat::Tar => Ok(ResolvedStreamSplit::MaterializedArchive(
ArchiveSplitKind::Tar,
)),
StreamSplitFormat::TarGz => Ok(ResolvedStreamSplit::MaterializedArchive(
ArchiveSplitKind::TarGz,
)),
_ => Ok(ResolvedStreamSplit::Incremental(resolve_incremental_codec(
&resolved,
)?)),
}
}
pub fn fragment_stream_exchange(parent: &Exchange, body: Body) -> Exchange {
let mut ex = fragment_exchange(parent, body);
ex.input.headers.remove("Content-Length");
ex.input.headers.remove("Content-Type");
ex
}
#[cfg(test)]
mod tests {
use super::*;
use camel_api::StreamMetadata;
fn default_metadata() -> StreamMetadata {
StreamMetadata::default()
}
#[test]
fn test_resolve_split_ndjson_is_incremental() {
let result = resolve_split(&StreamSplitFormat::Ndjson, &default_metadata()).unwrap();
assert!(matches!(result, ResolvedStreamSplit::Incremental(_)));
}
#[test]
fn test_resolve_split_zip_is_materialized() {
let result = resolve_split(&StreamSplitFormat::Zip, &default_metadata()).unwrap();
assert!(matches!(
result,
ResolvedStreamSplit::MaterializedArchive(ArchiveSplitKind::Zip)
));
}
#[test]
fn test_resolve_split_tar_is_materialized() {
let result = resolve_split(&StreamSplitFormat::Tar, &default_metadata()).unwrap();
assert!(matches!(
result,
ResolvedStreamSplit::MaterializedArchive(ArchiveSplitKind::Tar)
));
}
#[test]
fn test_resolve_split_tar_gz_is_materialized() {
let result = resolve_split(&StreamSplitFormat::TarGz, &default_metadata()).unwrap();
assert!(matches!(
result,
ResolvedStreamSplit::MaterializedArchive(ArchiveSplitKind::TarGz)
));
}
#[test]
fn test_resolve_incremental_codec_rejects_tar_formats() {
for format in [StreamSplitFormat::Tar, StreamSplitFormat::TarGz] {
let result = resolve_incremental_codec(&format);
let msg = match result {
Err(e) => e.to_string(),
Ok(_) => panic!("expected Err for {format:?}"),
};
assert!(
msg.contains("materialized archive format"),
"expected materialized-archive rejection, got: {msg}"
);
}
}
#[test]
fn test_auto_tar_content_types_suggest_explicit_format() {
for (ct, expected) in [
("application/x-tar", "format: tar"),
("application/gzip", "format: tar.gz"),
] {
let meta = StreamMetadata {
content_type: Some(ct.to_string()),
..Default::default()
};
let result = resolve_split(&StreamSplitFormat::Auto, &meta);
let msg = match result {
Err(e) => e.to_string(),
Ok(_) => panic!("expected Err for Auto + {ct}"),
};
assert!(
msg.contains(expected),
"expected suggestion '{expected}' in error, got: {msg}"
);
}
}
#[test]
fn test_resolve_incremental_codec_returns_codec_for_ndjson() {
let result = resolve_incremental_codec(&StreamSplitFormat::Ndjson);
assert!(result.is_ok());
}
#[test]
fn test_resolve_incremental_codec_rejects_zip() {
let result = resolve_incremental_codec(&StreamSplitFormat::Zip);
assert!(result.is_err());
}
#[test]
fn test_auto_application_zip_suggests_explicit_format() {
let meta = StreamMetadata {
content_type: Some("application/zip".to_string()),
..Default::default()
};
let result = resolve_split(&StreamSplitFormat::Auto, &meta);
let msg = match result {
Err(e) => e.to_string(),
Ok(_) => panic!("expected Err for Auto + application/zip"),
};
assert!(
msg.contains("format: zip"),
"expected suggestion in error, got: {msg}"
);
}
}