From 9a3fefc85615f0a530fed82c6afcb88286a0e385 Mon Sep 17 00:00:00 2001 From: Aster Seker Date: Wed, 1 Jul 2026 18:42:42 +0300 Subject: [PATCH 1/7] feat(client): add synchronous named pipe client Add a C++ NamedPipeClient with synchronous connect, read, write, timeout read, and basic callbacks. Wire a static client target, add a Windows server/client round-trip test, extend ODR coverage, and make .ipp definitions emit out-of-line symbols in static-library mode. --- CMakeLists.txt | 24 +- README.md | 22 + include/SimpleNamedPipe/NamedPipeClient.hpp | 134 ++++++ .../NamedPipeClient/ClientConfig.hpp | 36 ++ .../NamedPipeClient/NamedPipeClient.ipp | 429 ++++++++++++++++++ .../NamedPipeServer/NamedPipeServer.ipp | 70 +-- src/NamedPipeClient.cpp | 10 + tests/client_server_test.cpp | 133 ++++++ tests/odr/main1.cpp | 3 + tests/odr/main2.cpp | 3 + tests/stubs/windows.h | 9 +- 11 files changed, 839 insertions(+), 34 deletions(-) create mode 100644 include/SimpleNamedPipe/NamedPipeClient.hpp create mode 100644 include/SimpleNamedPipe/NamedPipeClient/ClientConfig.hpp create mode 100644 include/SimpleNamedPipe/NamedPipeClient/NamedPipeClient.ipp create mode 100644 src/NamedPipeClient.cpp create mode 100644 tests/client_server_test.cpp diff --git a/CMakeLists.txt b/CMakeLists.txt index f5ff738..465819f 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -6,6 +6,7 @@ set(CMAKE_CXX_STANDARD_REQUIRED ON) option(SIMPLE_NAMED_PIPE_BUILD_EXAMPLES "Build example programs" ON) option(SIMPLE_NAMED_PIPE_BUILD_STATIC "Build static library from .ipp implementation" ON) +option(SIMPLE_NAMED_PIPE_BUILD_TESTS "Build test programs" OFF) # Header-only library add_library(SimpleNamedPipe INTERFACE) @@ -17,10 +18,17 @@ if(SIMPLE_NAMED_PIPE_BUILD_STATIC) src/NamedPipeServer.cpp ) + add_library(SimpleNamedPipeClient STATIC + src/NamedPipeClient.cpp + ) + target_include_directories(SimpleNamedPipeServer PUBLIC ${CMAKE_CURRENT_SOURCE_DIR}/include) + target_include_directories(SimpleNamedPipeClient PUBLIC ${CMAKE_CURRENT_SOURCE_DIR}/include) target_compile_definitions(SimpleNamedPipeServer PUBLIC SIMPLE_NAMED_PIPE_STATIC_LIB) - - set_target_properties(SimpleNamedPipeServer PROPERTIES OUTPUT_NAME snp_server) + target_compile_definitions(SimpleNamedPipeClient PUBLIC SIMPLE_NAMED_PIPE_STATIC_LIB) + + set_target_properties(SimpleNamedPipeServer PROPERTIES OUTPUT_NAME snp_server) + set_target_properties(SimpleNamedPipeClient PROPERTIES OUTPUT_NAME snp_client) endif() # Examples @@ -34,6 +42,18 @@ if(SIMPLE_NAMED_PIPE_BUILD_EXAMPLES) target_link_libraries(${EXAMPLE_NAME} PRIVATE SimpleNamedPipe) if(SIMPLE_NAMED_PIPE_BUILD_STATIC) target_link_libraries(${EXAMPLE_NAME} PRIVATE SimpleNamedPipeServer) + target_link_libraries(${EXAMPLE_NAME} PRIVATE SimpleNamedPipeClient) endif() endforeach() endif() + +if(SIMPLE_NAMED_PIPE_BUILD_TESTS AND WIN32) + add_executable(client_server_test + tests/client_server_test.cpp + ) + target_include_directories(client_server_test PRIVATE ${CMAKE_CURRENT_SOURCE_DIR}/include) + target_link_libraries(client_server_test PRIVATE SimpleNamedPipe) + if(SIMPLE_NAMED_PIPE_BUILD_STATIC) + target_link_libraries(client_server_test PRIVATE SimpleNamedPipeServer SimpleNamedPipeClient) + endif() +endif() diff --git a/README.md b/README.md index 14364fd..27c3ba6 100644 --- a/README.md +++ b/README.md @@ -14,6 +14,7 @@ For the documentation in Russian, see [README-RU.md](README-RU.md). - supports up to 256 simultaneous clients; - send queue with limits on message size and count; - event notifications via callbacks or the `ServerEventHandler` class; +- synchronous C++ client for tests and lightweight local integrations; - lightweight MQL5 client with optional global callbacks; - the MQL5 client performs read/write synchronously; call `update()` for polling (e.g., in a timer). @@ -38,6 +39,27 @@ int main() { } ``` +### Minimal C++ client + +```cpp +#include "SimpleNamedPipe/NamedPipeClient.hpp" +using namespace SimpleNamedPipe; + +int main() { + NamedPipeClient client({"ExamplePipe"}); + std::error_code error; + + if (!client.connect(&error)) + return 1; + + client.write("ping", &error); + + std::string response; + client.read(response, 5000, &error); + client.close(); +} +``` + ### Minimal MQL5 client ```mql5 diff --git a/include/SimpleNamedPipe/NamedPipeClient.hpp b/include/SimpleNamedPipe/NamedPipeClient.hpp new file mode 100644 index 0000000..d2de8c6 --- /dev/null +++ b/include/SimpleNamedPipe/NamedPipeClient.hpp @@ -0,0 +1,134 @@ +#pragma once +#ifndef _SIMPLE_NAMED_PIPE_CLIENT_HPP_INCLUDED +#define _SIMPLE_NAMED_PIPE_CLIENT_HPP_INCLUDED + +/// \file NamedPipeClient.hpp +/// \brief Synchronous named pipe client for tests and lightweight integrations. + +#include "NamedPipeClient/ClientConfig.hpp" +#include "NamedPipeServer/errors.hpp" + +#include +#include +#include +#include +#include +#include +#include +#include + +namespace SimpleNamedPipe { + + /// \class NamedPipeClient + /// \brief Synchronous client for message-mode Windows named pipes. + class NamedPipeClient final { + public: + /// \brief Construct client with default configuration. + NamedPipeClient(); + + /// \brief Construct client with explicit configuration. + /// \param config Configuration to use. + explicit NamedPipeClient(const ClientConfig& config); + + /// \brief Destructor closes the pipe handle if it is open. + ~NamedPipeClient(); + + NamedPipeClient(const NamedPipeClient&) = delete; + NamedPipeClient& operator=(const NamedPipeClient&) = delete; + + /// \brief Applies a new client configuration. + /// \param config Configuration to use for subsequent connect calls. + void set_config(const ClientConfig& config); + + /// \brief Returns current client configuration. + /// \return Copy of the current configuration. + ClientConfig get_config() const; + + /// \brief Opens the pipe configured in ClientConfig. + /// \param error Optional output error code. + /// \return true on success. + bool connect(std::error_code* error = nullptr); + + /// \brief Opens the named pipe and stores that name in the configuration. + /// \param pipe_name Pipe name or full \\.\pipe\ path. + /// \param error Optional output error code. + /// \return true on success. + bool open(const std::string& pipe_name, std::error_code* error = nullptr); + + /// \brief Closes the pipe handle. + void close(); + + /// \brief Returns true while a pipe handle is open. + bool is_connected() const; + + /// \brief Alias for is_connected(). + bool connected() const; + + /// \brief Writes a complete message to the pipe. + /// \param message UTF-8 message. + /// \param error Optional output error code. + /// \return true when the message was fully written. + bool write(const std::string& message, std::error_code* error = nullptr); + + /// \brief Reads one complete message, blocking until data is available. + /// \param message Output message. + /// \param error Optional output error code. + /// \return true when a message was read. + bool read(std::string& message, std::error_code* error = nullptr); + + /// \brief Reads one message if bytes are already available. + /// \param message Output message. + /// \param error Optional output error code. + /// \return true when a message was read, false when no message is available or on error. + bool try_read(std::string& message, std::error_code* error = nullptr); + + /// \brief Waits for one message up to timeout_ms and reads it. + /// \param message Output message. + /// \param timeout_ms Timeout in milliseconds. + /// \param error Optional output error code. + /// \return true when a message was read. + bool read(std::string& message, size_t timeout_ms, std::error_code* error = nullptr); + + /// \brief Returns number of bytes currently available in the pipe. + /// \param error Optional output error code. + /// \return Available byte count, or 0 when disconnected or on error. + size_t available(std::error_code* error = nullptr) const; + + /// \brief Flushes pipe write buffers. + /// \param error Optional output error code. + /// \return true on success. + bool flush(std::error_code* error = nullptr); + + /// \brief Returns native Windows pipe handle. + HANDLE native_handle() const; + + std::function on_connected; ///< Called after a successful connect. + std::function on_disconnected; ///< Called after close of an active connection. + std::function on_message; ///< Called by try_read/read timeout overloads. + std::function on_error; ///< Called when an operation fails. + + private: + ClientConfig m_config; + mutable std::mutex m_mutex; + HANDLE m_pipe = INVALID_HANDLE_VALUE; + std::atomic m_is_connected{false}; + + static std::wstring make_pipe_path(const std::string& pipe_name); + static DWORD to_dword_timeout(size_t timeout_ms); + + bool connect_no_lock(std::error_code* error); + bool read_no_lock(std::string& message, std::error_code* error); + void close_no_lock(bool notify); + void set_error(std::error_code* out, const std::error_code& error) const; + void clear_error(std::error_code* out) const; + }; + +} // namespace SimpleNamedPipe + +/// \note Implementation is included only in header-only mode. +/// When building as a static library, do NOT include the .ipp here. +#ifndef SIMPLE_NAMED_PIPE_STATIC_LIB +#include "NamedPipeClient/NamedPipeClient.ipp" +#endif + +#endif // _SIMPLE_NAMED_PIPE_CLIENT_HPP_INCLUDED diff --git a/include/SimpleNamedPipe/NamedPipeClient/ClientConfig.hpp b/include/SimpleNamedPipe/NamedPipeClient/ClientConfig.hpp new file mode 100644 index 0000000..9efd76d --- /dev/null +++ b/include/SimpleNamedPipe/NamedPipeClient/ClientConfig.hpp @@ -0,0 +1,36 @@ +#pragma once +#ifndef _SIMPLE_NAMED_PIPE_CLIENT_CONFIG_HPP_INCLUDED +#define _SIMPLE_NAMED_PIPE_CLIENT_CONFIG_HPP_INCLUDED + +/// \file ClientConfig.hpp +/// \brief Configuration for the named pipe client. + +#include +#include + +namespace SimpleNamedPipe { + + /// \class ClientConfig + /// \brief Named pipe client configuration. + class ClientConfig { + public: + std::string pipe_name; ///< Named pipe name or full \\.\pipe\ path. + size_t buffer_size; ///< Size of I/O buffers. + size_t timeout; ///< Connect/read polling timeout in milliseconds. + + /// \brief Construct with optional parameters. + /// \param pipe_name Name of the pipe or full pipe path. + /// \param buffer_size Buffer size in bytes. + /// \param timeout Wait timeout in milliseconds. + ClientConfig( + const std::string& pipe_name = "server", + size_t buffer_size = 65536, + size_t timeout = 5000) + : pipe_name(pipe_name), + buffer_size(buffer_size), + timeout(timeout) {} + }; + +} // namespace SimpleNamedPipe + +#endif // _SIMPLE_NAMED_PIPE_CLIENT_CONFIG_HPP_INCLUDED diff --git a/include/SimpleNamedPipe/NamedPipeClient/NamedPipeClient.ipp b/include/SimpleNamedPipe/NamedPipeClient/NamedPipeClient.ipp new file mode 100644 index 0000000..ae1ddd5 --- /dev/null +++ b/include/SimpleNamedPipe/NamedPipeClient/NamedPipeClient.ipp @@ -0,0 +1,429 @@ +#ifdef SIMPLE_NAMED_PIPE_STATIC_LIB +#include "../NamedPipeClient.hpp" +#endif + +#include +#include +#include +#include +#include + +#ifndef SIMPLE_NAMED_PIPE_INLINE +#ifdef SIMPLE_NAMED_PIPE_STATIC_LIB +#define SIMPLE_NAMED_PIPE_INLINE +#else +#define SIMPLE_NAMED_PIPE_INLINE inline +#endif +#endif + +namespace SimpleNamedPipe { + + SIMPLE_NAMED_PIPE_INLINE NamedPipeClient::NamedPipeClient() = default; + + SIMPLE_NAMED_PIPE_INLINE NamedPipeClient::NamedPipeClient(const ClientConfig& config) + : m_config(config) {} + + SIMPLE_NAMED_PIPE_INLINE NamedPipeClient::~NamedPipeClient() { + close(); + } + + SIMPLE_NAMED_PIPE_INLINE void NamedPipeClient::set_config(const ClientConfig& config) { + std::lock_guard lock(m_mutex); + m_config = config; + } + + SIMPLE_NAMED_PIPE_INLINE ClientConfig NamedPipeClient::get_config() const { + std::lock_guard lock(m_mutex); + return m_config; + } + + SIMPLE_NAMED_PIPE_INLINE bool NamedPipeClient::connect(std::error_code* error) { + std::error_code ec; + bool result = false; + { + std::lock_guard lock(m_mutex); + result = connect_no_lock(&ec); + } + set_error(error, ec); + if (result) { + if (on_connected) on_connected(); + } else if (ec && on_error) { + on_error(ec); + } + return result; + } + + SIMPLE_NAMED_PIPE_INLINE bool NamedPipeClient::open(const std::string& pipe_name, std::error_code* error) { + std::error_code ec; + bool result = false; + { + std::lock_guard lock(m_mutex); + m_config.pipe_name = pipe_name; + result = connect_no_lock(&ec); + } + set_error(error, ec); + if (result) { + if (on_connected) on_connected(); + } else if (ec && on_error) { + on_error(ec); + } + return result; + } + + SIMPLE_NAMED_PIPE_INLINE void NamedPipeClient::close() { + bool notify = false; + { + std::lock_guard lock(m_mutex); + notify = m_is_connected.load(std::memory_order_acquire); + close_no_lock(false); + } + if (notify && on_disconnected) { + on_disconnected(); + } + } + + SIMPLE_NAMED_PIPE_INLINE bool NamedPipeClient::is_connected() const { + return m_is_connected.load(std::memory_order_acquire); + } + + SIMPLE_NAMED_PIPE_INLINE bool NamedPipeClient::connected() const { + return is_connected(); + } + + SIMPLE_NAMED_PIPE_INLINE bool NamedPipeClient::write(const std::string& message, std::error_code* error) { + std::error_code ec; + bool result = false; + { + std::lock_guard lock(m_mutex); + if (!m_is_connected.load(std::memory_order_acquire) || + m_pipe == INVALID_HANDLE_VALUE) { + ec = make_error_code(NamedPipeErrc::NotConnected); + } else if (message.empty()) { + ec = std::make_error_code(std::errc::invalid_argument); + } else if (message.size() > (std::numeric_limits::max)()) { + ec = std::make_error_code(std::errc::message_size); + } else { + DWORD bytes_written = 0; + const BOOL ok = WriteFile( + m_pipe, + message.data(), + static_cast(message.size()), + &bytes_written, + nullptr); + if (!ok || bytes_written != message.size()) { + ec = std::error_code(static_cast(GetLastError()), std::system_category()); + close_no_lock(false); + } else { + result = true; + } + } + } + set_error(error, ec); + if (!result && ec && on_error) { + on_error(ec); + } + return result; + } + + SIMPLE_NAMED_PIPE_INLINE bool NamedPipeClient::read(std::string& message, std::error_code* error) { + std::error_code ec; + bool result = false; + { + std::lock_guard lock(m_mutex); + result = read_no_lock(message, &ec); + } + set_error(error, ec); + if (result) { + if (on_message) on_message(message); + } else if (ec && on_error) { + on_error(ec); + } + return result; + } + + SIMPLE_NAMED_PIPE_INLINE bool NamedPipeClient::try_read(std::string& message, std::error_code* error) { + std::error_code ec; + bool result = false; + bool no_message = false; + { + std::lock_guard lock(m_mutex); + if (!m_is_connected.load(std::memory_order_acquire) || + m_pipe == INVALID_HANDLE_VALUE) { + ec = make_error_code(NamedPipeErrc::NotConnected); + } else { + DWORD bytes_available = 0; + const BOOL ok = PeekNamedPipe( + m_pipe, + nullptr, + 0, + nullptr, + &bytes_available, + nullptr); + if (!ok) { + ec = std::error_code(static_cast(GetLastError()), std::system_category()); + close_no_lock(false); + } else if (bytes_available == 0) { + no_message = true; + } else { + result = read_no_lock(message, &ec); + } + } + } + set_error(error, ec); + if (result) { + if (on_message) on_message(message); + } else if (!no_message && ec && on_error) { + on_error(ec); + } + return result; + } + + SIMPLE_NAMED_PIPE_INLINE bool NamedPipeClient::read( + std::string& message, + size_t timeout_ms, + std::error_code* error) { + const auto start = std::chrono::steady_clock::now(); + const auto timeout = std::chrono::milliseconds(timeout_ms); + + for (;;) { + std::error_code ec; + if (try_read(message, &ec)) { + clear_error(error); + return true; + } + if (ec) { + set_error(error, ec); + return false; + } + + const auto now = std::chrono::steady_clock::now(); + if (now - start >= timeout) { + const auto timeout_error = std::error_code(WAIT_TIMEOUT, std::system_category()); + set_error(error, timeout_error); + if (on_error) on_error(timeout_error); + return false; + } + + std::this_thread::sleep_for(std::chrono::milliseconds(1)); + } + } + + SIMPLE_NAMED_PIPE_INLINE size_t NamedPipeClient::available(std::error_code* error) const { + std::error_code ec; + DWORD bytes_available = 0; + { + std::lock_guard lock(m_mutex); + if (!m_is_connected.load(std::memory_order_acquire) || + m_pipe == INVALID_HANDLE_VALUE) { + ec = make_error_code(NamedPipeErrc::NotConnected); + } else { + const BOOL ok = PeekNamedPipe( + m_pipe, + nullptr, + 0, + nullptr, + &bytes_available, + nullptr); + if (!ok) { + ec = std::error_code(static_cast(GetLastError()), std::system_category()); + } + } + } + set_error(error, ec); + if (ec && on_error) { + on_error(ec); + } + return static_cast(bytes_available); + } + + SIMPLE_NAMED_PIPE_INLINE bool NamedPipeClient::flush(std::error_code* error) { + std::error_code ec; + bool result = false; + { + std::lock_guard lock(m_mutex); + if (!m_is_connected.load(std::memory_order_acquire) || + m_pipe == INVALID_HANDLE_VALUE) { + ec = make_error_code(NamedPipeErrc::NotConnected); + } else if (!FlushFileBuffers(m_pipe)) { + ec = std::error_code(static_cast(GetLastError()), std::system_category()); + close_no_lock(false); + } else { + result = true; + } + } + set_error(error, ec); + if (!result && ec && on_error) { + on_error(ec); + } + return result; + } + + SIMPLE_NAMED_PIPE_INLINE HANDLE NamedPipeClient::native_handle() const { + std::lock_guard lock(m_mutex); + return m_pipe; + } + + SIMPLE_NAMED_PIPE_INLINE std::wstring NamedPipeClient::make_pipe_path(const std::string& pipe_name) { + static const std::string prefix = "\\\\.\\pipe\\"; + const std::string full_name = + pipe_name.compare(0, prefix.size(), prefix) == 0 + ? pipe_name + : prefix + pipe_name; + + std::wstring_convert> conv; + return conv.from_bytes(full_name); + } + + SIMPLE_NAMED_PIPE_INLINE DWORD NamedPipeClient::to_dword_timeout(size_t timeout_ms) { + const size_t max_value = (std::numeric_limits::max)(); + return static_cast((std::min)(timeout_ms, max_value)); + } + + SIMPLE_NAMED_PIPE_INLINE bool NamedPipeClient::connect_no_lock(std::error_code* error) { + close_no_lock(false); + clear_error(error); + + if (m_config.pipe_name.empty()) { + set_error(error, std::make_error_code(std::errc::invalid_argument)); + return false; + } + if (m_config.buffer_size == 0) { + set_error(error, std::make_error_code(std::errc::invalid_argument)); + return false; + } + + const std::wstring pipe_path = make_pipe_path(m_config.pipe_name); + const auto start = std::chrono::steady_clock::now(); + const auto timeout = std::chrono::milliseconds(m_config.timeout); + + for (;;) { + HANDLE pipe = CreateFileW( + pipe_path.c_str(), + GENERIC_READ | GENERIC_WRITE, + 0, + nullptr, + OPEN_EXISTING, + 0, + nullptr); + + if (pipe != INVALID_HANDLE_VALUE) { + DWORD mode = PIPE_READMODE_MESSAGE; + if (!SetNamedPipeHandleState(pipe, &mode, nullptr, nullptr)) { + const auto ec = std::error_code( + static_cast(GetLastError()), + std::system_category()); + CloseHandle(pipe); + set_error(error, ec); + return false; + } + + m_pipe = pipe; + m_is_connected.store(true, std::memory_order_release); + clear_error(error); + return true; + } + + const DWORD open_error = GetLastError(); + if (open_error != ERROR_PIPE_BUSY && + open_error != ERROR_FILE_NOT_FOUND) { + set_error( + error, + std::error_code(static_cast(open_error), std::system_category())); + return false; + } + + const auto now = std::chrono::steady_clock::now(); + if (now - start >= timeout) { + set_error(error, std::error_code(WAIT_TIMEOUT, std::system_category())); + return false; + } + + const auto elapsed = std::chrono::duration_cast(now - start); + const auto remaining = + elapsed >= timeout + ? std::chrono::milliseconds(0) + : (timeout - elapsed); + const DWORD wait_ms = to_dword_timeout( + static_cast((std::min)(remaining, std::chrono::milliseconds(50)).count())); + + if (open_error == ERROR_PIPE_BUSY) { + WaitNamedPipeW(pipe_path.c_str(), wait_ms); + } else { + std::this_thread::sleep_for(std::chrono::milliseconds(wait_ms > 0 ? wait_ms : 1)); + } + } + } + + SIMPLE_NAMED_PIPE_INLINE bool NamedPipeClient::read_no_lock(std::string& message, std::error_code* error) { + clear_error(error); + message.clear(); + + if (!m_is_connected.load(std::memory_order_acquire) || + m_pipe == INVALID_HANDLE_VALUE) { + set_error(error, make_error_code(NamedPipeErrc::NotConnected)); + return false; + } + + std::vector buffer(m_config.buffer_size); + + for (;;) { + DWORD bytes_read = 0; + const BOOL ok = ReadFile( + m_pipe, + buffer.data(), + static_cast(buffer.size()), + &bytes_read, + nullptr); + const DWORD read_error = GetLastError(); + + if (bytes_read > 0) { + message.append(buffer.data(), bytes_read); + } + + if (ok) { + clear_error(error); + return true; + } + + if (read_error == ERROR_MORE_DATA) { + continue; + } + + const auto ec = std::error_code( + static_cast(read_error), + std::system_category()); + set_error(error, ec); + + if (read_error == ERROR_BROKEN_PIPE || + read_error == ERROR_NO_DATA) { + close_no_lock(false); + } + return false; + } + } + + SIMPLE_NAMED_PIPE_INLINE void NamedPipeClient::close_no_lock(bool notify) { + const bool was_connected = m_is_connected.load(std::memory_order_acquire); + if (m_pipe != INVALID_HANDLE_VALUE) { + CloseHandle(m_pipe); + m_pipe = INVALID_HANDLE_VALUE; + } + m_is_connected.store(false, std::memory_order_release); + if (notify && was_connected && on_disconnected) { + on_disconnected(); + } + } + + SIMPLE_NAMED_PIPE_INLINE void NamedPipeClient::set_error(std::error_code* out, const std::error_code& error) const { + if (out) { + *out = error; + } + } + + SIMPLE_NAMED_PIPE_INLINE void NamedPipeClient::clear_error(std::error_code* out) const { + if (out) { + out->clear(); + } + } + +} // namespace SimpleNamedPipe diff --git a/include/SimpleNamedPipe/NamedPipeServer/NamedPipeServer.ipp b/include/SimpleNamedPipe/NamedPipeServer/NamedPipeServer.ipp index bf2ea11..6d18467 100644 --- a/include/SimpleNamedPipe/NamedPipeServer/NamedPipeServer.ipp +++ b/include/SimpleNamedPipe/NamedPipeServer/NamedPipeServer.ipp @@ -6,31 +6,39 @@ #include #include +#ifndef SIMPLE_NAMED_PIPE_INLINE +#ifdef SIMPLE_NAMED_PIPE_STATIC_LIB +#define SIMPLE_NAMED_PIPE_INLINE +#else +#define SIMPLE_NAMED_PIPE_INLINE inline +#endif +#endif + namespace SimpleNamedPipe { - inline NamedPipeServer::NamedPipeServer() { + SIMPLE_NAMED_PIPE_INLINE NamedPipeServer::NamedPipeServer() { for (size_t i = 0; i < MAX_CLIENTS; ++i) { m_pipes[i] = INVALID_HANDLE_VALUE; } }; - inline NamedPipeServer::NamedPipeServer(const ServerConfig& config) { + SIMPLE_NAMED_PIPE_INLINE NamedPipeServer::NamedPipeServer(const ServerConfig& config) { set_config(config); }; - inline NamedPipeServer::~NamedPipeServer() { + SIMPLE_NAMED_PIPE_INLINE NamedPipeServer::~NamedPipeServer() { stop(); } - inline void NamedPipeServer::set_event_handler(std::shared_ptr handler) { + SIMPLE_NAMED_PIPE_INLINE void NamedPipeServer::set_event_handler(std::shared_ptr handler) { m_event_handler = handler; } - inline std::shared_ptr NamedPipeServer::get_event_handler() const { + SIMPLE_NAMED_PIPE_INLINE std::shared_ptr NamedPipeServer::get_event_handler() const { return m_event_handler; } - inline void NamedPipeServer::set_config(const ServerConfig& config) { + SIMPLE_NAMED_PIPE_INLINE void NamedPipeServer::set_config(const ServerConfig& config) { std::unique_lock lock(m_config_mutex); m_config = config; m_is_config_updated = true; @@ -46,12 +54,12 @@ namespace SimpleNamedPipe { } } - inline const ServerConfig NamedPipeServer::get_config() const { + SIMPLE_NAMED_PIPE_INLINE const ServerConfig NamedPipeServer::get_config() const { std::lock_guard lock(m_config_mutex); return m_config; } - inline void NamedPipeServer::start(bool run_async) { + SIMPLE_NAMED_PIPE_INLINE void NamedPipeServer::start(bool run_async) { std::lock_guard lock(m_mutex); if (m_server_thread.joinable()) { m_is_stop_server = true; @@ -70,7 +78,7 @@ namespace SimpleNamedPipe { } } - inline void NamedPipeServer::stop() { + SIMPLE_NAMED_PIPE_INLINE void NamedPipeServer::stop() { std::lock_guard lock(m_mutex); if (m_is_stop_server) return; m_is_stop_server = true; @@ -85,11 +93,11 @@ namespace SimpleNamedPipe { } } - inline bool NamedPipeServer::is_running() const { + SIMPLE_NAMED_PIPE_INLINE bool NamedPipeServer::is_running() const { return m_is_running.load(std::memory_order_acquire); } - inline void NamedPipeServer::send_to(int client_id, const std::string& message, DoneCallback on_done) { + SIMPLE_NAMED_PIPE_INLINE void NamedPipeServer::send_to(int client_id, const std::string& message, DoneCallback on_done) { HANDLE completion_port = m_completion_port.load(std::memory_order_acquire); if (!m_is_running.load(std::memory_order_acquire) || !completion_port) { @@ -111,7 +119,7 @@ namespace SimpleNamedPipe { PostQueuedCompletionStatus(completion_port, 0, CMD_TYPE_SEND | (index & CMD_INDEX_MASK), nullptr); } - inline void NamedPipeServer::close(int client_id, DoneCallback on_done) { + SIMPLE_NAMED_PIPE_INLINE void NamedPipeServer::close(int client_id, DoneCallback on_done) { HANDLE completion_port = m_completion_port.load(std::memory_order_acquire); if (!m_is_running.load(std::memory_order_acquire) || !completion_port) { @@ -128,19 +136,19 @@ namespace SimpleNamedPipe { PostQueuedCompletionStatus(completion_port, 0, CMD_TYPE_CLOSE | (index & CMD_INDEX_MASK), nullptr); } - inline bool NamedPipeServer::is_connected(int client_id) const { + SIMPLE_NAMED_PIPE_INLINE bool NamedPipeServer::is_connected(int client_id) const { size_t index = check_client_id(client_id); return m_is_connected[index].load(std::memory_order_acquire); } - inline size_t NamedPipeServer::check_client_id(int client_id) const { + SIMPLE_NAMED_PIPE_INLINE size_t NamedPipeServer::check_client_id(int client_id) const { if (client_id < 0 || static_cast(client_id) >= MAX_CLIENTS) { throw std::out_of_range("client_id is out of range"); } return static_cast(client_id); } - inline bool NamedPipeServer::check_write_limits(int client_id, const std::string& message, std::error_code& ec) { + SIMPLE_NAMED_PIPE_INLINE bool NamedPipeServer::check_write_limits(int client_id, const std::string& message, std::error_code& ec) { if (message.size() > m_write_limits.max_message_size) { ec = make_error_code(NamedPipeErrc::MessageTooLarge); return false; @@ -157,7 +165,7 @@ namespace SimpleNamedPipe { return true; } - inline void NamedPipeServer::init(const ServerConfig& config) { + SIMPLE_NAMED_PIPE_INLINE void NamedPipeServer::init(const ServerConfig& config) { m_write_limits = config.write_limits; HANDLE completion_port = CreateIoCompletionPort(INVALID_HANDLE_VALUE, nullptr, 0, 0); @@ -177,7 +185,7 @@ namespace SimpleNamedPipe { } } - inline void NamedPipeServer::create_pipe(size_t index, HANDLE completion_port, const ServerConfig& config) { + SIMPLE_NAMED_PIPE_INLINE void NamedPipeServer::create_pipe(size_t index, HANDLE completion_port, const ServerConfig& config) { std::wstring_convert> conv; std::wstring pipe_name_w = L"\\\\.\\pipe\\" + conv.from_bytes(config.pipe_name); @@ -206,7 +214,7 @@ namespace SimpleNamedPipe { } } - inline void NamedPipeServer::main_loop() { + SIMPLE_NAMED_PIPE_INLINE void NamedPipeServer::main_loop() { while (!m_is_stop_server) { std::unique_lock lock(m_config_mutex); m_config_cv.wait(lock, [this] { @@ -252,7 +260,7 @@ namespace SimpleNamedPipe { } } - inline void NamedPipeServer::run_server_loop(const ServerConfig& config) { + SIMPLE_NAMED_PIPE_INLINE void NamedPipeServer::run_server_loop(const ServerConfig& config) { notify_start(config); HANDLE completion_port = m_completion_port.load(std::memory_order_acquire); while (!m_is_stop_server) { @@ -349,7 +357,7 @@ namespace SimpleNamedPipe { } // Process all accumulated write commands - inline void NamedPipeServer::process_write_commands(size_t index) { + SIMPLE_NAMED_PIPE_INLINE void NamedPipeServer::process_write_commands(size_t index) { std::unique_lock lock(m_write_mutex); while (!m_pending_writes[index].empty()) { m_active_writes[index].push(std::move(m_pending_writes[index].front())); @@ -363,7 +371,7 @@ namespace SimpleNamedPipe { } } - inline void NamedPipeServer::handle_write_completion(size_t index, size_t bytes_transferred) { + SIMPLE_NAMED_PIPE_INLINE void NamedPipeServer::handle_write_completion(size_t index, size_t bytes_transferred) { if (!m_active_writes[index].empty()) { auto& cmd = m_active_writes[index].front(); cmd.offset += bytes_transferred; @@ -375,7 +383,7 @@ namespace SimpleNamedPipe { post_next_write(index); } - inline void NamedPipeServer::post_next_write(size_t index) { + SIMPLE_NAMED_PIPE_INLINE void NamedPipeServer::post_next_write(size_t index) { if (m_active_writes[index].empty()) { m_is_writing[index] = false; return; @@ -424,7 +432,7 @@ namespace SimpleNamedPipe { } } - inline bool NamedPipeServer::reconnect_client(size_t index, HANDLE completion_port, OVERLAPPED* ov) { + SIMPLE_NAMED_PIPE_INLINE bool NamedPipeServer::reconnect_client(size_t index, HANDLE completion_port, OVERLAPPED* ov) { memset(ov, 0, sizeof(OVERLAPPED)); BOOL connected = ConnectNamedPipe(m_pipes[index], ov); @@ -445,7 +453,7 @@ namespace SimpleNamedPipe { return true; } - inline void NamedPipeServer::handle_close(size_t index, HANDLE completion_port) { + SIMPLE_NAMED_PIPE_INLINE void NamedPipeServer::handle_close(size_t index, HANDLE completion_port) { std::unique_lock lock(m_write_mutex); if (m_pending_closes[index].empty()) return; auto on_done = std::move(m_pending_closes[index].front()); @@ -467,7 +475,7 @@ namespace SimpleNamedPipe { if (on_done) on_done(make_error_code(NamedPipeErrc::InvalidPipeHandle)); } - inline void NamedPipeServer::cleanup_pending_operations(const std::error_code& reason) { + SIMPLE_NAMED_PIPE_INLINE void NamedPipeServer::cleanup_pending_operations(const std::error_code& reason) { std::array, MAX_CLIENTS> pending_writes; std::array, MAX_CLIENTS> pending_closes; @@ -504,7 +512,7 @@ namespace SimpleNamedPipe { } } - inline void NamedPipeServer::notify_connected(size_t index) { + SIMPLE_NAMED_PIPE_INLINE void NamedPipeServer::notify_connected(size_t index) { if (m_is_connected[index].load(std::memory_order_acquire)) return; m_is_connected[index].store(true, std::memory_order_release); m_connections[index] = std::make_shared(index, this); @@ -513,7 +521,7 @@ namespace SimpleNamedPipe { if (on_event) on_event(ServerEvent::client_connected(static_cast(index), m_connections[index])); } - inline void NamedPipeServer::notify_disconnected(size_t index, const std::error_code& ec) { + SIMPLE_NAMED_PIPE_INLINE void NamedPipeServer::notify_disconnected(size_t index, const std::error_code& ec) { if (!m_is_connected[index].load(std::memory_order_acquire)) return; m_is_connected[index].store(false, std::memory_order_release); if (m_connections[index]) m_connections[index]->invalidate(); @@ -522,14 +530,14 @@ namespace SimpleNamedPipe { if (on_event) on_event(ServerEvent::client_disconnected(static_cast(index), m_connections[index], ec)); } - inline void NamedPipeServer::notify_message(size_t index) { + SIMPLE_NAMED_PIPE_INLINE void NamedPipeServer::notify_message(size_t index) { if (m_event_handler) m_event_handler->on_message(static_cast(index), m_message_buffers[index]); if (on_message) on_message(static_cast(index), m_message_buffers[index]); if (on_event) on_event(ServerEvent::message_received(static_cast(index), m_connections[index], std::move(m_message_buffers[index]))); m_message_buffers[index].clear(); } - inline void NamedPipeServer::notify_start(const ServerConfig& config) { + SIMPLE_NAMED_PIPE_INLINE void NamedPipeServer::notify_start(const ServerConfig& config) { if (m_is_running.load(std::memory_order_acquire)) return; m_is_running.store(true, std::memory_order_release); if (m_event_handler) m_event_handler->on_start(config); @@ -537,7 +545,7 @@ namespace SimpleNamedPipe { if (on_event) on_event(ServerEvent::server_started()); } - inline void NamedPipeServer::notify_stop(const ServerConfig& config) { + SIMPLE_NAMED_PIPE_INLINE void NamedPipeServer::notify_stop(const ServerConfig& config) { if (!m_is_running.load(std::memory_order_acquire)) return; m_is_running.store(false, std::memory_order_release); if (m_event_handler) m_event_handler->on_stop(config); @@ -545,7 +553,7 @@ namespace SimpleNamedPipe { if (on_event) on_event(ServerEvent::server_stopped()); } - inline void NamedPipeServer::notify_error(const std::error_code& ec) { + SIMPLE_NAMED_PIPE_INLINE void NamedPipeServer::notify_error(const std::error_code& ec) { if (m_event_handler) m_event_handler->on_error(ec); if (on_error) on_error(ec); if (on_event) on_event(ServerEvent::error_occurred(ec)); diff --git a/src/NamedPipeClient.cpp b/src/NamedPipeClient.cpp new file mode 100644 index 0000000..db8afef --- /dev/null +++ b/src/NamedPipeClient.cpp @@ -0,0 +1,10 @@ +#ifdef _MSC_VER +#pragma warning(push) +#pragma warning(disable:4267) +#endif + +#include "SimpleNamedPipe/NamedPipeClient/NamedPipeClient.ipp" + +#ifdef _MSC_VER +#pragma warning(pop) +#endif diff --git a/tests/client_server_test.cpp b/tests/client_server_test.cpp new file mode 100644 index 0000000..4354460 --- /dev/null +++ b/tests/client_server_test.cpp @@ -0,0 +1,133 @@ +#include +#include + +#include +#include +#include +#include +#include +#include + +namespace { + +bool wait_for( + std::condition_variable& cv, + std::mutex& mutex, + bool& flag, + std::chrono::milliseconds timeout) { + std::unique_lock lock(mutex); + return cv.wait_for(lock, timeout, [&flag] { return flag; }); +} + +std::string make_test_pipe_name() { + const auto stamp = std::chrono::duration_cast( + std::chrono::steady_clock::now().time_since_epoch()).count(); + return "SimpleNamedPipeClientServerTest_" + std::to_string(stamp); +} + +} // namespace + +int main() { + const std::string pipe_name = make_test_pipe_name(); + + SimpleNamedPipe::ServerConfig server_config(pipe_name, 4096, 50); + SimpleNamedPipe::NamedPipeServer server(server_config); + + std::mutex mutex; + std::condition_variable cv; + bool server_started = false; + bool client_connected = false; + bool server_received_message = false; + std::error_code server_error; + + server.on_start = [&](const SimpleNamedPipe::ServerConfig&) { + std::lock_guard lock(mutex); + server_started = true; + cv.notify_all(); + }; + + server.on_connected = [&](int) { + std::lock_guard lock(mutex); + client_connected = true; + cv.notify_all(); + }; + + server.on_message = [&](int client_id, const std::string& message) { + { + std::lock_guard lock(mutex); + server_received_message = true; + } + server.send_to(client_id, "Echo: " + message); + cv.notify_all(); + }; + + server.on_error = [&](const std::error_code& error) { + std::lock_guard lock(mutex); + server_error = error; + cv.notify_all(); + }; + + server.start(); + + if (!wait_for(cv, mutex, server_started, std::chrono::milliseconds(3000))) { + std::cerr << "server did not start\n"; + server.stop(); + return EXIT_FAILURE; + } + + SimpleNamedPipe::ClientConfig client_config(pipe_name, 4096, 3000); + SimpleNamedPipe::NamedPipeClient client(client_config); + + std::error_code client_error; + if (!client.connect(&client_error)) { + std::cerr << "client connect failed: " << client_error.message() << "\n"; + server.stop(); + return EXIT_FAILURE; + } + + if (!wait_for(cv, mutex, client_connected, std::chrono::milliseconds(3000))) { + std::cerr << "server did not report client connection\n"; + client.close(); + server.stop(); + return EXIT_FAILURE; + } + + if (!client.write("ping", &client_error)) { + std::cerr << "client write failed: " << client_error.message() << "\n"; + client.close(); + server.stop(); + return EXIT_FAILURE; + } + + if (!wait_for(cv, mutex, server_received_message, std::chrono::milliseconds(3000))) { + std::cerr << "server did not receive client message\n"; + client.close(); + server.stop(); + return EXIT_FAILURE; + } + + std::string response; + if (!client.read(response, 3000, &client_error)) { + std::cerr << "client read failed: " << client_error.message() << "\n"; + client.close(); + server.stop(); + return EXIT_FAILURE; + } + + if (response != "Echo: ping") { + std::cerr << "unexpected response: " << response << "\n"; + client.close(); + server.stop(); + return EXIT_FAILURE; + } + + client.close(); + server.stop(); + + if (server_error) { + std::cerr << "server error: " << server_error.message() << "\n"; + return EXIT_FAILURE; + } + + return EXIT_SUCCESS; +} diff --git a/tests/odr/main1.cpp b/tests/odr/main1.cpp index 37cef33..df23702 100644 --- a/tests/odr/main1.cpp +++ b/tests/odr/main1.cpp @@ -1,7 +1,10 @@ #include +#include int main() { SimpleNamedPipe::NamedPipeServer server; + SimpleNamedPipe::NamedPipeClient client; (void)server.is_running(); + (void)client.connected(); return 0; } diff --git a/tests/odr/main2.cpp b/tests/odr/main2.cpp index d61a2d8..5f0da53 100644 --- a/tests/odr/main2.cpp +++ b/tests/odr/main2.cpp @@ -1,6 +1,9 @@ #include +#include void run_server() { SimpleNamedPipe::NamedPipeServer server; + SimpleNamedPipe::NamedPipeClient client; server.stop(); + client.close(); } diff --git a/tests/stubs/windows.h b/tests/stubs/windows.h index 8f75796..9ff1580 100644 --- a/tests/stubs/windows.h +++ b/tests/stubs/windows.h @@ -49,8 +49,14 @@ using LPOVERLAPPED = OVERLAPPED*; #define PIPE_UNLIMITED_INSTANCES 255 #define PIPE_REJECT_REMOTE_CLIENTS 0x00000008 #define NMPWAIT_USE_DEFAULT_WAIT 0x00000000 +#define GENERIC_READ 0x80000000 +#define GENERIC_WRITE 0x40000000 +#define OPEN_EXISTING 3 +#define FILE_ATTRIBUTE_NORMAL 0x00000080 #define ERROR_IO_PENDING 997 #define ERROR_IO_INCOMPLETE 996 +#define ERROR_FILE_NOT_FOUND 2 +#define ERROR_PIPE_BUSY 231 #define ERROR_PIPE_CONNECTED 535 #define ERROR_IO_PENDING 997 #define ERROR_PIPE_LISTENING 536 @@ -68,6 +74,8 @@ inline DWORD GetLastError() { return 0; } inline HANDLE CreateIoCompletionPort(HANDLE, HANDLE, ULONG_PTR, DWORD) { return reinterpret_cast(1); } inline BOOL PostQueuedCompletionStatus(HANDLE, DWORD, ULONG_PTR, LPOVERLAPPED) { return 1; } inline HANDLE CreateNamedPipeW(const wchar_t*, DWORD, DWORD, DWORD, DWORD, DWORD, DWORD, const SECURITY_ATTRIBUTES*) { return reinterpret_cast(1); } +inline HANDLE CreateFileW(const wchar_t*, DWORD, DWORD, const SECURITY_ATTRIBUTES*, DWORD, DWORD, HANDLE) { return reinterpret_cast(1); } +inline BOOL WaitNamedPipeW(const wchar_t*, DWORD) { return 1; } inline BOOL ConnectNamedPipe(HANDLE, LPOVERLAPPED) { return 1; } inline BOOL DisconnectNamedPipe(HANDLE) { return 1; } inline BOOL CloseHandle(HANDLE) { return 1; } @@ -117,4 +125,3 @@ inline BOOL SetCommTimeouts(HANDLE, const void*) { return 1; } inline BOOL SetNamedPipeHandleState(HANDLE, DWORD*, DWORD*, DWORD*) { return 1; } #define CALLBACK - From 9f251c643b336b447ed136c27fa068d075dbb58a Mon Sep 17 00:00:00 2001 From: Aster Seker Date: Wed, 1 Jul 2026 19:23:21 +0300 Subject: [PATCH 2/7] fix: use WinAPI UTF-8 pipe path conversion --- .../NamedPipeClient/NamedPipeClient.ipp | 7 +-- .../NamedPipeServer/NamedPipeServer.ipp | 7 +-- .../SimpleNamedPipe/detail/string_utils.hpp | 59 +++++++++++++++++++ 3 files changed, 65 insertions(+), 8 deletions(-) create mode 100644 include/SimpleNamedPipe/detail/string_utils.hpp diff --git a/include/SimpleNamedPipe/NamedPipeClient/NamedPipeClient.ipp b/include/SimpleNamedPipe/NamedPipeClient/NamedPipeClient.ipp index ae1ddd5..56796e7 100644 --- a/include/SimpleNamedPipe/NamedPipeClient/NamedPipeClient.ipp +++ b/include/SimpleNamedPipe/NamedPipeClient/NamedPipeClient.ipp @@ -2,10 +2,10 @@ #include "../NamedPipeClient.hpp" #endif +#include "../detail/string_utils.hpp" + #include -#include #include -#include #include #ifndef SIMPLE_NAMED_PIPE_INLINE @@ -270,8 +270,7 @@ namespace SimpleNamedPipe { ? pipe_name : prefix + pipe_name; - std::wstring_convert> conv; - return conv.from_bytes(full_name); + return detail::utf8_to_wide(full_name); } SIMPLE_NAMED_PIPE_INLINE DWORD NamedPipeClient::to_dword_timeout(size_t timeout_ms) { diff --git a/include/SimpleNamedPipe/NamedPipeServer/NamedPipeServer.ipp b/include/SimpleNamedPipe/NamedPipeServer/NamedPipeServer.ipp index 6d18467..2173401 100644 --- a/include/SimpleNamedPipe/NamedPipeServer/NamedPipeServer.ipp +++ b/include/SimpleNamedPipe/NamedPipeServer/NamedPipeServer.ipp @@ -2,8 +2,8 @@ #include "../NamedPipeServer.hpp" #endif -#include -#include +#include "../detail/string_utils.hpp" + #include #ifndef SIMPLE_NAMED_PIPE_INLINE @@ -186,8 +186,7 @@ namespace SimpleNamedPipe { } SIMPLE_NAMED_PIPE_INLINE void NamedPipeServer::create_pipe(size_t index, HANDLE completion_port, const ServerConfig& config) { - std::wstring_convert> conv; - std::wstring pipe_name_w = L"\\\\.\\pipe\\" + conv.from_bytes(config.pipe_name); + std::wstring pipe_name_w = L"\\\\.\\pipe\\" + detail::utf8_to_wide(config.pipe_name); m_pipes[index] = CreateNamedPipeW( pipe_name_w.c_str(), diff --git a/include/SimpleNamedPipe/detail/string_utils.hpp b/include/SimpleNamedPipe/detail/string_utils.hpp new file mode 100644 index 0000000..3e53102 --- /dev/null +++ b/include/SimpleNamedPipe/detail/string_utils.hpp @@ -0,0 +1,59 @@ +#pragma once +#ifndef _SIMPLE_NAMED_PIPE_DETAIL_STRING_UTILS_HPP_INCLUDED +#define _SIMPLE_NAMED_PIPE_DETAIL_STRING_UTILS_HPP_INCLUDED + +#include + +#include +#include +#include +#include + +namespace SimpleNamedPipe::detail { + + inline std::wstring utf8_to_wide(const std::string& value) { + if (value.empty()) { + return {}; + } + if (value.size() > static_cast((std::numeric_limits::max)())) { + throw std::length_error("Named pipe string is too long"); + } + + const int input_size = static_cast(value.size()); + const int output_size = MultiByteToWideChar( + CP_UTF8, + MB_ERR_INVALID_CHARS, + value.data(), + input_size, + nullptr, + 0); + + if (output_size <= 0) { + throw std::system_error( + static_cast(GetLastError()), + std::system_category(), + "Failed to convert UTF-8 named pipe string to UTF-16"); + } + + std::wstring output(static_cast(output_size), L'\0'); + const int converted = MultiByteToWideChar( + CP_UTF8, + MB_ERR_INVALID_CHARS, + value.data(), + input_size, + &output[0], + output_size); + + if (converted != output_size) { + throw std::system_error( + static_cast(GetLastError()), + std::system_category(), + "Failed to write converted UTF-16 named pipe string"); + } + + return output; + } + +} // namespace SimpleNamedPipe::detail + +#endif // _SIMPLE_NAMED_PIPE_DETAIL_STRING_UTILS_HPP_INCLUDED From 57e516ea4267e7d7e500311b6c17783b14b276cf Mon Sep 17 00:00:00 2001 From: Aster Seker Date: Wed, 1 Jul 2026 22:24:11 +0300 Subject: [PATCH 3/7] test: harden named pipe client server behavior --- .github/workflows/windows-tests.yml | 36 ++ include/SimpleNamedPipe/NamedPipeClient.hpp | 7 +- .../NamedPipeClient/NamedPipeClient.ipp | 116 ++++- include/SimpleNamedPipe/NamedPipeServer.hpp | 2 +- .../NamedPipeServer/NamedPipeServer.ipp | 11 +- tests/client_server_test.cpp | 487 +++++++++++++++--- tests/stubs/windows.h | 23 + 7 files changed, 575 insertions(+), 107 deletions(-) create mode 100644 .github/workflows/windows-tests.yml diff --git a/.github/workflows/windows-tests.yml b/.github/workflows/windows-tests.yml new file mode 100644 index 0000000..8a801e6 --- /dev/null +++ b/.github/workflows/windows-tests.yml @@ -0,0 +1,36 @@ +name: Windows Named Pipe Tests + +on: + push: + pull_request: + +jobs: + client-server: + runs-on: windows-latest + strategy: + fail-fast: false + matrix: + static: [ON, OFF] + + steps: + - name: Checkout + uses: actions/checkout@v4 + + - name: Configure + run: > + cmake -S . -B build-${{ matrix.static }} + -DSIMPLE_NAMED_PIPE_BUILD_TESTS=ON + -DSIMPLE_NAMED_PIPE_BUILD_STATIC=${{ matrix.static }} + -DSIMPLE_NAMED_PIPE_BUILD_EXAMPLES=OFF + + - name: Build + run: cmake --build build-${{ matrix.static }} --config Release --target client_server_test + + - name: Run client-server test + shell: pwsh + run: | + $exe = "build-${{ matrix.static }}\Release\client_server_test.exe" + if (!(Test-Path $exe)) { + $exe = "build-${{ matrix.static }}\client_server_test.exe" + } + & $exe diff --git a/include/SimpleNamedPipe/NamedPipeClient.hpp b/include/SimpleNamedPipe/NamedPipeClient.hpp index d2de8c6..db76f81 100644 --- a/include/SimpleNamedPipe/NamedPipeClient.hpp +++ b/include/SimpleNamedPipe/NamedPipeClient.hpp @@ -92,7 +92,7 @@ namespace SimpleNamedPipe { /// \brief Returns number of bytes currently available in the pipe. /// \param error Optional output error code. /// \return Available byte count, or 0 when disconnected or on error. - size_t available(std::error_code* error = nullptr) const; + size_t available(std::error_code* error = nullptr); /// \brief Flushes pipe write buffers. /// \param error Optional output error code. @@ -117,8 +117,9 @@ namespace SimpleNamedPipe { static DWORD to_dword_timeout(size_t timeout_ms); bool connect_no_lock(std::error_code* error); - bool read_no_lock(std::string& message, std::error_code* error); - void close_no_lock(bool notify); + bool read_no_lock(std::string& message, std::error_code* error, bool* disconnected); + bool close_no_lock(); + bool validate_config_no_lock(std::error_code* error) const; void set_error(std::error_code* out, const std::error_code& error) const; void clear_error(std::error_code* out) const; }; diff --git a/include/SimpleNamedPipe/NamedPipeClient/NamedPipeClient.ipp b/include/SimpleNamedPipe/NamedPipeClient/NamedPipeClient.ipp index 56796e7..19934a7 100644 --- a/include/SimpleNamedPipe/NamedPipeClient/NamedPipeClient.ipp +++ b/include/SimpleNamedPipe/NamedPipeClient/NamedPipeClient.ipp @@ -40,12 +40,16 @@ namespace SimpleNamedPipe { SIMPLE_NAMED_PIPE_INLINE bool NamedPipeClient::connect(std::error_code* error) { std::error_code ec; bool result = false; + bool already_connected = false; { std::lock_guard lock(m_mutex); + already_connected = + m_is_connected.load(std::memory_order_acquire) && + m_pipe != INVALID_HANDLE_VALUE; result = connect_no_lock(&ec); } set_error(error, ec); - if (result) { + if (result && !already_connected) { if (on_connected) on_connected(); } else if (ec && on_error) { on_error(ec); @@ -56,13 +60,21 @@ namespace SimpleNamedPipe { SIMPLE_NAMED_PIPE_INLINE bool NamedPipeClient::open(const std::string& pipe_name, std::error_code* error) { std::error_code ec; bool result = false; + bool already_connected = false; { std::lock_guard lock(m_mutex); - m_config.pipe_name = pipe_name; - result = connect_no_lock(&ec); + already_connected = + m_is_connected.load(std::memory_order_acquire) && + m_pipe != INVALID_HANDLE_VALUE; + if (already_connected) { + ec = std::make_error_code(std::errc::already_connected); + } else { + m_config.pipe_name = pipe_name; + result = connect_no_lock(&ec); + } } set_error(error, ec); - if (result) { + if (result && !already_connected) { if (on_connected) on_connected(); } else if (ec && on_error) { on_error(ec); @@ -74,8 +86,7 @@ namespace SimpleNamedPipe { bool notify = false; { std::lock_guard lock(m_mutex); - notify = m_is_connected.load(std::memory_order_acquire); - close_no_lock(false); + notify = close_no_lock(); } if (notify && on_disconnected) { on_disconnected(); @@ -93,6 +104,7 @@ namespace SimpleNamedPipe { SIMPLE_NAMED_PIPE_INLINE bool NamedPipeClient::write(const std::string& message, std::error_code* error) { std::error_code ec; bool result = false; + bool disconnected = false; { std::lock_guard lock(m_mutex); if (!m_is_connected.load(std::memory_order_acquire) || @@ -110,15 +122,21 @@ namespace SimpleNamedPipe { static_cast(message.size()), &bytes_written, nullptr); - if (!ok || bytes_written != message.size()) { + if (!ok) { ec = std::error_code(static_cast(GetLastError()), std::system_category()); - close_no_lock(false); + disconnected = close_no_lock(); + } else if (bytes_written != message.size()) { + ec = std::make_error_code(std::errc::io_error); + disconnected = close_no_lock(); } else { result = true; } } } set_error(error, ec); + if (disconnected && on_disconnected) { + on_disconnected(); + } if (!result && ec && on_error) { on_error(ec); } @@ -128,11 +146,15 @@ namespace SimpleNamedPipe { SIMPLE_NAMED_PIPE_INLINE bool NamedPipeClient::read(std::string& message, std::error_code* error) { std::error_code ec; bool result = false; + bool disconnected = false; { std::lock_guard lock(m_mutex); - result = read_no_lock(message, &ec); + result = read_no_lock(message, &ec, &disconnected); } set_error(error, ec); + if (disconnected && on_disconnected) { + on_disconnected(); + } if (result) { if (on_message) on_message(message); } else if (ec && on_error) { @@ -145,6 +167,7 @@ namespace SimpleNamedPipe { std::error_code ec; bool result = false; bool no_message = false; + bool disconnected = false; { std::lock_guard lock(m_mutex); if (!m_is_connected.load(std::memory_order_acquire) || @@ -161,15 +184,18 @@ namespace SimpleNamedPipe { nullptr); if (!ok) { ec = std::error_code(static_cast(GetLastError()), std::system_category()); - close_no_lock(false); + disconnected = close_no_lock(); } else if (bytes_available == 0) { no_message = true; } else { - result = read_no_lock(message, &ec); + result = read_no_lock(message, &ec, &disconnected); } } } set_error(error, ec); + if (disconnected && on_disconnected) { + on_disconnected(); + } if (result) { if (on_message) on_message(message); } else if (!no_message && ec && on_error) { @@ -208,9 +234,10 @@ namespace SimpleNamedPipe { } } - SIMPLE_NAMED_PIPE_INLINE size_t NamedPipeClient::available(std::error_code* error) const { + SIMPLE_NAMED_PIPE_INLINE size_t NamedPipeClient::available(std::error_code* error) { std::error_code ec; DWORD bytes_available = 0; + bool disconnected = false; { std::lock_guard lock(m_mutex); if (!m_is_connected.load(std::memory_order_acquire) || @@ -226,10 +253,14 @@ namespace SimpleNamedPipe { nullptr); if (!ok) { ec = std::error_code(static_cast(GetLastError()), std::system_category()); + disconnected = close_no_lock(); } } } set_error(error, ec); + if (disconnected && on_disconnected) { + on_disconnected(); + } if (ec && on_error) { on_error(ec); } @@ -239,6 +270,7 @@ namespace SimpleNamedPipe { SIMPLE_NAMED_PIPE_INLINE bool NamedPipeClient::flush(std::error_code* error) { std::error_code ec; bool result = false; + bool disconnected = false; { std::lock_guard lock(m_mutex); if (!m_is_connected.load(std::memory_order_acquire) || @@ -246,12 +278,15 @@ namespace SimpleNamedPipe { ec = make_error_code(NamedPipeErrc::NotConnected); } else if (!FlushFileBuffers(m_pipe)) { ec = std::error_code(static_cast(GetLastError()), std::system_category()); - close_no_lock(false); + disconnected = close_no_lock(); } else { result = true; } } set_error(error, ec); + if (disconnected && on_disconnected) { + on_disconnected(); + } if (!result && ec && on_error) { on_error(ec); } @@ -279,19 +314,28 @@ namespace SimpleNamedPipe { } SIMPLE_NAMED_PIPE_INLINE bool NamedPipeClient::connect_no_lock(std::error_code* error) { - close_no_lock(false); clear_error(error); - if (m_config.pipe_name.empty()) { - set_error(error, std::make_error_code(std::errc::invalid_argument)); + if (m_is_connected.load(std::memory_order_acquire) && + m_pipe != INVALID_HANDLE_VALUE) { + return true; + } + + if (!validate_config_no_lock(error)) { return false; } - if (m_config.buffer_size == 0) { + + std::wstring pipe_path; + try { + pipe_path = make_pipe_path(m_config.pipe_name); + } catch (const std::system_error& ex) { + set_error(error, ex.code()); + return false; + } catch (const std::exception&) { set_error(error, std::make_error_code(std::errc::invalid_argument)); return false; } - const std::wstring pipe_path = make_pipe_path(m_config.pipe_name); const auto start = std::chrono::steady_clock::now(); const auto timeout = std::chrono::milliseconds(m_config.timeout); @@ -353,9 +397,15 @@ namespace SimpleNamedPipe { } } - SIMPLE_NAMED_PIPE_INLINE bool NamedPipeClient::read_no_lock(std::string& message, std::error_code* error) { + SIMPLE_NAMED_PIPE_INLINE bool NamedPipeClient::read_no_lock( + std::string& message, + std::error_code* error, + bool* disconnected) { clear_error(error); message.clear(); + if (disconnected) { + *disconnected = false; + } if (!m_is_connected.load(std::memory_order_acquire) || m_pipe == INVALID_HANDLE_VALUE) { @@ -363,6 +413,10 @@ namespace SimpleNamedPipe { return false; } + if (!validate_config_no_lock(error)) { + return false; + } + std::vector buffer(m_config.buffer_size); for (;;) { @@ -395,22 +449,38 @@ namespace SimpleNamedPipe { if (read_error == ERROR_BROKEN_PIPE || read_error == ERROR_NO_DATA) { - close_no_lock(false); + if (disconnected) { + *disconnected = close_no_lock(); + } else { + close_no_lock(); + } } return false; } } - SIMPLE_NAMED_PIPE_INLINE void NamedPipeClient::close_no_lock(bool notify) { + SIMPLE_NAMED_PIPE_INLINE bool NamedPipeClient::close_no_lock() { const bool was_connected = m_is_connected.load(std::memory_order_acquire); if (m_pipe != INVALID_HANDLE_VALUE) { CloseHandle(m_pipe); m_pipe = INVALID_HANDLE_VALUE; } m_is_connected.store(false, std::memory_order_release); - if (notify && was_connected && on_disconnected) { - on_disconnected(); + return was_connected; + } + + SIMPLE_NAMED_PIPE_INLINE bool NamedPipeClient::validate_config_no_lock(std::error_code* error) const { + if (m_config.pipe_name.empty()) { + set_error(error, std::make_error_code(std::errc::invalid_argument)); + return false; } + if (m_config.buffer_size == 0 || + m_config.buffer_size > static_cast((std::numeric_limits::max)())) { + set_error(error, std::make_error_code(std::errc::invalid_argument)); + return false; + } + clear_error(error); + return true; } SIMPLE_NAMED_PIPE_INLINE void NamedPipeClient::set_error(std::error_code* out, const std::error_code& error) const { diff --git a/include/SimpleNamedPipe/NamedPipeServer.hpp b/include/SimpleNamedPipe/NamedPipeServer.hpp index 0108683..432d343 100644 --- a/include/SimpleNamedPipe/NamedPipeServer.hpp +++ b/include/SimpleNamedPipe/NamedPipeServer.hpp @@ -102,7 +102,7 @@ namespace SimpleNamedPipe { }; static constexpr ULONG_PTR CMD_INDEX_MASK = 0x000000FF; - static constexpr size_t MAX_CLIENTS = 256; + static constexpr size_t MAX_CLIENTS = PIPE_UNLIMITED_INSTANCES; struct WriteCommand { size_t client_index; diff --git a/include/SimpleNamedPipe/NamedPipeServer/NamedPipeServer.ipp b/include/SimpleNamedPipe/NamedPipeServer/NamedPipeServer.ipp index 2173401..6412e98 100644 --- a/include/SimpleNamedPipe/NamedPipeServer/NamedPipeServer.ipp +++ b/include/SimpleNamedPipe/NamedPipeServer/NamedPipeServer.ipp @@ -296,7 +296,12 @@ namespace SimpleNamedPipe { } // ov != nullptr means the operation completed with an error - if (err == ERROR_BROKEN_PIPE) { + if (err == ERROR_OPERATION_ABORTED) { + continue; + } + + if (err == ERROR_BROKEN_PIPE || + err == ERROR_NO_DATA) { notify_disconnected(index, std::error_code(err, std::system_category())); DisconnectNamedPipe(m_pipes[index]); reconnect_client(index, completion_port, &m_read_overlapped[index]); @@ -342,9 +347,13 @@ namespace SimpleNamedPipe { if (!result && err != ERROR_IO_PENDING) { if (err == ERROR_BROKEN_PIPE || err == ERROR_NO_DATA) { + notify_disconnected(index, std::error_code(static_cast(err), std::system_category())); DisconnectNamedPipe(m_pipes[index]); reconnect_client(index, completion_port, new_ov); continue; + } else + if (err == ERROR_OPERATION_ABORTED) { + continue; } else { notify_error(std::error_code(static_cast(err), std::system_category())); continue; diff --git a/tests/client_server_test.cpp b/tests/client_server_test.cpp index 4354460..42e3334 100644 --- a/tests/client_server_test.cpp +++ b/tests/client_server_test.cpp @@ -1,133 +1,462 @@ #include #include +#include +#include #include #include #include +#include #include +#include #include +#include +#include #include +#include +#include namespace { -bool wait_for( +const std::chrono::milliseconds kDefaultWait(5000); +const std::chrono::milliseconds kHeavyWait(15000); + +std::string make_test_pipe_name(const std::string& suffix) { + static std::atomic counter{0}; + const auto stamp = std::chrono::duration_cast( + std::chrono::steady_clock::now().time_since_epoch()).count(); + std::ostringstream out; + out << "SimpleNamedPipeClientServerTest_" + << stamp << "_" << counter.fetch_add(1) << "_" << suffix; + return out.str(); +} + +template +bool wait_until( std::condition_variable& cv, std::mutex& mutex, - bool& flag, + Predicate predicate, std::chrono::milliseconds timeout) { std::unique_lock lock(mutex); - return cv.wait_for(lock, timeout, [&flag] { return flag; }); + return cv.wait_for(lock, timeout, predicate); } -std::string make_test_pipe_name() { - const auto stamp = std::chrono::duration_cast( - std::chrono::steady_clock::now().time_since_epoch()).count(); - return "SimpleNamedPipeClientServerTest_" + std::to_string(stamp); +template +bool poll_until(Predicate predicate, std::chrono::milliseconds timeout) { + const auto start = std::chrono::steady_clock::now(); + while (std::chrono::steady_clock::now() - start < timeout) { + if (predicate()) { + return true; + } + std::this_thread::sleep_for(std::chrono::milliseconds(5)); + } + return predicate(); } -} // namespace +std::string with_error(const std::string& message, const std::error_code& error) { + return message + error.message(); +} -int main() { - const std::string pipe_name = make_test_pipe_name(); +struct TestRunner { + int failures = 0; - SimpleNamedPipe::ServerConfig server_config(pipe_name, 4096, 50); - SimpleNamedPipe::NamedPipeServer server(server_config); + bool expect(bool condition, const std::string& message) { + if (!condition) { + ++failures; + std::cerr << "FAILED: " << message << "\n"; + } + return condition; + } - std::mutex mutex; - std::condition_variable cv; - bool server_started = false; - bool client_connected = false; - bool server_received_message = false; - std::error_code server_error; + void run(const std::string& name, const std::function& test) { + const int before = failures; + std::cerr << "[ RUN ] " << name << "\n"; + try { + test(*this); + } catch (const std::exception& ex) { + ++failures; + std::cerr << "FAILED: " << name << " threw: " << ex.what() << "\n"; + } catch (...) { + ++failures; + std::cerr << "FAILED: " << name << " threw an unknown exception\n"; + } - server.on_start = [&](const SimpleNamedPipe::ServerConfig&) { - std::lock_guard lock(mutex); - server_started = true; - cv.notify_all(); - }; + if (failures == before) { + std::cerr << "[ OK ] " << name << "\n"; + } else { + std::cerr << "[ FAILED ] " << name << "\n"; + } + } +}; - server.on_connected = [&](int) { - std::lock_guard lock(mutex); - client_connected = true; - cv.notify_all(); - }; +class ServerHarness { +public: + explicit ServerHarness( + const std::string& suffix, + size_t buffer_size = 4096, + size_t timeout_ms = 25) + : pipe_name(make_test_pipe_name(suffix)), + config(pipe_name, buffer_size, timeout_ms), + server(config) { + server.on_start = [this](const SimpleNamedPipe::ServerConfig&) { + std::lock_guard lock(mutex); + started = true; + cv.notify_all(); + }; - server.on_message = [&](int client_id, const std::string& message) { - { + server.on_stop = [this](const SimpleNamedPipe::ServerConfig&) { std::lock_guard lock(mutex); - server_received_message = true; - } - server.send_to(client_id, "Echo: " + message); - cv.notify_all(); - }; + stopped = true; + cv.notify_all(); + }; - server.on_error = [&](const std::error_code& error) { - std::lock_guard lock(mutex); - server_error = error; - cv.notify_all(); - }; + server.on_connected = [this](int client_id) { + std::lock_guard lock(mutex); + ++connected_events; + connected_ids.insert(client_id); + cv.notify_all(); + }; + + server.on_disconnected = [this](int client_id, const std::error_code&) { + std::lock_guard lock(mutex); + ++disconnected_events; + connected_ids.erase(client_id); + cv.notify_all(); + }; + + server.on_message = [this](int client_id, const std::string& message) { + { + std::lock_guard lock(mutex); + ++message_events; + messages.push_back(message); + cv.notify_all(); + } + server.send_to(client_id, "Echo: " + message); + }; - server.start(); + server.on_error = [this](const std::error_code& error) { + std::lock_guard lock(mutex); + ++error_events; + last_error = error; + cv.notify_all(); + }; + } - if (!wait_for(cv, mutex, server_started, std::chrono::milliseconds(3000))) { - std::cerr << "server did not start\n"; + ~ServerHarness() { server.stop(); - return EXIT_FAILURE; } - SimpleNamedPipe::ClientConfig client_config(pipe_name, 4096, 3000); - SimpleNamedPipe::NamedPipeClient client(client_config); + void start() { + server.start(); + } - std::error_code client_error; - if (!client.connect(&client_error)) { - std::cerr << "client connect failed: " << client_error.message() << "\n"; + void stop() { server.stop(); - return EXIT_FAILURE; } - if (!wait_for(cv, mutex, client_connected, std::chrono::milliseconds(3000))) { - std::cerr << "server did not report client connection\n"; - client.close(); - server.stop(); - return EXIT_FAILURE; + bool wait_started(std::chrono::milliseconds timeout = kDefaultWait) { + return wait_until(cv, mutex, [this] { return started; }, timeout); } - if (!client.write("ping", &client_error)) { - std::cerr << "client write failed: " << client_error.message() << "\n"; - client.close(); - server.stop(); - return EXIT_FAILURE; + bool wait_connected_at_least(int value, std::chrono::milliseconds timeout = kDefaultWait) { + return wait_until(cv, mutex, [this, value] { return connected_events >= value; }, timeout); } - if (!wait_for(cv, mutex, server_received_message, std::chrono::milliseconds(3000))) { - std::cerr << "server did not receive client message\n"; - client.close(); - server.stop(); - return EXIT_FAILURE; + bool wait_disconnected_at_least(int value, std::chrono::milliseconds timeout = kDefaultWait) { + return wait_until(cv, mutex, [this, value] { return disconnected_events >= value; }, timeout); + } + + bool wait_messages_at_least(int value, std::chrono::milliseconds timeout = kDefaultWait) { + return wait_until(cv, mutex, [this, value] { return message_events >= value; }, timeout); + } + + int connected_count() const { + std::lock_guard lock(mutex); + return connected_events; + } + + int disconnected_count() const { + std::lock_guard lock(mutex); + return disconnected_events; + } + + int error_count() const { + std::lock_guard lock(mutex); + return error_events; + } + + std::error_code error() const { + std::lock_guard lock(mutex); + return last_error; + } + + std::vector current_client_ids() const { + std::lock_guard lock(mutex); + return std::vector(connected_ids.begin(), connected_ids.end()); + } + + std::string pipe_name; + SimpleNamedPipe::ServerConfig config; + SimpleNamedPipe::NamedPipeServer server; + +private: + mutable std::mutex mutex; + std::condition_variable cv; + bool started = false; + bool stopped = false; + int connected_events = 0; + int disconnected_events = 0; + int message_events = 0; + int error_events = 0; + std::error_code last_error; + std::set connected_ids; + std::vector messages; +}; + +std::unique_ptr make_client( + const std::string& pipe_name, + size_t buffer_size = 4096, + size_t timeout_ms = 3000) { + return std::unique_ptr( + new SimpleNamedPipe::NamedPipeClient( + SimpleNamedPipe::ClientConfig(pipe_name, buffer_size, timeout_ms))); +} + +bool read_echo( + SimpleNamedPipe::NamedPipeClient& client, + const std::string& payload, + std::error_code& ec) { + if (!client.write(payload, &ec)) { + return false; } std::string response; - if (!client.read(response, 3000, &client_error)) { - std::cerr << "client read failed: " << client_error.message() << "\n"; - client.close(); - server.stop(); - return EXIT_FAILURE; + if (!client.read(response, 3000, &ec)) { + return false; } - if (response != "Echo: ping") { - std::cerr << "unexpected response: " << response << "\n"; - client.close(); - server.stop(); - return EXIT_FAILURE; + return response == "Echo: " + payload; +} + +void basic_io_edges(TestRunner& tr) { + ServerHarness harness("basic", 4096, 25); + harness.start(); + if (!tr.expect(harness.wait_started(), "server did not start")) return; + + auto client = make_client(harness.pipe_name, 16, 3000); + std::error_code ec; + if (!tr.expect(client->connect(&ec), with_error("client connect failed: ", ec))) return; + tr.expect(harness.wait_connected_at_least(1), "server did not observe client connect"); + + std::string response; + ec.clear(); + tr.expect(!client->try_read(response, &ec), "try_read should report no immediate message"); + tr.expect(!ec, "try_read without data should not set an error"); + + tr.expect(!client->read(response, 25, &ec), "timed read should fail with timeout"); + tr.expect(ec.value() == WAIT_TIMEOUT, "timed read should return WAIT_TIMEOUT"); + + const std::string long_payload(300, 'x'); + tr.expect(read_echo(*client, long_payload, ec), with_error("long message echo failed: ", ec)); + tr.expect(harness.wait_messages_at_least(1), "server did not receive long message"); + + client->close(); + tr.expect(harness.wait_disconnected_at_least(1), "server did not observe client close"); + harness.stop(); + tr.expect(harness.error_count() == 0, with_error("server reported error: ", harness.error())); +} + +void repeated_connect_and_open_contract(TestRunner& tr) { + ServerHarness harness("repeat", 1024, 25); + harness.start(); + if (!tr.expect(harness.wait_started(), "server did not start")) return; + + auto client = make_client(harness.pipe_name, 1024, 3000); + std::atomic connected_callbacks{0}; + client->on_connected = [&connected_callbacks] { + connected_callbacks.fetch_add(1); + }; + + std::error_code ec; + tr.expect(client->connect(&ec), with_error("first connect failed: ", ec)); + tr.expect(client->connect(&ec), with_error("second connect should be idempotent: ", ec)); + tr.expect(connected_callbacks.load() == 1, "second connect should not emit on_connected"); + tr.expect(harness.wait_connected_at_least(1), "server did not observe first connect"); + tr.expect(harness.connected_count() == 1, "second connect should not create a second server connection"); + + ec.clear(); + tr.expect(!client->open(harness.pipe_name + "_other", &ec), "open should reject re-open while connected"); + tr.expect(ec == std::make_error_code(std::errc::already_connected), + "open while connected should return already_connected"); + tr.expect(client->is_connected(), "client should remain connected after rejected open"); + + client->close(); + tr.expect(harness.wait_disconnected_at_least(1), "server did not observe client close"); + harness.stop(); + tr.expect(harness.error_count() == 0, with_error("server reported error: ", harness.error())); +} + +void server_disconnect_notifies_client(TestRunner& tr) { + ServerHarness harness("server_disconnect", 1024, 25); + harness.start(); + if (!tr.expect(harness.wait_started(), "server did not start")) return; + + auto client = make_client(harness.pipe_name, 1024, 3000); + std::atomic disconnected_callbacks{0}; + client->on_disconnected = [&disconnected_callbacks] { + disconnected_callbacks.fetch_add(1); + }; + + std::error_code ec; + tr.expect(client->connect(&ec), with_error("client connect failed: ", ec)); + tr.expect(harness.wait_connected_at_least(1), "server did not observe client connect"); + + const std::vector ids = harness.current_client_ids(); + if (!tr.expect(!ids.empty(), "server has no connected client id")) return; + + std::mutex done_mutex; + std::condition_variable done_cv; + bool close_done = false; + harness.server.close(ids.front(), [&](const std::error_code&) { + std::lock_guard lock(done_mutex); + close_done = true; + done_cv.notify_all(); + }); + + tr.expect(wait_until(done_cv, done_mutex, [&close_done] { return close_done; }, kDefaultWait), + "server close callback did not fire"); + tr.expect(harness.wait_disconnected_at_least(1), "server did not emit disconnect"); + + tr.expect(poll_until([&] { + std::error_code local_ec; + client->available(&local_ec); + return !client->is_connected() && disconnected_callbacks.load() == 1; + }, kDefaultWait), "client did not synchronize disconnected state"); + + tr.expect(disconnected_callbacks.load() == 1, "on_disconnected should fire exactly once"); + harness.stop(); + tr.expect(harness.error_count() == 0, with_error("server reported error: ", harness.error())); +} + +void churn_clients(TestRunner& tr) { + ServerHarness harness("churn", 2048, 25); + harness.start(); + if (!tr.expect(harness.wait_started(), "server did not start")) return; + + for (int cycle = 0; cycle < 10; ++cycle) { + const int before_connected = harness.connected_count(); + const int before_disconnected = harness.disconnected_count(); + std::vector > clients; + + for (int i = 0; i < 12; ++i) { + clients.push_back(make_client(harness.pipe_name, 512, 3000)); + std::error_code ec; + if (!tr.expect(clients.back()->connect(&ec), with_error("churn connect failed: ", ec))) { + return; + } + } + + tr.expect(harness.wait_connected_at_least(before_connected + 12), "server missed churn connects"); + + for (int i = 0; i < 4; ++i) { + std::error_code ec; + std::ostringstream payload; + payload << "cycle-" << cycle << "-client-" << i; + tr.expect(read_echo(*clients[static_cast(i)], payload.str(), ec), + with_error("churn echo failed: ", ec)); + } + + std::vector ids = harness.current_client_ids(); + const size_t server_close_count = (std::min)(ids.size(), static_cast(4)); + for (size_t i = 0; i < server_close_count; ++i) { + harness.server.close(ids[i]); + } + + for (size_t i = server_close_count; i < clients.size(); ++i) { + if (i % 2 == 0) { + std::error_code ec; + clients[i]->write("closing-soon", &ec); + } + clients[i]->close(); + } + + tr.expect(harness.wait_disconnected_at_least(before_disconnected + 12, kHeavyWait), + "server missed churn disconnects"); + } + + harness.stop(); + tr.expect(harness.error_count() == 0, with_error("server reported error during churn: ", harness.error())); +} + +void peak_255_clients_and_slot_reuse(TestRunner& tr) { + ServerHarness harness("peak", 512, 25); + harness.start(); + if (!tr.expect(harness.wait_started(), "server did not start")) return; + + std::vector > clients; + clients.reserve(255); + + for (int i = 0; i < 255; ++i) { + clients.push_back(make_client(harness.pipe_name, 128, 3000)); + std::error_code ec; + if (!tr.expect(clients.back()->connect(&ec), with_error("peak connect failed: ", ec))) { + return; + } + } + + tr.expect(harness.wait_connected_at_least(255, kHeavyWait), "server did not accept 255 clients"); + + auto extra = make_client(harness.pipe_name, 128, 150); + std::error_code ec; + tr.expect(!extra->connect(&ec), "256th client should not connect while all instances are busy"); + + const int before_disconnected = harness.disconnected_count(); + clients.back()->close(); + clients.pop_back(); + tr.expect(harness.wait_disconnected_at_least(before_disconnected + 1, kHeavyWait), + "server did not release a closed peak client slot"); + + extra = make_client(harness.pipe_name, 128, 3000); + tr.expect(extra->connect(&ec), with_error("extra client should connect after a slot is released: ", ec)); + tr.expect(harness.wait_connected_at_least(256, kHeavyWait), "server did not observe slot reuse connection"); + + extra->close(); + for (size_t i = 0; i < clients.size(); ++i) { + clients[i]->close(); } - client.close(); - server.stop(); + harness.stop(); + tr.expect(harness.error_count() == 0, with_error("server reported error during peak test: ", harness.error())); +} + +void invalid_utf8_pipe_name_fails(TestRunner& tr) { + const std::string invalid_name("\xC3\x28", 2); + SimpleNamedPipe::NamedPipeClient client( + SimpleNamedPipe::ClientConfig(invalid_name, 512, 10)); + + std::error_code ec; + tr.expect(!client.connect(&ec), "invalid UTF-8 pipe name should fail"); + tr.expect(static_cast(ec), "invalid UTF-8 pipe name should set an error"); + tr.expect(!client.is_connected(), "invalid UTF-8 connect should not leave client connected"); +} + +} // namespace + +int main() { + TestRunner runner; + + runner.run("basic_io_edges", basic_io_edges); + runner.run("repeated_connect_and_open_contract", repeated_connect_and_open_contract); + runner.run("server_disconnect_notifies_client", server_disconnect_notifies_client); + runner.run("churn_clients", churn_clients); + runner.run("peak_255_clients_and_slot_reuse", peak_255_clients_and_slot_reuse); + runner.run("invalid_utf8_pipe_name_fails", invalid_utf8_pipe_name_fails); - if (server_error) { - std::cerr << "server error: " << server_error.message() << "\n"; + if (runner.failures != 0) { + std::cerr << runner.failures << " failure(s)\n"; return EXIT_FAILURE; } + std::cerr << "All client/server named pipe tests passed\n"; return EXIT_SUCCESS; } diff --git a/tests/stubs/windows.h b/tests/stubs/windows.h index 9ff1580..f9b485a 100644 --- a/tests/stubs/windows.h +++ b/tests/stubs/windows.h @@ -69,8 +69,31 @@ using LPOVERLAPPED = OVERLAPPED*; #define WAIT_TIMEOUT 258 #define WAIT_OBJECT_0 0 #define WAIT_FAILED 0xFFFFFFFF +#define CP_UTF8 65001 +#define MB_ERR_INVALID_CHARS 0x00000008 inline DWORD GetLastError() { return 0; } +inline int MultiByteToWideChar( + unsigned int, + DWORD, + const char* input, + int input_size, + wchar_t* output, + int output_size) { + if (!input || input_size < 0 || output_size < 0) { + return 0; + } + + if (!output) { + return input_size; + } + + const int count = input_size < output_size ? input_size : output_size; + for (int i = 0; i < count; ++i) { + output[i] = static_cast(input[i]); + } + return count; +} inline HANDLE CreateIoCompletionPort(HANDLE, HANDLE, ULONG_PTR, DWORD) { return reinterpret_cast(1); } inline BOOL PostQueuedCompletionStatus(HANDLE, DWORD, ULONG_PTR, LPOVERLAPPED) { return 1; } inline HANDLE CreateNamedPipeW(const wchar_t*, DWORD, DWORD, DWORD, DWORD, DWORD, DWORD, const SECURITY_ATTRIBUTES*) { return reinterpret_cast(1); } From a0ca431ff7d7a760cea87c6387750972a67f1af9 Mon Sep 17 00:00:00 2001 From: Aster Seker Date: Wed, 1 Jul 2026 22:26:29 +0300 Subject: [PATCH 4/7] fix: keep string utilities C++11 compatible --- include/SimpleNamedPipe/detail/string_utils.hpp | 6 ++++-- 1 file changed, 4 insertions(+), 2 deletions(-) diff --git a/include/SimpleNamedPipe/detail/string_utils.hpp b/include/SimpleNamedPipe/detail/string_utils.hpp index 3e53102..c3370a1 100644 --- a/include/SimpleNamedPipe/detail/string_utils.hpp +++ b/include/SimpleNamedPipe/detail/string_utils.hpp @@ -9,7 +9,8 @@ #include #include -namespace SimpleNamedPipe::detail { +namespace SimpleNamedPipe { +namespace detail { inline std::wstring utf8_to_wide(const std::string& value) { if (value.empty()) { @@ -54,6 +55,7 @@ namespace SimpleNamedPipe::detail { return output; } -} // namespace SimpleNamedPipe::detail +} // namespace detail +} // namespace SimpleNamedPipe #endif // _SIMPLE_NAMED_PIPE_DETAIL_STRING_UTILS_HPP_INCLUDED From 6a9e0029bd6a73b245d1247d24fec384ff2481eb Mon Sep 17 00:00:00 2001 From: Aster Seker Date: Thu, 2 Jul 2026 00:12:42 +0300 Subject: [PATCH 5/7] docs: correct maximum client count --- README-RU.md | 2 +- README.md | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/README-RU.md b/README-RU.md index 35fe410..8f93a1e 100644 --- a/README-RU.md +++ b/README-RU.md @@ -10,7 +10,7 @@ SimpleNamedPipe — легковесная библиотека C++ для со - асинхронная обработка клиентов через IO Completion Port; - работа либо в отдельном потоке, либо блокирующе в текущем (параметр `start()`); -- поддержка до 256 одновременных клиентов; +- поддержка до 255 одновременных клиентов; - очередь отправки с ограничением размера и количества сообщений; - уведомления о событиях через колбэки или класс `ServerEventHandler`; - лёгкий клиент для MQL5 с опциональными глобальными обратными вызовами. diff --git a/README.md b/README.md index 27c3ba6..6017268 100644 --- a/README.md +++ b/README.md @@ -11,7 +11,7 @@ For the documentation in Russian, see [README-RU.md](README-RU.md). - asynchronous client handling through IO Completion Port; - runs either in a separate thread or blocking the current one (the `start()` parameter); -- supports up to 256 simultaneous clients; +- supports up to 255 simultaneous clients; - send queue with limits on message size and count; - event notifications via callbacks or the `ServerEventHandler` class; - synchronous C++ client for tests and lightweight local integrations; From 14afdcaab518eda9e4c49357575d258621ed20ec Mon Sep 17 00:00:00 2001 From: Aster Seker Date: Thu, 2 Jul 2026 00:16:01 +0300 Subject: [PATCH 6/7] fix: tolerate stale pipe reconnect races --- .../NamedPipeServer/NamedPipeServer.ipp | 37 ++++++++++++------- 1 file changed, 24 insertions(+), 13 deletions(-) diff --git a/include/SimpleNamedPipe/NamedPipeServer/NamedPipeServer.ipp b/include/SimpleNamedPipe/NamedPipeServer/NamedPipeServer.ipp index 6412e98..830f120 100644 --- a/include/SimpleNamedPipe/NamedPipeServer/NamedPipeServer.ipp +++ b/include/SimpleNamedPipe/NamedPipeServer/NamedPipeServer.ipp @@ -441,24 +441,35 @@ namespace SimpleNamedPipe { } SIMPLE_NAMED_PIPE_INLINE bool NamedPipeServer::reconnect_client(size_t index, HANDLE completion_port, OVERLAPPED* ov) { - memset(ov, 0, sizeof(OVERLAPPED)); + for (int attempt = 0; attempt < 2; ++attempt) { + memset(ov, 0, sizeof(OVERLAPPED)); - BOOL connected = ConnectNamedPipe(m_pipes[index], ov); - if (connected) { - PostQueuedCompletionStatus(completion_port, 0, static_cast(index), ov); - return true; - } + BOOL connected = ConnectNamedPipe(m_pipes[index], ov); + if (connected) { + PostQueuedCompletionStatus(completion_port, 0, static_cast(index), ov); + return true; + } + + DWORD err = GetLastError(); + if (err == ERROR_PIPE_CONNECTED) { + PostQueuedCompletionStatus(completion_port, 0, static_cast(index), ov); + return true; + } + if (err == ERROR_IO_PENDING) { + return true; + } + if (err == ERROR_BROKEN_PIPE || + err == ERROR_NO_DATA || + err == ERROR_OPERATION_ABORTED) { + DisconnectNamedPipe(m_pipes[index]); + continue; + } - DWORD err = GetLastError(); - if (err == ERROR_PIPE_CONNECTED) { - PostQueuedCompletionStatus(completion_port, 0, static_cast(index), ov); - return true; - } else - if (err != ERROR_IO_PENDING) { notify_error(std::error_code(static_cast(err), std::system_category())); return false; } - return true; + + return false; } SIMPLE_NAMED_PIPE_INLINE void NamedPipeServer::handle_close(size_t index, HANDLE completion_port) { From fd4737a7c1bbd6310fd31f28e24ef7afc5d9497d Mon Sep 17 00:00:00 2001 From: Aster Seker Date: Thu, 2 Jul 2026 00:43:09 +0300 Subject: [PATCH 7/7] refactor: normalize implementation include paths --- README-RU.md | 2 +- README.md | 2 +- .../NamedPipeClient/NamedPipeClient.ipp | 4 ++-- include/SimpleNamedPipe/NamedPipeServer.hpp | 2 +- .../NamedPipeServer/NamedPipeServer.ipp | 4 ++-- src/NamedPipeClient.cpp | 2 +- src/NamedPipeServer.cpp | 4 ++-- tests/client_server_test.cpp | 14 +++++++------- 8 files changed, 17 insertions(+), 17 deletions(-) diff --git a/README-RU.md b/README-RU.md index 8f93a1e..35fe410 100644 --- a/README-RU.md +++ b/README-RU.md @@ -10,7 +10,7 @@ SimpleNamedPipe — легковесная библиотека C++ для со - асинхронная обработка клиентов через IO Completion Port; - работа либо в отдельном потоке, либо блокирующе в текущем (параметр `start()`); -- поддержка до 255 одновременных клиентов; +- поддержка до 256 одновременных клиентов; - очередь отправки с ограничением размера и количества сообщений; - уведомления о событиях через колбэки или класс `ServerEventHandler`; - лёгкий клиент для MQL5 с опциональными глобальными обратными вызовами. diff --git a/README.md b/README.md index 6017268..27c3ba6 100644 --- a/README.md +++ b/README.md @@ -11,7 +11,7 @@ For the documentation in Russian, see [README-RU.md](README-RU.md). - asynchronous client handling through IO Completion Port; - runs either in a separate thread or blocking the current one (the `start()` parameter); -- supports up to 255 simultaneous clients; +- supports up to 256 simultaneous clients; - send queue with limits on message size and count; - event notifications via callbacks or the `ServerEventHandler` class; - synchronous C++ client for tests and lightweight local integrations; diff --git a/include/SimpleNamedPipe/NamedPipeClient/NamedPipeClient.ipp b/include/SimpleNamedPipe/NamedPipeClient/NamedPipeClient.ipp index 19934a7..3a62f6d 100644 --- a/include/SimpleNamedPipe/NamedPipeClient/NamedPipeClient.ipp +++ b/include/SimpleNamedPipe/NamedPipeClient/NamedPipeClient.ipp @@ -1,8 +1,8 @@ #ifdef SIMPLE_NAMED_PIPE_STATIC_LIB -#include "../NamedPipeClient.hpp" +#include #endif -#include "../detail/string_utils.hpp" +#include #include #include diff --git a/include/SimpleNamedPipe/NamedPipeServer.hpp b/include/SimpleNamedPipe/NamedPipeServer.hpp index 432d343..0108683 100644 --- a/include/SimpleNamedPipe/NamedPipeServer.hpp +++ b/include/SimpleNamedPipe/NamedPipeServer.hpp @@ -102,7 +102,7 @@ namespace SimpleNamedPipe { }; static constexpr ULONG_PTR CMD_INDEX_MASK = 0x000000FF; - static constexpr size_t MAX_CLIENTS = PIPE_UNLIMITED_INSTANCES; + static constexpr size_t MAX_CLIENTS = 256; struct WriteCommand { size_t client_index; diff --git a/include/SimpleNamedPipe/NamedPipeServer/NamedPipeServer.ipp b/include/SimpleNamedPipe/NamedPipeServer/NamedPipeServer.ipp index 830f120..2e4f978 100644 --- a/include/SimpleNamedPipe/NamedPipeServer/NamedPipeServer.ipp +++ b/include/SimpleNamedPipe/NamedPipeServer/NamedPipeServer.ipp @@ -1,8 +1,8 @@ #ifdef SIMPLE_NAMED_PIPE_STATIC_LIB -#include "../NamedPipeServer.hpp" +#include #endif -#include "../detail/string_utils.hpp" +#include #include diff --git a/src/NamedPipeClient.cpp b/src/NamedPipeClient.cpp index db8afef..e855b0b 100644 --- a/src/NamedPipeClient.cpp +++ b/src/NamedPipeClient.cpp @@ -3,7 +3,7 @@ #pragma warning(disable:4267) #endif -#include "SimpleNamedPipe/NamedPipeClient/NamedPipeClient.ipp" +#include #ifdef _MSC_VER #pragma warning(pop) diff --git a/src/NamedPipeServer.cpp b/src/NamedPipeServer.cpp index c831b66..ff8d5b9 100644 --- a/src/NamedPipeServer.cpp +++ b/src/NamedPipeServer.cpp @@ -3,8 +3,8 @@ #pragma warning(disable:4267) #endif -#include "SimpleNamedPipe/NamedPipeServer/NamedPipeServer.ipp" +#include #ifdef _MSC_VER #pragma warning(pop) -#endif \ No newline at end of file +#endif diff --git a/tests/client_server_test.cpp b/tests/client_server_test.cpp index 42e3334..a92725f 100644 --- a/tests/client_server_test.cpp +++ b/tests/client_server_test.cpp @@ -388,15 +388,15 @@ void churn_clients(TestRunner& tr) { tr.expect(harness.error_count() == 0, with_error("server reported error during churn: ", harness.error())); } -void peak_255_clients_and_slot_reuse(TestRunner& tr) { +void peak_256_clients_and_slot_reuse(TestRunner& tr) { ServerHarness harness("peak", 512, 25); harness.start(); if (!tr.expect(harness.wait_started(), "server did not start")) return; std::vector > clients; - clients.reserve(255); + clients.reserve(256); - for (int i = 0; i < 255; ++i) { + for (int i = 0; i < 256; ++i) { clients.push_back(make_client(harness.pipe_name, 128, 3000)); std::error_code ec; if (!tr.expect(clients.back()->connect(&ec), with_error("peak connect failed: ", ec))) { @@ -404,11 +404,11 @@ void peak_255_clients_and_slot_reuse(TestRunner& tr) { } } - tr.expect(harness.wait_connected_at_least(255, kHeavyWait), "server did not accept 255 clients"); + tr.expect(harness.wait_connected_at_least(256, kHeavyWait), "server did not accept 256 clients"); auto extra = make_client(harness.pipe_name, 128, 150); std::error_code ec; - tr.expect(!extra->connect(&ec), "256th client should not connect while all instances are busy"); + tr.expect(!extra->connect(&ec), "257th client should not connect while all slots are busy"); const int before_disconnected = harness.disconnected_count(); clients.back()->close(); @@ -418,7 +418,7 @@ void peak_255_clients_and_slot_reuse(TestRunner& tr) { extra = make_client(harness.pipe_name, 128, 3000); tr.expect(extra->connect(&ec), with_error("extra client should connect after a slot is released: ", ec)); - tr.expect(harness.wait_connected_at_least(256, kHeavyWait), "server did not observe slot reuse connection"); + tr.expect(harness.wait_connected_at_least(257, kHeavyWait), "server did not observe slot reuse connection"); extra->close(); for (size_t i = 0; i < clients.size(); ++i) { @@ -449,7 +449,7 @@ int main() { runner.run("repeated_connect_and_open_contract", repeated_connect_and_open_contract); runner.run("server_disconnect_notifies_client", server_disconnect_notifies_client); runner.run("churn_clients", churn_clients); - runner.run("peak_255_clients_and_slot_reuse", peak_255_clients_and_slot_reuse); + runner.run("peak_256_clients_and_slot_reuse", peak_256_clients_and_slot_reuse); runner.run("invalid_utf8_pipe_name_fails", invalid_utf8_pipe_name_fails); if (runner.failures != 0) {