From 411b784fba9a91c1b5dd18b93d3f8adcb29928df Mon Sep 17 00:00:00 2001 From: Serra Date: Fri, 6 Nov 2026 14:24:18 -0800 Subject: [PATCH 1/4] Adds handler settings struct built from the Rust settings loader --- src/core/settings.rs | 27 +++ src/handler/mod.rs | 1 + src/handler/settings.rs | 453 ++++++++++++++++++++++++++++++++++++++++ 3 files changed, 481 insertions(+) create mode 100644 src/handler/settings.rs diff --git a/src/core/settings.rs b/src/core/settings.rs index 2a6624e9b..d19ce00c9 100644 --- a/src/core/settings.rs +++ b/src/core/settings.rs @@ -49,6 +49,9 @@ pub enum SettingsError { #[error("invalid tcp address: {0}")] InvalidTcpAddress(String), + + #[error("must set {0}")] + Missing(&'static str), } pub struct Settings { @@ -306,6 +309,30 @@ impl Settings { } } +pub struct CommonSettings { + pub services: Vec, + pub proxy_workers: u32, + pub client_maxconn: u32, + pub stats_connection_send: bool, + pub stats_connection_ttl: u32, +} + +impl CommonSettings { + pub fn from_settings(settings: &Settings) -> Result { + Ok(Self { + services: settings.list("runner/services")?, + proxy_workers: settings.number("proxy/workers")?.unwrap_or(1), + client_maxconn: settings.number("runner/client_maxconn")?.unwrap_or(50000), + stats_connection_send: settings + .bool("global/stats_connection_send")? + .unwrap_or(true), + stats_connection_ttl: settings + .number("global/stats_connection_ttl")? + .unwrap_or(120), + }) + } +} + #[cfg(test)] mod tests { use super::*; diff --git a/src/handler/mod.rs b/src/handler/mod.rs index abfa32fb9..9bfbb68c8 100644 --- a/src/handler/mod.rs +++ b/src/handler/mod.rs @@ -15,6 +15,7 @@ */ pub mod cliargs; +pub mod settings; #[cfg(test)] mod tests { diff --git a/src/handler/settings.rs b/src/handler/settings.rs new file mode 100644 index 000000000..b83ec09d4 --- /dev/null +++ b/src/handler/settings.rs @@ -0,0 +1,453 @@ +/* + * Copyright (C) 2026 Fastly, Inc. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +use crate::core::settings::{CommonSettings, Settings, SettingsError}; +use std::net::IpAddr; + +const DEFAULT_HTTP_MAX_HEADERS_SIZE: u32 = 10000; +const DEFAULT_HTTP_MAX_BODY_SIZE: u32 = 1000000; + +pub struct HandlerSettings { + pub common: CommonSettings, + pub server_in_stream_specs: Vec, + pub server_out_specs: Vec, + pub client_out_specs: Vec, + pub client_out_stream_specs: Vec, + pub client_in_specs: Vec, + pub inspect_specs: Vec, + pub accept_specs: Vec, + pub retry_out_specs: Vec, + pub ws_control_init_specs: Vec, + pub ws_control_stream_specs: Vec, + pub stats_spec: Option, + pub command_spec: Option, + pub state_spec: Option, + pub proxy_stats_specs: Vec, + pub proxy_command_spec: Option, + pub push_in_spec: Option, + pub push_in_sub_specs: Vec, + pub push_in_sub_connect: bool, + pub push_in_http_addr: Option, + pub push_in_http_port: Option, + pub push_in_http_max_headers_size: u32, + pub push_in_http_max_body_size: u32, + pub ipc_file_mode: Option, + pub share_all: bool, + pub message_rate: Option, + pub message_hwm: Option, + pub message_block_size: Option, + pub message_wait: u32, + pub id_cache_ttl: u32, + pub update_on_first_subscription: bool, + pub connection_subscription_max: u32, + pub subscription_linger: u32, + pub stats_subscription_ttl: u32, + pub stats_report_interval: u32, + pub stats_format: Option, + pub prometheus_port: Option, + pub prometheus_prefix: Option, +} + +impl HandlerSettings { + pub fn from_settings(settings: &Settings) -> Result { + let common = CommonSettings::from_settings(settings)?; + let workers = common.proxy_workers; + let has_service = |name: &str| common.services.iter().any(|s| s == name); + + let mut connmgr_in_stream_specs = settings.list("proxy/connmgr_in_stream_specs")?; + connmgr_in_stream_specs.extend(settings.list("proxy/condure_in_stream_specs")?); + + let mut connmgr_out_specs = settings.list("proxy/connmgr_out_specs")?; + connmgr_out_specs.extend(settings.list("proxy/condure_out_specs")?); + + let m2a_in_stream_specs = settings.list("handler/m2a_in_stream_specs")?; + let m2a_out_specs = settings.list("handler/m2a_out_specs")?; + + if m2a_in_stream_specs.is_empty() || m2a_out_specs.is_empty() { + return Err(SettingsError::Missing( + "m2a_in_stream_specs and m2a_out_specs", + )); + } + + let inspect_specs = combined_specs( + settings, + "handler/proxy_inspect_specs", + "handler/proxy_inspect_spec", + )?; + let accept_specs = combined_specs( + settings, + "handler/proxy_accept_specs", + "handler/proxy_accept_spec", + )?; + let retry_out_specs = combined_specs( + settings, + "handler/proxy_retry_out_specs", + "handler/proxy_retry_out_spec", + )?; + + if inspect_specs.is_empty() || accept_specs.is_empty() || retry_out_specs.is_empty() { + return Err(SettingsError::Missing( + "proxy_inspect_specs, proxy_accept_specs, and proxy_retry_out_specs", + )); + } + + let (server_in_stream_specs, server_out_specs) = if !has_service("mongrel2") + && (!connmgr_in_stream_specs.is_empty() || !connmgr_out_specs.is_empty()) + { + (connmgr_in_stream_specs, connmgr_out_specs) + } else { + (m2a_in_stream_specs, m2a_out_specs) + }; + + let push_in_http_addr = settings.ip_addr("handler/push_in_http_addr")?; + let push_in_http_port = settings.adjust_port("handler/push_in_http_port")?; + + // Only serve the API here if the api component is not enabled + let (push_in_http_addr, push_in_http_port) = if has_service("api") { + (None, None) + } else { + (push_in_http_addr, push_in_http_port) + }; + + Ok(Self { + server_in_stream_specs, + server_out_specs, + client_out_specs: expand_specs( + settings.list("handler/proxy_intreq_out_specs")?, + workers, + ), + client_out_stream_specs: expand_specs( + settings.list("handler/proxy_intreq_out_stream_specs")?, + workers, + ), + client_in_specs: expand_specs(settings.list("handler/proxy_intreq_in_specs")?, workers), + inspect_specs: expand_specs(inspect_specs, workers), + accept_specs: expand_specs(accept_specs, workers), + retry_out_specs: expand_specs(retry_out_specs, workers), + ws_control_init_specs: expand_specs( + settings.list("handler/proxy_ws_control_init_specs")?, + workers, + ), + ws_control_stream_specs: expand_specs( + settings.list("handler/proxy_ws_control_stream_specs")?, + workers, + ), + stats_spec: settings.string("handler/stats_spec")?, + command_spec: settings.string("handler/command_spec")?, + state_spec: settings.string("handler/state_spec")?, + proxy_stats_specs: expand_specs( + combined_specs( + settings, + "handler/proxy_stats_specs", + "handler/proxy_stats_spec", + )?, + workers, + ), + proxy_command_spec: settings.string("handler/proxy_command_spec")?.map(|spec| { + if spec.starts_with("ipc:") && workers > 1 { + format!("{spec}-0") + } else { + spec + } + }), + push_in_spec: settings.string("handler/push_in_spec")?, + push_in_sub_specs: combined_specs( + settings, + "handler/push_in_sub_specs", + "handler/push_in_sub_spec", + )?, + push_in_sub_connect: settings + .bool("handler/push_in_sub_connect")? + .unwrap_or(false), + push_in_http_addr, + push_in_http_port, + push_in_http_max_headers_size: settings + .number("handler/push_in_http_max_headers_size")? + .unwrap_or(DEFAULT_HTTP_MAX_HEADERS_SIZE), + push_in_http_max_body_size: settings + .number("handler/push_in_http_max_body_size")? + .unwrap_or(DEFAULT_HTTP_MAX_BODY_SIZE), + ipc_file_mode: settings.octal("handler/ipc_file_mode")?, + share_all: settings.bool("handler/share_all")?.unwrap_or(false), + message_rate: settings.number("handler/message_rate")?, + message_hwm: settings.number("handler/message_hwm")?, + message_block_size: settings.number("handler/message_block_size")?, + message_wait: settings.number("handler/message_wait")?.unwrap_or(5000), + id_cache_ttl: settings.number("handler/id_cache_ttl")?.unwrap_or(0), + update_on_first_subscription: settings + .bool("handler/update_on_first_subscription")? + .unwrap_or(true), + connection_subscription_max: settings + .number("handler/connection_subscription_max")? + .unwrap_or(20), + subscription_linger: settings + .number("handler/subscription_linger")? + .unwrap_or(60), + stats_subscription_ttl: settings + .number("handler/stats_subscription_ttl")? + .unwrap_or(60), + stats_report_interval: settings + .number("handler/stats_report_interval")? + .unwrap_or(10), + stats_format: settings.string("handler/stats_format")?, + prometheus_port: settings.string("handler/prometheus_port")?, + prometheus_prefix: settings.string("handler/prometheus_prefix")?, + common, + }) + } +} + +fn combined_specs( + settings: &Settings, + list_key: &str, + single_key: &str, +) -> Result, SettingsError> { + let mut specs = settings.list(list_key)?; + specs.extend(settings.string(single_key)?); + + Ok(specs) +} + +fn expand_specs(specs: Vec, workers: u32) -> Vec { + match specs.as_slice() { + [base] if base.starts_with("ipc:") && workers > 1 => { + (0..workers).map(|i| format!("{base}-{i}")).collect() + } + _ => specs, + } +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::core::test_dir; + use std::fs; + use std::net::Ipv4Addr; + use std::path::Path; + + fn load_conf(test: &str, contents: &str) -> Result { + let dir = test_dir().join("handler-settings").join(test); + fs::create_dir_all(&dir).unwrap(); + + let path = dir.join("pushpin.conf"); + fs::write(&path, contents).unwrap(); + + HandlerSettings::from_settings(&Settings::load(&path)?) + } + + const GLOBAL: &str = "[global]\ninclude=internal.conf\nrundir=run\n"; + + #[test] + fn example_config() { + let settings = Settings::load(Path::new("examples/config/pushpin.conf")).unwrap(); + let s = HandlerSettings::from_settings(&settings).unwrap(); + + assert_eq!( + s.server_in_stream_specs, + vec!["ipc://run/pushpin-connmgr-out-stream"] + ); + assert_eq!(s.server_out_specs, vec!["ipc://run/pushpin-connmgr-in"]); + assert_eq!(s.client_out_specs, vec!["ipc://run/pushpin-intreq-in"]); + assert_eq!(s.inspect_specs, vec!["ipc://run/pushpin-inspect"]); + assert_eq!(s.accept_specs, vec!["ipc://run/pushpin-accept"]); + assert_eq!(s.retry_out_specs, vec!["ipc://run/pushpin-retry"]); + assert_eq!(s.proxy_stats_specs, vec!["ipc://run/pushpin-proxy-stats"]); + assert_eq!( + s.proxy_command_spec.as_deref(), + Some("ipc://run/pushpin-proxy-command") + ); + assert_eq!(s.stats_spec.as_deref(), Some("ipc://run/pushpin-stats")); + assert_eq!(s.command_spec.as_deref(), Some("tcp://127.0.0.1:5563")); + assert_eq!(s.state_spec, None); + assert_eq!(s.push_in_spec.as_deref(), Some("tcp://127.0.0.1:5560")); + assert_eq!(s.push_in_sub_specs, vec!["tcp://127.0.0.1:5562"]); + assert!(!s.push_in_sub_connect); + assert_eq!( + s.push_in_http_addr, + Some(IpAddr::V4(Ipv4Addr::new(127, 0, 0, 1))) + ); + assert_eq!(s.push_in_http_port, Some(5561)); + assert_eq!(s.ipc_file_mode, None); + assert_eq!(s.message_rate, Some(2500)); + assert_eq!(s.message_hwm, Some(25000)); + assert_eq!(s.message_block_size, None); + assert_eq!(s.id_cache_ttl, 60); + assert_eq!(s.stats_format.as_deref(), Some("tnetstring")); + assert_eq!(s.common.services, vec!["connmgr", "proxy", "handler"]); + assert_eq!(s.common.client_maxconn, 50000); + assert_eq!(s.common.stats_connection_ttl, 120); + } + + #[test] + fn defaults_when_unset() { + let s = load_conf("defaults", GLOBAL).unwrap(); + + assert_eq!(s.common.proxy_workers, 1); + assert_eq!(s.common.client_maxconn, 50000); + assert!(s.common.stats_connection_send); + assert_eq!(s.common.stats_connection_ttl, 120); + assert_eq!(s.push_in_http_port, None); + assert_eq!(s.push_in_http_max_headers_size, 10000); + assert_eq!(s.push_in_http_max_body_size, 1000000); + assert_eq!(s.message_rate, None); + assert_eq!(s.message_wait, 5000); + assert_eq!(s.id_cache_ttl, 0); + assert!(s.update_on_first_subscription); + assert_eq!(s.connection_subscription_max, 20); + assert_eq!(s.subscription_linger, 60); + assert_eq!(s.stats_subscription_ttl, 60); + assert_eq!(s.stats_report_interval, 10); + } + + #[test] + fn specs_expanded_per_proxy_worker() { + let s = load_conf("workers", &format!("{GLOBAL}[proxy]\nworkers=3\n")).unwrap(); + + assert_eq!( + s.inspect_specs, + vec![ + "ipc://run/pushpin-inspect-0", + "ipc://run/pushpin-inspect-1", + "ipc://run/pushpin-inspect-2" + ] + ); + assert_eq!(s.client_out_specs.len(), 3); + assert_eq!(s.ws_control_init_specs.len(), 3); + assert_eq!( + s.proxy_command_spec.as_deref(), + Some("ipc://run/pushpin-proxy-command-0") + ); + assert_eq!( + s.server_in_stream_specs, + vec!["ipc://run/pushpin-connmgr-out-stream"] + ); + } + + #[test] + fn condure_specs_merged() { + let s = load_conf( + "condure", + &format!("{GLOBAL}[proxy]\ncondure_in_stream_specs=ipc://run/condure-out-stream\n"), + ) + .unwrap(); + + assert_eq!( + s.server_in_stream_specs, + vec![ + "ipc://run/pushpin-connmgr-out-stream", + "ipc://run/condure-out-stream" + ] + ); + } + + #[test] + fn single_spec_appended() { + let s = load_conf( + "single_spec", + &format!("{GLOBAL}[handler]\nproxy_inspect_spec=ipc://run/extra-inspect\n"), + ) + .unwrap(); + + assert_eq!( + s.inspect_specs, + vec!["ipc://run/pushpin-inspect", "ipc://run/extra-inspect"] + ); + } + + #[test] + fn mongrel2_uses_m2a_specs() { + let s = load_conf( + "mongrel2", + &format!("{GLOBAL}[runner]\nservices=mongrel2,m2adapter,proxy,handler\n"), + ) + .unwrap(); + + assert_eq!( + s.server_in_stream_specs, + vec![ + "ipc://run/pushpin-m2zhttp-out-stream", + "ipc://run/pushpin-m2zws-out-stream" + ] + ); + assert_eq!( + s.server_out_specs, + vec!["ipc://run/pushpin-m2zhttp-in", "ipc://run/pushpin-m2zws-in"] + ); + } + + #[test] + fn api_service_disables_push_http() { + let s = load_conf( + "api", + &format!( + "{GLOBAL}[runner]\nservices=connmgr,proxy,handler,api\n[handler]\npush_in_http_addr=127.0.0.1\npush_in_http_port=5561\n" + ), + ) + .unwrap(); + + assert_eq!(s.push_in_http_addr, None); + assert_eq!(s.push_in_http_port, None); + } + + #[test] + fn missing_m2a_specs_errors() { + let r = load_conf("missing_m2a", "[global]\nrundir=run\n"); + assert!( + matches!(r, Err(SettingsError::Missing(m)) if m.starts_with("m2a_in_stream_specs")) + ); + } + + #[test] + fn missing_proxy_specs_errors() { + let r = load_conf( + "missing_proxy", + "[global]\nrundir=run\n[handler]\nm2a_in_stream_specs=ipc://run/a\nm2a_out_specs=ipc://run/b\n", + ); + assert!( + matches!(r, Err(SettingsError::Missing(m)) if m.starts_with("proxy_inspect_specs")) + ); + } + + #[test] + fn invalid_push_in_http_addr_errors() { + let r = load_conf( + "bad_addr", + &format!("{GLOBAL}[handler]\npush_in_http_addr=127.0.0.l\n"), + ); + assert!(matches!( + r, + Err(SettingsError::InvalidValue { key, .. }) if key == "handler/push_in_http_addr" + )); + } + + #[test] + fn invalid_ipc_file_mode_errors() { + let r = load_conf( + "bad_mode", + &format!("{GLOBAL}[handler]\nipc_file_mode=abc\n"), + ); + assert!(matches!( + r, + Err(SettingsError::InvalidValue { key, .. }) if key == "handler/ipc_file_mode" + )); + } + + #[test] + fn ipc_file_mode_octal() { + let s = load_conf("mode", &format!("{GLOBAL}[handler]\nipc_file_mode=777\n")).unwrap(); + assert_eq!(s.ipc_file_mode, Some(0o777)); + } +} From ba19bda3ff2a79806dd544149afb05dcbe4240c9 Mon Sep 17 00:00:00 2001 From: Serra Date: Fri, 6 Nov 2026 14:25:31 -0800 Subject: [PATCH 2/4] Moves handler logging setup from C++ to the Rust main --- src/bin/pushpin-handler.rs | 22 +++++++++++++++++++++- src/handler/handlerapp.cpp | 10 ---------- 2 files changed, 21 insertions(+), 11 deletions(-) diff --git a/src/bin/pushpin-handler.rs b/src/bin/pushpin-handler.rs index 816731519..b3e34c6f5 100644 --- a/src/bin/pushpin-handler.rs +++ b/src/bin/pushpin-handler.rs @@ -1,5 +1,5 @@ /* - * Copyright (C) 2023-2025 Fastly, Inc. + * Copyright (C) 2023-2026 Fastly, Inc. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -15,10 +15,14 @@ */ use clap::Parser; +use log::debug; +use pushpin::core::log::ffi::{log_init, log_set_level}; use pushpin::handler::cliargs::ffi::HandlerCliArgs; use pushpin::handler::cliargs::{destroy_handler_cli_args, CliArgs}; use pushpin::import_cpp; +use std::ffi::CString; use std::process::ExitCode; +use std::ptr; import_cpp! { fn handler_init(args: *const HandlerCliArgs) -> libc::c_int; @@ -26,6 +30,22 @@ import_cpp! { fn main() -> ExitCode { let cli_args = CliArgs::parse().verify(); + + let log_file = cli_args + .log_file + .as_deref() + .filter(|f| !f.is_empty()) + .map(|f| CString::new(f).unwrap()); + + // SAFETY: the pointer is either null or a valid C string + if unsafe { log_init(log_file.as_ref().map_or(ptr::null(), |f| f.as_ptr())) } != 0 { + return ExitCode::from(1); + } + + log_set_level(cli_args.log_level as i32); + + debug!("starting..."); + let cli_args_ffi = cli_args.to_ffi(); let exit_code = unsafe { handler_init(&cli_args_ffi) }; diff --git a/src/handler/handlerapp.cpp b/src/handler/handlerapp.cpp index 13d174906..682fcb5dd 100644 --- a/src/handler/handlerapp.cpp +++ b/src/handler/handlerapp.cpp @@ -159,16 +159,6 @@ extern "C" { int handler_init(const ffi::HandlerCliArgs *argsFfi) { HandlerArgsData args(argsFfi); - if (!log_init(args.logFile)) - return 1; - - if (args.logLevel != -1) - log_setOutputLevel(args.logLevel); - else - log_setOutputLevel(LOG_LEVEL_INFO); - - log_debug("starting..."); - // QSettings doesn't inform us if the config file can't be opened, so do that ourselves { QFile file(args.configFile); From 4954eac19ae46b4141808d85ab70e1ce72a3aec8 Mon Sep 17 00:00:00 2001 From: Serra Date: Fri, 6 Nov 2026 14:28:23 -0800 Subject: [PATCH 3/4] Switches handler to settings from the Rust loader over cxx --- Cargo.lock | 121 ++++++++++++++++- Cargo.toml | 2 + build.rs | 10 ++ src/bin/pushpin-handler.rs | 41 +++++- src/handler/handler.pri | 3 +- src/handler/handlerapp.cpp | 262 ++++++++----------------------------- src/handler/handlerinit.h | 31 +++++ src/handler/settings.rs | 111 ++++++++++++++++ 8 files changed, 359 insertions(+), 222 deletions(-) create mode 100644 src/handler/handlerinit.h diff --git a/Cargo.lock b/Cargo.lock index f1517c8c2..c20d1c9df 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -233,6 +233,17 @@ version = "1.1.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "1c133bc6a41be0d194c306b5506d15e6feeea7b1d6604bd3f8310dfb2ca96486" +[[package]] +name = "codespan-reporting" +version = "0.13.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "af491d569909a7e4dee0ad7db7f5341fef5c614d5b8ec8cf765732aba3cff681" +dependencies = [ + "serde", + "termcolor", + "unicode-width", +] + [[package]] name = "colorchoice" version = "1.0.0" @@ -373,6 +384,68 @@ dependencies = [ "typenum", ] +[[package]] +name = "cxx" +version = "1.0.198" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6fe442a792c7c736eea18b32a7f8a3b63cf8aafabda6760042dc2fdeda456291" +dependencies = [ + "cc", + "cxx-build", + "cxxbridge-cmd", + "cxxbridge-flags", + "cxxbridge-macro", + "foldhash", + "link-cplusplus", +] + +[[package]] +name = "cxx-build" +version = "1.0.198" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e3184a94384c663718698311a78a51ac00c484c10b4eeac06fb0a068c5f64fa2" +dependencies = [ + "cc", + "codespan-reporting", + "indexmap", + "proc-macro2", + "quote", + "scratch", + "syn 3.0.5", +] + +[[package]] +name = "cxxbridge-cmd" +version = "1.0.198" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0148d8fd1199329ddf1d157a5e134e51ceff37c6a7ddd38615c399d81cb05d8d" +dependencies = [ + "clap", + "codespan-reporting", + "indexmap", + "proc-macro2", + "quote", + "syn 3.0.5", +] + +[[package]] +name = "cxxbridge-flags" +version = "1.0.198" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "52850339faed2eaadd24e286dc1d8268cc6f8a7bd9524d713adc9099566b4c89" + +[[package]] +name = "cxxbridge-macro" +version = "1.0.198" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2c77c856545d886c9bd5215409ebb63b925e262135248b50c79e5a5f194ee47c" +dependencies = [ + "indexmap", + "proc-macro2", + "quote", + "syn 3.0.5", +] + [[package]] name = "deranged" version = "0.4.0" @@ -467,6 +540,12 @@ version = "1.0.7" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "3f9eec918d3f24069decb9af1554cad7c880e2da24a9afd88aca000531ab82c1" +[[package]] +name = "foldhash" +version = "0.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "77ce24cb58228fbb8aa041425bb1050850ac19177686ea6e0f41a70416f56fdb" + [[package]] name = "foreign-types" version = "0.3.2" @@ -540,9 +619,9 @@ checksum = "e5274423e17b7c9fc20b6e7e208532f9b19825d82dfd615708b70edd83df41f1" [[package]] name = "hashbrown" -version = "0.15.2" +version = "0.17.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "bf151400ff0baff5465007dd2f3e717f3fe502074ca563069ce3a6629d07b289" +checksum = "ed5909b6e89a2db4456e54cd5f673791d7eca6732202bbf2a9cc504fe2f9b84a" [[package]] name = "heck" @@ -715,12 +794,12 @@ dependencies = [ [[package]] name = "indexmap" -version = "2.6.0" +version = "2.14.2" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "707907fe3c25f5424cce2cb7e1cbcafee6bdbe735ca90ef77c29e84591e5b9da" +checksum = "cc4e190f5d26ca7051642629da2c52fc03bde85a03197c99408dcd291734c855" dependencies = [ "equivalent", - "hashbrown 0.15.2", + "hashbrown 0.17.1", ] [[package]] @@ -837,6 +916,15 @@ version = "0.2.170" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "875b3680cb2f8f71bdcf9a30f38d48282f5d3c95cbf9b3fa57269bb5d5c06828" +[[package]] +name = "link-cplusplus" +version = "1.0.12" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7f78c730aaa7d0b9336a299029ea49f9ee53b0ed06e9202e8cb7db9bae7b8c82" +dependencies = [ + "cc", +] + [[package]] name = "linux-raw-sys" version = "0.4.11" @@ -1221,6 +1309,8 @@ dependencies = [ "clap", "config", "criterion", + "cxx", + "cxx-build", "env_logger", "httparse", "ipnet", @@ -1451,6 +1541,12 @@ version = "1.2.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "94143f37725109f92c262ed2cf5e59bce7498c01bcc1502d7b9afe439a4e9f49" +[[package]] +name = "scratch" +version = "1.0.9" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d68f2ec51b097e4c1a75b681a8bec621909b5e91f15bb7b840c4f2f7b01148b2" + [[package]] name = "security-framework" version = "3.7.0" @@ -1668,6 +1764,15 @@ dependencies = [ "windows-sys 0.59.0", ] +[[package]] +name = "termcolor" +version = "1.4.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "06794f8f6c5c898b3275aebefa6b8a1cb24cd2c6c79397ab15774837a0bc5755" +dependencies = [ + "winapi-util", +] + [[package]] name = "terminal_size" version = "0.4.4" @@ -1843,6 +1948,12 @@ version = "1.0.12" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "3354b9ac3fae1ff6755cb6db53683adb661634f67557942dea4facebec0fee4b" +[[package]] +name = "unicode-width" +version = "0.2.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ca718471876992059f251e33aafc334346ada08042b7661dd1fe313a9bced557" + [[package]] name = "untrusted" version = "0.9.0" diff --git a/Cargo.toml b/Cargo.toml index 936f3ab98..d3a0031c2 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -73,6 +73,7 @@ arrayvec = "0.7" base64 = "0.13" clap = { version = "4.6", features = ["cargo", "string", "wrap_help", "derive"] } config = { version = "=0.15.25", default-features = false, features = ["ini"] } +cxx = "=1.0.198" httparse = "1.7" ipnet = "2" jsonwebtoken = "9" @@ -109,6 +110,7 @@ test-log = "0.2" pkg-config = "0.3" time = { version = "0.3.36", features = ["formatting", "local-offset", "macros"] } cbindgen = "0.27" +cxx-build = "=1.0.198" [[bench]] name = "memorypool" diff --git a/build.rs b/build.rs index 6cc74c340..21e18b427 100644 --- a/build.rs +++ b/build.rs @@ -454,6 +454,10 @@ fn main() -> Result<(), Box> { }, ); + cxx_build::bridge("src/handler/settings.rs") + .flag_if_supported("-std=c++17") + .compile("pushpin-cxxbridge"); + let (qmake_path, qt_version) = get_qmake()?; let qt_install_libs = { @@ -516,6 +520,12 @@ fn main() -> Result<(), Box> { include_paths.push(cpp_build_dir.as_ref()); + // Headers generated by cxx_build + let cxx_include_dir = out_dir.join("cxxbridge").join("include"); + let cxx_crate_dir = out_dir.join("cxxbridge").join("crate"); + include_paths.push(cxx_include_dir.as_ref()); + include_paths.push(cxx_crate_dir.as_ref()); + if boost_include_dir != Path::new("/usr/include") { include_paths.push(boost_include_dir.as_ref()); } diff --git a/src/bin/pushpin-handler.rs b/src/bin/pushpin-handler.rs index b3e34c6f5..06a458b2c 100644 --- a/src/bin/pushpin-handler.rs +++ b/src/bin/pushpin-handler.rs @@ -15,17 +15,28 @@ */ use clap::Parser; -use log::debug; +use log::{debug, error}; use pushpin::core::log::ffi::{log_init, log_set_level}; -use pushpin::handler::cliargs::ffi::HandlerCliArgs; +use pushpin::core::settings::{Settings, SettingsError}; use pushpin::handler::cliargs::{destroy_handler_cli_args, CliArgs}; -use pushpin::import_cpp; +use pushpin::handler::settings::{ffi, HandlerSettings}; use std::ffi::CString; +use std::path::Path; use std::process::ExitCode; use std::ptr; -import_cpp! { - fn handler_init(args: *const HandlerCliArgs) -> libc::c_int; +fn load_settings(args: &CliArgs, config_file: &str) -> Result { + let mut settings = Settings::load(Path::new(config_file))?; + + if let Some(ipc_prefix) = args.ipc_prefix.as_deref().filter(|p| !p.is_empty()) { + settings.set_ipc_prefix(ipc_prefix); + } + + if let Some(port_offset) = args.port_offset { + settings.set_port_offset(port_offset as i32); + } + + HandlerSettings::from_settings(&settings) } fn main() -> ExitCode { @@ -46,9 +57,27 @@ fn main() -> ExitCode { debug!("starting..."); + let config_file = cli_args + .config_file + .clone() + .expect("verify sets the config file"); + + let settings = match load_settings(&cli_args, &config_file) { + Ok(settings) => settings.to_ffi(), + Err(e) => { + error!("failed to load config file {config_file}: {e}"); + return ExitCode::from(1); + } + }; + let cli_args_ffi = cli_args.to_ffi(); - let exit_code = unsafe { handler_init(&cli_args_ffi) }; + let exit_code = unsafe { + ffi::handler_init( + &cli_args_ffi as *const _ as *const ffi::HandlerCliArgs, + &settings, + ) + }; // Clean up the allocated memory unsafe { destroy_handler_cli_args(cli_args_ffi) }; diff --git a/src/handler/handler.pri b/src/handler/handler.pri index 4ecbfba41..8ead28918 100644 --- a/src/handler/handler.pri +++ b/src/handler/handler.pri @@ -27,7 +27,8 @@ HEADERS += \ $$PWD/filter.h \ $$PWD/filterstack.h \ $$PWD/handlerengine.h \ - $$PWD/handlerargsdata.h + $$PWD/handlerargsdata.h \ + $$PWD/handlerinit.h SOURCES += \ $$PWD/deferred.cpp \ diff --git a/src/handler/handlerapp.cpp b/src/handler/handlerapp.cpp index 682fcb5dd..d1764211c 100644 --- a/src/handler/handlerapp.cpp +++ b/src/handler/handlerapp.cpp @@ -31,51 +31,26 @@ #include "httpsessionupdatemanager.h" #include "log.h" #include "processquit.h" +#include "pushpin/src/handler/settings.rs.h" #include "reactor.h" #include "rust/bindings.h" -#include "settings.h" #include "simplehttpserver.h" #include "timer.h" #include "wssession.h" -#include -#include -#include #include #include -#include #include -#define DEFAULT_HTTP_MAX_HEADERS_SIZE 10000 -#define DEFAULT_HTTP_MAX_BODY_SIZE 1000000 - -static void trimlist(QStringList *list) { - for (int n = 0; n < list->count(); ++n) { - if ((*list)[n].isEmpty()) { - list->removeAt(n); - --n; // Adjust position - } - } +static QString toQString(const rust::String &s) { + return QString::fromUtf8(s.data(), (qsizetype)s.size()); } -static QStringList expandSpecs(const QStringList &l, int peerCount) { - if (l.count() == 1 && l[0].startsWith("ipc:") && peerCount > 1) { - QString base = l[0]; - - QStringList out; - for (int i = 0; i < peerCount; ++i) - out += base + QString("-%1").arg(i); - - return out; - } +static QStringList toQStringList(const rust::Vec &v) { + QStringList out; + for (const rust::String &s : v) + out += toQString(s); - return l; -} - -static QString firstSpec(const QString &s, int peerCount) { - if (s.startsWith("ipc:") && peerCount > 1) - return s + "-0"; - - return s; + return out; } static int runLoop(const QString &logFile, const HandlerEngine::Configuration &config) { @@ -154,189 +129,56 @@ static int runLoop(const QString &logFile, const HandlerEngine::Configuration &c return exitCode.value(); } -extern "C" { +namespace ffi { -int handler_init(const ffi::HandlerCliArgs *argsFfi) { +int handler_init(const HandlerCliArgs *argsFfi, const HandlerSettings &s) { HandlerArgsData args(argsFfi); - // QSettings doesn't inform us if the config file can't be opened, so do that ourselves - { - QFile file(args.configFile); - if (!file.open(QIODevice::ReadOnly)) { - log_error("failed to open config file: %s", qPrintable(args.configFile)); - return 1; - } - } - - Settings settings(args.configFile); - if (!args.ipcPrefix.isEmpty()) - settings.setIpcPrefix(args.ipcPrefix); - if (args.portOffset != -1) - settings.setPortOffset(args.portOffset); - - QStringList services = settings.value("runner/services").toStringList(); - - QStringList connmgr_in_stream_specs = - settings.value("proxy/connmgr_in_stream_specs").toStringList(); - trimlist(&connmgr_in_stream_specs); - QStringList connmgr_out_specs = settings.value("proxy/connmgr_out_specs").toStringList(); - trimlist(&connmgr_out_specs); - QStringList condure_in_stream_specs = - settings.value("proxy/condure_in_stream_specs").toStringList(); - trimlist(&condure_in_stream_specs); - connmgr_in_stream_specs += condure_in_stream_specs; - QStringList condure_out_specs = settings.value("proxy/condure_out_specs").toStringList(); - trimlist(&condure_out_specs); - connmgr_out_specs += condure_out_specs; - int proxyWorkerCount = settings.value("proxy/workers", 1).toInt(); - QStringList m2a_in_stream_specs = settings.value("handler/m2a_in_stream_specs").toStringList(); - trimlist(&m2a_in_stream_specs); - QStringList m2a_out_specs = settings.value("handler/m2a_out_specs").toStringList(); - trimlist(&m2a_out_specs); - QStringList intreq_out_specs = settings.value("handler/proxy_intreq_out_specs").toStringList(); - trimlist(&intreq_out_specs); - QStringList intreq_out_stream_specs = - settings.value("handler/proxy_intreq_out_stream_specs").toStringList(); - trimlist(&intreq_out_stream_specs); - QStringList intreq_in_specs = settings.value("handler/proxy_intreq_in_specs").toStringList(); - trimlist(&intreq_in_specs); - QStringList proxy_inspect_specs = settings.value("handler/proxy_inspect_specs").toStringList(); - trimlist(&proxy_inspect_specs); - QString proxy_inspect_spec = settings.value("handler/proxy_inspect_spec").toString(); - if (!proxy_inspect_spec.isEmpty()) - proxy_inspect_specs += proxy_inspect_spec; - QStringList proxy_accept_specs = settings.value("handler/proxy_accept_specs").toStringList(); - trimlist(&proxy_accept_specs); - QString proxy_accept_spec = settings.value("handler/proxy_accept_spec").toString(); - if (!proxy_accept_spec.isEmpty()) - proxy_accept_specs += proxy_accept_spec; - QStringList proxy_retry_out_specs = - settings.value("handler/proxy_retry_out_specs").toStringList(); - trimlist(&proxy_retry_out_specs); - QString proxy_retry_out_spec = settings.value("handler/proxy_retry_out_spec").toString(); - if (!proxy_retry_out_spec.isEmpty()) - proxy_retry_out_specs += proxy_retry_out_spec; - QStringList ws_control_init_specs = - settings.value("handler/proxy_ws_control_init_specs").toStringList(); - trimlist(&ws_control_init_specs); - QStringList ws_control_stream_specs = - settings.value("handler/proxy_ws_control_stream_specs").toStringList(); - trimlist(&ws_control_stream_specs); - QString stats_spec = settings.value("handler/stats_spec").toString(); - QString command_spec = settings.value("handler/command_spec").toString(); - QString state_spec = settings.value("handler/state_spec").toString(); - QStringList proxy_stats_specs = settings.value("handler/proxy_stats_specs").toStringList(); - trimlist(&proxy_stats_specs); - QString proxy_stats_spec = settings.value("handler/proxy_stats_spec").toString(); - if (!proxy_stats_spec.isEmpty()) - proxy_stats_specs += proxy_stats_spec; - QString proxy_command_spec = settings.value("handler/proxy_command_spec").toString(); - QString push_in_spec = settings.value("handler/push_in_spec").toString(); - QStringList push_in_sub_specs = settings.value("handler/push_in_sub_specs").toStringList(); - trimlist(&push_in_sub_specs); - QString push_in_sub_spec = settings.value("handler/push_in_sub_spec").toString(); - if (!push_in_sub_spec.isEmpty()) - push_in_sub_specs += push_in_sub_spec; - bool push_in_sub_connect = settings.value("handler/push_in_sub_connect").toBool(); - QString push_in_http_addr = settings.value("handler/push_in_http_addr").toString(); - int push_in_http_port = settings.adjustedPort("handler/push_in_http_port"); - int push_in_http_max_headers_size = - settings.value("handler/push_in_http_max_headers_size", DEFAULT_HTTP_MAX_HEADERS_SIZE) - .toInt(); - int push_in_http_max_body_size = - settings.value("handler/push_in_http_max_body_size", DEFAULT_HTTP_MAX_BODY_SIZE).toInt(); - bool ok; - int ipcFileMode = settings.value("handler/ipc_file_mode", -1).toString().toInt(&ok, 8); - bool shareAll = settings.value("handler/share_all").toBool(); - int messageRate = settings.value("handler/message_rate", -1).toInt(); - int messageHwm = settings.value("handler/message_hwm", -1).toInt(); - int messageBlockSize = settings.value("handler/message_block_size", -1).toInt(); - int messageWait = settings.value("handler/message_wait", 5000).toInt(); - int idCacheTtl = settings.value("handler/id_cache_ttl", 0).toInt(); - bool updateOnFirstSubscription = - settings.value("handler/update_on_first_subscription", true).toBool(); - int clientMaxconn = settings.value("runner/client_maxconn", 50000).toInt(); - int connectionSubscriptionMax = - settings.value("handler/connection_subscription_max", 20).toInt(); - int subscriptionLinger = settings.value("handler/subscription_linger", 60).toInt(); - int statsConnectionSend = settings.value("global/stats_connection_send", true).toBool(); - int statsConnectionTtl = settings.value("global/stats_connection_ttl", 120).toInt(); - int statsSubscriptionTtl = settings.value("handler/stats_subscription_ttl", 60).toInt(); - int statsReportInterval = settings.value("handler/stats_report_interval", 10).toInt(); - QString statsFormat = settings.value("handler/stats_format").toString(); - QString prometheusPort = settings.value("handler/prometheus_port").toString(); - QString prometheusPrefix = settings.value("handler/prometheus_prefix").toString(); - - if (m2a_in_stream_specs.isEmpty() || m2a_out_specs.isEmpty()) { - log_error("must set m2a_in_stream_specs and m2a_out_specs"); - return 1; - } - - if (proxy_inspect_specs.isEmpty() || proxy_accept_specs.isEmpty() || - proxy_retry_out_specs.isEmpty()) { - log_error("must set proxy_inspect_specs, proxy_accept_specs, and " - "proxy_retry_out_specs"); - return 1; - } - - if (!push_in_http_addr.isEmpty() && QHostAddress(push_in_http_addr).isNull()) { - log_error("invalid push_in_http_addr: %s", qPrintable(push_in_http_addr)); - return 1; - } - HandlerEngine::Configuration config; config.appVersion = Config::get().version; config.instanceId = "handler_" + QByteArray::number(getpid()); - if (!services.contains("mongrel2") && - (!connmgr_in_stream_specs.isEmpty() || !connmgr_out_specs.isEmpty())) { - config.serverInStreamSpecs = connmgr_in_stream_specs; - config.serverOutSpecs = connmgr_out_specs; - } else { - config.serverInStreamSpecs = m2a_in_stream_specs; - config.serverOutSpecs = m2a_out_specs; - } - config.clientOutSpecs = expandSpecs(intreq_out_specs, proxyWorkerCount); - config.clientOutStreamSpecs = expandSpecs(intreq_out_stream_specs, proxyWorkerCount); - config.clientInSpecs = expandSpecs(intreq_in_specs, proxyWorkerCount); - config.inspectSpecs = expandSpecs(proxy_inspect_specs, proxyWorkerCount); - config.acceptSpecs = expandSpecs(proxy_accept_specs, proxyWorkerCount); - config.retryOutSpecs = expandSpecs(proxy_retry_out_specs, proxyWorkerCount); - config.wsControlInitSpecs = expandSpecs(ws_control_init_specs, proxyWorkerCount); - config.wsControlStreamSpecs = expandSpecs(ws_control_stream_specs, proxyWorkerCount); - config.statsSpec = stats_spec; - config.commandSpec = command_spec; - config.stateSpec = state_spec; - config.proxyStatsSpecs = expandSpecs(proxy_stats_specs, proxyWorkerCount); - config.proxyCommandSpec = firstSpec(proxy_command_spec, proxyWorkerCount); - config.pushInSpec = push_in_spec; - config.pushInSubSpecs = push_in_sub_specs; - config.pushInSubConnect = push_in_sub_connect; - if (!services.contains("api")) { - // Only serve the API here if the api component is not enabled - config.pushInHttpAddr = QHostAddress(push_in_http_addr); - config.pushInHttpPort = push_in_http_port; - } - config.pushInHttpMaxHeadersSize = push_in_http_max_headers_size; - config.pushInHttpMaxBodySize = push_in_http_max_body_size; - config.ipcFileMode = ipcFileMode; - config.shareAll = shareAll; - config.messageRate = messageRate; - config.messageHwm = messageHwm; - config.messageBlockSize = messageBlockSize; - config.messageWait = messageWait; - config.idCacheTtl = idCacheTtl; - config.updateOnFirstSubscription = updateOnFirstSubscription; - config.connectionsMax = clientMaxconn; - config.connectionSubscriptionMax = connectionSubscriptionMax; - config.subscriptionLinger = subscriptionLinger; - config.statsConnectionSend = statsConnectionSend; - config.statsConnectionTtl = statsConnectionTtl; - config.statsSubscriptionTtl = statsSubscriptionTtl; - config.statsReportInterval = statsReportInterval; - config.statsFormat = statsFormat; - config.prometheusPort = prometheusPort; - config.prometheusPrefix = prometheusPrefix; + config.serverInStreamSpecs = toQStringList(s.server_in_stream_specs); + config.serverOutSpecs = toQStringList(s.server_out_specs); + config.clientOutSpecs = toQStringList(s.client_out_specs); + config.clientOutStreamSpecs = toQStringList(s.client_out_stream_specs); + config.clientInSpecs = toQStringList(s.client_in_specs); + config.inspectSpecs = toQStringList(s.inspect_specs); + config.acceptSpecs = toQStringList(s.accept_specs); + config.retryOutSpecs = toQStringList(s.retry_out_specs); + config.wsControlInitSpecs = toQStringList(s.ws_control_init_specs); + config.wsControlStreamSpecs = toQStringList(s.ws_control_stream_specs); + config.statsSpec = toQString(s.stats_spec); + config.commandSpec = toQString(s.command_spec); + config.stateSpec = toQString(s.state_spec); + config.proxyStatsSpecs = toQStringList(s.proxy_stats_specs); + config.proxyCommandSpec = toQString(s.proxy_command_spec); + config.pushInSpec = toQString(s.push_in_spec); + config.pushInSubSpecs = toQStringList(s.push_in_sub_specs); + config.pushInSubConnect = s.push_in_sub_connect; + config.pushInHttpAddr = QHostAddress(toQString(s.push_in_http_addr)); + config.pushInHttpPort = s.push_in_http_port; + config.pushInHttpMaxHeadersSize = s.push_in_http_max_headers_size; + config.pushInHttpMaxBodySize = s.push_in_http_max_body_size; + config.ipcFileMode = s.ipc_file_mode; + config.shareAll = s.share_all; + config.messageRate = s.message_rate; + config.messageHwm = s.message_hwm; + config.messageBlockSize = s.message_block_size; + config.messageWait = s.message_wait; + config.idCacheTtl = s.id_cache_ttl; + config.updateOnFirstSubscription = s.update_on_first_subscription; + config.connectionsMax = s.connections_max; + config.connectionSubscriptionMax = s.connection_subscription_max; + config.subscriptionLinger = s.subscription_linger; + config.statsConnectionSend = s.stats_connection_send; + config.statsConnectionTtl = s.stats_connection_ttl; + config.statsSubscriptionTtl = s.stats_subscription_ttl; + config.statsReportInterval = s.stats_report_interval; + config.statsFormat = toQString(s.stats_format); + config.prometheusPort = toQString(s.prometheus_port); + config.prometheusPrefix = toQString(s.prometheus_prefix); return runLoop(args.logFile, config); } -} + +} // namespace ffi diff --git a/src/handler/handlerinit.h b/src/handler/handlerinit.h new file mode 100644 index 000000000..f72c84960 --- /dev/null +++ b/src/handler/handlerinit.h @@ -0,0 +1,31 @@ +/* + * Copyright (C) 2026 Fastly, Inc. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +#ifndef HANDLERINIT_H +#define HANDLERINIT_H + +// Included by the cxx bridge, so it must not include Qt headers + +namespace ffi { + +struct HandlerCliArgs; +struct HandlerSettings; + +int handler_init(const HandlerCliArgs *args, const HandlerSettings &settings); + +} // namespace ffi + +#endif diff --git a/src/handler/settings.rs b/src/handler/settings.rs index b83ec09d4..37ebc39a0 100644 --- a/src/handler/settings.rs +++ b/src/handler/settings.rs @@ -15,11 +15,69 @@ */ use crate::core::settings::{CommonSettings, Settings, SettingsError}; +use std::convert::TryFrom; use std::net::IpAddr; const DEFAULT_HTTP_MAX_HEADERS_SIZE: u32 = 10000; const DEFAULT_HTTP_MAX_BODY_SIZE: u32 = 1000000; +/// cbindgen:ignore +#[allow(clippy::missing_safety_doc)] +#[cxx::bridge(namespace = "ffi")] +pub mod ffi { + // -1 or empty means unset + struct HandlerSettings { + server_in_stream_specs: Vec, + server_out_specs: Vec, + client_out_specs: Vec, + client_out_stream_specs: Vec, + client_in_specs: Vec, + inspect_specs: Vec, + accept_specs: Vec, + retry_out_specs: Vec, + ws_control_init_specs: Vec, + ws_control_stream_specs: Vec, + stats_spec: String, + command_spec: String, + state_spec: String, + proxy_stats_specs: Vec, + proxy_command_spec: String, + push_in_spec: String, + push_in_sub_specs: Vec, + push_in_sub_connect: bool, + push_in_http_addr: String, + push_in_http_port: i32, + push_in_http_max_headers_size: i32, + push_in_http_max_body_size: i32, + ipc_file_mode: i32, + share_all: bool, + message_rate: i32, + message_hwm: i32, + message_block_size: i32, + message_wait: i32, + id_cache_ttl: i32, + update_on_first_subscription: bool, + connections_max: i32, + connection_subscription_max: i32, + subscription_linger: i32, + stats_connection_send: bool, + stats_connection_ttl: i32, + stats_subscription_ttl: i32, + stats_report_interval: i32, + stats_format: String, + prometheus_port: String, + prometheus_prefix: String, + } + + unsafe extern "C++" { + include!("pushpin/src/handler/handlerinit.h"); + + type HandlerCliArgs; + + unsafe fn handler_init(args: *const HandlerCliArgs, settings: &HandlerSettings) -> i32; + } +} + pub struct HandlerSettings { pub common: CommonSettings, pub server_in_stream_specs: Vec, @@ -210,6 +268,59 @@ impl HandlerSettings { } } +impl HandlerSettings { + pub fn to_ffi(&self) -> ffi::HandlerSettings { + let int = |x: u32| i32::try_from(x).unwrap_or(i32::MAX); + let opt = |x: Option| x.map_or(-1, int); + let string = |s: &Option| s.clone().unwrap_or_default(); + + ffi::HandlerSettings { + server_in_stream_specs: self.server_in_stream_specs.clone(), + server_out_specs: self.server_out_specs.clone(), + client_out_specs: self.client_out_specs.clone(), + client_out_stream_specs: self.client_out_stream_specs.clone(), + client_in_specs: self.client_in_specs.clone(), + inspect_specs: self.inspect_specs.clone(), + accept_specs: self.accept_specs.clone(), + retry_out_specs: self.retry_out_specs.clone(), + ws_control_init_specs: self.ws_control_init_specs.clone(), + ws_control_stream_specs: self.ws_control_stream_specs.clone(), + stats_spec: string(&self.stats_spec), + command_spec: string(&self.command_spec), + state_spec: string(&self.state_spec), + proxy_stats_specs: self.proxy_stats_specs.clone(), + proxy_command_spec: string(&self.proxy_command_spec), + push_in_spec: string(&self.push_in_spec), + push_in_sub_specs: self.push_in_sub_specs.clone(), + push_in_sub_connect: self.push_in_sub_connect, + push_in_http_addr: self + .push_in_http_addr + .map_or_else(String::new, |a| a.to_string()), + push_in_http_port: self.push_in_http_port.map_or(-1, i32::from), + push_in_http_max_headers_size: int(self.push_in_http_max_headers_size), + push_in_http_max_body_size: int(self.push_in_http_max_body_size), + ipc_file_mode: opt(self.ipc_file_mode), + share_all: self.share_all, + message_rate: opt(self.message_rate), + message_hwm: opt(self.message_hwm), + message_block_size: opt(self.message_block_size), + message_wait: int(self.message_wait), + id_cache_ttl: int(self.id_cache_ttl), + update_on_first_subscription: self.update_on_first_subscription, + connections_max: int(self.common.client_maxconn), + connection_subscription_max: int(self.connection_subscription_max), + subscription_linger: int(self.subscription_linger), + stats_connection_send: self.common.stats_connection_send, + stats_connection_ttl: int(self.common.stats_connection_ttl), + stats_subscription_ttl: int(self.stats_subscription_ttl), + stats_report_interval: int(self.stats_report_interval), + stats_format: string(&self.stats_format), + prometheus_port: string(&self.prometheus_port), + prometheus_prefix: string(&self.prometheus_prefix), + } + } +} + fn combined_specs( settings: &Settings, list_key: &str, From c11a57c94f524d55e6da82c912f952b62ec50758 Mon Sep 17 00:00:00 2001 From: Serra Date: Tue, 10 Nov 2026 11:05:39 -0800 Subject: [PATCH 4/4] Adds tests for the handler's shared settings and its conversion for C++ --- src/handler/settings.rs | 52 +++++++++++++++++++++++++++++++++++++++++ 1 file changed, 52 insertions(+) diff --git a/src/handler/settings.rs b/src/handler/settings.rs index 37ebc39a0..cf9e9d085 100644 --- a/src/handler/settings.rs +++ b/src/handler/settings.rs @@ -556,6 +556,58 @@ mod tests { )); } + #[test] + fn common_settings_read_from_config() { + let s = load_conf( + "common", + &format!( + "{GLOBAL}stats_connection_send=false\nstats_connection_ttl=30\n[runner]\nclient_maxconn=100\n" + ), + ) + .unwrap(); + + assert_eq!(s.common.client_maxconn, 100); + assert!(!s.common.stats_connection_send); + assert_eq!(s.common.stats_connection_ttl, 30); + } + + #[test] + fn to_ffi_unset_values() { + let f = load_conf("ffi_unset", GLOBAL).unwrap().to_ffi(); + + assert_eq!(f.push_in_http_addr, ""); + assert_eq!(f.push_in_http_port, -1); + assert_eq!(f.ipc_file_mode, -1); + assert_eq!(f.message_rate, -1); + assert_eq!(f.message_hwm, -1); + assert_eq!(f.message_block_size, -1); + assert_eq!(f.state_spec, ""); + assert_eq!(f.stats_format, ""); + } + + #[test] + fn to_ffi_set_values() { + let f = load_conf( + "ffi_set", + &format!( + "{GLOBAL}stats_connection_send=false\nstats_connection_ttl=30\n[runner]\nclient_maxconn=100\n[handler]\npush_in_http_addr=127.0.0.1\npush_in_http_port=5561\nipc_file_mode=770\nmessage_rate=2500\nmessage_wait=3000000000\nstats_format=json\n" + ), + ) + .unwrap() + .to_ffi(); + + assert_eq!(f.push_in_http_addr, "127.0.0.1"); + assert_eq!(f.push_in_http_port, 5561); + assert_eq!(f.ipc_file_mode, 0o770); + assert_eq!(f.message_rate, 2500); + assert_eq!(f.message_wait, i32::MAX); + assert_eq!(f.stats_format, "json"); + assert_eq!(f.connections_max, 100); + assert!(!f.stats_connection_send); + assert_eq!(f.stats_connection_ttl, 30); + assert_eq!(f.inspect_specs, vec!["ipc://run/pushpin-inspect"]); + } + #[test] fn ipc_file_mode_octal() { let s = load_conf("mode", &format!("{GLOBAL}[handler]\nipc_file_mode=777\n")).unwrap();