use bytes::Bytes;
use tracing::{debug, trace, warn};
use super::{
FilterPipeline,
http_utils::{
BodyFilterOutcome, accumulate_body_bytes, as_request_body_filter, as_response_body_filter,
dispatch_body_result, released_or_continue, run_response_filter, skip_by_response_conditions,
},
};
use crate::{
FilterError, actions::FilterAction, any_filter::AnyFilter, condition::should_execute, context::HttpFilterContext,
};
impl FilterPipeline {
pub async fn execute_http_request(&self, ctx: &mut HttpFilterContext<'_>) -> Result<FilterAction, FilterError> {
for (filter, conditions, _resp_conditions) in &self.filters {
let http_filter = match filter {
AnyFilter::Http(f) => f.as_ref(),
AnyFilter::Tcp(_) => continue,
};
if !should_execute(conditions, ctx.request) {
trace!(filter = http_filter.name(), "skipped by conditions");
continue;
}
trace!(filter = http_filter.name(), "on_request");
match http_filter.on_request(ctx).await {
Ok(FilterAction::Continue | FilterAction::Release) => {},
Ok(FilterAction::Reject(rejection)) => {
debug!(
filter = http_filter.name(),
status = rejection.status,
"filter rejected request"
);
return Ok(FilterAction::Reject(rejection));
},
Err(e) => {
warn!(filter = http_filter.name(), error = %e, "filter error during request");
return Err(e);
},
}
}
Ok(FilterAction::Continue)
}
pub async fn execute_http_response(&self, ctx: &mut HttpFilterContext<'_>) -> Result<FilterAction, FilterError> {
for (filter, _req_conditions, resp_conditions) in self.filters.iter().rev() {
let http_filter = match filter {
AnyFilter::Http(f) => f.as_ref(),
AnyFilter::Tcp(_) => continue,
};
if skip_by_response_conditions(http_filter, resp_conditions, ctx) {
continue;
}
trace!(filter = http_filter.name(), "on_response");
let action = run_response_filter(http_filter, ctx).await?;
if let Some(rejection) = action {
return Ok(FilterAction::Reject(rejection));
}
}
Ok(FilterAction::Continue)
}
pub async fn execute_http_request_body(
&self,
ctx: &mut HttpFilterContext<'_>,
body: &mut Option<Bytes>,
end_of_stream: bool,
) -> Result<FilterAction, FilterError> {
accumulate_body_bytes(&mut ctx.request_body_bytes, body);
let mut released = false;
for (filter, conditions, _resp_conditions) in &self.filters {
let Some(http_filter) = as_request_body_filter(filter, conditions, ctx.request) else {
continue;
};
trace!(filter = http_filter.name(), "on_request_body");
match dispatch_body_result(
http_filter.on_request_body(ctx, body, end_of_stream).await,
http_filter.name(),
"request body",
)? {
BodyFilterOutcome::Continue => {},
BodyFilterOutcome::Released => released = true,
BodyFilterOutcome::Rejected(r) => return Ok(FilterAction::Reject(r)),
}
}
Ok(released_or_continue(released))
}
pub fn execute_http_response_body(
&self,
ctx: &mut HttpFilterContext<'_>,
body: &mut Option<Bytes>,
end_of_stream: bool,
) -> Result<FilterAction, FilterError> {
accumulate_body_bytes(&mut ctx.response_body_bytes, body);
let mut released = false;
for (filter, _req_conditions, resp_conditions) in self.filters.iter().rev() {
let Some(http_filter) = as_response_body_filter(filter, resp_conditions, ctx) else {
continue;
};
trace!(filter = http_filter.name(), "on_response_body");
match dispatch_body_result(
http_filter.on_response_body(ctx, body, end_of_stream),
http_filter.name(),
"response body",
)? {
BodyFilterOutcome::Continue => {},
BodyFilterOutcome::Released => released = true,
BodyFilterOutcome::Rejected(r) => return Ok(FilterAction::Reject(r)),
}
}
Ok(released_or_continue(released))
}
}
#[cfg(test)]
mod tests {
use bytes::Bytes;
use praxis_core::config::{ResponseCondition, ResponseConditionMatch};
use super::super::http_utils::{
accumulate_body_bytes, dispatch_body_result, released_or_continue, skip_by_response_conditions,
};
use crate::{FilterAction, FilterError, actions::Rejection};
#[test]
fn accumulate_body_bytes_increments_with_some() {
let mut counter = 0u64;
let body = Some(Bytes::from_static(b"hello"));
accumulate_body_bytes(&mut counter, &body);
assert_eq!(counter, 5, "counter should equal body length");
}
#[test]
fn accumulate_body_bytes_multiple_chunks() {
let mut counter = 0u64;
accumulate_body_bytes(&mut counter, &Some(Bytes::from_static(b"abc")));
accumulate_body_bytes(&mut counter, &Some(Bytes::from_static(b"de")));
assert_eq!(counter, 5, "counter should accumulate across calls");
}
#[test]
fn accumulate_body_bytes_noop_with_none() {
let mut counter = 10u64;
accumulate_body_bytes(&mut counter, &None);
assert_eq!(counter, 10, "counter should be unchanged when body is None");
}
#[test]
fn accumulate_body_bytes_noop_with_empty() {
let mut counter = 0u64;
accumulate_body_bytes(&mut counter, &Some(Bytes::new()));
assert_eq!(counter, 0, "counter should be unchanged when body is empty");
}
#[test]
fn released_or_continue_true_returns_release() {
assert!(
matches!(released_or_continue(true), FilterAction::Release),
"true should yield Release"
);
}
#[test]
fn released_or_continue_false_returns_continue() {
assert!(
matches!(released_or_continue(false), FilterAction::Continue),
"false should yield Continue"
);
}
#[test]
fn skip_by_response_conditions_empty_conditions() {
let filter = crate::builtins::StaticResponseFilter::from_config(
&serde_yaml::from_str::<serde_yaml::Value>("status: 200").unwrap(),
)
.unwrap();
let req = crate::test_utils::make_request(http::Method::GET, "/");
let mut resp = crate::test_utils::make_response();
let mut ctx = crate::test_utils::make_filter_context(&req);
ctx.response_header = Some(&mut resp);
assert!(
!skip_by_response_conditions(filter.as_ref(), &[], &ctx),
"empty conditions should not skip"
);
}
#[test]
fn skip_by_response_conditions_matching_when_does_not_skip() {
let filter = crate::builtins::StaticResponseFilter::from_config(
&serde_yaml::from_str::<serde_yaml::Value>("status: 200").unwrap(),
)
.unwrap();
let conds = vec![ResponseCondition::When(ResponseConditionMatch {
status: Some(vec![200]),
headers: None,
})];
let req = crate::test_utils::make_request(http::Method::GET, "/");
let mut resp = crate::test_utils::make_response();
let mut ctx = crate::test_utils::make_filter_context(&req);
ctx.response_header = Some(&mut resp);
assert!(
!skip_by_response_conditions(filter.as_ref(), &conds, &ctx),
"matching 'when' condition should not skip"
);
}
#[test]
fn skip_by_response_conditions_non_matching_when_skips() {
let filter = crate::builtins::StaticResponseFilter::from_config(
&serde_yaml::from_str::<serde_yaml::Value>("status: 200").unwrap(),
)
.unwrap();
let conds = vec![ResponseCondition::When(ResponseConditionMatch {
status: Some(vec![404]),
headers: None,
})];
let req = crate::test_utils::make_request(http::Method::GET, "/");
let mut resp = crate::test_utils::make_response();
let mut ctx = crate::test_utils::make_filter_context(&req);
ctx.response_header = Some(&mut resp);
assert!(
skip_by_response_conditions(filter.as_ref(), &conds, &ctx),
"non-matching 'when' condition should skip"
);
}
#[test]
fn skip_by_response_conditions_no_response_header_does_not_skip() {
let filter = crate::builtins::StaticResponseFilter::from_config(
&serde_yaml::from_str::<serde_yaml::Value>("status: 200").unwrap(),
)
.unwrap();
let conds = vec![ResponseCondition::When(ResponseConditionMatch {
status: Some(vec![200]),
headers: None,
})];
let req = crate::test_utils::make_request(http::Method::GET, "/");
let ctx = crate::test_utils::make_filter_context(&req);
assert!(
!skip_by_response_conditions(filter.as_ref(), &conds, &ctx),
"no response header should not skip"
);
}
#[test]
fn dispatch_body_result_continue() {
let outcome = dispatch_body_result(Ok(FilterAction::Continue), "test", "request body").unwrap();
assert!(
matches!(outcome, super::super::http_utils::BodyFilterOutcome::Continue),
"Continue action should produce Continue outcome"
);
}
#[test]
fn dispatch_body_result_release() {
let outcome = dispatch_body_result(Ok(FilterAction::Release), "test", "request body").unwrap();
assert!(
matches!(outcome, super::super::http_utils::BodyFilterOutcome::Released),
"Release action should produce Released outcome"
);
}
#[test]
fn dispatch_body_result_reject() {
let outcome =
dispatch_body_result(Ok(FilterAction::Reject(Rejection::status(403))), "test", "request body").unwrap();
assert!(
matches!(outcome, super::super::http_utils::BodyFilterOutcome::Rejected(r) if r.status == 403),
"Reject action should produce Rejected outcome with correct status"
);
}
#[test]
fn dispatch_body_result_error() {
let err: FilterError = "boom".into();
let result = dispatch_body_result(Err(err), "test", "request body");
assert!(result.is_err(), "error result should propagate as Err");
assert!(
result.unwrap_err().to_string().contains("boom"),
"error message should be preserved"
);
}
}