From 665caf11387b746257c94bff8c287731ce80508f Mon Sep 17 00:00:00 2001 From: Chris Beck Date: Fri, 5 Dec 2025 04:11:34 -0700 Subject: [PATCH] bring back support for unix domain sockets --- signal-gateway/src/gateway/mod.rs | 55 +++++++++++++++++++--------- signal-gateway/src/jsonrpc.rs | 10 +++++ signal-gateway/src/transports/ipc.rs | 23 ++++++++++++ signal-gateway/src/transports/mod.rs | 2 + 4 files changed, 72 insertions(+), 18 deletions(-) create mode 100644 signal-gateway/src/transports/ipc.rs diff --git a/signal-gateway/src/gateway/mod.rs b/signal-gateway/src/gateway/mod.rs index edc5338..006bd93 100644 --- a/signal-gateway/src/gateway/mod.rs +++ b/signal-gateway/src/gateway/mod.rs @@ -5,6 +5,8 @@ use crate::{ message_handler::{AdminMessageResponse, MessageHandler, MessageHandlerResult}, prometheus::{Prometheus, PrometheusConfig}, }; +#[cfg(unix)] +use crate::jsonrpc::connect_ipc; use chrono::Utc; use conf::{Conf, Subcommands}; use futures_util::FutureExt; @@ -33,9 +35,19 @@ mod rate_limiter; use rate_limiter::{MultiRateLimiter, RateThreshold, SourceLocationRateLimiter}; #[derive(Conf, Debug)] +#[cfg_attr( + unix, + conf(one_of_fields(signal_cli_tcp_addr, signal_cli_socket_path)) +)] +#[cfg_attr(not(unix), conf(one_of_fields(signal_cli_tcp_addr)))] pub struct GatewayConfig { - #[conf(long, env, default_value = "127.0.0.1:7583")] - pub signal_cli_tcp_addr: SocketAddr, + /// TCP address of signal-cli JSON-RPC server + #[conf(long, env)] + pub signal_cli_tcp_addr: Option, + /// Unix socket path of signal-cli JSON-RPC server + #[cfg(unix)] + #[conf(long, env)] + pub signal_cli_socket_path: Option, #[conf(long, env)] pub signal_account: String, #[conf(repeat, long, env)] @@ -187,26 +199,33 @@ impl Gateway { if self.token.is_cancelled() { return; } - match connect_tcp(&self.config.signal_cli_tcp_addr).await { - Err(err) => { - error!( - "Could not connect to signal_cli @ ({}): {err}", - self.config.signal_cli_tcp_addr - ); - tokio::time::sleep(Duration::from_secs(5)).await; - } - Ok(client) => { - if let Err(err) = self.do_run(&client).await { - error!("Error with signal cli, reconnecting: {err}"); - tokio::time::sleep(Duration::from_secs(5)).await; - } else { - continue; - } - } + + if let Err(err) = self.connect_and_run().await { + error!("Error with signal-cli, reconnecting: {err}"); + tokio::time::sleep(Duration::from_secs(5)).await; } } } + /// Connect to signal-cli and run the main loop. + async fn connect_and_run(&self) -> Result<(), Box> { + #[cfg(unix)] + if let Some(path) = &self.config.signal_cli_socket_path { + info!("Connecting to signal-cli via unix socket: {}", path.display()); + let client = connect_ipc(path).await?; + return Ok(self.do_run(&client).await?); + } + + if let Some(addr) = &self.config.signal_cli_tcp_addr { + info!("Connecting to signal-cli via TCP: {addr}"); + let client = connect_tcp(addr).await?; + return Ok(self.do_run(&client).await?); + } + + // This shouldn't happen due to one_of_fields validation + unreachable!("one_of_fields should ensure exactly one transport is configured") + } + async fn do_run(&self, signal_cli: &impl RpcClient) -> Result<(), RpcClientError> { let mut admin_mq_rx = self .admin_mq_rx diff --git a/signal-gateway/src/jsonrpc.rs b/signal-gateway/src/jsonrpc.rs index aa21727..d6172f2 100644 --- a/signal-gateway/src/jsonrpc.rs +++ b/signal-gateway/src/jsonrpc.rs @@ -441,6 +441,16 @@ pub async fn connect_tcp( Ok(ClientBuilder::default().build_with_tokio(sender, receiver)) } +/// Connect to signal-cli over unix domain socket +#[cfg(unix)] +pub async fn connect_ipc( + path: impl AsRef, +) -> Result { + let (sender, receiver) = super::transports::ipc::connect(path).await?; + + Ok(ClientBuilder::default().build_with_tokio(sender, receiver)) +} + impl Envelope { /// Send a read-receipt for an envelope pub async fn send_read_receipt( diff --git a/signal-gateway/src/transports/ipc.rs b/signal-gateway/src/transports/ipc.rs new file mode 100644 index 0000000..c5b5800 --- /dev/null +++ b/signal-gateway/src/transports/ipc.rs @@ -0,0 +1,23 @@ +use std::io::Error; +use std::path::Path; + +use futures_util::stream::StreamExt; +use jsonrpsee::core::client::{TransportReceiverT, TransportSenderT}; +use tokio::net::UnixStream; +use tokio_util::codec::Decoder; + +use super::stream_codec::StreamCodec; +use super::{Receiver, Sender}; + +/// Connect to a JSON-RPC server via Unix domain socket. +pub async fn connect( + socket: impl AsRef, +) -> Result<(impl TransportSenderT + Send, impl TransportReceiverT + Send), Error> { + let connection = UnixStream::connect(socket).await?; + let (sink, stream) = StreamCodec::stream_incoming().framed(connection).split(); + + let sender = Sender { inner: sink }; + let receiver = Receiver { inner: stream }; + + Ok((sender, receiver)) +} diff --git a/signal-gateway/src/transports/mod.rs b/signal-gateway/src/transports/mod.rs index 24772a9..07964ae 100644 --- a/signal-gateway/src/transports/mod.rs +++ b/signal-gateway/src/transports/mod.rs @@ -5,6 +5,8 @@ use jsonrpsee::core::client::{ReceivedMessage, TransportReceiverT, TransportSend use thiserror::Error; mod stream_codec; +#[cfg(unix)] +pub mod ipc; pub mod tcp; #[derive(Debug, Error)]