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/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..db76f81 --- /dev/null +++ b/include/SimpleNamedPipe/NamedPipeClient.hpp @@ -0,0 +1,135 @@ +#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); + + /// \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, 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; + }; + +} // 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..3a62f6d --- /dev/null +++ b/include/SimpleNamedPipe/NamedPipeClient/NamedPipeClient.ipp @@ -0,0 +1,498 @@ +#ifdef SIMPLE_NAMED_PIPE_STATIC_LIB +#include +#endif + +#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; + 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 && !already_connected) { + 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; + 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; + 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 && !already_connected) { + 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 = close_no_lock(); + } + 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; + bool disconnected = 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) { + ec = std::error_code(static_cast(GetLastError()), std::system_category()); + 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); + } + return result; + } + + 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, &disconnected); + } + set_error(error, ec); + if (disconnected && on_disconnected) { + on_disconnected(); + } + 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; + bool disconnected = 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()); + disconnected = close_no_lock(); + } else if (bytes_available == 0) { + no_message = true; + } else { + 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) { + 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) { + 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) || + 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()); + disconnected = close_no_lock(); + } + } + } + set_error(error, ec); + if (disconnected && on_disconnected) { + on_disconnected(); + } + 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; + bool disconnected = 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()); + 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); + } + 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; + + return detail::utf8_to_wide(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) { + clear_error(error); + + if (m_is_connected.load(std::memory_order_acquire) && + m_pipe != INVALID_HANDLE_VALUE) { + return true; + } + + if (!validate_config_no_lock(error)) { + return false; + } + + 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 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, + 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) { + set_error(error, make_error_code(NamedPipeErrc::NotConnected)); + return false; + } + + if (!validate_config_no_lock(error)) { + 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) { + if (disconnected) { + *disconnected = close_no_lock(); + } else { + close_no_lock(); + } + } + return false; + } + } + + 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); + 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 { + 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..2e4f978 100644 --- a/include/SimpleNamedPipe/NamedPipeServer/NamedPipeServer.ipp +++ b/include/SimpleNamedPipe/NamedPipeServer/NamedPipeServer.ipp @@ -1,36 +1,44 @@ #ifdef SIMPLE_NAMED_PIPE_STATIC_LIB -#include "../NamedPipeServer.hpp" +#include #endif -#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 { - 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,9 +185,8 @@ namespace SimpleNamedPipe { } } - 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); + SIMPLE_NAMED_PIPE_INLINE void NamedPipeServer::create_pipe(size_t index, HANDLE completion_port, const ServerConfig& config) { + std::wstring pipe_name_w = L"\\\\.\\pipe\\" + detail::utf8_to_wide(config.pipe_name); m_pipes[index] = CreateNamedPipeW( pipe_name_w.c_str(), @@ -206,7 +213,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 +259,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) { @@ -289,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]); @@ -335,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; @@ -349,7 +365,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 +379,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 +391,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,28 +440,39 @@ namespace SimpleNamedPipe { } } - inline bool NamedPipeServer::reconnect_client(size_t index, HANDLE completion_port, OVERLAPPED* ov) { - memset(ov, 0, sizeof(OVERLAPPED)); + SIMPLE_NAMED_PIPE_INLINE bool NamedPipeServer::reconnect_client(size_t index, HANDLE completion_port, OVERLAPPED* ov) { + 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; } - 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 +494,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 +531,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 +540,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 +549,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 +564,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 +572,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/include/SimpleNamedPipe/detail/string_utils.hpp b/include/SimpleNamedPipe/detail/string_utils.hpp new file mode 100644 index 0000000..c3370a1 --- /dev/null +++ b/include/SimpleNamedPipe/detail/string_utils.hpp @@ -0,0 +1,61 @@ +#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 { +namespace 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 detail +} // namespace SimpleNamedPipe + +#endif // _SIMPLE_NAMED_PIPE_DETAIL_STRING_UTILS_HPP_INCLUDED diff --git a/src/NamedPipeClient.cpp b/src/NamedPipeClient.cpp new file mode 100644 index 0000000..e855b0b --- /dev/null +++ b/src/NamedPipeClient.cpp @@ -0,0 +1,10 @@ +#ifdef _MSC_VER +#pragma warning(push) +#pragma warning(disable:4267) +#endif + +#include + +#ifdef _MSC_VER +#pragma warning(pop) +#endif 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 new file mode 100644 index 0000000..a92725f --- /dev/null +++ b/tests/client_server_test.cpp @@ -0,0 +1,462 @@ +#include +#include + +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +namespace { + +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, + Predicate predicate, + std::chrono::milliseconds timeout) { + std::unique_lock lock(mutex); + return cv.wait_for(lock, timeout, predicate); +} + +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(); +} + +std::string with_error(const std::string& message, const std::error_code& error) { + return message + error.message(); +} + +struct TestRunner { + int failures = 0; + + bool expect(bool condition, const std::string& message) { + if (!condition) { + ++failures; + std::cerr << "FAILED: " << message << "\n"; + } + return condition; + } + + 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"; + } + + if (failures == before) { + std::cerr << "[ OK ] " << name << "\n"; + } else { + std::cerr << "[ FAILED ] " << name << "\n"; + } + } +}; + +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_stop = [this](const SimpleNamedPipe::ServerConfig&) { + std::lock_guard lock(mutex); + stopped = true; + 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.on_error = [this](const std::error_code& error) { + std::lock_guard lock(mutex); + ++error_events; + last_error = error; + cv.notify_all(); + }; + } + + ~ServerHarness() { + server.stop(); + } + + void start() { + server.start(); + } + + void stop() { + server.stop(); + } + + bool wait_started(std::chrono::milliseconds timeout = kDefaultWait) { + return wait_until(cv, mutex, [this] { return started; }, timeout); + } + + bool wait_connected_at_least(int value, std::chrono::milliseconds timeout = kDefaultWait) { + return wait_until(cv, mutex, [this, value] { return connected_events >= value; }, timeout); + } + + 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, &ec)) { + return false; + } + + 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_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(256); + + 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))) { + return; + } + } + + 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), "257th client should not connect while all slots 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(257, kHeavyWait), "server did not observe slot reuse connection"); + + extra->close(); + for (size_t i = 0; i < clients.size(); ++i) { + clients[i]->close(); + } + + 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_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) { + 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/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..f9b485a 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 @@ -63,11 +69,36 @@ 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); } +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 +148,3 @@ inline BOOL SetCommTimeouts(HANDLE, const void*) { return 1; } inline BOOL SetNamedPipeHandleState(HANDLE, DWORD*, DWORD*, DWORD*) { return 1; } #define CALLBACK -