feat: Properly support IPv6 TCP servers.

Author: kcexn

include/net/service/impl/async_tcp_service_impl.hpp

@@ -46,7 +46,7 @@ auto async_tcp_service<TCPStreamHandler>::start(async_context &ctx) noexcept
   using namespace io;
   using namespace io::socket;
 
-  auto sock = socket_handle(AF_INET, SOCK_STREAM, IPPROTO_TCP);
+  auto sock = socket_handle(address_->sin6_family, SOCK_STREAM, 0);
   if (auto error = initialize_(sock))
   {
     ctx.scope.request_stop();
@@ -57,7 +57,7 @@ auto async_tcp_service<TCPStreamHandler>::start(async_context &ctx) noexcept
     using namespace stdexec;
     ctx.scope.request_stop();
     sender auto connect =
-        io::connect(ctx.poller.emplace(AF_INET, SOCK_STREAM, IPPROTO_TCP),
+        io::connect(ctx.poller.emplace(address_->sin6_family, SOCK_STREAM, 0),
                     address_) |
         then([](auto status) {}) | upon_error([](auto &&error) {});
     ctx.scope.spawn(std::move(connect));

tests/test_async_tcp_service.cpp

@@ -72,9 +72,9 @@ struct echo_block_service : public async_tcp_service<echo_block_service> {
   }
 };
 
-class AsyncTcpServiceTest : public ::testing::Test {};
+class AsyncTcpServiceV4Test : public ::testing::Test {};
 
-TEST_F(AsyncTcpServiceTest, StartTest)
+TEST_F(AsyncTcpServiceV4Test, StartTest)
 {
   using namespace io::socket;
 
@@ -97,7 +97,7 @@ TEST_F(AsyncTcpServiceTest, StartTest)
   while (ctx.poller.wait());
 }
 
-TEST_F(AsyncTcpServiceTest, EchoTest)
+TEST_F(AsyncTcpServiceV4Test, EchoTest)
 {
   using namespace io::socket;
 
@@ -149,7 +149,7 @@ TEST_F(AsyncTcpServiceTest, EchoTest)
   while (ctx.poller.wait());
 }
 
-TEST_F(AsyncTcpServiceTest, InitializeError)
+TEST_F(AsyncTcpServiceV4Test, InitializeError)
 {
   using namespace io::socket;
 
@@ -176,7 +176,7 @@ TEST_F(AsyncTcpServiceTest, InitializeError)
   while (ctx.poller.wait());
 }
 
-TEST_F(AsyncTcpServiceTest, AsyncServiceTest)
+TEST_F(AsyncTcpServiceV4Test, AsyncServiceTest)
 {
   using namespace io::socket;
   using service_type = async_service<echo_block_service>;
@@ -218,4 +218,151 @@ TEST_F(AsyncTcpServiceTest, AsyncServiceTest)
     }
   }
 }
+
+class AsyncTcpServiceV6Test : public ::testing::Test {};
+
+TEST_F(AsyncTcpServiceV6Test, StartTestV6)
+{
+  using namespace io::socket;
+
+  auto ctx = async_context{};
+  auto addr = socket_address<sockaddr_in>();
+  addr->sin_family = AF_INET6;
+  auto service = echo_block_service{addr};
+
+  ctx.interrupt = [&] {
+    auto sigmask = ctx.sigmask.exchange(0);
+    for (int signum = 0; auto mask = (sigmask >> signum); ++signum)
+    {
+      if (mask & (1 << 0))
+        service.signal_handler(signum);
+    }
+  };
+
+  service.start(ctx);
+  ctx.signal(ctx.terminate);
+  while (ctx.poller.wait());
+}
+
+TEST_F(AsyncTcpServiceV6Test, EchoTest)
+{
+  using namespace io::socket;
+
+  auto ctx = async_context();
+  auto addr = socket_address<sockaddr_in6>();
+  addr->sin6_family = AF_INET6;
+  addr->sin6_port = htons(8080);
+  auto service = echo_block_service(addr);
+
+  ctx.interrupt = [&] {
+    auto sigmask = ctx.sigmask.exchange(0);
+    for (int signum = 0; auto mask = (sigmask >> signum); ++signum)
+    {
+      if (mask & (1 << 0))
+        service.signal_handler(signum);
+    }
+  };
+
+  ASSERT_FALSE(service.initialized);
+  service.start(ctx);
+  {
+    ASSERT_TRUE(service.initialized);
+    ASSERT_FALSE(ctx.scope.get_stop_token().stop_requested());
+
+    using namespace io;
+    auto sock = socket_handle(AF_INET6, SOCK_STREAM, 0);
+    addr->sin6_addr = IN6ADDR_LOOPBACK_INIT;
+
+    ASSERT_EQ(connect(sock, addr), 0);
+    ctx.poller.wait();
+
+    auto buf = std::array<char, 1>{'x'};
+    auto msg = socket_message{.buffers = buf};
+
+    const char *alphabet = "abcdefghijklmnopqrstuvwxyz";
+    auto *end = alphabet + 26;
+
+    for (auto *it = alphabet; it != end; ++it)
+    {
+      ASSERT_EQ(sendmsg(sock, socket_message{.buffers = std::span(it, 1)}, 0),
+                1);
+      ctx.poller.wait();
+      ASSERT_EQ(recvmsg(sock, msg, 0), 1);
+      EXPECT_EQ(buf[0], *it);
+    }
+  }
+
+  ctx.signal(ctx.terminate);
+  while (ctx.poller.wait());
+}
+
+TEST_F(AsyncTcpServiceV6Test, InitializeError)
+{
+  using namespace io::socket;
+
+  auto ctx = async_context();
+  auto addr = socket_address<sockaddr_in6>();
+  addr->sin6_family = AF_INET6;
+  addr->sin6_port = htons(8080);
+  auto service = echo_block_service(addr);
+  service.initialized = true;
+
+  ctx.interrupt = [&] {
+    auto sigmask = ctx.sigmask.exchange(0);
+    for (int signum = 0; auto mask = (sigmask >> signum); ++signum)
+    {
+      if (mask & (1 << 0))
+        service.signal_handler(signum);
+    }
+  };
+
+  service.start(ctx);
+  EXPECT_TRUE(ctx.scope.get_stop_token().stop_requested());
+
+  ctx.signal(ctx.terminate);
+  while (ctx.poller.wait());
+}
+
+TEST_F(AsyncTcpServiceV6Test, AsyncServiceTest)
+{
+  using namespace io::socket;
+  using service_type = async_service<echo_block_service>;
+
+  auto list = std::list<service_type>{};
+  auto &service = list.emplace_back();
+
+  std::mutex mtx;
+  std::condition_variable cvar;
+  auto addr = socket_address<sockaddr_in6>();
+  addr->sin6_family = AF_INET6;
+  addr->sin6_port = htons(8081);
+
+  service.start(mtx, cvar, addr);
+  {
+    auto lock = std::unique_lock{mtx};
+    cvar.wait(lock, [&] { return service.interrupt || service.stopped; });
+  }
+  ASSERT_TRUE(static_cast<bool>(service.interrupt));
+  {
+    using namespace io;
+    auto sock = socket_handle(AF_INET6, SOCK_STREAM, 0);
+    addr->sin6_addr = IN6ADDR_LOOPBACK_INIT;
+
+    ASSERT_EQ(connect(sock, addr), 0);
+
+    auto buf = std::array<char, 1>{'x'};
+    auto msg = socket_message{.buffers = buf};
+
+    const char *alphabet = "abcdefghijklmnopqrstuvwxyz";
+    auto *end = alphabet + 26;
+
+    for (auto *it = alphabet; it != end; ++it)
+    {
+      ASSERT_EQ(sendmsg(sock, socket_message{.buffers = std::span(it, 1)}, 0),
+                1);
+      ASSERT_EQ(recvmsg(sock, msg, 0), 1);
+      EXPECT_EQ(buf[0], *it);
+    }
+  }
+}
 // NOLINTEND