proxima_primitives/pipe/
routing.rs

1use bytes::Bytes;
2use std::collections::BTreeSet;
3use std::sync::Arc;
4
5use crate::pipe::handler::PipeHandle;
6use crate::pipe::path_pattern::PathPattern;
7use crate::pipe::request::Request;
8
9/// Mount entry: pattern + handle + filters. Generic over the handle so
10/// `Mount<PipeHandle>` (default) carries a cross-thread pipe while
11/// `Mount<ThreadLocalPipeHandle>` carries a per-thread pipe.
12#[derive(Clone)]
13pub struct Mount<Handle = PipeHandle> {
14    pub path: PathPattern,
15    pub pipe: Handle,
16    pub methods: MethodFilter,
17    pub host: HostFilter,
18    pub label: Arc<[u8]>,
19    /// Mount-site label — set via [`Mount::named`], `App::mount*`'s
20    /// `MountTarget::Named(name)` arm, or defaulted to `"anonymous"`. This
21    /// is the mount-site's own name, distinct from `label` (the raw pattern
22    /// bytes): mounting the same handle at two paths gives it two
23    /// `pipe_name`s, one per mount-site.
24    pub pipe_name: Arc<str>,
25}
26
27// host matcher: optional set of host header values that must match.
28// nginx-style `server_name`. case-insensitive; `None` matches any.
29// missing Host header matches `None` only.
30#[derive(Debug, Clone, Default)]
31pub struct HostFilter {
32    allowed: Option<Vec<String>>,
33}
34
35impl HostFilter {
36    #[must_use]
37    pub fn any() -> Self {
38        Self { allowed: None }
39    }
40
41    #[must_use]
42    pub fn only<I: IntoIterator<Item = String>>(hosts: I) -> Self {
43        let collected: Vec<String> = hosts
44            .into_iter()
45            .map(|host| host.to_ascii_lowercase())
46            .collect();
47        if collected.is_empty() {
48            Self::any()
49        } else {
50            Self {
51                allowed: Some(collected),
52            }
53        }
54    }
55
56    #[must_use]
57    pub fn matches(&self, host_header: Option<&str>) -> bool {
58        let Some(allowed) = self.allowed.as_ref() else {
59            return true;
60        };
61        let Some(host) = host_header else {
62            return false;
63        };
64        // strip port suffix per RFC: "example.com:8080" -> "example.com"
65        let host_no_port = host
66            .split_once(':')
67            .map_or(host, |(name, _)| name)
68            .to_ascii_lowercase();
69        allowed.iter().any(|candidate| candidate == &host_no_port)
70    }
71}
72
73#[derive(Debug, Clone)]
74pub struct MethodFilter {
75    allowed: Option<BTreeSet<String>>,
76}
77
78impl MethodFilter {
79    #[must_use]
80    pub fn any() -> Self {
81        Self { allowed: None }
82    }
83
84    #[must_use]
85    pub fn only<I: IntoIterator<Item = String>>(methods: I) -> Self {
86        Self {
87            allowed: Some(
88                methods
89                    .into_iter()
90                    .map(|method| method.to_uppercase())
91                    .collect(),
92            ),
93        }
94    }
95
96    #[must_use]
97    pub fn matches(&self, method: &[u8]) -> bool {
98        match &self.allowed {
99            Some(set) => set
100                .iter()
101                .any(|allowed| allowed.as_bytes().eq_ignore_ascii_case(method)),
102            None => true,
103        }
104    }
105}
106
107// generic over `Handle`: `pipe_name` (not the erased handle) carries the
108// printable identity now, so the Debug impl no longer needs a concrete,
109// erasure-specific handle type.
110impl<Handle> std::fmt::Debug for Mount<Handle> {
111    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
112        formatter
113            .debug_struct("Mount")
114            .field("path", &self.path.raw())
115            .field("pipe", &self.pipe_name)
116            .field("methods", &self.methods)
117            .field("label", &self.label)
118            .finish()
119    }
120}
121
122impl<Handle> std::fmt::Debug for Router<Handle> {
123    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
124        formatter
125            .debug_struct("Router")
126            .field("mounts", &self.mounts)
127            .finish()
128    }
129}
130
131impl<Handle> Mount<Handle> {
132    #[must_use]
133    pub fn new(pattern: &str, pipe: Handle) -> Self {
134        Self {
135            path: PathPattern::parse(pattern),
136            pipe,
137            methods: MethodFilter::any(),
138            host: HostFilter::any(),
139            label: Arc::from(pattern.as_bytes()),
140            pipe_name: Arc::from("anonymous"),
141        }
142    }
143
144    #[must_use]
145    pub fn with_methods(mut self, methods: MethodFilter) -> Self {
146        self.methods = methods;
147        self
148    }
149
150    #[must_use]
151    pub fn with_host(mut self, host: HostFilter) -> Self {
152        self.host = host;
153        self
154    }
155
156    /// Set the mount-site label (default `"anonymous"`). `App::mount*`'s
157    /// `MountTarget::Named(name)` arm calls this; a `MountTarget::Handle(_)`
158    /// mount leaves the default.
159    #[must_use]
160    pub fn named(mut self, name: impl Into<Arc<str>>) -> Self {
161        self.pipe_name = name.into();
162        self
163    }
164
165    #[must_use]
166    pub fn matches(&self, request: &Request<Bytes>) -> bool {
167        self.matches_with_params(request).is_some()
168    }
169
170    /// Like `matches` but returns the extracted path-pattern parameters
171    /// when the mount accepts this request.
172    #[must_use]
173    pub fn matches_with_params(
174        &self,
175        request: &Request<Bytes>,
176    ) -> Option<std::collections::BTreeMap<String, String>> {
177        if !self.methods.matches(request.method.as_bytes()) {
178            return None;
179        }
180        let host_header = request
181            .metadata
182            .iter()
183            .find(|(name, _)| name.eq_ignore_ascii_case(b"host"))
184            .and_then(|(_, value)| std::str::from_utf8(value).ok());
185        if !self.host.matches(host_header) {
186            return None;
187        }
188        let path_view = std::str::from_utf8(&request.path).unwrap_or("");
189        self.path.matches(path_view)
190    }
191}
192
193#[derive(Clone)]
194pub struct Router<Handle = PipeHandle> {
195    mounts: Vec<Mount<Handle>>,
196}
197
198impl<Handle> Default for Router<Handle> {
199    fn default() -> Self {
200        Self { mounts: Vec::new() }
201    }
202}
203
204impl<Handle> Router<Handle> {
205    #[must_use]
206    pub fn new() -> Self {
207        Self::default()
208    }
209
210    #[must_use]
211    pub fn with_capacity(capacity: usize) -> Self {
212        Self {
213            mounts: Vec::with_capacity(capacity),
214        }
215    }
216
217    pub fn add(&mut self, mount: Mount<Handle>) {
218        self.mounts.push(mount);
219    }
220
221    pub fn remove(&mut self, pattern: &str) -> usize {
222        let before = self.mounts.len();
223        let needle = pattern.as_bytes();
224        self.mounts.retain(|mount| mount.label.as_ref() != needle);
225        before - self.mounts.len()
226    }
227
228    pub fn replace(&mut self, pattern: &str, mount: Mount<Handle>) -> bool {
229        let removed = self.remove(pattern);
230        self.mounts.push(mount);
231        removed > 0
232    }
233
234    #[must_use]
235    pub fn route(&self, request: &Request<Bytes>) -> Option<&Mount<Handle>> {
236        self.mounts.iter().find(|mount| mount.matches(request))
237    }
238
239    /// Find a matching mount and return the extracted path params.
240    pub fn route_with_params(
241        &self,
242        request: &Request<Bytes>,
243    ) -> Option<(&Mount<Handle>, std::collections::BTreeMap<String, String>)> {
244        for mount in &self.mounts {
245            if let Some(params) = mount.matches_with_params(request) {
246                return Some((mount, params));
247            }
248        }
249        None
250    }
251
252    #[must_use]
253    pub fn mounts(&self) -> &[Mount<Handle>] {
254        &self.mounts
255    }
256}
257
258use core::future::Future;
259
260/// Pattern-routed dispatcher. Generic over the handle:
261/// `RoutingPipe<PipeHandle>` impls [`Handler`](crate::pipe::handler::Handler);
262/// `RoutingPipe<ThreadLocalPipeHandle>` impls
263/// [`ThreadLocalHandler`](crate::pipe::handler::ThreadLocalHandler).
264pub struct RoutingPipe<Handle = PipeHandle> {
265    router: Router<Handle>,
266    label: String,
267    fallback: Option<Handle>,
268}
269
270impl<Handle> RoutingPipe<Handle> {
271    #[must_use]
272    pub fn new(label: impl Into<String>) -> Self {
273        Self {
274            router: Router::new(),
275            label: label.into(),
276            fallback: None,
277        }
278    }
279
280    #[must_use]
281    pub fn route(mut self, pattern: &str, pipe: Handle) -> Self {
282        self.router.add(Mount::new(pattern, pipe));
283        self
284    }
285
286    #[must_use]
287    pub fn route_with_methods(
288        mut self,
289        pattern: &str,
290        pipe: Handle,
291        methods: MethodFilter,
292    ) -> Self {
293        self.router
294            .add(Mount::new(pattern, pipe).with_methods(methods));
295        self
296    }
297
298    #[must_use]
299    pub fn fallback(mut self, pipe: Handle) -> Self {
300        self.fallback = Some(pipe);
301        self
302    }
303
304    /// The routing pipe's own label (set at construction, not a mount-site
305    /// name — this labels the `RoutingPipe` node itself, e.g. for tracing).
306    #[must_use]
307    pub fn label(&self) -> &str {
308        &self.label
309    }
310}
311
312impl<Handle> crate::pipe::primitives::SendPipe for RoutingPipe<Handle>
313where
314    Handle: crate::pipe::handler::Handler + Clone + Send + Sync + 'static,
315{
316    type In = Request<Bytes>;
317    type Out = crate::pipe::request::Response<Bytes>;
318    type Err = crate::pipe::ProximaError;
319
320    fn call(
321        &self,
322        request: Request<Bytes>,
323    ) -> impl Future<
324        Output = Result<crate::pipe::request::Response<Bytes>, crate::pipe::ProximaError>,
325    > + Send {
326        let routed = self.router.route(&request).map(|mount| mount.pipe.clone());
327        let fallback = self.fallback.clone();
328        async move {
329            if let Some(pipe) = routed {
330                return crate::pipe::primitives::SendPipe::call(&pipe, request).await;
331            }
332            if let Some(pipe) = fallback {
333                return crate::pipe::primitives::SendPipe::call(&pipe, request).await;
334            }
335            Ok(crate::pipe::request::Response::not_found().with_body("no route matched"))
336        }
337    }
338}
339
340impl crate::pipe::primitives::Pipe for RoutingPipe<crate::pipe::handler::ThreadLocalPipeHandle> {
341    type In = Request<Bytes>;
342    type Out = crate::pipe::request::Response<Bytes>;
343    type Err = crate::pipe::ProximaError;
344
345    fn call(
346        &self,
347        request: Request<Bytes>,
348    ) -> impl Future<Output = Result<crate::pipe::request::Response<Bytes>, crate::pipe::ProximaError>>
349    {
350        let routed = self.router.route(&request).map(|mount| mount.pipe.clone());
351        let fallback = self.fallback.clone();
352        async move {
353            if let Some(pipe) = routed {
354                return crate::pipe::primitives::Pipe::call(&pipe, request).await;
355            }
356            if let Some(pipe) = fallback {
357                return crate::pipe::primitives::Pipe::call(&pipe, request).await;
358            }
359            Ok(crate::pipe::request::Response::not_found().with_body("no route matched"))
360        }
361    }
362}
363
364// `#[proxima::test]` pulls in the `proxima` dev-dependency, which the
365// loom build keeps out of the graph (see
366// `[target.'cfg(not(loom))'.dev-dependencies]` in Cargo.toml); these
367// tests are unrelated to the Notify/watch loom protocol.
368#[cfg(all(test, not(loom)))]
369#[allow(clippy::unwrap_used, clippy::expect_used)]
370mod routing_pipe_tests {
371    use super::*;
372    use crate::pipe::SendPipe;
373    use crate::pipe::handler::into_handle;
374    use crate::pipe::request::Response;
375
376    struct Static(&'static str);
377
378    impl SendPipe for Static {
379        type In = Request<Bytes>;
380        type Out = Response<Bytes>;
381        type Err = crate::pipe::ProximaError;
382
383        fn call(
384            &self,
385            _request: Request<Bytes>,
386        ) -> impl Future<Output = Result<Response<Bytes>, crate::pipe::ProximaError>> + Send
387        {
388            let body = self.0;
389            async move { Ok(Response::ok(bytes::Bytes::from_static(body.as_bytes()))) }
390        }
391    }
392
393    fn build_request(method: &str, path: &str) -> Request<Bytes> {
394        Request::builder()
395            .method(method)
396            .path(path)
397            .build()
398            .expect("builder")
399    }
400
401    #[proxima::test]
402    async fn route_dispatches_to_matching_pipe() {
403        let svc = RoutingPipe::new("api")
404            .route("/users/{id}", into_handle(Static("users")))
405            .route("/posts/{id}", into_handle(Static("posts")));
406
407        let response = SendPipe::call(&svc, build_request("GET", "/users/42"))
408            .await
409            .expect("call");
410        let body = response.collect_body().await.expect("collect");
411        assert_eq!(&body[..], b"users");
412    }
413
414    #[proxima::test]
415    async fn unmatched_path_falls_through_to_fallback() {
416        let svc = RoutingPipe::new("api")
417            .route("/known", into_handle(Static("known")))
418            .fallback(into_handle(Static("fallback")));
419        let response = SendPipe::call(&svc, build_request("GET", "/anything-else"))
420            .await
421            .expect("call");
422        let body = response.collect_body().await.expect("collect");
423        assert_eq!(&body[..], b"fallback");
424    }
425
426    #[proxima::test]
427    async fn unmatched_path_without_fallback_returns_404() {
428        let svc = RoutingPipe::new("api").route("/known", into_handle(Static("known")));
429        let response = SendPipe::call(&svc, build_request("GET", "/missing"))
430            .await
431            .expect("call");
432        assert_eq!(response.status, 404);
433    }
434
435    #[proxima::test]
436    async fn method_filter_excludes_unmatched_method() {
437        let svc = RoutingPipe::new("api").route_with_methods(
438            "/users",
439            into_handle(Static("users")),
440            MethodFilter::only(["GET".into()]),
441        );
442        let response = SendPipe::call(&svc, build_request("DELETE", "/users"))
443            .await
444            .expect("call");
445        assert_eq!(response.status, 404);
446    }
447}
448
449// `#[proxima::test]` pulls in the `proxima` dev-dependency, which the
450// loom build keeps out of the graph (see
451// `[target.'cfg(not(loom))'.dev-dependencies]` in Cargo.toml); these
452// tests are unrelated to the Notify/watch loom protocol.
453#[cfg(all(test, not(loom)))]
454#[allow(clippy::unwrap_used, clippy::expect_used)]
455mod tests {
456    use super::*;
457    use crate::pipe::ProximaError;
458    use crate::pipe::SendPipe;
459    use crate::pipe::handler::into_handle;
460    use crate::pipe::request::Response;
461    use std::future::Future;
462
463    struct StubPipe(&'static str);
464
465    impl SendPipe for StubPipe {
466        type In = Request<Bytes>;
467        type Out = Response<Bytes>;
468        type Err = ProximaError;
469
470        fn call(
471            &self,
472            _request: Request<Bytes>,
473        ) -> impl Future<Output = Result<Response<Bytes>, ProximaError>> + Send {
474            let body = self.0;
475            async move { Ok(Response::ok(body)) }
476        }
477    }
478
479    fn build_request(method: &str, path: &str) -> Request<Bytes> {
480        Request::builder()
481            .method(method)
482            .path(path)
483            .build()
484            .expect("builder should succeed")
485    }
486
487    #[test]
488    fn router_returns_first_matching_mount() {
489        let mut router = Router::new();
490        router.add(Mount::new("/users/{id}", into_handle(StubPipe("users"))).named("users"));
491        router.add(Mount::new("/posts/{id}", into_handle(StubPipe("posts"))).named("posts"));
492        let request = build_request("GET", "/users/42");
493        let mount = router.route(&request).expect("should match");
494        assert_eq!(mount.pipe_name.as_ref(), "users");
495    }
496
497    #[test]
498    fn method_filter_excludes_unlisted() {
499        let mut router = Router::new();
500        let mount = Mount::new("/users", into_handle(StubPipe("users")))
501            .with_methods(MethodFilter::only(["GET".into(), "POST".into()]));
502        router.add(mount);
503        assert!(router.route(&build_request("GET", "/users")).is_some());
504        assert!(router.route(&build_request("POST", "/users")).is_some());
505        assert!(router.route(&build_request("DELETE", "/users")).is_none());
506    }
507
508    #[test]
509    fn no_match_returns_none() {
510        let router: Router = Router::new();
511        assert!(router.route(&build_request("GET", "/missing")).is_none());
512    }
513
514    #[test]
515    fn unnamed_mount_defaults_to_anonymous() {
516        let mount = Mount::new("/x", into_handle(StubPipe("x")));
517        assert_eq!(mount.pipe_name.as_ref(), "anonymous");
518    }
519}
proxima_primitives/pipe/routing.rs

proxima_primitives/pipe/
routing.rs
