commit a3b8214eff9d3e846b7832e5723bd3d914e69045
| author | 斟酌 鵬兄 <tgckpg@gmail.com> |
|---|
| date | 2026-05-29T02:24:26Z |
|---|
| subject | Because windows |
|---|
commit a3b8214eff9d3e846b7832e5723bd3d914e69045
Author: 斟酌 鵬兄 <tgckpg@gmail.com>
Date: 2026-05-29T02:24:26Z
Because windows
---
src/stream_listener.c | 94 ++++++++++++++++++++++++++++++++------------
tests/test_unix_listeners.py | 75 +++++++++++++++++++++++------------
2 files changed, 118 insertions(+), 51 deletions(-)
diff --git a/src/stream_listener.c b/src/stream_listener.c
index 07e1c33..3813fa4 100644
--- a/src/stream_listener.c
+++ b/src/stream_listener.c
@@ -85,7 +85,10 @@ static int bind_unix_stream_listener(struct stream_route_ctx *ctx)
const struct endpoint *ep = &ctx->route->listen;
const char *path = ep->path;
struct sockaddr_un sa;
+ evutil_socket_t fd = -1;
struct evconnlistener *listener;
+ size_t path_len;
+ int rc;
if (path[0] == '\0') {
LOG_ERROR("empty unix stream listen path",
@@ -97,14 +100,15 @@ static int bind_unix_stream_listener(struct stream_route_ctx *ctx)
memset(&sa, 0, sizeof(sa));
sa.sun_family = AF_UNIX;
- if (strlen(path) >= sizeof(sa.sun_path)) {
+ path_len = strlen(path);
+ if (path_len >= sizeof(sa.sun_path)) {
LOG_ERROR("unix stream listen path too long",
"line", _LOGV(ctx->route->line_no),
"path", _LOGV(path));
return -ENAMETOOLONG;
}
- memcpy(sa.sun_path, path, strlen(path) + 1);
+ memcpy(sa.sun_path, path, path_len + 1);
if (compat_unlink(path) < 0 && errno != ENOENT) {
int err = errno;
@@ -116,35 +120,75 @@ static int bind_unix_stream_listener(struct stream_route_ctx *ctx)
return -err;
}
- listener = evconnlistener_new_bind(
+ fd = socket(AF_UNIX, SOCK_STREAM, 0);
+ if (fd < 0) {
+ int err = EVUTIL_SOCKET_ERROR();
+
+ LOG_ERROR("failed to create unix stream listener socket",
+ "line", _LOGV(ctx->route->line_no),
+ "path", _LOGV(path),
+ "err", _LOGV(evutil_socket_error_to_string(err)));
+ return err ? -err : -EIO;
+ }
+
+#ifdef _WIN32
+ /*
+ * Windows AF_UNIX is picky. Do not apply TCP-ish listener options
+ * such as SO_KEEPALIVE / SO_REUSEADDR here.
+ */
+#else
+ if (evutil_make_listen_socket_reuseable(fd) < 0) {
+ int err = errno;
+
+ LOG_ERROR("failed to make unix stream listener reusable",
+ "line", _LOGV(ctx->route->line_no),
+ "path", _LOGV(path),
+ "err", _LOGV(strerror(err)));
+ evutil_closesocket(fd);
+ return -err;
+ }
+#endif
+
+ if (evutil_make_socket_nonblocking(fd) < 0) {
+ int err = EVUTIL_SOCKET_ERROR();
+
+ LOG_ERROR("failed to make unix stream listener nonblocking",
+ "line", _LOGV(ctx->route->line_no),
+ "path", _LOGV(path),
+ "err", _LOGV(evutil_socket_error_to_string(err)));
+ evutil_closesocket(fd);
+ return err ? -err : -EIO;
+ }
+
+ rc = bind(fd, (struct sockaddr *)&sa, sizeof(sa));
+ if (rc < 0) {
+ int err = EVUTIL_SOCKET_ERROR();
+
+ LOG_ERROR("failed to bind unix stream listener socket",
+ "line", _LOGV(ctx->route->line_no),
+ "path", _LOGV(path),
+ "err", _LOGV(evutil_socket_error_to_string(err)));
+ evutil_closesocket(fd);
+ return err ? -err : -EIO;
+ }
+
+ listener = evconnlistener_new(
ctx->accept_base,
accept_cb,
ctx,
- LEV_OPT_CLOSE_ON_FREE | LEV_OPT_REUSEABLE,
+ LEV_OPT_CLOSE_ON_FREE,
-1,
- (struct sockaddr *)&sa,
- sizeof(sa)
+ fd
);
-
if (listener == NULL) {
- int sockerr = EVUTIL_SOCKET_ERROR();
- int syserr = errno;
-
- LOG_ERROR("failed to bind unix stream listener",
- "line", _LOGV(ctx->route->line_no),
- "path", _LOGV(path),
- "sockerr", _LOGV(sockerr),
- "sockerrstr", _LOGV(evutil_socket_error_to_string(sockerr)),
- "errno", _LOGV(syserr),
- "errnostr", _LOGV(strerror(syserr)));
-
- if (sockerr != 0) {
- return -sockerr;
- }
- if (syserr != 0) {
- return -syserr;
- }
- return -EIO;
+ int err = EVUTIL_SOCKET_ERROR();
+
+ LOG_ERROR("failed to create unix stream listener",
+ "line", _LOGV(ctx->route->line_no),
+ "path", _LOGV(path),
+ "err", _LOGV(evutil_socket_error_to_string(err)));
+ evutil_closesocket(fd);
+ return err ? -err : -EIO;
}
ctx->listener = listener;
diff --git a/tests/test_unix_listeners.py b/tests/test_unix_listeners.py
index ccae43f..e9427e0 100644
--- a/tests/test_unix_listeners.py
+++ b/tests/test_unix_listeners.py
@@ -64,6 +64,39 @@ async def close_writer(writer: asyncio.StreamWriter) -> None:
async def open_unix_connection(path: str):
return await asyncio.open_unix_connection(path)
+def unix_stream_roundtrip_sync(path: str, payload: bytes) -> bytes:
+ with socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) as s:
+ s.settimeout(3.0)
+ s.connect(path)
+ s.sendall(payload)
+
+ chunks = []
+ remaining = len(payload)
+
+ while remaining > 0:
+ data = s.recv(remaining)
+ if not data:
+ break
+ chunks.append(data)
+ remaining -= len(data)
+
+ return b"".join(chunks)
+
+
+async def unix_stream_roundtrip(path: str, payload: bytes) -> bytes:
+ return await asyncio.to_thread(unix_stream_roundtrip_sync, path, payload)
+
+
+def unix_stream_read_sync(path: str) -> bytes:
+ with socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) as s:
+ s.settimeout(3.0)
+ s.connect(path)
+ return s.recv(65536)
+
+
+async def unix_stream_read(path: str) -> bytes:
+ return await asyncio.to_thread(unix_stream_read_sync, path)
+
async def udp_echo_server(host: str, port: int):
loop = asyncio.get_running_loop()
@@ -121,24 +154,16 @@ async def test_unix_stream_to_tcp() -> None:
conf_text=conf_text,
proto="unix",
):
- reader, writer = await open_unix_connection(UNIX_STREAM_SOCK)
-
- try:
- payload = b"hello over unix stream\n"
+ payload = b"hello over unix stream\n"
- writer.write(payload)
- await writer.drain()
-
- got = await asyncio.wait_for(
- reader.readexactly(len(payload)),
- timeout=3.0,
- )
+ got = await asyncio.wait_for(
+ unix_stream_roundtrip(UNIX_STREAM_SOCK, payload),
+ timeout=3.0,
+ )
- assert got == payload, (
- f"unix stream roundtrip mismatch: got={got!r} expected={payload!r}"
- )
- finally:
- await close_writer(writer)
+ assert got == payload, (
+ f"unix stream roundtrip mismatch: got={got!r} expected={payload!r}"
+ )
finally:
backend_server.close()
await backend_server.wait_closed()
@@ -162,17 +187,15 @@ async def test_unix_stream_to_builtin_client_addr() -> None:
conf_text=conf_text,
proto="unix",
):
- reader, writer = await open_unix_connection(UNIX_BUILTIN_SOCK)
-
- try:
- got = await asyncio.wait_for(reader.read(65536), timeout=3.0)
+ got = await asyncio.wait_for(
+ unix_stream_read(UNIX_BUILTIN_SOCK),
+ timeout=3.0,
+ )
- assert got, "expected builtin client_addr response"
- assert b"unix" in got.lower() or b"unknown" in got.lower() or got.strip(), (
- f"unexpected builtin client_addr response: {got!r}"
- )
- finally:
- await close_writer(writer)
+ assert got, "expected builtin client_addr response"
+ assert b"unix" in got.lower() or b"unknown" in got.lower() or got.strip(), (
+ f"unexpected builtin client_addr response: {got!r}"
+ )
finally:
unlink_if_exists(UNIX_BUILTIN_SOCK)