praxis_protocol/tcp/
mod.rs

1// SPDX-License-Identifier: MIT
2// Copyright (c) 2024 Praxis Contributors
3
4//! Raw TCP/L4 bidirectional forwarding protocol.
5
6use std::{sync::Arc, time::Duration};
7
8use arc_swap::ArcSwap;
9use pingora_core::services::listening::Service;
10use praxis_core::{ProxyError, config::Config};
11use praxis_filter::{FilterPipeline, FilterRegistry};
12use tokio::sync::{Semaphore, watch};
13
14use crate::{ListenerPipelines, Protocol};
15
16/// Bidirectional TCP proxy application.
17pub(crate) mod proxy;
18/// TLS configuration and listener grouping utilities.
19mod tls_setup;
20
21// -----------------------------------------------------------------------------
22// PingoraTcp
23// -----------------------------------------------------------------------------
24
25/// Pingora-backed raw TCP/L4 protocol implementation.
26///
27/// Groups TCP listeners by `(upstream address, idle timeout, max duration)`,
28/// creating one bidirectional forwarder per unique combination. Implements [`Protocol`].
29///
30/// [`Protocol`]: crate::Protocol
31pub struct PingoraTcp;
32
33#[expect(clippy::too_many_lines, reason = "linear registration with shutdown collection")]
34impl Protocol for PingoraTcp {
35    fn register(
36        self: Box<Self>,
37        server: &mut praxis_core::PingoraServerRuntime,
38        config: &Config,
39        pipelines: &ListenerPipelines,
40    ) -> Result<Vec<watch::Sender<bool>>, ProxyError> {
41        let groups = tls_setup::group_tcp_listeners(config);
42        tls_setup::validate_tcp_group_consistency(&groups)?;
43        #[expect(clippy::expect_used, reason = "empty pipeline is infallible")]
44        let fallback_pipeline = Arc::new(ArcSwap::from_pointee(
45            FilterPipeline::build(&mut [], &FilterRegistry::with_builtins()).expect("empty pipeline is valid"),
46        ));
47
48        let mut cert_watcher_shutdowns = Vec::new();
49
50        for ((upstream_opt, cluster_opt, timeout_ms, max_dur_secs), listeners) in groups {
51            let pipeline = listeners
52                .first()
53                .and_then(|l| pipelines.get(&l.name))
54                .map_or_else(|| Arc::clone(&fallback_pipeline), Arc::clone);
55
56            let session_timeout = timeout_ms.map(Duration::from_millis);
57            let max_duration = max_dur_secs.map(Duration::from_secs);
58            let service_name = match (upstream_opt.as_deref(), cluster_opt.as_deref()) {
59                (Some(addr), _) => format!("tcp-proxy:{addr}"),
60                (_, Some(cluster)) => format!("tcp-proxy:cluster:{cluster}"),
61                _ => "tcp-proxy:filter-routed".to_owned(),
62            };
63            let connection_semaphore = listeners
64                .first()
65                .and_then(|l| l.max_connections)
66                .map(|max| Arc::new(Semaphore::new(max as usize)));
67            let app = proxy::PingoraTcpProxy::new(
68                upstream_opt.clone(),
69                cluster_opt.map(Arc::from),
70                pipeline,
71                session_timeout,
72                max_duration,
73                connection_semaphore,
74            );
75            let mut service = Service::new(service_name, app);
76
77            cert_watcher_shutdowns.extend(tls_setup::register_tcp_listeners(
78                &mut service,
79                &listeners,
80                upstream_opt.as_deref(),
81            )?);
82            server.server_mut().add_service(service);
83        }
84
85        Ok(cert_watcher_shutdowns)
86    }
87}
praxis_protocol/tcp/mod.rs

praxis_protocol/tcp/
mod.rs
