1#![forbid(unsafe_code)]
2
3use blazingly_core::{
4 AppDefinition, BackgroundTask, BackgroundTaskError, BodyStreamError, HttpMethod, HttpUpgrade,
5 InputSource, OperationDescriptor, ResponseHeader, SecuritySchemeDescriptor, StreamingBody,
6};
7use blazingly_executor::{
8 DependencyError, ExecutableApp, ExecutionOutcome, FromInvocation,
9 HttpRequestParts as InvocationRequestParts, InputRejection, InvocationControl, InvocationInput,
10};
11use blazingly_json::{Value, json};
12use blazingly_openapi::{OpenApiAssetResponse, OpenApiConfig, OpenApiService};
13use serde::Serialize;
14use serde::de::DeserializeOwned;
15use std::any::{Any, TypeId};
16use std::borrow::Cow;
17use std::cell::{OnceCell, RefCell};
18use std::collections::{BTreeMap, HashMap};
19use std::fmt;
20use std::future::Future;
21use std::net::{IpAddr, SocketAddr};
22use std::rc::Rc;
23use std::str::Utf8Error;
24
25pub const DEFAULT_MAX_BODY_BYTES: usize = 1024 * 1024;
26
27pub const EMIT_VARIABLE: &str = "BLAZINGLY_EMIT";
30
31#[derive(Clone, Debug, Eq, PartialEq)]
33pub struct Request {
34 method: HttpMethod,
35 target: String,
36 headers: BTreeMap<String, String>,
37 body: Vec<u8>,
38 peer_addr: Option<SocketAddr>,
39 scheme: String,
40}
41
42pub trait HttpRequestView {
47 fn method(&self) -> HttpMethod;
48 fn target(&self) -> &str;
49 fn header_value(&self, name: &str, index: usize) -> Option<&str>;
50 fn body(&self) -> &[u8];
51
52 fn take_body_stream(&self) -> Option<StreamingBody> {
54 None
55 }
56
57 fn peer_addr(&self) -> Option<SocketAddr> {
59 None
60 }
61
62 #[allow(clippy::unnecessary_literal_bound)]
64 fn scheme(&self) -> &str {
65 "http"
66 }
67}
68
69pub struct HttpRequestContext<'request> {
75 request: &'request dyn HttpRequestView,
76 client_ip: Option<IpAddr>,
77 scheme: Cow<'request, str>,
78 host: Option<Cow<'request, str>>,
79 extensions: Vec<(TypeId, Box<dyn Any>)>,
80}
81
82impl<'request> HttpRequestContext<'request> {
83 fn new(request: &'request dyn HttpRequestView) -> Self {
84 Self {
85 request,
86 client_ip: request.peer_addr().map(|address| address.ip()),
87 scheme: Cow::Borrowed(request.scheme()),
88 host: None,
89 extensions: Vec::new(),
90 }
91 }
92
93 #[must_use]
94 pub fn request(&self) -> &dyn HttpRequestView {
95 self.request
96 }
97
98 #[must_use]
99 pub fn client_ip(&self) -> Option<IpAddr> {
100 self.client_ip
101 }
102
103 pub fn set_client_ip(&mut self, client_ip: IpAddr) {
104 self.client_ip = Some(client_ip);
105 }
106
107 #[must_use]
108 pub fn scheme(&self) -> &str {
109 &self.scheme
110 }
111
112 pub fn set_scheme(&mut self, scheme: impl Into<String>) {
113 self.scheme = Cow::Owned(scheme.into());
114 }
115
116 #[must_use]
117 pub fn host(&self) -> Option<&str> {
118 self.host
119 .as_deref()
120 .or_else(|| self.request.header_value("host", 0))
121 }
122
123 pub fn set_host(&mut self, host: impl Into<String>) {
124 self.host = Some(Cow::Owned(host.into()));
125 }
126
127 pub fn insert_extension<T: 'static>(&mut self, value: T) {
128 let type_id = TypeId::of::<T>();
129 if let Some((_, existing)) = self
130 .extensions
131 .iter_mut()
132 .find(|(existing, _)| *existing == type_id)
133 {
134 *existing = Box::new(value);
135 } else {
136 self.extensions.push((type_id, Box::new(value)));
137 }
138 }
139
140 #[must_use]
141 pub fn extension<T: 'static>(&self) -> Option<&T> {
142 self.extension_by_id(TypeId::of::<T>())?.downcast_ref()
143 }
144
145 #[must_use]
148 pub fn connection_info(&self) -> ConnectionInfo {
149 ConnectionInfo {
150 client_ip: self.client_ip,
151 scheme: self.scheme.as_ref().to_owned(),
152 host: self.host().map(str::to_owned),
153 }
154 }
155
156 fn extension_by_id(&self, type_id: TypeId) -> Option<&dyn Any> {
157 self.extensions
158 .iter()
159 .find(|(existing, _)| *existing == type_id)
160 .map(|(_, value)| value.as_ref())
161 }
162}
163
164#[derive(Clone, Debug, Eq, PartialEq)]
170pub struct ConnectionInfo {
171 client_ip: Option<IpAddr>,
172 scheme: String,
173 host: Option<String>,
174}
175
176impl ConnectionInfo {
177 #[must_use]
179 pub const fn client_ip(&self) -> Option<IpAddr> {
180 self.client_ip
181 }
182
183 #[must_use]
185 pub fn scheme(&self) -> &str {
186 &self.scheme
187 }
188
189 #[must_use]
191 pub fn host(&self) -> Option<&str> {
192 self.host.as_deref()
193 }
194
195 fn from_request(request: &(impl HttpRequestView + ?Sized)) -> Self {
196 Self {
197 client_ip: request.peer_addr().map(|address| address.ip()),
198 scheme: request.scheme().to_owned(),
199 host: request.header_value("host", 0).map(str::to_owned),
200 }
201 }
202}
203
204pub trait HttpMiddleware {
210 fn on_request(&self, _context: &mut HttpRequestContext<'_>) -> Option<Response> {
212 None
213 }
214
215 fn on_operation(
217 &self,
218 _context: &mut HttpRequestContext<'_>,
219 _operation: &OperationDescriptor,
220 _security_schemes: &[SecuritySchemeDescriptor],
221 ) -> Option<Response> {
222 None
223 }
224
225 fn on_response(
228 &self,
229 _context: &HttpRequestContext<'_>,
230 _operation: Option<&OperationDescriptor>,
231 _response: &mut Response,
232 ) {
233 }
234
235 fn verifies_security(&self) -> bool {
243 true
244 }
245}
246
247type OperationFilter = Rc<dyn Fn(&str) -> bool>;
248
249#[derive(Clone)]
250enum OperationPredicate {
251 Exact(Box<str>),
252 Prefix(Box<str>),
253 Filter(OperationFilter),
254}
255
256impl OperationPredicate {
257 fn matches(&self, operation_id: &str) -> bool {
258 match self {
259 Self::Exact(expected) => operation_id == expected.as_ref(),
260 Self::Prefix(prefix) => operation_id.starts_with(prefix.as_ref()),
261 Self::Filter(filter) => filter(operation_id),
262 }
263 }
264}
265
266#[derive(Clone, Default)]
280pub struct MiddlewareScope {
281 prefixes: Vec<Box<str>>,
282 operations: Vec<OperationPredicate>,
283}
284
285impl MiddlewareScope {
286 #[must_use]
288 pub const fn all() -> Self {
289 Self {
290 prefixes: Vec::new(),
291 operations: Vec::new(),
292 }
293 }
294
295 #[must_use]
297 pub fn prefix(prefix: &str) -> Self {
298 Self::all().with_prefix(prefix)
299 }
300
301 #[must_use]
303 pub fn operation(operation_id: &str) -> Self {
304 Self::all().with_operation(operation_id)
305 }
306
307 #[must_use]
311 pub fn with_prefix(mut self, prefix: &str) -> Self {
312 self.prefixes.push(normalize_prefix(prefix));
313 self
314 }
315
316 #[must_use]
318 pub fn with_operation(mut self, operation_id: &str) -> Self {
319 self.operations
320 .push(OperationPredicate::Exact(Box::from(operation_id)));
321 self
322 }
323
324 #[must_use]
327 pub fn with_operation_prefix(mut self, prefix: &str) -> Self {
328 self.operations
329 .push(OperationPredicate::Prefix(Box::from(prefix)));
330 self
331 }
332
333 #[must_use]
336 pub fn with_operation_filter<Filter>(mut self, filter: Filter) -> Self
337 where
338 Filter: Fn(&str) -> bool + 'static,
339 {
340 self.operations
341 .push(OperationPredicate::Filter(Rc::new(filter)));
342 self
343 }
344
345 #[must_use]
347 pub fn is_global(&self) -> bool {
348 self.prefixes.is_empty() && self.operations.is_empty()
349 }
350
351 #[must_use]
353 pub fn matches_request(&self, path: &str) -> bool {
354 self.operations.is_empty() && self.matches_path(path)
355 }
356
357 #[must_use]
359 pub fn matches_operation(&self, path: &str, operation_id: &str) -> bool {
360 self.matches_path(path)
361 && (self.operations.is_empty()
362 || self
363 .operations
364 .iter()
365 .any(|predicate| predicate.matches(operation_id)))
366 }
367
368 fn matches_path(&self, path: &str) -> bool {
369 self.prefixes.is_empty()
370 || self
371 .prefixes
372 .iter()
373 .any(|prefix| path_has_prefix(path, prefix))
374 }
375}
376
377impl fmt::Debug for MiddlewareScope {
378 fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
379 formatter
380 .debug_struct("MiddlewareScope")
381 .field("prefixes", &self.prefixes)
382 .field("operations", &self.operations.len())
383 .finish()
384 }
385}
386
387fn normalize_prefix(prefix: &str) -> Box<str> {
388 let trimmed = prefix.trim_end_matches('/');
389 if trimmed.is_empty() {
390 Box::from("")
391 } else if trimmed.starts_with('/') {
392 Box::from(trimmed)
393 } else {
394 Box::from(format!("/{trimmed}"))
395 }
396}
397
398fn path_has_prefix(path: &str, prefix: &str) -> bool {
399 let Some(rest) = path.strip_prefix(prefix) else {
400 return false;
401 };
402 rest.is_empty() || rest.starts_with('/')
403}
404
405struct ScopedMiddleware {
406 scope: MiddlewareScope,
407 layer: Rc<dyn HttpMiddleware>,
408}
409
410impl ScopedMiddleware {
411 fn matches(&self, path: &str, operation: Option<&OperationDescriptor>) -> bool {
412 operation.map_or_else(
413 || self.scope.matches_request(path),
414 |operation| {
415 self.scope
416 .matches_operation(path, operation.contract.id.as_str())
417 },
418 )
419 }
420}
421
422#[derive(Clone, Copy, Debug, Eq, PartialEq)]
424pub enum HttpErrorSource {
425 Routing,
427 Request,
429 Rejection,
431 Domain,
433 Internal,
436}
437
438pub struct HttpError<'dispatch> {
440 source: HttpErrorSource,
441 status: u16,
442 code: &'dispatch str,
443 message: &'dispatch str,
444 method: HttpMethod,
445 path: &'dispatch str,
446 operation: Option<&'dispatch OperationDescriptor>,
447}
448
449impl HttpError<'_> {
450 #[must_use]
452 pub const fn source(&self) -> HttpErrorSource {
453 self.source
454 }
455
456 #[must_use]
458 pub const fn status(&self) -> u16 {
459 self.status
460 }
461
462 #[must_use]
464 pub const fn code(&self) -> &str {
465 self.code
466 }
467
468 #[must_use]
470 pub const fn message(&self) -> &str {
471 self.message
472 }
473
474 #[must_use]
476 pub const fn method(&self) -> HttpMethod {
477 self.method
478 }
479
480 #[must_use]
482 pub const fn path(&self) -> &str {
483 self.path
484 }
485
486 #[must_use]
488 pub const fn operation(&self) -> Option<&OperationDescriptor> {
489 self.operation
490 }
491}
492
493pub trait HttpErrorHandler {
509 fn on_error(&self, error: &HttpError<'_>, response: &mut Response);
511}
512
513#[derive(Clone, Debug, Default)]
525pub struct BackgroundTasks {
526 tasks: Rc<RefCell<Vec<BackgroundTask>>>,
527}
528
529impl BackgroundTasks {
530 #[must_use]
531 pub fn new() -> Self {
532 Self::default()
533 }
534
535 pub fn add_task(&self, task: BackgroundTask) {
537 self.tasks.borrow_mut().push(task);
538 }
539
540 pub fn add<Task, TaskFuture>(&self, task: Task)
542 where
543 Task: FnOnce() -> TaskFuture + 'static,
544 TaskFuture: Future<Output = Result<(), BackgroundTaskError>> + 'static,
545 {
546 self.add_task(BackgroundTask::new(task));
547 }
548
549 pub fn add_infallible<Task, TaskFuture>(&self, task: Task)
551 where
552 Task: FnOnce() -> TaskFuture + 'static,
553 TaskFuture: Future<Output = ()> + 'static,
554 {
555 self.add_task(BackgroundTask::infallible(task));
556 }
557
558 #[must_use]
560 pub fn len(&self) -> usize {
561 self.tasks.borrow().len()
562 }
563
564 #[must_use]
566 pub fn is_empty(&self) -> bool {
567 self.tasks.borrow().is_empty()
568 }
569
570 #[must_use]
572 pub fn take(&self) -> Vec<BackgroundTask> {
573 std::mem::take(&mut self.tasks.borrow_mut())
574 }
575}
576
577impl FromInvocation for BackgroundTasks {
578 fn from_invocation(
579 input: &InvocationInput<'_>,
580 name: &str,
581 _required: bool,
582 ) -> Result<Self, InputRejection> {
583 let InvocationInput::Http(request) = input else {
584 return Err(InputRejection::new(
585 500,
586 "background_tasks_transport_mismatch",
587 "after-response tasks are available only through HTTP",
588 ));
589 };
590 request
591 .extension(TypeId::of::<Self>())
592 .and_then(<dyn Any>::downcast_ref::<Self>)
593 .cloned()
594 .ok_or_else(|| {
595 InputRejection::new(
596 500,
597 "background_tasks_unavailable",
598 format!("this transport installed no after-response tasks for `{name}`"),
599 )
600 })
601 }
602}
603
604impl Request {
605 #[must_use]
606 pub fn new(method: HttpMethod, target: impl Into<String>) -> Self {
607 Self {
608 method,
609 target: target.into(),
610 headers: BTreeMap::new(),
611 body: Vec::new(),
612 peer_addr: None,
613 scheme: "http".to_owned(),
614 }
615 }
616
617 #[must_use]
618 pub fn get(target: impl Into<String>) -> Self {
619 Self::new(HttpMethod::Get, target)
620 }
621
622 #[must_use]
623 pub fn head(target: impl Into<String>) -> Self {
624 Self::new(HttpMethod::Head, target)
625 }
626
627 #[must_use]
628 pub fn post(target: impl Into<String>) -> Self {
629 Self::new(HttpMethod::Post, target)
630 }
631
632 #[must_use]
633 pub fn put(target: impl Into<String>) -> Self {
634 Self::new(HttpMethod::Put, target)
635 }
636
637 #[must_use]
638 pub fn patch(target: impl Into<String>) -> Self {
639 Self::new(HttpMethod::Patch, target)
640 }
641
642 #[must_use]
643 pub fn delete(target: impl Into<String>) -> Self {
644 Self::new(HttpMethod::Delete, target)
645 }
646
647 #[must_use]
648 pub fn options(target: impl Into<String>) -> Self {
649 Self::new(HttpMethod::Options, target)
650 }
651
652 #[must_use]
653 pub fn trace(target: impl Into<String>) -> Self {
654 Self::new(HttpMethod::Trace, target)
655 }
656
657 #[must_use]
658 pub fn connect(target: impl Into<String>) -> Self {
659 Self::new(HttpMethod::Connect, target)
660 }
661
662 #[must_use]
663 pub const fn method(&self) -> HttpMethod {
664 self.method
665 }
666
667 #[must_use]
668 pub fn target(&self) -> &str {
669 &self.target
670 }
671
672 #[must_use]
673 pub fn path(&self) -> &str {
674 self.target
675 .split_once('?')
676 .map_or(self.target.as_str(), |(path, _)| path)
677 }
678
679 #[must_use]
680 pub fn header(mut self, name: impl AsRef<str>, value: impl Into<String>) -> Self {
681 self.headers
682 .insert(normalize_header_name(name.as_ref()), value.into());
683 self
684 }
685
686 #[must_use]
687 pub fn body(mut self, body: impl Into<Vec<u8>>) -> Self {
688 self.body = body.into();
689 self
690 }
691
692 #[must_use]
694 pub const fn peer_addr(mut self, peer_addr: SocketAddr) -> Self {
695 self.peer_addr = Some(peer_addr);
696 self
697 }
698
699 #[must_use]
701 pub fn scheme(mut self, scheme: impl Into<String>) -> Self {
702 self.scheme = scheme.into();
703 self
704 }
705
706 pub fn json(mut self, value: &impl Serialize) -> Result<Self, blazingly_json::Error> {
712 self.body = blazingly_json::to_vec(value)?;
713 self.headers
714 .insert("content-type".to_owned(), "application/json".to_owned());
715 Ok(self)
716 }
717
718 #[must_use]
719 pub fn get_header(&self, name: &str) -> Option<&str> {
720 self.headers
721 .get(name)
722 .or_else(|| {
723 self.headers
724 .iter()
725 .find(|(header, _)| header.eq_ignore_ascii_case(name))
726 .map(|(_, value)| value)
727 })
728 .map(String::as_str)
729 }
730
731 #[must_use]
732 pub fn headers(&self) -> &BTreeMap<String, String> {
733 &self.headers
734 }
735
736 #[must_use]
737 pub fn body_bytes(&self) -> &[u8] {
738 &self.body
739 }
740}
741
742impl HttpRequestView for Request {
743 fn method(&self) -> HttpMethod {
744 self.method()
745 }
746
747 fn target(&self) -> &str {
748 self.target()
749 }
750
751 fn header_value(&self, name: &str, index: usize) -> Option<&str> {
752 self.headers
753 .iter()
754 .filter(|(header, _)| header_name_matches(header, name))
755 .nth(index)
756 .map(|(_, value)| value.as_str())
757 }
758
759 fn body(&self) -> &[u8] {
760 self.body_bytes()
761 }
762
763 fn peer_addr(&self) -> Option<SocketAddr> {
764 self.peer_addr
765 }
766
767 fn scheme(&self) -> &str {
768 &self.scheme
769 }
770}
771
772#[derive(Debug)]
774pub struct Response {
775 status: u16,
776 headers: ResponseHeaders,
777 body: Vec<u8>,
778 stream: Option<StreamingBody>,
779 upgrade: Option<HttpUpgrade>,
780 background: Vec<BackgroundTask>,
781}
782
783impl Response {
784 #[must_use]
786 pub fn from_bytes(status: u16, body: impl Into<Vec<u8>>) -> Self {
787 Self {
788 status,
789 headers: ResponseHeaders::empty(),
790 body: body.into(),
791 stream: None,
792 upgrade: None,
793 background: Vec::new(),
794 }
795 }
796
797 #[must_use]
799 pub fn empty(status: u16) -> Self {
800 Self::from_bytes(status, Vec::new())
801 }
802
803 #[must_use]
806 pub fn payload_too_large(max_body_bytes: usize) -> Self {
807 BodyRejection::PayloadTooLarge { max_body_bytes }.into_response()
808 }
809
810 #[must_use]
811 pub const fn status(&self) -> u16 {
812 self.status
813 }
814
815 pub const fn set_status(&mut self, status: u16) {
817 self.status = status;
818 }
819
820 #[must_use]
821 pub fn get_header(&self, name: &str) -> Option<&str> {
822 self.headers.get(name)
823 }
824
825 pub fn headers(&self) -> impl Iterator<Item = (&str, &str)> {
826 self.headers.iter()
827 }
828
829 pub fn set_header(&mut self, name: impl AsRef<str>, value: impl Into<String>) {
832 self.headers.insert(
833 Cow::Owned(normalize_header_name(name.as_ref())),
834 Cow::Owned(value.into()),
835 );
836 }
837
838 pub fn remove_header(&mut self, name: &str) {
840 self.headers.remove(name);
841 }
842
843 #[must_use]
844 pub fn body(&self) -> &[u8] {
845 &self.body
846 }
847
848 pub fn replace_body(&mut self, body: impl Into<Vec<u8>>) -> bool {
851 if self.stream.is_some() {
852 return false;
853 }
854 self.body = body.into();
855 true
856 }
857
858 #[must_use]
860 pub const fn is_streaming(&self) -> bool {
861 self.stream.is_some()
862 }
863
864 #[must_use]
870 pub fn take_body_stream(&mut self) -> Option<StreamingBody> {
871 self.stream.take()
872 }
873
874 pub fn set_body_stream(&mut self, stream: StreamingBody) {
879 self.body.clear();
880 self.stream = Some(stream);
881 }
882
883 #[must_use]
884 pub const fn is_upgrade(&self) -> bool {
885 self.upgrade.is_some()
886 }
887
888 pub fn take_upgrade(&mut self) -> Option<HttpUpgrade> {
890 self.upgrade.take()
891 }
892
893 pub fn take_background_tasks(&mut self) -> Vec<BackgroundTask> {
896 std::mem::take(&mut self.background)
897 }
898
899 pub async fn run_background(&mut self) -> Result<(), BackgroundTaskError> {
906 let mut first_error = None;
907 for task in self.take_background_tasks() {
908 if let Err(error) = task.run().await
909 && first_error.is_none()
910 {
911 first_error = Some(error);
912 }
913 }
914 first_error.map_or(Ok(()), Err)
915 }
916
917 #[must_use]
919 pub fn exact_body_length(&self) -> Option<u64> {
920 self.stream.as_ref().map_or_else(
921 || u64::try_from(self.body.len()).ok(),
922 StreamingBody::exact_length,
923 )
924 }
925
926 pub async fn next_body_chunk(&mut self) -> Option<Result<Vec<u8>, BodyStreamError>> {
931 match self.stream.as_mut() {
932 Some(stream) => stream.next_chunk().await,
933 None => None,
934 }
935 }
936
937 pub async fn collect_body(mut self, limit: usize) -> Result<Vec<u8>, CollectBodyError> {
947 if self.stream.is_none() {
948 if self.body.len() > limit {
949 return Err(CollectBodyError::LimitExceeded { limit });
950 }
951 return Ok(self.body);
952 }
953
954 let initial_capacity = self
955 .exact_body_length()
956 .and_then(|length| usize::try_from(length).ok())
957 .unwrap_or(0)
958 .min(limit);
959 let mut body = Vec::with_capacity(initial_capacity);
960 while let Some(chunk) = self.next_body_chunk().await {
961 let chunk = chunk.map_err(CollectBodyError::Stream)?;
962 if body.len().saturating_add(chunk.len()) > limit {
963 return Err(CollectBodyError::LimitExceeded { limit });
964 }
965 body.extend_from_slice(&chunk);
966 }
967 Ok(body)
968 }
969
970 pub fn text(&self) -> Result<&str, Utf8Error> {
976 std::str::from_utf8(&self.body)
977 }
978
979 pub fn json<T: DeserializeOwned>(&self) -> Result<T, blazingly_json::Error> {
985 blazingly_json::from_slice(&self.body)
986 }
987}
988
989#[derive(Clone, Debug, Eq, PartialEq)]
991pub enum CollectBodyError {
992 Stream(BodyStreamError),
993 LimitExceeded { limit: usize },
994}
995
996impl fmt::Display for CollectBodyError {
997 fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
998 match self {
999 Self::Stream(error) => error.fmt(formatter),
1000 Self::LimitExceeded { limit } => {
1001 write!(
1002 formatter,
1003 "response body exceeds the {limit}-byte collection limit"
1004 )
1005 }
1006 }
1007 }
1008}
1009
1010impl std::error::Error for CollectBodyError {}
1011
1012pub struct HttpApp {
1017 app: ExecutableApp,
1018 router: Router,
1019 max_body_bytes: usize,
1020 openapi: Option<OpenApiService>,
1021 middleware: Vec<ScopedMiddleware>,
1022 error_handlers: Vec<Rc<dyn HttpErrorHandler>>,
1023 allow_unverified_security_schemes: bool,
1024}
1025
1026impl HttpApp {
1027 #[must_use]
1047 pub fn new(app: ExecutableApp) -> Self {
1048 emit_and_exit_if_requested(&app);
1049 let router = Router::new(&app);
1050 Self {
1051 app,
1052 router,
1053 max_body_bytes: DEFAULT_MAX_BODY_BYTES,
1054 openapi: None,
1055 middleware: Vec::new(),
1056 error_handlers: Vec::new(),
1057 allow_unverified_security_schemes: false,
1058 }
1059 }
1060
1061 #[must_use]
1062 pub const fn with_max_body_bytes(mut self, max_body_bytes: usize) -> Self {
1063 self.max_body_bytes = max_body_bytes;
1064 self
1065 }
1066
1067 #[must_use]
1075 pub const fn with_unverified_security_schemes(mut self, allow: bool) -> Self {
1076 self.allow_unverified_security_schemes = allow;
1077 self
1078 }
1079
1080 pub async fn startup(&self) -> Result<(), DependencyError> {
1086 self.app.startup().await
1087 }
1088
1089 pub async fn shutdown(&self) -> Result<(), DependencyError> {
1095 self.app.shutdown().await
1096 }
1097
1098 #[must_use]
1100 pub fn with_openapi(mut self, config: OpenApiConfig) -> Self {
1101 self.openapi = Some(OpenApiService::new(self.app.definition(), config));
1102 self
1103 }
1104
1105 #[must_use]
1107 pub fn with_middleware(mut self, middleware: impl HttpMiddleware + 'static) -> Self {
1108 self.middleware.push(ScopedMiddleware {
1109 scope: MiddlewareScope::all(),
1110 layer: Rc::new(middleware),
1111 });
1112 self
1113 }
1114
1115 #[must_use]
1117 pub fn with_shared_middleware(mut self, middleware: Rc<dyn HttpMiddleware>) -> Self {
1118 self.middleware.push(ScopedMiddleware {
1119 scope: MiddlewareScope::all(),
1120 layer: middleware,
1121 });
1122 self
1123 }
1124
1125 #[must_use]
1128 pub fn with_scoped_middleware(
1129 mut self,
1130 scope: MiddlewareScope,
1131 middleware: impl HttpMiddleware + 'static,
1132 ) -> Self {
1133 self.middleware.push(ScopedMiddleware {
1134 scope,
1135 layer: Rc::new(middleware),
1136 });
1137 self
1138 }
1139
1140 #[must_use]
1143 pub fn with_shared_scoped_middleware(
1144 mut self,
1145 scope: MiddlewareScope,
1146 middleware: Rc<dyn HttpMiddleware>,
1147 ) -> Self {
1148 self.middleware.push(ScopedMiddleware {
1149 scope,
1150 layer: middleware,
1151 });
1152 self
1153 }
1154
1155 #[must_use]
1157 pub fn with_error_handler(mut self, handler: impl HttpErrorHandler + 'static) -> Self {
1158 self.error_handlers.push(Rc::new(handler));
1159 self
1160 }
1161
1162 #[must_use]
1164 pub fn with_shared_error_handler(mut self, handler: Rc<dyn HttpErrorHandler>) -> Self {
1165 self.error_handlers.push(handler);
1166 self
1167 }
1168
1169 pub async fn call(&self, request: Request) -> Response {
1170 self.call_view(&request).await
1171 }
1172
1173 pub async fn call_view(&self, request: &impl HttpRequestView) -> Response {
1174 let dispatcher = self.dispatcher();
1175 dispatcher.dispatch(request, None).await
1176 }
1177
1178 pub async fn call_view_controlled(
1179 &self,
1180 request: &impl HttpRequestView,
1181 control: InvocationControl,
1182 ) -> Response {
1183 let dispatcher = self.dispatcher();
1184 dispatcher.dispatch(request, Some(control)).await
1185 }
1186
1187 fn dispatcher(&self) -> Dispatcher<'_> {
1188 Dispatcher {
1189 app: &self.app,
1190 router: &self.router,
1191 max_body_bytes: self.max_body_bytes,
1192 openapi: self.openapi.as_ref(),
1193 middleware: &self.middleware,
1194 error_handlers: &self.error_handlers,
1195 allow_unverified_security_schemes: self.allow_unverified_security_schemes,
1196 }
1197 }
1198
1199 #[must_use]
1204 pub fn request_body_source(&self, method: HttpMethod, target: &str) -> Option<InputSource> {
1205 let path = target.split_once('?').map_or(target, |(path, _)| path);
1206 self.router
1207 .recognize(method, path)
1208 .ok()
1209 .and_then(|route| route.body_source())
1210 }
1211}
1212
1213fn emit_and_exit_if_requested(app: &ExecutableApp) {
1221 static EMITTED: std::sync::Once = std::sync::Once::new();
1222 let Some(mode) = std::env::var_os(EMIT_VARIABLE) else {
1223 return;
1224 };
1225 if mode.is_empty() {
1226 return;
1227 }
1228 let mut code = 0;
1229 EMITTED.call_once(|| code = emit(app, &mode.to_string_lossy()));
1230 std::process::exit(code);
1231}
1232
1233fn emit(app: &ExecutableApp, mode: &str) -> i32 {
1237 use std::io::Write as _;
1238 let output = match mode {
1239 "openapi" => match openapi_document_text(app.definition()) {
1240 Ok(document) => document,
1241 Err(error) => {
1242 eprintln!("error: the OpenAPI document could not be serialized: {error}");
1243 return 1;
1244 }
1245 },
1246 "routes" => routes_table_text(app.definition()),
1247 unknown => {
1248 eprintln!("error: {EMIT_VARIABLE} must be `openapi` or `routes`, not `{unknown}`");
1249 return 2;
1250 }
1251 };
1252 let mut stdout = std::io::stdout().lock();
1253 if stdout.write_all(output.as_bytes()).is_err() || stdout.flush().is_err() {
1254 return 1;
1255 }
1256 0
1257}
1258
1259fn openapi_document_text(definition: &AppDefinition) -> Result<String, blazingly_json::Error> {
1261 let document = blazingly_openapi::to_value(definition);
1262 let mut text = blazingly_json::to_string_pretty(&document)?;
1263 text.push('\n');
1264 Ok(text)
1265}
1266
1267fn routes_table_text(definition: &AppDefinition) -> String {
1269 use std::fmt::Write as _;
1270 let mut table = String::from("METHOD\tPATH\tOPERATION\tSUMMARY\n");
1271 for operation in definition.operations() {
1272 let _ = writeln!(
1273 table,
1274 "{}\t{}\t{}\t{}",
1275 operation.http.method.as_str(),
1276 operation.http.path,
1277 operation.contract.id.as_str(),
1278 operation.contract.summary
1279 );
1280 }
1281 table
1282}
1283
1284pub struct TestApp<'app> {
1286 app: &'app ExecutableApp,
1287 router: Router,
1288 max_body_bytes: usize,
1289 openapi: Option<OpenApiService>,
1290 middleware: Vec<ScopedMiddleware>,
1291 error_handlers: Vec<Rc<dyn HttpErrorHandler>>,
1292 allow_unverified_security_schemes: bool,
1293}
1294
1295impl<'app> TestApp<'app> {
1296 #[must_use]
1297 pub fn new(app: &'app ExecutableApp) -> Self {
1298 Self {
1299 app,
1300 router: Router::new(app),
1301 max_body_bytes: DEFAULT_MAX_BODY_BYTES,
1302 openapi: None,
1303 middleware: Vec::new(),
1304 error_handlers: Vec::new(),
1305 allow_unverified_security_schemes: false,
1306 }
1307 }
1308
1309 #[must_use]
1310 pub const fn with_max_body_bytes(mut self, max_body_bytes: usize) -> Self {
1311 self.max_body_bytes = max_body_bytes;
1312 self
1313 }
1314
1315 #[must_use]
1321 pub const fn with_unverified_security_schemes(mut self, allow: bool) -> Self {
1322 self.allow_unverified_security_schemes = allow;
1323 self
1324 }
1325
1326 pub async fn startup(&self) -> Result<(), DependencyError> {
1332 self.app.startup().await
1333 }
1334
1335 pub async fn shutdown(&self) -> Result<(), DependencyError> {
1341 self.app.shutdown().await
1342 }
1343
1344 #[must_use]
1346 pub fn with_openapi(mut self, config: OpenApiConfig) -> Self {
1347 self.openapi = Some(OpenApiService::new(self.app.definition(), config));
1348 self
1349 }
1350
1351 #[must_use]
1353 pub fn with_middleware(mut self, middleware: impl HttpMiddleware + 'static) -> Self {
1354 self.middleware.push(ScopedMiddleware {
1355 scope: MiddlewareScope::all(),
1356 layer: Rc::new(middleware),
1357 });
1358 self
1359 }
1360
1361 #[must_use]
1363 pub fn with_shared_middleware(mut self, middleware: Rc<dyn HttpMiddleware>) -> Self {
1364 self.middleware.push(ScopedMiddleware {
1365 scope: MiddlewareScope::all(),
1366 layer: middleware,
1367 });
1368 self
1369 }
1370
1371 #[must_use]
1374 pub fn with_scoped_middleware(
1375 mut self,
1376 scope: MiddlewareScope,
1377 middleware: impl HttpMiddleware + 'static,
1378 ) -> Self {
1379 self.middleware.push(ScopedMiddleware {
1380 scope,
1381 layer: Rc::new(middleware),
1382 });
1383 self
1384 }
1385
1386 #[must_use]
1389 pub fn with_shared_scoped_middleware(
1390 mut self,
1391 scope: MiddlewareScope,
1392 middleware: Rc<dyn HttpMiddleware>,
1393 ) -> Self {
1394 self.middleware.push(ScopedMiddleware {
1395 scope,
1396 layer: middleware,
1397 });
1398 self
1399 }
1400
1401 #[must_use]
1403 pub fn with_error_handler(mut self, handler: impl HttpErrorHandler + 'static) -> Self {
1404 self.error_handlers.push(Rc::new(handler));
1405 self
1406 }
1407
1408 #[must_use]
1410 pub fn with_shared_error_handler(mut self, handler: Rc<dyn HttpErrorHandler>) -> Self {
1411 self.error_handlers.push(handler);
1412 self
1413 }
1414
1415 pub async fn call(&self, request: Request) -> Response {
1416 let dispatcher = self.dispatcher();
1417 dispatcher.dispatch(&request, None).await
1418 }
1419
1420 pub async fn call_controlled(&self, request: Request, control: InvocationControl) -> Response {
1421 let dispatcher = self.dispatcher();
1422 dispatcher.dispatch(&request, Some(control)).await
1423 }
1424
1425 fn dispatcher(&self) -> Dispatcher<'_> {
1426 Dispatcher {
1427 app: self.app,
1428 router: &self.router,
1429 max_body_bytes: self.max_body_bytes,
1430 openapi: self.openapi.as_ref(),
1431 middleware: &self.middleware,
1432 error_handlers: &self.error_handlers,
1433 allow_unverified_security_schemes: self.allow_unverified_security_schemes,
1434 }
1435 }
1436}
1437
1438struct Dispatcher<'app> {
1440 app: &'app ExecutableApp,
1441 router: &'app Router,
1442 max_body_bytes: usize,
1443 openapi: Option<&'app OpenApiService>,
1444 middleware: &'app [ScopedMiddleware],
1445 error_handlers: &'app [Rc<dyn HttpErrorHandler>],
1446 allow_unverified_security_schemes: bool,
1447}
1448
1449struct DispatchSite<'dispatch> {
1451 method: HttpMethod,
1452 path: &'dispatch str,
1453 operation: Option<&'dispatch OperationDescriptor>,
1454}
1455
1456impl Dispatcher<'_> {
1457 async fn dispatch<RequestView>(
1458 &self,
1459 request: &RequestView,
1460 control: Option<InvocationControl>,
1461 ) -> Response
1462 where
1463 RequestView: HttpRequestView,
1464 {
1465 if self.middleware.is_empty() {
1466 return self.dispatch_unlayered(request, control).await;
1467 }
1468 let middleware = self.middleware;
1469 let target = request.target();
1470 let mut site = DispatchSite {
1471 method: request.method(),
1472 path: target.split_once('?').map_or(target, |(path, _)| path),
1473 operation: None,
1474 };
1475 let mut context = HttpRequestContext::new(request);
1476 for layer in middleware {
1477 if layer.scope.matches_request(site.path)
1478 && let Some(response) = layer.layer.on_request(&mut context)
1479 {
1480 return complete_response(middleware, &context, &site, response);
1481 }
1482 }
1483
1484 if validate_url_encoding(target).is_err() {
1485 let response = self.fail(&site, invalid_url_encoding_failure());
1486 return complete_response(middleware, &context, &site, response);
1487 }
1488 if let Some(response) = self
1489 .openapi
1490 .and_then(|service| service.handle(site.method, site.path))
1491 {
1492 return complete_response(middleware, &context, &site, openapi_response(response));
1493 }
1494 let recognized = match self.router.recognize(site.method, site.path) {
1495 Ok(recognized) => recognized,
1496 Err(error) => {
1497 let response = self.fail(&site, route_miss_failure(&error));
1498 return complete_response(middleware, &context, &site, response);
1499 }
1500 };
1501 let Some(operation) = self.app.operation_at(recognized.operation_index()) else {
1502 let response = self.fail(&site, internal_failure());
1503 return complete_response(middleware, &context, &site, response);
1504 };
1505 let descriptor = operation.descriptor();
1506 site.operation = Some(descriptor);
1507 for layer in middleware {
1508 if layer.matches(site.path, site.operation)
1509 && let Some(response) = layer.layer.on_operation(
1510 &mut context,
1511 descriptor,
1512 self.app.definition().security_schemes(),
1513 )
1514 {
1515 return complete_response(middleware, &context, &site, response);
1516 }
1517 }
1518 if let Some(failure) = self.security_guard(site.path, descriptor) {
1519 let response = self.fail(&site, failure);
1520 return complete_response(middleware, &context, &site, response);
1521 }
1522 if let Some(body_source) = recognized.body_source() {
1523 match validate_body(request, self.max_body_bytes, body_source) {
1524 Ok(()) => {}
1525 Err(rejection) => {
1526 let response = self.fail(&site, rejection.into_failure());
1527 return complete_response(middleware, &context, &site, response);
1528 }
1529 }
1530 }
1531 let request_parts = RoutedRequestParts {
1532 request,
1533 route: &recognized,
1534 context: Some(&context),
1535 connection: OnceCell::new(),
1536 background: OnceCell::new(),
1537 };
1538 let outcome = if let Some(control) = control {
1539 operation
1540 .invoke_http_controlled(&request_parts, control)
1541 .await
1542 } else {
1543 operation.invoke_http(&request_parts).await
1544 };
1545 let mut response = match outcome_result(outcome) {
1546 Ok(response) => response,
1547 Err(failure) => self.fail(&site, failure),
1548 };
1549 response.background.extend(request_parts.scheduled_tasks());
1550 complete_response(middleware, &context, &site, response)
1551 }
1552
1553 async fn dispatch_unlayered<RequestView>(
1554 &self,
1555 request: &RequestView,
1556 control: Option<InvocationControl>,
1557 ) -> Response
1558 where
1559 RequestView: HttpRequestView + ?Sized,
1560 {
1561 let target = request.target();
1562 let mut site = DispatchSite {
1563 method: request.method(),
1564 path: target.split_once('?').map_or(target, |(path, _)| path),
1565 operation: None,
1566 };
1567 if validate_url_encoding(target).is_err() {
1568 return self.fail(&site, invalid_url_encoding_failure());
1569 }
1570 if let Some(response) = self
1571 .openapi
1572 .and_then(|service| service.handle(site.method, site.path))
1573 {
1574 return openapi_response(response);
1575 }
1576 let recognized = match self.router.recognize(site.method, site.path) {
1577 Ok(recognized) => recognized,
1578 Err(error) => return self.fail(&site, route_miss_failure(&error)),
1579 };
1580 let Some(operation) = self.app.operation_at(recognized.operation_index()) else {
1581 return self.fail(&site, internal_failure());
1582 };
1583 site.operation = Some(operation.descriptor());
1584 if let Some(failure) = self.security_guard(site.path, operation.descriptor()) {
1585 return self.fail(&site, failure);
1586 }
1587 if let Some(body_source) = recognized.body_source() {
1588 match validate_body(request, self.max_body_bytes, body_source) {
1589 Ok(()) => {}
1590 Err(rejection) => return self.fail(&site, rejection.into_failure()),
1591 }
1592 }
1593 let request_parts = RoutedRequestParts {
1594 request,
1595 route: &recognized,
1596 context: None,
1597 connection: OnceCell::new(),
1598 background: OnceCell::new(),
1599 };
1600 let outcome = if let Some(control) = control {
1601 operation
1602 .invoke_http_controlled(&request_parts, control)
1603 .await
1604 } else {
1605 operation.invoke_http(&request_parts).await
1606 };
1607 let mut response = match outcome_result(outcome) {
1608 Ok(response) => response,
1609 Err(failure) => self.fail(&site, failure),
1610 };
1611 response.background.extend(request_parts.scheduled_tasks());
1612 response
1613 }
1614
1615 fn fail(&self, site: &DispatchSite<'_>, mut failure: Failure) -> Response {
1618 let mut response = failure.response();
1619 if self.error_handlers.is_empty() {
1620 return response;
1621 }
1622 let error = HttpError {
1623 source: failure.source,
1624 status: failure.status,
1625 code: &failure.code,
1626 message: &failure.message,
1627 method: site.method,
1628 path: site.path,
1629 operation: site.operation,
1630 };
1631 for handler in self.error_handlers {
1632 handler.on_error(&error, &mut response);
1633 }
1634 if failure.source == HttpErrorSource::Domain {
1636 response.status = failure.status;
1637 }
1638 response
1639 }
1640
1641 fn security_guard(&self, path: &str, descriptor: &OperationDescriptor) -> Option<Failure> {
1648 if self.allow_unverified_security_schemes || descriptor.contract.security.is_empty() {
1649 return None;
1650 }
1651 if self
1652 .middleware
1653 .iter()
1654 .any(|layer| layer.layer.verifies_security() && layer.matches(path, Some(descriptor)))
1655 {
1656 return None;
1657 }
1658 Some(Failure::new(
1659 HttpErrorSource::Internal,
1660 500,
1661 "security_verifier_missing",
1662 "the operation declares a security scheme with no registered verifier",
1663 ))
1664 }
1665}
1666
1667struct Failure {
1669 source: HttpErrorSource,
1670 status: u16,
1671 code: Cow<'static, str>,
1672 message: Cow<'static, str>,
1673 details: Option<Value>,
1674 headers: Vec<ResponseHeader>,
1675}
1676
1677impl Failure {
1678 fn new(
1679 source: HttpErrorSource,
1680 status: u16,
1681 code: impl Into<Cow<'static, str>>,
1682 message: impl Into<Cow<'static, str>>,
1683 ) -> Self {
1684 Self {
1685 source,
1686 status,
1687 code: code.into(),
1688 message: message.into(),
1689 details: None,
1690 headers: Vec::new(),
1691 }
1692 }
1693
1694 fn with_details(mut self, details: Value) -> Self {
1695 self.details = Some(details);
1696 self
1697 }
1698
1699 fn with_headers(mut self, headers: Vec<ResponseHeader>) -> Self {
1700 self.headers = headers;
1701 self
1702 }
1703
1704 fn response(&mut self) -> Response {
1705 let response = error_response(self.status, &self.code, &self.message, self.details.take());
1706 with_outcome_headers(response, std::mem::take(&mut self.headers))
1707 }
1708
1709 fn into_response(mut self) -> Response {
1710 self.response()
1711 }
1712}
1713
1714fn invalid_url_encoding_failure() -> Failure {
1715 Failure::new(
1716 HttpErrorSource::Request,
1717 400,
1718 "invalid_url_encoding",
1719 "request target contains invalid percent encoding",
1720 )
1721}
1722
1723fn route_miss_failure(error: &RouteError) -> Failure {
1724 match error {
1725 RouteError::MethodNotAllowed { allowed } => {
1726 let allow = allowed
1727 .iter()
1728 .map(|method| method.as_str())
1729 .collect::<Vec<_>>()
1730 .join(", ");
1731 Failure::new(
1732 HttpErrorSource::Routing,
1733 405,
1734 "method_not_allowed",
1735 "HTTP method not allowed",
1736 )
1737 .with_headers(vec![ResponseHeader::new("allow", allow)])
1738 }
1739 RouteError::NotFound => Failure::new(
1740 HttpErrorSource::Routing,
1741 404,
1742 "not_found",
1743 "HTTP route not found",
1744 ),
1745 }
1746}
1747
1748fn complete_response(
1749 middleware: &[ScopedMiddleware],
1750 context: &HttpRequestContext<'_>,
1751 site: &DispatchSite<'_>,
1752 mut response: Response,
1753) -> Response {
1754 for layer in middleware.iter().rev() {
1755 if layer.matches(site.path, site.operation) {
1756 layer
1757 .layer
1758 .on_response(context, site.operation, &mut response);
1759 }
1760 }
1761 response
1762}
1763
1764fn openapi_response(asset: OpenApiAssetResponse) -> Response {
1765 let mut response = Response {
1766 status: asset.status,
1767 headers: ResponseHeaders::empty(),
1768 body: asset.body,
1769 stream: None,
1770 upgrade: None,
1771 background: Vec::new(),
1772 };
1773 for (name, value) in asset.headers {
1774 response = response.with_header(name, value);
1775 }
1776 response
1777}
1778
1779#[derive(Default)]
1780struct RouteNode {
1781 static_children: HashMap<String, usize>,
1782 parameter_child: Option<usize>,
1783 endpoints: BTreeMap<HttpMethod, CompiledEndpoint>,
1784}
1785
1786#[derive(Clone)]
1787struct CompiledEndpoint {
1788 operation_index: usize,
1789 parameter_names: Vec<String>,
1790 body_source: Option<InputSource>,
1791}
1792
1793pub struct Router {
1795 nodes: Vec<RouteNode>,
1796 static_routes: HashMap<HttpMethod, HashMap<String, CompiledEndpoint>>,
1797}
1798
1799impl Router {
1800 #[must_use]
1801 pub fn new(app: &ExecutableApp) -> Self {
1802 let mut router = Self {
1803 nodes: vec![RouteNode::default()],
1804 static_routes: HashMap::new(),
1805 };
1806 for descriptor in app.definition().operations() {
1807 let Some(operation_index) = app.operation_index(&descriptor.contract.id) else {
1808 continue;
1809 };
1810 router.insert(descriptor, operation_index);
1811 }
1812 router
1813 }
1814
1815 fn insert(&mut self, descriptor: &OperationDescriptor, operation_index: usize) {
1816 let endpoint = CompiledEndpoint {
1817 operation_index,
1818 parameter_names: route_segments(&descriptor.http.path)
1819 .filter_map(path_parameter_name)
1820 .map(str::to_owned)
1821 .collect(),
1822 body_source: body_source(descriptor),
1823 };
1824 if endpoint.parameter_names.is_empty() {
1825 self.static_routes
1826 .entry(descriptor.http.method)
1827 .or_default()
1828 .insert(descriptor.http.path.clone(), endpoint);
1829 return;
1830 }
1831
1832 let mut node_index = 0;
1833 for segment in route_segments(&descriptor.http.path) {
1834 if path_parameter_name(segment).is_some() {
1835 node_index = if let Some(child) = self.nodes[node_index].parameter_child {
1836 child
1837 } else {
1838 let child = self.nodes.len();
1839 self.nodes.push(RouteNode::default());
1840 self.nodes[node_index].parameter_child = Some(child);
1841 child
1842 };
1843 } else if let Some(child) = self.nodes[node_index].static_children.get(segment) {
1844 node_index = *child;
1845 } else {
1846 let child = self.nodes.len();
1847 self.nodes.push(RouteNode::default());
1848 self.nodes[node_index]
1849 .static_children
1850 .insert(segment.to_owned(), child);
1851 node_index = child;
1852 }
1853 }
1854 self.nodes[node_index]
1855 .endpoints
1856 .insert(descriptor.http.method, endpoint);
1857 }
1858
1859 pub fn recognize<'router, 'path>(
1866 &'router self,
1867 method: HttpMethod,
1868 path: &'path str,
1869 ) -> Result<RouteMatch<'router, 'path>, RouteError> {
1870 if let Some(endpoint) = self
1871 .static_routes
1872 .get(&method)
1873 .and_then(|routes| routes.get(path))
1874 {
1875 return Ok(RouteMatch {
1876 endpoint,
1877 captures: CapturedSegments::new(),
1878 });
1879 }
1880
1881 if let Some((endpoint, captures)) =
1882 self.find_dynamic(0, route_segments(path), method, CapturedSegments::new())
1883 {
1884 return Ok(RouteMatch { endpoint, captures });
1885 }
1886
1887 let mut allowed = self
1888 .static_routes
1889 .iter()
1890 .filter_map(|(allowed_method, routes)| {
1891 routes.contains_key(path).then_some(*allowed_method)
1892 })
1893 .collect::<Vec<_>>();
1894 self.collect_dynamic_methods(0, route_segments(path), &mut allowed);
1895 allowed.sort_unstable();
1896 allowed.dedup();
1897 if allowed.is_empty() {
1898 Err(RouteError::NotFound)
1899 } else {
1900 Err(RouteError::MethodNotAllowed { allowed })
1901 }
1902 }
1903
1904 fn find_dynamic<'router, 'path, I>(
1905 &'router self,
1906 node_index: usize,
1907 mut segments: I,
1908 method: HttpMethod,
1909 captures: CapturedSegments<'path>,
1910 ) -> Option<(&'router CompiledEndpoint, CapturedSegments<'path>)>
1911 where
1912 I: Iterator<Item = &'path str> + Clone,
1913 {
1914 let Some(segment) = segments.next() else {
1915 return self.nodes[node_index]
1916 .endpoints
1917 .get(&method)
1918 .map(|endpoint| (endpoint, captures));
1919 };
1920
1921 if let Some(child) = self.nodes[node_index].static_children.get(segment)
1922 && let Some(found) =
1923 self.find_dynamic(*child, segments.clone(), method, captures.clone())
1924 {
1925 return Some(found);
1926 }
1927 let child = self.nodes[node_index].parameter_child?;
1928 let mut captures = captures;
1929 captures.push(segment);
1930 self.find_dynamic(child, segments, method, captures)
1931 }
1932
1933 fn collect_dynamic_methods<'path, I>(
1934 &self,
1935 node_index: usize,
1936 mut segments: I,
1937 methods: &mut Vec<HttpMethod>,
1938 ) where
1939 I: Iterator<Item = &'path str> + Clone,
1940 {
1941 let Some(segment) = segments.next() else {
1942 methods.extend(self.nodes[node_index].endpoints.keys().copied());
1943 return;
1944 };
1945 if let Some(child) = self.nodes[node_index].static_children.get(segment) {
1946 self.collect_dynamic_methods(*child, segments.clone(), methods);
1947 }
1948 if let Some(child) = self.nodes[node_index].parameter_child {
1949 self.collect_dynamic_methods(child, segments, methods);
1950 }
1951 }
1952}
1953
1954fn route_segments(path: &str) -> std::str::Split<'_, char> {
1955 path.strip_prefix('/').unwrap_or(path).split('/')
1956}
1957
1958#[derive(Clone, Debug, Eq, PartialEq)]
1960pub enum RouteError {
1961 MethodNotAllowed { allowed: Vec<HttpMethod> },
1962 NotFound,
1963}
1964
1965pub struct RouteMatch<'router, 'path> {
1967 endpoint: &'router CompiledEndpoint,
1968 captures: CapturedSegments<'path>,
1969}
1970
1971impl RouteMatch<'_, '_> {
1972 #[must_use]
1973 pub const fn operation_index(&self) -> usize {
1974 self.endpoint.operation_index
1975 }
1976
1977 #[must_use]
1978 pub const fn requires_json(&self) -> bool {
1979 matches!(self.endpoint.body_source, Some(InputSource::Json))
1980 }
1981
1982 #[must_use]
1983 pub const fn body_source(&self) -> Option<InputSource> {
1984 self.endpoint.body_source
1985 }
1986
1987 #[must_use]
1988 pub fn path_parameter(&self, name: &str) -> Option<Cow<'_, str>> {
1989 self.endpoint
1990 .parameter_names
1991 .iter()
1992 .position(|parameter| parameter == name)
1993 .and_then(|position| self.captures.get(position))
1994 .and_then(|value| decode_url_component(value, false).ok())
1995 }
1996}
1997
1998const INLINE_PATH_PARAMETERS: usize = 8;
1999
2000#[derive(Clone)]
2001enum CapturedSegments<'path> {
2002 Inline {
2003 values: [Option<&'path str>; INLINE_PATH_PARAMETERS],
2004 len: usize,
2005 },
2006 Heap(Vec<&'path str>),
2007}
2008
2009impl<'path> CapturedSegments<'path> {
2010 const fn new() -> Self {
2011 Self::Inline {
2012 values: [None; INLINE_PATH_PARAMETERS],
2013 len: 0,
2014 }
2015 }
2016
2017 fn push(&mut self, value: &'path str) {
2018 match self {
2019 Self::Inline { values, len } if *len < INLINE_PATH_PARAMETERS => {
2020 values[*len] = Some(value);
2021 *len += 1;
2022 }
2023 Self::Inline { values, len } => {
2024 let mut heap = values[..*len]
2025 .iter()
2026 .filter_map(|value| *value)
2027 .collect::<Vec<_>>();
2028 heap.push(value);
2029 *self = Self::Heap(heap);
2030 }
2031 Self::Heap(values) => values.push(value),
2032 }
2033 }
2034
2035 fn get(&self, index: usize) -> Option<&'path str> {
2036 match self {
2037 Self::Inline { values, len } if index < *len => values[index],
2038 Self::Inline { .. } => None,
2039 Self::Heap(values) => values.get(index).copied(),
2040 }
2041 }
2042}
2043
2044struct RoutedRequestParts<'request, 'router, 'path, 'context, RequestView: ?Sized> {
2045 request: &'request RequestView,
2046 route: &'request RouteMatch<'router, 'path>,
2047 context: Option<&'context HttpRequestContext<'request>>,
2048 connection: OnceCell<ConnectionInfo>,
2049 background: OnceCell<BackgroundTasks>,
2050}
2051
2052impl<RequestView> RoutedRequestParts<'_, '_, '_, '_, RequestView>
2053where
2054 RequestView: HttpRequestView + ?Sized,
2055{
2056 fn connection_info(&self) -> &ConnectionInfo {
2058 self.connection.get_or_init(|| {
2059 self.context.map_or_else(
2060 || ConnectionInfo::from_request(self.request),
2061 HttpRequestContext::connection_info,
2062 )
2063 })
2064 }
2065
2066 fn background_tasks(&self) -> &BackgroundTasks {
2069 self.background.get_or_init(BackgroundTasks::new)
2070 }
2071
2072 fn scheduled_tasks(&self) -> Vec<BackgroundTask> {
2074 self.background
2075 .get()
2076 .map_or_else(Vec::new, BackgroundTasks::take)
2077 }
2078}
2079
2080impl<RequestView> InvocationRequestParts for RoutedRequestParts<'_, '_, '_, '_, RequestView>
2081where
2082 RequestView: HttpRequestView + ?Sized,
2083{
2084 fn value(&self, source: InputSource, name: &str, index: usize) -> Option<Cow<'_, str>> {
2085 match source {
2086 InputSource::Path if index == 0 => self.route.path_parameter(name),
2087 InputSource::Query => query_value(self.request.target(), name, index),
2088 InputSource::Header => self.request.header_value(name, index).map(Cow::Borrowed),
2089 InputSource::Cookie => cookie_value(self.request, name, index),
2090 InputSource::Form => form_value(self.request.body(), name, index),
2091 InputSource::Path
2092 | InputSource::Json
2093 | InputSource::Multipart
2094 | InputSource::File
2095 | InputSource::Stream => None,
2096 }
2097 }
2098
2099 fn body(&self) -> &[u8] {
2100 self.request.body()
2101 }
2102
2103 fn take_body_stream(&self) -> Option<StreamingBody> {
2104 self.request.take_body_stream()
2105 }
2106
2107 fn extension(&self, type_id: TypeId) -> Option<&dyn Any> {
2108 if let Some(value) = self
2109 .context
2110 .and_then(|context| context.extension_by_id(type_id))
2111 {
2112 return Some(value);
2113 }
2114 if type_id == TypeId::of::<ConnectionInfo>() {
2115 return Some(self.connection_info());
2116 }
2117 if type_id == TypeId::of::<BackgroundTasks>() {
2118 return Some(self.background_tasks());
2119 }
2120 None
2121 }
2122}
2123
2124fn body_source(descriptor: &OperationDescriptor) -> Option<InputSource> {
2125 descriptor
2126 .contract
2127 .inputs
2128 .iter()
2129 .map(|input| input.source)
2130 .find(|source| {
2131 matches!(
2132 source,
2133 InputSource::Json
2134 | InputSource::Form
2135 | InputSource::Multipart
2136 | InputSource::File
2137 | InputSource::Stream
2138 )
2139 })
2140}
2141
2142fn cookie_value<'request>(
2143 request: &'request (impl HttpRequestView + ?Sized),
2144 name: &str,
2145 index: usize,
2146) -> Option<Cow<'request, str>> {
2147 let mut header_index = 0;
2148 let mut found = 0;
2149 while let Some(header) = request.header_value("cookie", header_index) {
2150 for cookie in header.split(';') {
2151 let (cookie_name, value) = cookie.trim().split_once('=').unwrap_or((cookie.trim(), ""));
2152 if cookie_name == name {
2153 if found == index {
2154 return Some(Cow::Borrowed(value));
2155 }
2156 found += 1;
2157 }
2158 }
2159 header_index += 1;
2160 }
2161 None
2162}
2163
2164fn form_value<'body>(body: &'body [u8], name: &str, index: usize) -> Option<Cow<'body, str>> {
2165 let body = std::str::from_utf8(body).ok()?;
2166 let mut found = 0;
2167 for pair in body.split('&').filter(|pair| !pair.is_empty()) {
2168 let (raw_name, raw_value) = pair.split_once('=').unwrap_or((pair, ""));
2169 let decoded_name = decode_url_component(raw_name, true).ok()?;
2170 if decoded_name == name {
2171 if found == index {
2172 return decode_url_component(raw_value, true).ok();
2173 }
2174 found += 1;
2175 }
2176 }
2177 None
2178}
2179
2180fn path_parameter_name(segment: &str) -> Option<&str> {
2181 segment
2182 .strip_prefix('{')
2183 .and_then(|segment| segment.strip_suffix('}'))
2184 .filter(|name| !name.is_empty())
2185}
2186
2187fn query_value<'target>(
2188 target: &'target str,
2189 name: &str,
2190 index: usize,
2191) -> Option<Cow<'target, str>> {
2192 let (_, query) = target.split_once('?')?;
2193 let mut found = 0;
2194 for pair in query.split('&').filter(|pair| !pair.is_empty()) {
2195 let (raw_name, raw_value) = pair.split_once('=').unwrap_or((pair, ""));
2196 let decoded_name = decode_url_component(raw_name, true).ok()?;
2197 if decoded_name == name {
2198 if found == index {
2199 return decode_url_component(raw_value, true).ok();
2200 }
2201 found += 1;
2202 }
2203 }
2204 None
2205}
2206
2207fn header_name_matches(header: &str, argument: &str) -> bool {
2208 header.bytes().eq(argument.bytes().map(|byte| {
2209 if byte == b'_' {
2210 b'-'
2211 } else {
2212 byte.to_ascii_lowercase()
2213 }
2214 }))
2215}
2216
2217fn validate_url_encoding(target: &str) -> Result<(), ()> {
2218 let path = target.split_once('?').map_or(target, |(path, _)| path);
2219 decode_url_component(path, false)?;
2220 if let Some((_, query)) = target.split_once('?') {
2221 for pair in query.split('&') {
2222 let (name, value) = pair.split_once('=').unwrap_or((pair, ""));
2223 decode_url_component(name, true)?;
2224 decode_url_component(value, true)?;
2225 }
2226 }
2227 Ok(())
2228}
2229
2230fn decode_url_component(value: &str, plus_as_space: bool) -> Result<Cow<'_, str>, ()> {
2231 if !value.as_bytes().contains(&b'%') && (!plus_as_space || !value.as_bytes().contains(&b'+')) {
2232 return Ok(Cow::Borrowed(value));
2233 }
2234 let bytes = value.as_bytes();
2235 let mut decoded = Vec::with_capacity(bytes.len());
2236 let mut index = 0;
2237 while index < bytes.len() {
2238 match bytes[index] {
2239 b'%' => {
2240 let high = bytes
2241 .get(index + 1)
2242 .copied()
2243 .and_then(hex_value)
2244 .ok_or(())?;
2245 let low = bytes
2246 .get(index + 2)
2247 .copied()
2248 .and_then(hex_value)
2249 .ok_or(())?;
2250 decoded.push((high << 4) | low);
2251 index += 3;
2252 }
2253 b'+' if plus_as_space => {
2254 decoded.push(b' ');
2255 index += 1;
2256 }
2257 byte => {
2258 decoded.push(byte);
2259 index += 1;
2260 }
2261 }
2262 }
2263 String::from_utf8(decoded).map(Cow::Owned).map_err(|_| ())
2264}
2265
2266const fn hex_value(byte: u8) -> Option<u8> {
2267 match byte {
2268 b'0'..=b'9' => Some(byte - b'0'),
2269 b'a'..=b'f' => Some(byte - b'a' + 10),
2270 b'A'..=b'F' => Some(byte - b'A' + 10),
2271 _ => None,
2272 }
2273}
2274
2275enum BodyRejection {
2276 PayloadTooLarge { max_body_bytes: usize },
2277 UnsupportedMediaType,
2278}
2279
2280impl BodyRejection {
2281 fn into_failure(self) -> Failure {
2282 match self {
2283 Self::PayloadTooLarge { max_body_bytes } => Failure::new(
2284 HttpErrorSource::Request,
2285 413,
2286 "payload_too_large",
2287 "request body exceeds the configured limit",
2288 )
2289 .with_details(json!({ "maxBytes": max_body_bytes })),
2290 Self::UnsupportedMediaType => Failure::new(
2291 HttpErrorSource::Request,
2292 415,
2293 "unsupported_media_type",
2294 "request body media type does not match the operation input",
2295 ),
2296 }
2297 }
2298
2299 fn into_response(self) -> Response {
2300 self.into_failure().into_response()
2301 }
2302}
2303
2304fn validate_body(
2305 request: &(impl HttpRequestView + ?Sized),
2306 max_body_bytes: usize,
2307 source: InputSource,
2308) -> Result<(), BodyRejection> {
2309 if request.body().len() > max_body_bytes {
2310 return Err(BodyRejection::PayloadTooLarge { max_body_bytes });
2311 }
2312 if source == InputSource::Stream {
2313 return Ok(());
2314 }
2315
2316 let valid_media_type = request
2317 .header_value("content-type", 0)
2318 .is_some_and(|content_type| match source {
2319 InputSource::Json => is_json_media_type(content_type),
2320 InputSource::Form => media_type_is(content_type, "application/x-www-form-urlencoded"),
2321 InputSource::Multipart | InputSource::File => {
2322 media_type_is(content_type, "multipart/form-data")
2323 }
2324 InputSource::Stream => unreachable!("streaming bodies do not require a media type"),
2325 InputSource::Path | InputSource::Query | InputSource::Header | InputSource::Cookie => {
2326 true
2327 }
2328 });
2329 if !valid_media_type {
2330 return Err(BodyRejection::UnsupportedMediaType);
2331 }
2332
2333 Ok(())
2334}
2335
2336fn media_type_is(value: &str, expected: &str) -> bool {
2337 value
2338 .split(';')
2339 .next()
2340 .is_some_and(|media_type| media_type.trim().eq_ignore_ascii_case(expected))
2341}
2342
2343fn outcome_result(outcome: ExecutionOutcome) -> Result<Response, Failure> {
2344 match outcome {
2345 ExecutionOutcome::Success {
2346 status,
2347 headers,
2348 body,
2349 background,
2350 } => {
2351 let response = match body {
2352 Some(body) => Response {
2353 status,
2354 headers: json_headers(),
2355 body,
2356 stream: None,
2357 upgrade: None,
2358 background: Vec::new(),
2359 },
2360 None => Response {
2361 status,
2362 headers: ResponseHeaders::empty(),
2363 body: Vec::new(),
2364 stream: None,
2365 upgrade: None,
2366 background: Vec::new(),
2367 },
2368 };
2369 let mut response = with_outcome_headers(response, headers);
2370 response.background = background;
2371 Ok(response)
2372 }
2373 ExecutionOutcome::StreamingSuccess {
2374 status,
2375 headers,
2376 body,
2377 background,
2378 } => Ok(with_outcome_headers(
2379 Response {
2380 status,
2381 headers: ResponseHeaders::empty(),
2382 body: Vec::new(),
2383 stream: Some(body),
2384 upgrade: None,
2385 background,
2386 },
2387 headers,
2388 )),
2389 ExecutionOutcome::Upgrade {
2390 upgrade,
2391 background,
2392 } => {
2393 let headers = upgrade.headers().to_vec();
2394 Ok(with_outcome_headers(
2395 Response {
2396 status: 101,
2397 headers: ResponseHeaders::empty(),
2398 body: Vec::new(),
2399 stream: None,
2400 upgrade: Some(upgrade),
2401 background,
2402 },
2403 headers,
2404 ))
2405 }
2406 ExecutionOutcome::Rejected {
2407 status,
2408 code,
2409 message,
2410 details,
2411 } => Err(Failure {
2412 source: HttpErrorSource::Rejection,
2413 status,
2414 code: Cow::Owned(code),
2415 message: Cow::Owned(message),
2416 details,
2417 headers: Vec::new(),
2418 }),
2419 ExecutionOutcome::DomainError(error) => {
2420 let details = match error.details {
2421 Some(details) => match blazingly_json::from_slice(&details) {
2422 Ok(details) => Some(details),
2423 Err(_) => return Err(internal_failure()),
2424 },
2425 None => None,
2426 };
2427 Err(Failure {
2428 source: HttpErrorSource::Domain,
2429 status: error.status,
2430 code: Cow::Owned(error.code),
2431 message: Cow::Owned(error.message),
2432 details,
2433 headers: error.headers,
2434 })
2435 }
2436 ExecutionOutcome::InternalError { .. } => Err(internal_failure()),
2437 }
2438}
2439
2440fn with_outcome_headers(mut response: Response, headers: Vec<ResponseHeader>) -> Response {
2441 for header in headers {
2442 response = response.with_header(header.name, header.value);
2443 }
2444 response
2445}
2446
2447fn error_response(status: u16, code: &str, message: &str, details: Option<Value>) -> Response {
2448 let mut error = json!({
2449 "error": {
2450 "code": code,
2451 "message": message,
2452 }
2453 });
2454 if let Some(details) = details {
2455 error["error"]["details"] = details;
2456 }
2457 json_response(status, &error)
2458}
2459
2460fn internal_failure() -> Failure {
2461 Failure::new(
2462 HttpErrorSource::Internal,
2463 500,
2464 "internal_error",
2465 "the operation could not be completed",
2466 )
2467}
2468
2469fn json_response(status: u16, value: &Value) -> Response {
2470 let Ok(body) = blazingly_json::to_vec(value) else {
2471 return Response {
2472 status: 500,
2473 headers: json_headers(),
2474 body: br#"{"error":{"code":"internal_error","message":"the operation could not be completed"}}"#
2475 .to_vec(),
2476 stream: None,
2477 upgrade: None,
2478 background: Vec::new(),
2479 };
2480 };
2481
2482 Response {
2483 status,
2484 headers: json_headers(),
2485 body,
2486 stream: None,
2487 upgrade: None,
2488 background: Vec::new(),
2489 }
2490}
2491
2492fn json_headers() -> ResponseHeaders {
2493 ResponseHeaders::one("content-type", "application/json")
2494}
2495
2496fn is_json_media_type(value: &str) -> bool {
2497 value
2498 .split(';')
2499 .next()
2500 .is_some_and(|media_type| media_type.trim().eq_ignore_ascii_case("application/json"))
2501}
2502
2503fn normalize_header_name(name: &str) -> String {
2504 name.to_ascii_lowercase()
2505}
2506
2507impl Response {
2508 #[must_use]
2509 pub fn with_header(mut self, name: impl AsRef<str>, value: impl Into<String>) -> Self {
2510 self.set_header(name, value);
2511 self
2512 }
2513}
2514
2515const INLINE_RESPONSE_HEADERS: usize = 4;
2516type OwnedHeader = (Cow<'static, str>, Cow<'static, str>);
2517
2518#[derive(Clone, Debug, Eq, PartialEq)]
2519struct ResponseHeaders {
2520 inline: [Option<OwnedHeader>; INLINE_RESPONSE_HEADERS],
2521 overflow: Vec<OwnedHeader>,
2522}
2523
2524impl ResponseHeaders {
2525 fn empty() -> Self {
2526 Self {
2527 inline: std::array::from_fn(|_| None),
2528 overflow: Vec::new(),
2529 }
2530 }
2531
2532 fn one(name: &'static str, value: &'static str) -> Self {
2533 let mut headers = Self::empty();
2534 headers.inline[0] = Some((Cow::Borrowed(name), Cow::Borrowed(value)));
2535 headers
2536 }
2537
2538 fn insert(&mut self, name: Cow<'static, str>, value: Cow<'static, str>) {
2539 if !name.eq_ignore_ascii_case("set-cookie") {
2542 if let Some((_, existing)) = self
2543 .inline
2544 .iter_mut()
2545 .filter_map(Option::as_mut)
2546 .chain(self.overflow.iter_mut())
2547 .find(|(existing, _)| existing.eq_ignore_ascii_case(&name))
2548 {
2549 *existing = value;
2550 return;
2551 }
2552 }
2553 if let Some(slot) = self.inline.iter_mut().find(|slot| slot.is_none()) {
2554 *slot = Some((name, value));
2555 } else {
2556 self.overflow.push((name, value));
2557 }
2558 }
2559
2560 fn remove(&mut self, name: &str) {
2561 for slot in &mut self.inline {
2562 if slot
2563 .as_ref()
2564 .is_some_and(|(existing, _)| existing.eq_ignore_ascii_case(name))
2565 {
2566 *slot = None;
2567 }
2568 }
2569 self.overflow
2570 .retain(|(existing, _)| !existing.eq_ignore_ascii_case(name));
2571 }
2572
2573 fn get(&self, name: &str) -> Option<&str> {
2574 self.inline
2575 .iter()
2576 .filter_map(Option::as_ref)
2577 .chain(&self.overflow)
2578 .find(|(header, _)| header.eq_ignore_ascii_case(name))
2579 .map(|(_, value)| value.as_ref())
2580 }
2581
2582 fn iter(&self) -> impl Iterator<Item = (&str, &str)> {
2583 self.inline
2584 .iter()
2585 .filter_map(Option::as_ref)
2586 .chain(&self.overflow)
2587 .map(|(name, value)| (name.as_ref(), value.as_ref()))
2588 }
2589}
2590
2591#[cfg(test)]
2592mod tests {
2593 use super::{
2594 BackgroundTasks, ConnectionInfo, HttpApp, HttpError, HttpErrorHandler, HttpErrorSource,
2595 HttpMiddleware, HttpRequestContext, MiddlewareScope, Request, Response, TestApp,
2596 };
2597 use blazingly_core::{
2598 HttpMethod, OperationDescriptor, OperationFailure, PreparedJson, ResponseDescriptor,
2599 SecurityLocation, SecurityRequirement, SecuritySchemeDescriptor, SecuritySchemeKind,
2600 TypeDescriptor,
2601 };
2602 use blazingly_executor::{
2603 ExecutableApp, ExecutableOperation, ExecutionOutcome, Extension, FromInvocation,
2604 InputRejection, InvocationControl, InvocationInput, OperationFuture, OperationOutput,
2605 };
2606 use blazingly_json::{Value, json};
2607 use futures_lite::future;
2608 use std::cell::{Cell, RefCell};
2609 use std::future::Future;
2610 use std::net::{IpAddr, Ipv4Addr, SocketAddrV4};
2611 use std::pin::Pin;
2612 use std::rc::Rc;
2613 use std::task::{Context, Poll};
2614
2615 struct PassthroughLayer;
2616
2617 impl HttpMiddleware for PassthroughLayer {}
2618
2619 struct AuditLayer;
2620
2621 impl HttpMiddleware for AuditLayer {
2622 fn verifies_security(&self) -> bool {
2623 false
2624 }
2625 }
2626
2627 struct NormalizingProxy;
2628
2629 impl HttpMiddleware for NormalizingProxy {
2630 fn on_request(&self, context: &mut HttpRequestContext<'_>) -> Option<Response> {
2631 context.set_client_ip(IpAddr::V4(Ipv4Addr::new(198, 51, 100, 7)));
2632 context.set_scheme("https");
2633 context.set_host("api.example");
2634 None
2635 }
2636 }
2637
2638 fn connection_operation(
2639 path: &str,
2640 id: &str,
2641 security: Vec<SecurityRequirement>,
2642 ) -> ExecutableOperation {
2643 let descriptor = OperationDescriptor::new(
2644 HttpMethod::Get,
2645 path,
2646 id,
2647 "Reports the normalized connection",
2648 None,
2649 vec![ResponseDescriptor::success(
2650 200,
2651 Some(TypeDescriptor::new("Connection")),
2652 )],
2653 )
2654 .expect("test operation id should be valid")
2655 .with_security(security);
2656 ExecutableOperation::typed(descriptor, |input| {
2657 let Extension(connection) =
2658 Extension::<ConnectionInfo>::from_invocation(&input, "connection", true)?;
2659 let body = json!({
2660 "scheme": connection.scheme(),
2661 "host": connection.host(),
2662 "clientIp": connection.client_ip().map(|address| address.to_string()),
2663 });
2664 Ok(Box::pin(async move {
2665 ExecutionOutcome::Success {
2666 status: 200,
2667 headers: Vec::new(),
2668 body: Some(blazingly_json::to_vec(&body).expect("connection body")),
2669 background: Vec::new(),
2670 }
2671 }) as OperationFuture)
2672 })
2673 }
2674
2675 fn executable() -> ExecutableApp {
2676 ExecutableApp::with_security_schemes(
2677 [
2678 connection_operation(
2679 "/secure",
2680 "http.secure",
2681 vec![SecurityRequirement::new("api_key")],
2682 ),
2683 connection_operation("/plain", "http.plain", Vec::new()),
2684 ],
2685 [SecuritySchemeDescriptor::new(
2686 "api_key",
2687 SecuritySchemeKind::ApiKey {
2688 location: SecurityLocation::Header,
2689 name: "x-api-key".to_owned(),
2690 },
2691 )],
2692 )
2693 .expect("secured operation graph should compile")
2694 }
2695
2696 fn error_code(response: &Response) -> String {
2697 let body: Value = response.json().expect("json error body");
2698 body["error"]["code"]
2699 .as_str()
2700 .expect("stable error code")
2701 .to_owned()
2702 }
2703
2704 fn prepared_operation() -> ExecutableOperation {
2708 let descriptor = OperationDescriptor::new(
2709 HttpMethod::Get,
2710 "/prepared",
2711 "http.prepared",
2712 "Encodes a borrowed view inside the operation",
2713 None,
2714 vec![ResponseDescriptor::success(
2715 200,
2716 Some(TypeDescriptor::new("Titles")),
2717 )],
2718 )
2719 .expect("test operation id should be valid");
2720 ExecutableOperation::typed(descriptor, |_| {
2721 Ok(Box::pin(async move {
2722 let owned = [String::from("first"), String::from("second")];
2723 let borrowed: Vec<&str> = owned.iter().map(String::as_str).collect();
2724 let body = PreparedJson::<TitlesSchema>::encode(&borrowed)
2725 .expect("the borrowed view encodes");
2726 OperationOutput::into_execution_outcome(body)
2727 }) as OperationFuture)
2728 })
2729 }
2730
2731 struct TitlesSchema;
2732
2733 impl blazingly_core::ApiSchema for TitlesSchema {
2734 fn type_descriptor() -> TypeDescriptor {
2735 TypeDescriptor::new("Titles")
2736 }
2737 }
2738
2739 #[test]
2740 fn a_prepared_body_reaches_the_wire_verbatim_as_json() {
2741 let executable = ExecutableApp::new([prepared_operation()])
2742 .expect("prepared operation graph should compile");
2743 let response = future::block_on(TestApp::new(&executable).call(Request::get("/prepared")));
2744
2745 assert_eq!(response.status(), 200);
2746 assert_eq!(
2747 response.get_header("content-type"),
2748 Some("application/json")
2749 );
2750 assert_eq!(response.body(), br#"["first","second"]"#);
2751 }
2752
2753 #[test]
2754 fn unlayered_dispatch_fails_closed_for_a_declared_scheme() {
2755 let executable = executable();
2756 let response = future::block_on(TestApp::new(&executable).call(Request::get("/secure")));
2757
2758 assert_eq!(response.status(), 500);
2759 assert_eq!(error_code(&response), "security_verifier_missing");
2760 }
2761
2762 #[test]
2763 fn owned_adapter_fails_closed_for_a_declared_scheme() {
2764 let app = HttpApp::new(executable());
2765 let response = future::block_on(app.call(Request::get("/secure")));
2766
2767 assert_eq!(response.status(), 500);
2768 assert_eq!(error_code(&response), "security_verifier_missing");
2769 }
2770
2771 #[test]
2772 fn unverified_scheme_opt_out_executes_the_operation() {
2773 let executable = executable();
2774 let response = future::block_on(
2775 TestApp::new(&executable)
2776 .with_unverified_security_schemes(true)
2777 .call(Request::get("/secure")),
2778 );
2779 assert_eq!(response.status(), 200);
2780
2781 let owned = HttpApp::new(executable).with_unverified_security_schemes(true);
2782 let response = future::block_on(owned.call(Request::get("/secure")));
2783 assert_eq!(response.status(), 200);
2784 }
2785
2786 #[test]
2787 fn routes_emission_lists_every_operation() {
2788 let executable = executable();
2789 let table = super::routes_table_text(executable.definition());
2790 assert!(table.starts_with("METHOD\tPATH\tOPERATION\tSUMMARY\n"));
2791 assert!(table.contains("GET\t/secure\thttp.secure\tReports the normalized connection\n"));
2792 assert!(table.contains("GET\t/plain\thttp.plain\tReports the normalized connection\n"));
2793 }
2794
2795 #[test]
2796 fn openapi_emission_serializes_the_default_document() {
2797 let executable = executable();
2798 let text = super::openapi_document_text(executable.definition())
2799 .expect("the OpenAPI document serializes");
2800 assert!(text.ends_with('\n'));
2801 let document: Value = blazingly_json::from_str(&text).expect("emitted JSON parses");
2802 assert_eq!(document["openapi"].as_str(), Some("3.1.0"));
2803 assert_eq!(
2804 document["paths"]["/secure"]["get"]["operationId"].as_str(),
2805 Some("http.secure")
2806 );
2807 assert_eq!(
2808 document["paths"]["/plain"]["get"]["operationId"].as_str(),
2809 Some("http.plain")
2810 );
2811 }
2812
2813 #[test]
2814 fn unsecured_operations_are_untouched_by_the_guard() {
2815 let executable = executable();
2816 let response = future::block_on(TestApp::new(&executable).call(Request::get("/plain")));
2817
2818 assert_eq!(response.status(), 200);
2819 }
2820
2821 #[test]
2822 fn layered_dispatch_runs_the_same_guard() {
2823 let executable = executable();
2824 let audited = future::block_on(
2825 TestApp::new(&executable)
2826 .with_middleware(AuditLayer)
2827 .call(Request::get("/secure")),
2828 );
2829 assert_eq!(audited.status(), 500);
2830 assert_eq!(error_code(&audited), "security_verifier_missing");
2831
2832 let verified = future::block_on(
2833 TestApp::new(&executable)
2834 .with_middleware(PassthroughLayer)
2835 .call(Request::get("/secure")),
2836 );
2837 assert_eq!(verified.status(), 200);
2838 }
2839
2840 type EventLog = Rc<RefCell<Vec<String>>>;
2841 type HandleExtractor = fn(&InvocationInput<'_>) -> Result<BackgroundTasks, InputRejection>;
2842
2843 struct RecordingLayer {
2844 name: &'static str,
2845 events: EventLog,
2846 }
2847
2848 impl HttpMiddleware for RecordingLayer {
2849 fn on_request(&self, _context: &mut HttpRequestContext<'_>) -> Option<Response> {
2850 self.events
2851 .borrow_mut()
2852 .push(format!("{}:request", self.name));
2853 None
2854 }
2855
2856 fn on_operation(
2857 &self,
2858 _context: &mut HttpRequestContext<'_>,
2859 operation: &OperationDescriptor,
2860 _security_schemes: &[SecuritySchemeDescriptor],
2861 ) -> Option<Response> {
2862 self.events.borrow_mut().push(format!(
2863 "{}:operation:{}",
2864 self.name,
2865 operation.contract.id.as_str()
2866 ));
2867 None
2868 }
2869
2870 fn on_response(
2871 &self,
2872 _context: &HttpRequestContext<'_>,
2873 _operation: Option<&OperationDescriptor>,
2874 _response: &mut Response,
2875 ) {
2876 self.events
2877 .borrow_mut()
2878 .push(format!("{}:response", self.name));
2879 }
2880 }
2881
2882 struct StatusStamp;
2883
2884 impl HttpMiddleware for StatusStamp {
2885 fn on_response(
2886 &self,
2887 _context: &HttpRequestContext<'_>,
2888 _operation: Option<&OperationDescriptor>,
2889 response: &mut Response,
2890 ) {
2891 let status = response.status().to_string();
2892 response.set_header("x-final-status", status);
2893 }
2894
2895 fn verifies_security(&self) -> bool {
2896 false
2897 }
2898 }
2899
2900 struct HouseStyle;
2901
2902 impl HttpErrorHandler for HouseStyle {
2903 fn on_error(&self, error: &HttpError<'_>, response: &mut Response) {
2904 response.set_header("x-error-source", format!("{:?}", error.source()));
2905 match error.source() {
2906 HttpErrorSource::Internal => {
2907 response.set_status(503);
2908 response.replace_body(br#"{"error":{"code":"unavailable"}}"#.to_vec());
2909 }
2910 HttpErrorSource::Domain => response.set_status(500),
2912 HttpErrorSource::Routing
2913 | HttpErrorSource::Request
2914 | HttpErrorSource::Rejection => {}
2915 }
2916 }
2917 }
2918
2919 fn scoped_executable() -> ExecutableApp {
2920 ExecutableApp::new([
2921 connection_operation("/ingest/events", "ingest.events", Vec::new()),
2922 connection_operation("/status", "status.read", Vec::new()),
2923 ])
2924 .expect("scoped operation graph should compile")
2925 }
2926
2927 fn ack_descriptor(path: &str, id: &str) -> OperationDescriptor {
2928 OperationDescriptor::new(
2929 HttpMethod::Get,
2930 path,
2931 id,
2932 "Reports an acknowledgement",
2933 None,
2934 vec![ResponseDescriptor::success(
2935 200,
2936 Some(TypeDescriptor::new("Ack")),
2937 )],
2938 )
2939 .expect("test operation id should be valid")
2940 }
2941
2942 fn outcome_operation(
2943 path: &str,
2944 id: &str,
2945 outcome: fn() -> ExecutionOutcome,
2946 ) -> ExecutableOperation {
2947 ExecutableOperation::typed(ack_descriptor(path, id), move |_| {
2948 Ok(Box::pin(async move { outcome() }) as OperationFuture)
2949 })
2950 }
2951
2952 fn extension_handle(input: &InvocationInput<'_>) -> Result<BackgroundTasks, InputRejection> {
2953 Extension::<BackgroundTasks>::from_invocation(input, "background", true)
2954 .map(|Extension(tasks)| tasks)
2955 }
2956
2957 fn bare_handle(input: &InvocationInput<'_>) -> Result<BackgroundTasks, InputRejection> {
2958 BackgroundTasks::from_invocation(input, "background", true)
2959 }
2960
2961 fn scheduling_operation(
2964 path: &str,
2965 id: &str,
2966 log: &EventLog,
2967 extract: HandleExtractor,
2968 outcome: fn() -> ExecutionOutcome,
2969 ) -> ExecutableOperation {
2970 let log = Rc::clone(log);
2971 ExecutableOperation::typed(ack_descriptor(path, id), move |input| {
2972 let tasks = extract(&input)?;
2973 let log = Rc::clone(&log);
2974 tasks.add_infallible(move || async move {
2975 log.borrow_mut().push("task".to_owned());
2976 });
2977 Ok(Box::pin(async move { outcome() }) as OperationFuture)
2978 })
2979 }
2980
2981 fn accepted() -> ExecutionOutcome {
2982 ExecutionOutcome::Success {
2983 status: 200,
2984 headers: Vec::new(),
2985 body: None,
2986 background: Vec::new(),
2987 }
2988 }
2989
2990 fn conflict() -> ExecutionOutcome {
2991 ExecutionOutcome::DomainError(OperationFailure::new(
2992 409,
2993 "conflict",
2994 "the event was already ingested",
2995 ))
2996 }
2997
2998 #[test]
2999 fn a_path_prefix_matches_only_on_segment_boundaries() {
3000 let scope = MiddlewareScope::prefix("/ingest");
3001 assert!(scope.matches_request("/ingest"));
3002 assert!(scope.matches_request("/ingest/events"));
3003 assert!(!scope.matches_request("/ingested"));
3004 assert!(!scope.matches_request("/"));
3005 assert!(MiddlewareScope::prefix("ingest/").matches_request("/ingest/events"));
3006 assert!(MiddlewareScope::all().is_global());
3007 assert!(!scope.is_global());
3008 }
3009
3010 #[test]
3011 fn operation_predicates_select_by_id_after_routing() {
3012 let scope = MiddlewareScope::all().with_operation_prefix("ingest.");
3013 assert!(scope.matches_operation("/anything", "ingest.events"));
3014 assert!(!scope.matches_operation("/anything", "status.read"));
3015 assert!(!scope.matches_request("/anything"));
3016
3017 let scope = MiddlewareScope::all()
3018 .with_operation_filter(|id| id.split('.').next_back() == Some("read"));
3019 assert!(scope.matches_operation("/anything", "status.read"));
3020 assert!(!scope.matches_operation("/anything", "status.write"));
3021
3022 let scope = MiddlewareScope::prefix("/ingest").with_operation("status.read");
3023 assert!(!scope.matches_operation("/status", "status.read"));
3024 }
3025
3026 #[test]
3027 fn a_prefix_scoped_layer_observes_only_its_subtree() {
3028 let executable = scoped_executable();
3029 let events: EventLog = Rc::new(RefCell::new(Vec::new()));
3030 let app = TestApp::new(&executable).with_scoped_middleware(
3031 MiddlewareScope::prefix("/ingest"),
3032 RecordingLayer {
3033 name: "ingest",
3034 events: Rc::clone(&events),
3035 },
3036 );
3037
3038 future::block_on(app.call(Request::get("/ingest/events")));
3039 let recorded = events.borrow().clone();
3040 assert_eq!(
3041 recorded,
3042 [
3043 "ingest:request",
3044 "ingest:operation:ingest.events",
3045 "ingest:response"
3046 ]
3047 );
3048
3049 events.borrow_mut().clear();
3050 future::block_on(app.call(Request::get("/status")));
3051 assert!(events.borrow().is_empty());
3052 }
3053
3054 #[test]
3055 fn an_operation_scoped_layer_starts_after_routing() {
3056 let executable = scoped_executable();
3057 let events: EventLog = Rc::new(RefCell::new(Vec::new()));
3058 let app = TestApp::new(&executable).with_scoped_middleware(
3059 MiddlewareScope::operation("ingest.events"),
3060 RecordingLayer {
3061 name: "op",
3062 events: Rc::clone(&events),
3063 },
3064 );
3065
3066 future::block_on(app.call(Request::get("/ingest/events")));
3067 let recorded = events.borrow().clone();
3068 assert_eq!(recorded, ["op:operation:ingest.events", "op:response"]);
3069
3070 events.borrow_mut().clear();
3071 future::block_on(app.call(Request::get("/status")));
3072 assert!(events.borrow().is_empty());
3073 }
3074
3075 #[test]
3076 fn a_scoped_verifier_does_not_cover_operations_outside_its_scope() {
3077 let executable = executable();
3078 let outside = future::block_on(
3079 TestApp::new(&executable)
3080 .with_scoped_middleware(MiddlewareScope::prefix("/other"), PassthroughLayer)
3081 .call(Request::get("/secure")),
3082 );
3083 assert_eq!(outside.status(), 500);
3084 assert_eq!(error_code(&outside), "security_verifier_missing");
3085
3086 let inside = future::block_on(
3087 TestApp::new(&executable)
3088 .with_scoped_middleware(MiddlewareScope::prefix("/secure"), PassthroughLayer)
3089 .call(Request::get("/secure")),
3090 );
3091 assert_eq!(inside.status(), 200);
3092 }
3093
3094 #[test]
3095 fn an_injected_handle_schedules_work_from_the_handler_body() {
3096 let log: EventLog = Rc::new(RefCell::new(Vec::new()));
3097 let executable = ExecutableApp::new([scheduling_operation(
3098 "/ingest",
3099 "tasks.ok",
3100 &log,
3101 extension_handle,
3102 accepted,
3103 )])
3104 .expect("scheduling operation graph should compile");
3105
3106 let mut response =
3107 future::block_on(TestApp::new(&executable).call(Request::get("/ingest")));
3108 assert_eq!(response.status(), 200);
3109 assert!(log.borrow().is_empty());
3110
3111 future::block_on(response.run_background()).expect("scheduled task");
3112 assert_eq!(log.borrow().clone(), ["task"]);
3113 }
3114
3115 #[test]
3116 fn scheduled_work_survives_a_failed_outcome() {
3117 let log: EventLog = Rc::new(RefCell::new(Vec::new()));
3118 let executable = ExecutableApp::new([scheduling_operation(
3119 "/ingest",
3120 "tasks.conflict",
3121 &log,
3122 bare_handle,
3123 conflict,
3124 )])
3125 .expect("scheduling operation graph should compile");
3126
3127 let mut response =
3128 future::block_on(TestApp::new(&executable).call(Request::get("/ingest")));
3129 assert_eq!(response.status(), 409);
3130 assert_eq!(error_code(&response), "conflict");
3131
3132 future::block_on(response.run_background()).expect("scheduled task");
3133 assert_eq!(log.borrow().clone(), ["task"]);
3134 }
3135
3136 struct ReadyWhenScheduled {
3139 scheduled: Rc<Cell<bool>>,
3140 }
3141
3142 impl Future for ReadyWhenScheduled {
3143 type Output = ();
3144
3145 fn poll(self: Pin<&mut Self>, _context: &mut Context<'_>) -> Poll<()> {
3146 if self.scheduled.get() {
3147 Poll::Ready(())
3148 } else {
3149 Poll::Pending
3150 }
3151 }
3152 }
3153
3154 fn aborting_operation(log: &EventLog, scheduled: &Rc<Cell<bool>>) -> ExecutableOperation {
3155 let log = Rc::clone(log);
3156 let scheduled = Rc::clone(scheduled);
3157 ExecutableOperation::typed(ack_descriptor("/ingest", "tasks.aborted"), move |input| {
3158 let tasks = bare_handle(&input)?;
3159 let log = Rc::clone(&log);
3160 tasks.add_infallible(move || async move {
3161 log.borrow_mut().push("task".to_owned());
3162 });
3163 scheduled.set(true);
3164 Ok(Box::pin(async move { accepted() }) as OperationFuture)
3165 })
3166 }
3167
3168 #[test]
3169 fn scheduled_work_survives_an_aborted_invocation() {
3170 let log: EventLog = Rc::new(RefCell::new(Vec::new()));
3171 let scheduled = Rc::new(Cell::new(false));
3172 let executable = ExecutableApp::new([aborting_operation(&log, &scheduled)])
3173 .expect("scheduling operation graph should compile");
3174 let control = InvocationControl::new().with_timeout(ReadyWhenScheduled {
3175 scheduled: Rc::clone(&scheduled),
3176 });
3177
3178 let mut response = future::block_on(
3179 TestApp::new(&executable).call_controlled(Request::get("/ingest"), control),
3180 );
3181 assert_eq!(response.status(), 504);
3182
3183 future::block_on(response.run_background()).expect("scheduled task");
3184 assert_eq!(log.borrow().clone(), ["task"]);
3185 }
3186
3187 #[test]
3188 fn an_unused_handle_leaves_the_response_untouched() {
3189 let executable = scoped_executable();
3190 let mut response =
3191 future::block_on(TestApp::new(&executable).call(Request::get("/status")));
3192
3193 assert_eq!(response.status(), 200);
3194 assert!(response.take_background_tasks().is_empty());
3195 }
3196
3197 #[test]
3198 fn an_error_handler_restyles_every_dispatch_failure() {
3199 let executable = executable();
3200 let app = TestApp::new(&executable).with_error_handler(HouseStyle);
3201
3202 let missing = future::block_on(app.call(Request::get("/nope")));
3203 assert_eq!(missing.status(), 404);
3204 assert_eq!(missing.get_header("x-error-source"), Some("Routing"));
3205
3206 let unverified = future::block_on(app.call(Request::get("/secure")));
3207 assert_eq!(unverified.status(), 503);
3208 assert_eq!(unverified.get_header("x-error-source"), Some("Internal"));
3209 assert_eq!(error_code(&unverified), "unavailable");
3210
3211 let allowed = future::block_on(app.call(Request::post("/plain")));
3212 assert_eq!(allowed.status(), 405);
3213 assert_eq!(allowed.get_header("allow"), Some("GET"));
3214 assert_eq!(allowed.get_header("x-error-source"), Some("Routing"));
3215 }
3216
3217 #[test]
3218 fn a_typed_domain_status_survives_the_error_handler() {
3219 let executable = ExecutableApp::new([
3220 outcome_operation("/conflict", "errors.conflict", conflict),
3221 outcome_operation("/broken", "errors.broken", || {
3222 ExecutionOutcome::InternalError {
3223 code: "boom".to_owned(),
3224 message: "boom".to_owned(),
3225 }
3226 }),
3227 ])
3228 .expect("error operation graph should compile");
3229 let app = TestApp::new(&executable).with_error_handler(HouseStyle);
3230
3231 let domain = future::block_on(app.call(Request::get("/conflict")));
3232 assert_eq!(domain.status(), 409);
3233 assert_eq!(error_code(&domain), "conflict");
3234 assert_eq!(domain.get_header("x-error-source"), Some("Domain"));
3235
3236 let internal = future::block_on(app.call(Request::get("/broken")));
3237 assert_eq!(internal.status(), 503);
3238 assert_eq!(error_code(&internal), "unavailable");
3239 }
3240
3241 #[test]
3242 fn error_handlers_run_before_middleware_sees_the_response() {
3243 let executable = executable();
3244 let response = future::block_on(
3245 TestApp::new(&executable)
3246 .with_error_handler(HouseStyle)
3247 .with_middleware(StatusStamp)
3248 .call(Request::get("/secure")),
3249 );
3250
3251 assert_eq!(response.status(), 503);
3252 assert_eq!(response.get_header("x-final-status"), Some("503"));
3253 }
3254
3255 #[test]
3256 fn the_owned_adapter_registers_the_same_seams() {
3257 let app = HttpApp::new(executable())
3258 .with_scoped_middleware(MiddlewareScope::prefix("/other"), PassthroughLayer)
3259 .with_error_handler(HouseStyle);
3260 let response = future::block_on(app.call(Request::get("/secure")));
3261
3262 assert_eq!(response.status(), 503);
3263 assert_eq!(response.get_header("x-error-source"), Some("Internal"));
3264 }
3265
3266 #[test]
3267 fn normalized_connection_reaches_the_operation_context() {
3268 let executable = executable();
3269 let request = || {
3270 Request::get("/plain")
3271 .peer_addr(SocketAddrV4::new(Ipv4Addr::LOCALHOST, 9000).into())
3272 .header("host", "internal:8080")
3273 };
3274
3275 let unlayered = future::block_on(TestApp::new(&executable).call(request()));
3276 let body: Value = unlayered.json().expect("connection body");
3277 assert_eq!(body["scheme"], "http");
3278 assert_eq!(body["host"], "internal:8080");
3279 assert_eq!(body["clientIp"], "127.0.0.1");
3280
3281 let layered = future::block_on(
3282 TestApp::new(&executable)
3283 .with_middleware(NormalizingProxy)
3284 .call(request()),
3285 );
3286 let body: Value = layered.json().expect("connection body");
3287 assert_eq!(body["scheme"], "https");
3288 assert_eq!(body["host"], "api.example");
3289 assert_eq!(body["clientIp"], "198.51.100.7");
3290 }
3291}