Adds support for UNIX sockets (#272)

close #246

Author: anarthal

.github/workflows/ci.yml

@@ -196,6 +196,13 @@ jobs:
             build-type: 'Debug'
             cxxflags: '-fsanitize=address -fsanitize=undefined -fno-sanitize-recover=all'
             ldflags: '-fsanitize=address -fsanitize=undefined'
+
+          - toolset: gcc-14
+            install: 'g++-14'
+            container: ubuntu:24.04
+            cxxstd: '23'
+            build-type: 'Debug'
+            cxxflags: '-DBOOST_ASIO_DISABLE_LOCAL_SOCKETS=1' # If a system had no UNIX socket support, we build correctly
 
           - toolset: gcc-14
             install: 'g++-14'

README.md

@@ -414,6 +414,7 @@ The examples below show how to use the features discussed so far
 
 * cpp20_intro.cpp: Does not use awaitable operators.
 * cpp20_intro_tls.cpp: Communicates over TLS.
+* cpp20_unix_sockets.cpp: Communicates over UNIX domain sockets.
 * cpp20_containers.cpp: Shows how to send and receive STL containers and how to use transactions.
 * cpp20_json.cpp: Shows how to serialize types using Boost.Json.
 * cpp20_protobuf.cpp: Shows how to serialize types using protobuf.

example/CMakeLists.txt

@@ -29,6 +29,7 @@ make_testable_example(cpp20_intro 20)
 make_testable_example(cpp20_containers 20)
 make_testable_example(cpp20_json 20)
 make_testable_example(cpp20_intro_tls 20)
+make_testable_example(cpp20_unix_sockets 20)
 
 make_example(cpp20_subscriber 20)
 make_example(cpp20_streams 20)

example/cpp20_unix_sockets.cpp

@@ -0,0 +1,60 @@
+//
+// Copyright (c) 2025 Marcelo Zimbres Silva ([email protected]),
+// Ruben Perez Hidalgo (rubenperez038 at gmail dot com)
+//
+// Distributed under the Boost Software License, Version 1.0. (See accompanying
+// file LICENSE_1_0.txt or copy at http://www.boost.org/LICENSE_1_0.txt)
+//
+#include <boost/asio/awaitable.hpp>
+
+#if defined(BOOST_ASIO_HAS_CO_AWAIT)
+
+#include <boost/redis/connection.hpp>
+
+#include <boost/asio/consign.hpp>
+#include <boost/asio/detached.hpp>
+#include <boost/asio/this_coro.hpp>
+
+#include <iostream>
+
+namespace asio = boost::asio;
+using boost::redis::request;
+using boost::redis::response;
+using boost::redis::config;
+using boost::redis::logger;
+using boost::redis::connection;
+
+#ifdef BOOST_ASIO_HAS_LOCAL_SOCKETS
+
+auto co_main(config cfg) -> asio::awaitable<void>
+{
+   // If unix_socket is set to a non-empty string, UNIX domain sockets will be used
+   // instead of TCP. Set this value to the path where your server is listening.
+   // UNIX domain socket connections work in the same way as TCP connections.
+   cfg.unix_socket = "/tmp/redis-socks/redis.sock";
+
+   auto conn = std::make_shared<connection>(co_await asio::this_coro::executor);
+   conn->async_run(cfg, {}, asio::consign(asio::detached, conn));
+
+   request req;
+   req.push("PING");
+
+   response<std::string> resp;
+
+   co_await conn->async_exec(req, resp);
+   conn->cancel();
+
+   std::cout << "Response: " << std::get<0>(resp).value() << std::endl;
+}
+
+#else
+
+auto co_main(config) -> asio::awaitable<void>
+{
+   std::cout << "Sorry, your system does not support UNIX domain sockets\n";
+   co_return;
+}
+
+#endif
+
+#endif

include/boost/redis/config.hpp

@@ -34,6 +34,11 @@ struct config {
    /// Address of the Redis server.
    address addr = address{"127.0.0.1", "6379"};
 
+   /// The UNIX domain socket path where the server is listening. If non-empty,
+   /// communication with the server will happen using UNIX domain sockets, and addr will be ignored.
+   /// UNIX domain sockets can't be used with SSL: if `unix_socket` is non-empty, `use_ssl` must be false.
+   std::string unix_socket;
+
    /** @brief Username passed to the
     * [HELLO](https://redis.io/commands/hello/) command.  If left
     * empty `HELLO` will be sent without authentication parameters.

include/boost/redis/connection.hpp

@@ -290,12 +290,25 @@ struct reader_op {
    }
 };
 
+inline system::error_code check_config(const config& cfg)
+{
+   if (!cfg.unix_socket.empty()) {
+#ifndef BOOST_ASIO_HAS_LOCAL_SOCKETS
+      return error::unix_sockets_unsupported;
+#endif
+      if (cfg.use_ssl)
+         return error::unix_sockets_ssl_unsupported;
+   }
+   return system::error_code{};
+}
+
 template <class Conn, class Logger>
 class run_op {
 private:
    Conn* conn_ = nullptr;
    Logger logger_;
    asio::coroutine coro_{};
+   system::error_code stored_ec_;
 
    using order_t = std::array<std::size_t, 5>;
 
@@ -321,75 +334,87 @@ class run_op {
       system::error_code ec3 = {},
       system::error_code = {})
    {
-      BOOST_ASIO_CORO_REENTER(coro_) for (;;)
+      BOOST_ASIO_CORO_REENTER(coro_)
       {
-         // Try to connect
-         BOOST_ASIO_CORO_YIELD
-         conn_->stream_.async_connect(&conn_->cfg_, logger_, std::move(self));
-
-         // If we failed, try again
+         // Check config
+         ec0 = check_config(conn_->cfg_);
          if (ec0) {
-            self.complete(ec0);
+            logger_.log_error("Invalid configuration", ec0);
+            stored_ec_ = ec0;
+            BOOST_ASIO_CORO_YIELD asio::async_immediate(self.get_io_executor(), std::move(self));
+            self.complete(stored_ec_);
             return;
          }
 
-         conn_->mpx_.reset();
+         for (;;) {
+            // Try to connect
+            BOOST_ASIO_CORO_YIELD
+            conn_->stream_.async_connect(&conn_->cfg_, logger_, std::move(self));
 
-         // Note: Order is important here because the writer might
-         // trigger an async_write before the async_hello thereby
-         // causing an authentication problem.
-         BOOST_ASIO_CORO_YIELD
-         asio::experimental::make_parallel_group(
-            [this](auto token) {
-               return conn_->handshaker_.async_hello(*conn_, logger_, token);
-            },
-            [this](auto token) {
-               return conn_->health_checker_.async_ping(*conn_, logger_, token);
-            },
-            [this](auto token) {
-               return conn_->health_checker_.async_check_timeout(*conn_, logger_, token);
-            },
-            [this](auto token) {
-               return conn_->reader(logger_, token);
-            },
-            [this](auto token) {
-               return conn_->writer(logger_, token);
-            })
-            .async_wait(asio::experimental::wait_for_one_error(), std::move(self));
-
-         if (order[0] == 0 && !!ec0) {
-            self.complete(ec0);
-            return;
-         }
+            // If we failed, try again
+            if (ec0) {
+               self.complete(ec0);
+               return;
+            }
 
-         if (order[0] == 2 && ec2 == error::pong_timeout) {
-            self.complete(ec1);
-            return;
-         }
+            conn_->mpx_.reset();
 
-         // The receive operation must be cancelled because channel
-         // subscription does not survive a reconnection but requires
-         // re-subscription.
-         conn_->cancel(operation::receive);
+            // Note: Order is important here because the writer might
+            // trigger an async_write before the async_hello thereby
+            // causing an authentication problem.
+            BOOST_ASIO_CORO_YIELD
+            asio::experimental::make_parallel_group(
+               [this](auto token) {
+                  return conn_->handshaker_.async_hello(*conn_, logger_, token);
+               },
+               [this](auto token) {
+                  return conn_->health_checker_.async_ping(*conn_, logger_, token);
+               },
+               [this](auto token) {
+                  return conn_->health_checker_.async_check_timeout(*conn_, logger_, token);
+               },
+               [this](auto token) {
+                  return conn_->reader(logger_, token);
+               },
+               [this](auto token) {
+                  return conn_->writer(logger_, token);
+               })
+               .async_wait(asio::experimental::wait_for_one_error(), std::move(self));
+
+            if (order[0] == 0 && !!ec0) {
+               self.complete(ec0);
+               return;
+            }
 
-         if (!conn_->will_reconnect()) {
-            conn_->cancel(operation::reconnection);
-            self.complete(ec3);
-            return;
-         }
+            if (order[0] == 2 && ec2 == error::pong_timeout) {
+               self.complete(ec1);
+               return;
+            }
 
-         conn_->reconnect_timer_.expires_after(conn_->cfg_.reconnect_wait_interval);
+            // The receive operation must be cancelled because channel
+            // subscription does not survive a reconnection but requires
+            // re-subscription.
+            conn_->cancel(operation::receive);
 
-         BOOST_ASIO_CORO_YIELD
-         conn_->reconnect_timer_.async_wait(asio::prepend(std::move(self), order_t{}));
-         if (ec0) {
-            self.complete(ec0);
-            return;
-         }
+            if (!conn_->will_reconnect()) {
+               conn_->cancel(operation::reconnection);
+               self.complete(ec3);
+               return;
+            }
 
-         if (!conn_->will_reconnect()) {
-            self.complete(asio::error::operation_aborted);
-            return;
+            conn_->reconnect_timer_.expires_after(conn_->cfg_.reconnect_wait_interval);
+
+            BOOST_ASIO_CORO_YIELD
+            conn_->reconnect_timer_.async_wait(asio::prepend(std::move(self), order_t{}));
+            if (ec0) {
+               self.complete(ec0);
+               return;
+            }
+
+            if (!conn_->will_reconnect()) {
+               self.complete(asio::error::operation_aborted);
+               return;
+            }
          }
       }
    }
@@ -753,9 +778,7 @@ class basic_connection {
          writer_timer_);
    }
 
-   auto is_open() const noexcept { return stream_.is_open(); }
-
-   [[nodiscard]] bool trigger_write() const noexcept { return is_open() && !mpx_.is_writing(); }
+   bool is_open() const noexcept { return stream_.is_open(); }
 
    detail::redis_stream<Executor> stream_;