bring back support for unix domain sockets

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<SocketAddr>,
+    /// Unix socket path of signal-cli JSON-RPC server
+    #[cfg(unix)]
+    #[conf(long, env)]
+    pub signal_cli_socket_path: Option<PathBuf>,
     #[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<dyn std::error::Error>> {
+        #[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

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<std::path::Path>,
+) -> Result<impl SubscriptionClientT, std::io::Error> {
+    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(

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<Path>,
+) -> 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))
+}

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)]