diff --git a/core/CMakeLists.txt b/core/CMakeLists.txt index d6f8ebae..7d4c5058 100644 --- a/core/CMakeLists.txt +++ b/core/CMakeLists.txt @@ -39,6 +39,7 @@ set(PUBLIC_HEADERS include/join/socket_stream.hpp include/join/acceptor.hpp include/join/async_socket.hpp + include/join/async_datagram_socket.hpp include/join/async_stream_socket.hpp include/join/async_acceptor.hpp include/join/cpu.hpp diff --git a/core/include/join/async_acceptor.hpp b/core/include/join/async_acceptor.hpp index 2fdcdeaa..56980664 100644 --- a/core/include/join/async_acceptor.hpp +++ b/core/include/join/async_acceptor.hpp @@ -58,7 +58,7 @@ namespace join using AsyncSocket = BasicAsyncStreamSocket; /// handler invoked on acceptation completion. - using AcceptHandler = Function; + using AcceptHandler = Function; /** * @brief create the acceptor instance. @@ -86,39 +86,14 @@ namespace join * @brief move constructor. * @param other other object to move. */ - BasicAsyncStreamAcceptor (BasicAsyncStreamAcceptor&& other) noexcept - : _acceptor (std::move (other._acceptor)) - , _engine (other._engine) - , _ops (std::move (other._ops)) - , _onAccept (std::move (other._onAccept)) - { - if (_ops) - { - _ops->accept.op.handler = this; - } - } + BasicAsyncStreamAcceptor (BasicAsyncStreamAcceptor&& other) = delete; /** * @brief move assignment operator. * @param other other object to assign. * @return assigned object. */ - BasicAsyncStreamAcceptor& operator= (BasicAsyncStreamAcceptor&& other) noexcept - { - close (); - - _acceptor = std::move (other._acceptor); - _engine = other._engine; - _ops = std::move (other._ops); - _onAccept = std::move (other._onAccept); - - if (_ops) - { - _ops->accept.op.handler = this; - } - - return *this; - } + BasicAsyncStreamAcceptor& operator= (BasicAsyncStreamAcceptor&& other) = delete; /** * @brief destroy the acceptor instance. @@ -146,26 +121,40 @@ namespace join { cancelAccept (); - if (_ops && !_engine->isProactorThread ()) + if (_engine->isProactorThread ()) { - Backoff backoff; + _acceptor.close (); + return; + } + + Backoff backoff; + AsyncOperation::State expected = AsyncOperation::Idle; - while (_ops->accept.state.load (std::memory_order_acquire) != AsyncOperation::Idle) + while (!_ops->accept.state.compare_exchange_strong (expected, AsyncOperation::Closing, + std::memory_order_acq_rel, std::memory_order_acquire)) + { + if (expected == AsyncOperation::Pending) { - backoff (); + cancelAccept (); } + + backoff (); + expected = AsyncOperation::Idle; } _acceptor.close (); + + _ops->accept.state.store (AsyncOperation::Idle, std::memory_order_release); } /** * @brief start an asynchronous acceptation. + * @param peer closed socket receiving the accepted connection, valid until the handler is invoked. * @param handler handler invoked on completion. * @param flags accepted socket creation flags. * @return 0 on success, -1 on failure. */ - int asyncAccept (AcceptHandler handler, int flags = SOCK_NONBLOCK | SOCK_CLOEXEC) noexcept + int asyncAccept (AsyncSocket& peer, AcceptHandler handler, int flags = SOCK_NONBLOCK | SOCK_CLOEXEC) noexcept { if (JOIN_UNLIKELY (!_acceptor.opened ())) { @@ -173,6 +162,12 @@ namespace join return -1; } + if (JOIN_UNLIKELY (peer.opened ())) + { + lastError = make_error_code (Errc::InUse); + return -1; + } + AsyncOperation::State expected = AsyncOperation::Idle; if (!_ops->accept.state.compare_exchange_strong (expected, AsyncOperation::Pending, @@ -189,6 +184,7 @@ namespace join _ops->peerLen = sizeof (struct sockaddr_storage); _onAccept = std::move (handler); + _peer = &peer; _ops->accept.op = IoOperation::makeAccept (_acceptor.handle (), _ops->peer.addr (), &_ops->peerLen, flags | SOCK_NONBLOCK, this); @@ -197,6 +193,7 @@ namespace join // LCOV_EXCL_START _ops->accept.state.store (AsyncOperation::Idle, std::memory_order_release); _onAccept.reset (); + _peer = nullptr; return -1; // LCOV_EXCL_STOP } @@ -210,7 +207,7 @@ namespace join */ int cancelAccept () noexcept { - if (!_ops || (_ops->accept.state.load (std::memory_order_acquire) == AsyncOperation::Idle)) + if (_ops->accept.state.load (std::memory_order_acquire) != AsyncOperation::Pending) { return 0; } @@ -341,20 +338,24 @@ namespace join AcceptHandler handler = std::move (_onAccept); - if (JOIN_UNLIKELY (!handler)) + AsyncSocket* peer = _peer; + _peer = nullptr; + + if (JOIN_UNLIKELY (result < 0)) { - if (result > -1) + if (JOIN_LIKELY (handler)) { - ::close (result); + handler (std::error_code (-result, std::generic_category ())); } } - else if (result < 0) - { - handler (std::error_code (-result, std::generic_category ()), AsyncSocket (*_engine)); - } else { - handler (std::error_code (), AsyncSocket (Socket (result, ops->peer), *_engine)); + peer->_socket = Socket (result, ops->peer); + + if (JOIN_LIKELY (handler)) + { + handler (std::error_code ()); + } } AsyncOperation::State expected = AsyncOperation::Dispatching; @@ -379,10 +380,13 @@ namespace join Engine* _engine; /// operation block. - std::unique_ptr _ops{new Ops ()}; + const std::unique_ptr _ops{new Ops ()}; /// handler invoked on acceptation completion. AcceptHandler _onAccept; + + /// socket receiving the accepted connection. + AsyncSocket* _peer = nullptr; }; } diff --git a/core/include/join/async_datagram_socket.hpp b/core/include/join/async_datagram_socket.hpp new file mode 100644 index 00000000..db9fe603 --- /dev/null +++ b/core/include/join/async_datagram_socket.hpp @@ -0,0 +1,357 @@ +/** + * MIT License + * + * Copyright (c) 2026 Mathieu Rabine + * + * Permission is hereby granted, free of charge, to any person obtaining a copy + * of this software and associated documentation files (the "Software"), to deal + * in the Software without restriction, including without limitation the rights + * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell + * copies of the Software, and to permit persons to whom the Software is + * furnished to do so, subject to the following conditions: + * + * The above copyright notice and this permission notice shall be included in all + * copies or substantial portions of the Software. + * + * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR + * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, + * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE + * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER + * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, + * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE + * SOFTWARE. + */ + +#ifndef JOIN_CORE_ASYNC_DATAGRAM_SOCKET_HPP +#define JOIN_CORE_ASYNC_DATAGRAM_SOCKET_HPP + +// libjoin. +#include +#include + +// C++. +#include +#include + +namespace join +{ + /** + * @brief asynchronous datagram socket class. + */ + template + class BasicAsyncDatagramSocket : public BasicAsyncSocket + { + public: + using Socket = BasicDatagramSocket; + using Endpoint = typename Protocol::Endpoint; + + /// handler invoked on read completion. + using ReadHandler = typename BasicAsyncSocket::ReadHandler; + + /// handler invoked on write completion. + using WriteHandler = typename BasicAsyncSocket::WriteHandler; + + /** + * @brief create the socket instance. + * @param engine engine driving the operations. + */ + explicit BasicAsyncDatagramSocket (Engine& engine = ProactorThread::proactor ()) + : BasicAsyncSocket (engine) + { + } + + /** + * @brief create the socket instance specifying the time to live. + * @param ttl packet time to live. + * @param engine engine driving the operations. + */ + explicit BasicAsyncDatagramSocket (int ttl, Engine& engine = ProactorThread::proactor ()) + : BasicAsyncSocket (Socket (ttl), engine) + { + } + + /** + * @brief create the socket instance adopting an already opened socket. + * @param sock socket to adopt. + * @param engine engine driving the operations. + */ + explicit BasicAsyncDatagramSocket (Socket&& sock, Engine& engine = ProactorThread::proactor ()) + : BasicAsyncSocket (std::move (sock), engine) + { + } + + /** + * @brief copy constructor. + * @param other other object to copy. + */ + BasicAsyncDatagramSocket (const BasicAsyncDatagramSocket& other) = delete; + + /** + * @brief copy assignment operator. + * @param other other object to assign. + * @return assigned object. + */ + BasicAsyncDatagramSocket& operator= (const BasicAsyncDatagramSocket& other) = delete; + + /** + * @brief move constructor. + * @param other other object to move. + */ + BasicAsyncDatagramSocket (BasicAsyncDatagramSocket&& other) = delete; + + /** + * @brief move assignment operator. + * @param other other object to assign. + * @return assigned object. + */ + BasicAsyncDatagramSocket& operator= (BasicAsyncDatagramSocket&& other) = delete; + + /** + * @brief destroy the socket instance. + */ + ~BasicAsyncDatagramSocket () + { + this->close (); + } + + /** + * @brief assign the default remote endpoint for this socket. + * @param endpoint endpoint to assign. + * @return 0 on success, -1 on failure. + */ + int connect (const Endpoint& endpoint) noexcept + { + return this->_socket.connect (endpoint); + } + + /** + * @brief remove the default remote endpoint. + * @return 0 on success, -1 on failure. + */ + int disconnect () noexcept + { + return this->_socket.disconnect (); + } + + /** + * @brief start an asynchronous read, reporting the endpoint the data are coming from. + * @param data buffer used to store the data received, valid until the handler is invoked. + * @param maxSize maximum number of bytes to read. + * @param endpoint endpoint from where data are coming, valid until the handler is invoked. + * @param handler handler invoked on completion. + * @return 0 on success, -1 on failure. + */ + int asyncReadFrom (char* data, size_t maxSize, Endpoint& endpoint, ReadHandler handler) noexcept + { + if (JOIN_UNLIKELY (!this->_socket.opened ())) + { + lastError = make_error_code (Errc::OperationFailed); + return -1; + } + + AsyncOperation::State expected = AsyncOperation::Idle; + + if (!this->_ops->read.state.compare_exchange_strong (expected, AsyncOperation::Pending, + std::memory_order_acquire, std::memory_order_acquire)) + { + if ((expected != AsyncOperation::Dispatching) || !this->_engine->isProactorThread ()) + { + lastError = make_error_code (Errc::InUse); + return -1; + } + + this->_ops->read.state.store (AsyncOperation::Pending, std::memory_order_release); + } + + this->_onRead = std::move (handler); + this->_ops->readIov.iov_base = data; + this->_ops->readIov.iov_len = maxSize; + this->_ops->readMsg.msg_name = endpoint.addr (); + this->_ops->readMsg.msg_namelen = sizeof (struct sockaddr_storage); + this->_ops->readMsg.msg_iov = &this->_ops->readIov; + this->_ops->readMsg.msg_iovlen = 1; + this->_ops->readMsg.msg_control = nullptr; + this->_ops->readMsg.msg_controllen = 0; + this->_ops->readMsg.msg_flags = 0; + this->_ops->read.op = IoOperation::makeRecvmsg (this->_socket.handle (), &this->_ops->readMsg, 0, this); + + if (this->_engine->submit (&this->_ops->read.op, true, false) == -1) + { + // LCOV_EXCL_START + this->_ops->read.state.store (AsyncOperation::Idle, std::memory_order_release); + this->_onRead.reset (); + return -1; + // LCOV_EXCL_STOP + } + + return 0; + } + + /** + * @brief start an asynchronous write to the given endpoint. + * @param data data buffer to send, valid until the handler is invoked. + * @param size number of bytes to write. + * @param endpoint endpoint where to write the data, valid until the handler is invoked. + * @param handler handler invoked on completion. + * @return 0 on success, -1 on failure. + */ + int asyncWriteTo (const char* data, size_t size, Endpoint& endpoint, WriteHandler handler) noexcept + { + if (!this->_socket.opened () && (this->_socket.open (endpoint.protocol ()) == -1)) + { + return -1; // LCOV_EXCL_LINE + } + + AsyncOperation::State expected = AsyncOperation::Idle; + + if (!this->_ops->write.state.compare_exchange_strong (expected, AsyncOperation::Pending, + std::memory_order_acquire, std::memory_order_acquire)) + { + if ((expected != AsyncOperation::Dispatching) || !this->_engine->isProactorThread ()) + { + lastError = make_error_code (Errc::InUse); + return -1; + } + + this->_ops->write.state.store (AsyncOperation::Pending, std::memory_order_release); + } + + this->_onWrite = std::move (handler); + this->_ops->writeIov.iov_base = const_cast (data); + this->_ops->writeIov.iov_len = size; + this->_ops->writeMsg.msg_name = endpoint.addr (); + this->_ops->writeMsg.msg_namelen = endpoint.length (); + this->_ops->writeMsg.msg_iov = &this->_ops->writeIov; + this->_ops->writeMsg.msg_iovlen = 1; + this->_ops->writeMsg.msg_control = nullptr; + this->_ops->writeMsg.msg_controllen = 0; + this->_ops->writeMsg.msg_flags = 0; + this->_ops->write.op = + IoOperation::makeSendmsg (this->_socket.handle (), &this->_ops->writeMsg, MSG_NOSIGNAL, this); + + if (this->_engine->submit (&this->_ops->write.op, true, false) == -1) + { + // LCOV_EXCL_START + this->_ops->write.state.store (AsyncOperation::Idle, std::memory_order_release); + this->_onWrite.reset (); + return -1; + // LCOV_EXCL_STOP + } + + return 0; + } + + /** + * @brief determine the remote endpoint associated with this socket. + * @return remote endpoint. + */ + const Endpoint& remoteEndpoint () const noexcept + { + return this->_socket.remoteEndpoint (); + } + + /** + * @brief check if the socket is connected. + * @return true if connected, false otherwise. + */ + bool connected () const noexcept + { + return this->_socket.connected (); + } + + /** + * @brief get socket mtu. + * @return mtu on success, -1 on failure. + */ + int mtu () const noexcept + { + return this->_socket.mtu (); + } + + /** + * @brief returns the Time-To-Live value. + * @return the Time-To-Live value. + */ + int ttl () const noexcept + { + return this->_socket.ttl (); + } + + protected: + /** + * @brief method called when an operation completes. + * @param op completed operation. + * @param result number of bytes transferred, or operation specific value. + */ + void onComplete (IoOperation* op, int result) override + { + dispatch (op, (result < 0) ? std::error_code (-result, std::generic_category ()) : std::error_code (), + (result > 0) ? static_cast (result) : 0); + } + + /** + * @brief method called when an operation is cancelled. + * @param op cancelled operation. + * @param result negative errno. + */ + void onCancel (IoOperation* op, [[maybe_unused]] int result) override + { + dispatch (op, make_error_code (std::errc::operation_canceled), 0); + } + + /** + * @brief invoke the handler owning the given operation slot. + * @param op completed or cancelled operation. + * @param code error code to report. + * @param size number of bytes transferred. + */ + void dispatch (IoOperation* op, const std::error_code& code, size_t size) noexcept + { + if (op == &this->_ops->read.op) + { + this->_ops->read.state.store (AsyncOperation::Dispatching, std::memory_order_release); + + ReadHandler handler = std::move (this->_onRead); + std::error_code result = code; + + if (JOIN_LIKELY (!result)) + { + if (JOIN_UNLIKELY (size == 0)) + { + result = make_error_code (Errc::ConnectionClosed); + this->_socket._state = Socket::Disconnected; + } + else if (JOIN_UNLIKELY (this->_ops->readMsg.msg_flags & MSG_TRUNC)) + { + result = make_error_code (Errc::MessageTooLong); + } + } + + if (JOIN_LIKELY (handler)) + { + handler (result, size); + } + + AsyncOperation::State expected = AsyncOperation::Dispatching; + this->_ops->read.state.compare_exchange_strong (expected, AsyncOperation::Idle, + std::memory_order_release, std::memory_order_relaxed); + return; + } + + this->_ops->write.state.store (AsyncOperation::Dispatching, std::memory_order_release); + + WriteHandler handler = std::move (this->_onWrite); + + if (JOIN_LIKELY (handler)) + { + handler (code, size); + } + + AsyncOperation::State expected = AsyncOperation::Dispatching; + this->_ops->write.state.compare_exchange_strong (expected, AsyncOperation::Idle, std::memory_order_release, + std::memory_order_relaxed); + } + }; +} + +#endif diff --git a/core/include/join/async_socket.hpp b/core/include/join/async_socket.hpp index 7bfb2d25..2190f291 100644 --- a/core/include/join/async_socket.hpp +++ b/core/include/join/async_socket.hpp @@ -61,6 +61,7 @@ namespace join Idle, /**< no operation in flight and no completion handler running. */ Pending, /**< an operation is in flight. */ Dispatching, /**< the completion handler is running. */ + Closing, /**< the socket is closing, no operation may be armed. */ }; /// operation. @@ -124,43 +125,14 @@ namespace join * @brief move constructor. * @param other other object to move. */ - BasicAsyncSocket (BasicAsyncSocket&& other) noexcept - : _socket (std::move (other._socket)) - , _engine (other._engine) - , _ops (std::move (other._ops)) - , _onRead (std::move (other._onRead)) - , _onWrite (std::move (other._onWrite)) - { - if (_ops) - { - _ops->read.op.handler = this; - _ops->write.op.handler = this; - } - } + BasicAsyncSocket (BasicAsyncSocket&& other) = delete; /** * @brief move assignment operator. * @param other other object to assign. * @return assigned object. */ - BasicAsyncSocket& operator= (BasicAsyncSocket&& other) noexcept - { - close (); - - _socket = std::move (other._socket); - _engine = other._engine; - _ops = std::move (other._ops); - _onRead = std::move (other._onRead); - _onWrite = std::move (other._onWrite); - - if (_ops) - { - _ops->read.op.handler = this; - _ops->write.op.handler = this; - } - - return *this; - } + BasicAsyncSocket& operator= (BasicAsyncSocket&& other) = delete; /** * @brief destroy the socket instance. @@ -188,18 +160,45 @@ namespace join cancelRead (); cancelWrite (); - if (_ops && !_engine->isProactorThread ()) + if (_engine->isProactorThread ()) + { + _socket.close (); + return; + } + + Backoff backoff; + AsyncOperation::State expected = AsyncOperation::Idle; + + while (!_ops->read.state.compare_exchange_strong (expected, AsyncOperation::Closing, + std::memory_order_acq_rel, std::memory_order_acquire)) { - Backoff backoff; + if (expected == AsyncOperation::Pending) + { + cancelRead (); + } - while ((_ops->read.state.load (std::memory_order_acquire) != AsyncOperation::Idle) || - (_ops->write.state.load (std::memory_order_acquire) != AsyncOperation::Idle)) + backoff (); + expected = AsyncOperation::Idle; + } + + expected = AsyncOperation::Idle; + + while (!_ops->write.state.compare_exchange_strong (expected, AsyncOperation::Closing, + std::memory_order_acq_rel, std::memory_order_acquire)) + { + if (expected == AsyncOperation::Pending) { - backoff (); + cancelWrite (); } + + backoff (); + expected = AsyncOperation::Idle; } _socket.close (); + + _ops->read.state.store (AsyncOperation::Idle, std::memory_order_release); + _ops->write.state.store (AsyncOperation::Idle, std::memory_order_release); } /** @@ -211,7 +210,7 @@ namespace join */ int asyncRead (char* data, size_t maxSize, ReadHandler handler) noexcept { - if (JOIN_UNLIKELY (!_ops || !_socket.opened ())) + if (JOIN_UNLIKELY (!_socket.opened ())) { lastError = make_error_code (Errc::OperationFailed); return -1; @@ -267,7 +266,7 @@ namespace join */ int asyncWrite (const char* data, size_t size, WriteHandler handler) noexcept { - if (JOIN_UNLIKELY (!_ops || !_socket.opened ())) + if (JOIN_UNLIKELY (!_socket.opened ())) { lastError = make_error_code (Errc::OperationFailed); return -1; @@ -320,7 +319,7 @@ namespace join */ int cancelRead () noexcept { - if (!_ops || (_ops->read.state.load (std::memory_order_acquire) == AsyncOperation::Idle)) + if (_ops->read.state.load (std::memory_order_acquire) != AsyncOperation::Pending) { return 0; } @@ -339,7 +338,7 @@ namespace join */ int cancelWrite () noexcept { - if (!_ops || (_ops->write.state.load (std::memory_order_acquire) == AsyncOperation::Idle)) + if (_ops->write.state.load (std::memory_order_acquire) != AsyncOperation::Pending) { return 0; } @@ -504,7 +503,7 @@ namespace join Engine* _engine; /// operation block. - std::unique_ptr _ops{new Ops ()}; + const std::unique_ptr _ops{new Ops ()}; /// handler invoked on read completion. ReadHandler _onRead; @@ -552,7 +551,7 @@ namespace join { if (JOIN_UNLIKELY (size == 0)) { - result = make_error_code (Errc::ConnectionClosed); + result = make_error_code (Errc::ConnectionClosed); // LCOV_EXCL_LINE } else if (JOIN_UNLIKELY (_ops->readMsg.msg_flags & MSG_TRUNC)) { diff --git a/core/include/join/async_stream_socket.hpp b/core/include/join/async_stream_socket.hpp index 4d99860f..ecf8c0fd 100644 --- a/core/include/join/async_stream_socket.hpp +++ b/core/include/join/async_stream_socket.hpp @@ -32,7 +32,6 @@ // C++. #include #include -#include namespace join { @@ -42,6 +41,10 @@ namespace join template class BasicAsyncStreamSocket : public BasicAsyncSocket { + /// friendship with basic asynchronous stream acceptor + template + friend class BasicAsyncStreamAcceptor; + public: using Socket = BasicStreamSocket; using Endpoint = typename Protocol::Endpoint; @@ -91,29 +94,14 @@ namespace join * @brief move constructor. * @param other other object to move. */ - BasicAsyncStreamSocket (BasicAsyncStreamSocket&& other) noexcept - : BasicAsyncSocket (std::move (other)) - , _remote (std::move (other._remote)) - , _onConnect (std::move (other._onConnect)) - { - } + BasicAsyncStreamSocket (BasicAsyncStreamSocket&& other) = delete; /** * @brief move assignment operator. * @param other other object to assign. * @return assigned object. */ - BasicAsyncStreamSocket& operator= (BasicAsyncStreamSocket&& other) noexcept - { - this->close (); - - BasicAsyncSocket::operator= (std::move (other)); - - _remote = std::move (other._remote); - _onConnect = std::move (other._onConnect); - - return *this; - } + BasicAsyncStreamSocket& operator= (BasicAsyncStreamSocket&& other) = delete; /** * @brief destroy the socket instance. @@ -131,12 +119,6 @@ namespace join */ int asyncConnect (const Endpoint& endpoint, ConnectHandler handler) noexcept { - if (!this->_ops || !_remote) - { - lastError = make_error_code (Errc::OperationFailed); - return -1; - } - if (!this->_socket.opened () && (this->_socket.open (endpoint.protocol ()) == -1)) { return -1; // LCOV_EXCL_LINE @@ -159,9 +141,8 @@ namespace join this->_socket._state = Socket::Connecting; this->_socket._remote = endpoint; _onConnect = std::move (handler); - *_remote = endpoint; - this->_ops->write.op = - IoOperation::makeConnect (this->_socket.handle (), _remote->addr (), _remote->length (), this); + this->_ops->write.op = IoOperation::makeConnect (this->_socket.handle (), this->_socket._remote.addr (), + this->_socket._remote.length (), this); if (this->_engine->submit (&this->_ops->write.op, true, false) == -1) { @@ -300,9 +281,6 @@ namespace join std::memory_order_relaxed); } - /// remote endpoint. - std::unique_ptr _remote{new Endpoint ()}; - /// handler invoked on connect completion. ConnectHandler _onConnect; }; diff --git a/core/include/join/datagram_socket.hpp b/core/include/join/datagram_socket.hpp index d441981f..b6eed432 100644 --- a/core/include/join/datagram_socket.hpp +++ b/core/include/join/datagram_socket.hpp @@ -39,6 +39,10 @@ namespace join template class BasicDatagramSocket final : public BasicSocket { + /// friendship with basic asynchronous datagram socket + template + friend class BasicAsyncDatagramSocket; + public: using Ptr = std::unique_ptr>; using Mode = typename BasicSocket::Mode; @@ -247,9 +251,20 @@ namespace join int readFrom (char* data, unsigned long maxSize, Endpoint* endpoint = nullptr) noexcept { struct sockaddr_storage sa; - socklen_t sa_len = sizeof (struct sockaddr_storage); - int size = ::recvfrom (this->_handle, data, maxSize, 0, reinterpret_cast (&sa), &sa_len); + struct iovec iov; + iov.iov_base = data; + iov.iov_len = maxSize; + + struct msghdr message; + message.msg_name = &sa; + message.msg_namelen = sizeof (struct sockaddr_storage); + message.msg_iov = &iov; + message.msg_iovlen = 1; + message.msg_control = nullptr; + message.msg_controllen = 0; + + int size = ::recvmsg (this->_handle, &message, 0); if (size < 1) { if (size == -1) @@ -265,9 +280,15 @@ namespace join return -1; } + if (message.msg_flags & MSG_TRUNC) + { + lastError = make_error_code (Errc::MessageTooLong); + return -1; + } + if (endpoint != nullptr) { - *endpoint = Endpoint (reinterpret_cast (&sa), sa_len); + *endpoint = Endpoint (reinterpret_cast (&sa), message.msg_namelen); } return size; diff --git a/core/include/join/protocol.hpp b/core/include/join/protocol.hpp index 24978cdb..15d99ebc 100644 --- a/core/include/join/protocol.hpp +++ b/core/include/join/protocol.hpp @@ -57,6 +57,9 @@ namespace join template > class BasicAsyncSocket; + template > + class BasicAsyncDatagramSocket; + template > class BasicAsyncStreamSocket; @@ -68,6 +71,9 @@ namespace join template class BasicAsyncSocket; + template + class BasicAsyncDatagramSocket; + template class BasicAsyncStreamSocket; @@ -83,6 +89,7 @@ namespace join public: using Endpoint = BasicUnixEndpoint; using Socket = BasicDatagramSocket; + using AsyncSocket = BasicAsyncDatagramSocket; /** * @brief construct the unix datagram protocol instance by default. @@ -218,6 +225,7 @@ namespace join public: using Endpoint = BasicInternetEndpoint; using Socket = BasicDatagramSocket; + using AsyncSocket = BasicAsyncDatagramSocket; /** * @brief construct the udp protocol instance. @@ -310,6 +318,7 @@ namespace join public: using Endpoint = BasicInternetEndpoint; using Socket = BasicDatagramSocket; + using AsyncSocket = BasicAsyncDatagramSocket; /** * @brief create the icmp protocol instance. diff --git a/core/tests/CMakeLists.txt b/core/tests/CMakeLists.txt index 3ce26d2b..fa606e7f 100644 --- a/core/tests/CMakeLists.txt +++ b/core/tests/CMakeLists.txt @@ -231,16 +231,6 @@ target_link_libraries(unix_stream_acceptor.gtest ${JOIN_CORE} GTest::gtest_main) add_test(NAME unix_stream_acceptor.gtest COMMAND unix_stream_acceptor.gtest) install(TARGETS unix_stream_acceptor.gtest RUNTIME DESTINATION ${CMAKE_INSTALL_DATADIR}/${PROJECT_NAME}/test) -add_executable(unix_async_stream_socket.gtest unix_async_stream_socket_test.cpp) -target_link_libraries(unix_async_stream_socket.gtest ${JOIN_CORE} GTest::gtest_main) -add_test(NAME unix_async_stream_socket.gtest COMMAND unix_async_stream_socket.gtest) -install(TARGETS unix_async_stream_socket.gtest RUNTIME DESTINATION ${CMAKE_INSTALL_DATADIR}/${PROJECT_NAME}/test) - -add_executable(unix_async_acceptor.gtest unix_async_acceptor_test.cpp) -target_link_libraries(unix_async_acceptor.gtest ${JOIN_CORE} GTest::gtest_main) -add_test(NAME unix_async_acceptor.gtest COMMAND unix_async_acceptor.gtest) -install(TARGETS unix_async_acceptor.gtest RUNTIME DESTINATION ${CMAKE_INSTALL_DATADIR}/${PROJECT_NAME}/test) - add_executable(raw_socket.gtest raw_socket_test.cpp) target_link_libraries(raw_socket.gtest ${JOIN_CORE} GTest::gtest_main) add_test(NAME raw_socket.gtest COMMAND raw_socket.gtest) @@ -271,11 +261,36 @@ target_link_libraries(tcp_acceptor.gtest ${JOIN_CORE} GTest::gtest_main) add_test(NAME tcp_acceptor.gtest COMMAND tcp_acceptor.gtest) install(TARGETS tcp_acceptor.gtest RUNTIME DESTINATION ${CMAKE_INSTALL_DATADIR}/${PROJECT_NAME}/test) +add_executable(unix_async_datagram_socket.gtest unix_async_datagram_socket_test.cpp) +target_link_libraries(unix_async_datagram_socket.gtest ${JOIN_CORE} GTest::gtest_main) +add_test(NAME unix_async_datagram_socket.gtest COMMAND unix_async_datagram_socket.gtest) +install(TARGETS unix_async_datagram_socket.gtest RUNTIME DESTINATION ${CMAKE_INSTALL_DATADIR}/${PROJECT_NAME}/test) + +add_executable(unix_async_stream_socket.gtest unix_async_stream_socket_test.cpp) +target_link_libraries(unix_async_stream_socket.gtest ${JOIN_CORE} GTest::gtest_main) +add_test(NAME unix_async_stream_socket.gtest COMMAND unix_async_stream_socket.gtest) +install(TARGETS unix_async_stream_socket.gtest RUNTIME DESTINATION ${CMAKE_INSTALL_DATADIR}/${PROJECT_NAME}/test) + +add_executable(unix_async_acceptor.gtest unix_async_acceptor_test.cpp) +target_link_libraries(unix_async_acceptor.gtest ${JOIN_CORE} GTest::gtest_main) +add_test(NAME unix_async_acceptor.gtest COMMAND unix_async_acceptor.gtest) +install(TARGETS unix_async_acceptor.gtest RUNTIME DESTINATION ${CMAKE_INSTALL_DATADIR}/${PROJECT_NAME}/test) + add_executable(raw_async_socket.gtest raw_async_socket_test.cpp) target_link_libraries(raw_async_socket.gtest ${JOIN_CORE} GTest::gtest_main) add_test(NAME raw_async_socket.gtest COMMAND raw_async_socket.gtest) install(TARGETS raw_async_socket.gtest RUNTIME DESTINATION ${CMAKE_INSTALL_DATADIR}/${PROJECT_NAME}/test) +add_executable(udp_async_datagram_socket.gtest udp_async_datagram_socket_test.cpp) +target_link_libraries(udp_async_datagram_socket.gtest ${JOIN_CORE} GTest::gtest_main) +add_test(NAME udp_async_datagram_socket.gtest COMMAND udp_async_datagram_socket.gtest) +install(TARGETS udp_async_datagram_socket.gtest RUNTIME DESTINATION ${CMAKE_INSTALL_DATADIR}/${PROJECT_NAME}/test) + +add_executable(icmp_async_datagram_socket.gtest icmp_async_datagram_socket_test.cpp) +target_link_libraries(icmp_async_datagram_socket.gtest ${JOIN_CORE} GTest::gtest_main) +add_test(NAME icmp_async_datagram_socket.gtest COMMAND icmp_async_datagram_socket.gtest) +install(TARGETS icmp_async_datagram_socket.gtest RUNTIME DESTINATION ${CMAKE_INSTALL_DATADIR}/${PROJECT_NAME}/test) + add_executable(tcp_async_stream_socket.gtest tcp_async_stream_socket_test.cpp) target_link_libraries(tcp_async_stream_socket.gtest ${JOIN_CORE} GTest::gtest_main) add_test(NAME tcp_async_stream_socket.gtest COMMAND tcp_async_stream_socket.gtest) diff --git a/core/tests/icmp_async_datagram_socket_test.cpp b/core/tests/icmp_async_datagram_socket_test.cpp new file mode 100644 index 00000000..50efbb1f --- /dev/null +++ b/core/tests/icmp_async_datagram_socket_test.cpp @@ -0,0 +1,536 @@ +/** + * MIT License + * + * Copyright (c) 2026 Mathieu Rabine + * + * Permission is hereby granted, free of charge, to any person obtaining a copy + * of this software and associated documentation files (the "Software"), to deal + * in the Software without restriction, including without limitation the rights + * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell + * copies of the Software, and to permit persons to whom the Software is + * furnished to do so, subject to the following conditions: + * + * The above copyright notice and this permission notice shall be included in all + * copies or substantial portions of the Software. + * + * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR + * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, + * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE + * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER + * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, + * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE + * SOFTWARE. + */ + +// libjoin. +#include +#include + +// Libraries. +#include + +using join::Errc; +using join::Mutex; +using join::Condition; +using join::ScopedLock; +using join::IpAddress; +using join::Icmp; + +/** + * @brief Class used to test the icmp asynchronous datagram socket API. + */ +class IcmpAsyncDatagramSocket : public ::testing::Test +{ +public: + /** + * @brief set up test case. + */ + static void SetUpTestCase () + { + struct icmphdr* icmp = reinterpret_cast (_data); + + icmp->type = ICMP_ECHO; + icmp->code = 0; + icmp->checksum = 0; + icmp->un.echo.sequence = htons (1); + icmp->un.echo.id = htons (getpid () & 0xFFFF); + icmp->checksum = Icmp::Socket::checksum (reinterpret_cast (icmp), sizeof (struct icmphdr), 0); + } + +protected: + /** + * @brief Sets up the test fixture. + */ + void SetUp () override + { + ScopedLock lock (_mut); + + _code = {}; + _completions = 0; + _transferred = 0; + } + + /** + * @brief report a completion to the test thread. + * @param ec error reported by the socket. + * @param size number of bytes transferred. + */ + static void onReport (const std::error_code& ec, size_t size) + { + ScopedLock lock (_mut); + + _code = ec; + _transferred = size; + ++_completions; + _cond.signal (); + } + + /// condition mutex. + static Mutex _mut; + + /// condition variable. + static Condition _cond; + + /// last reported error. + static std::error_code _code; + + /// number of completions reported. + static int _completions; + + /// number of bytes reported by the last completion. + static size_t _transferred; + + /// read buffer. + static char _buf[1024]; + + /// endpoint the last datagram was received from. + static Icmp::Endpoint _from; + + /// echo request sent by the tests. + static char _data[sizeof (struct icmphdr)]; + + /// host. + static const std::string _host; + + /// timeout. + static const int _timeout; +}; + +Mutex IcmpAsyncDatagramSocket::_mut; +Condition IcmpAsyncDatagramSocket::_cond; +std::error_code IcmpAsyncDatagramSocket::_code; +int IcmpAsyncDatagramSocket::_completions = 0; +size_t IcmpAsyncDatagramSocket::_transferred = 0; +char IcmpAsyncDatagramSocket::_buf[1024] = {}; +Icmp::Endpoint IcmpAsyncDatagramSocket::_from; +char IcmpAsyncDatagramSocket::_data[sizeof (struct icmphdr)] = {}; +const std::string IcmpAsyncDatagramSocket::_host = "127.0.0.1"; +const int IcmpAsyncDatagramSocket::_timeout = 1000; + +/** + * @brief Test open method. + */ +TEST_F (IcmpAsyncDatagramSocket, open) +{ + Icmp::AsyncSocket client; + + ASSERT_EQ (client.open (Icmp::v4 ()), 0) << join::lastError.message (); + ASSERT_EQ (client.open (Icmp::v4 ()), -1); + ASSERT_EQ (join::lastError, Errc::InUse); + client.close (); + + ASSERT_EQ (client.open (Icmp::v6 ()), 0) << join::lastError.message (); + ASSERT_EQ (client.open (Icmp::v6 ()), -1); + ASSERT_EQ (join::lastError, Errc::InUse); + client.close (); +} + +/** + * @brief Test close method. + */ +TEST_F (IcmpAsyncDatagramSocket, close) +{ + Icmp::AsyncSocket client; + + ASSERT_EQ (client.open (Icmp::v4 ()), 0) << join::lastError.message (); + ASSERT_TRUE (client.opened ()); + client.close (); + ASSERT_FALSE (client.opened ()); +} + +/** + * @brief Test bind method. + */ +TEST_F (IcmpAsyncDatagramSocket, bind) +{ + Icmp::AsyncSocket client; + + ASSERT_EQ (client.connect (_host), 0) << join::lastError.message (); + ASSERT_EQ (client.bind (_host), -1); + ASSERT_EQ (client.disconnect (), 0) << join::lastError.message (); + + ASSERT_EQ (client.bind (_host), 0) << join::lastError.message (); + client.close (); +} + +/** + * @brief Test bindToDevice method. + */ +TEST_F (IcmpAsyncDatagramSocket, bindToDevice) +{ + Icmp::AsyncSocket client; + + ASSERT_EQ (client.bindToDevice ("lo"), -1); + ASSERT_EQ (client.open (Icmp::v4 ()), 0) << join::lastError.message (); + ASSERT_EQ (client.bindToDevice ("lo"), 0) << join::lastError.message (); + ASSERT_EQ (client.bindToDevice ("foo"), -1); + client.close (); +} + +/** + * @brief Test connect method. + */ +TEST_F (IcmpAsyncDatagramSocket, connect) +{ + Icmp::AsyncSocket client; + + ASSERT_EQ (client.connect ("255.255.255.255"), -1); + + ASSERT_EQ (client.connect (_host), 0) << join::lastError.message (); + ASSERT_TRUE (client.connected ()); + ASSERT_EQ (client.connect (_host), -1); + ASSERT_EQ (join::lastError, Errc::InUse); + client.close (); +} + +/** + * @brief Test disconnect method. + */ +TEST_F (IcmpAsyncDatagramSocket, disconnect) +{ + Icmp::AsyncSocket client; + + ASSERT_EQ (client.disconnect (), 0) << join::lastError.message (); + ASSERT_EQ (client.connect (_host), 0) << join::lastError.message (); + ASSERT_TRUE (client.connected ()); + ASSERT_EQ (client.disconnect (), 0) << join::lastError.message (); + ASSERT_FALSE (client.connected ()); + client.close (); +} + +/** + * @brief Test asyncWriteTo method. + */ +TEST_F (IcmpAsyncDatagramSocket, asyncWriteTo) +{ + Icmp::AsyncSocket client; + Icmp::Endpoint dest (_host); + + ASSERT_FALSE (client.opened ()); + ASSERT_EQ (client.asyncWriteTo (_data, sizeof (_data), dest, onReport), 0) << join::lastError.message (); + ASSERT_TRUE (client.opened ()); + + ASSERT_EQ (client.asyncWriteTo (_data, sizeof (_data), dest, nullptr), -1); + ASSERT_EQ (join::lastError, Errc::InUse); + + { + ScopedLock lock (_mut); + ASSERT_TRUE (_cond.timedWait (lock, std::chrono::milliseconds (_timeout), [] () { + return _completions >= 1; + })); + ASSERT_FALSE (_code) << _code.message (); + ASSERT_EQ (_transferred, sizeof (_data)); + } + + client.close (); +} + +/** + * @brief Test asyncReadFrom method. + */ +TEST_F (IcmpAsyncDatagramSocket, asyncReadFrom) +{ + Icmp::AsyncSocket client, server; + + ASSERT_EQ (server.asyncReadFrom (_buf, sizeof (_buf), _from, nullptr), -1); + ASSERT_EQ (join::lastError, Errc::OperationFailed); + + ASSERT_EQ (server.bind (_host), 0) << join::lastError.message (); + ASSERT_EQ (server.asyncReadFrom (_buf, sizeof (_buf), _from, onReport), 0) << join::lastError.message (); + + ASSERT_EQ (server.asyncReadFrom (_buf, sizeof (_buf), _from, nullptr), -1); + ASSERT_EQ (join::lastError, Errc::InUse); + + ASSERT_EQ (client.connect (_host), 0) << join::lastError.message (); + ASSERT_EQ (client.asyncWrite (_data, sizeof (_data), nullptr), 0) << join::lastError.message (); + + { + ScopedLock lock (_mut); + ASSERT_TRUE (_cond.timedWait (lock, std::chrono::milliseconds (_timeout), [] () { + return _completions >= 1; + })); + ASSERT_FALSE (_code) << _code.message (); + ASSERT_GT (_transferred, 0u); + } + + client.close (); + server.close (); + + ASSERT_EQ (_from, Icmp::Endpoint (_host)); +} + +/** + * @brief Test asyncWrite method. + */ +TEST_F (IcmpAsyncDatagramSocket, asyncWrite) +{ + Icmp::AsyncSocket client; + + ASSERT_EQ (client.asyncWrite (_data, sizeof (_data), nullptr), -1); + ASSERT_EQ (join::lastError, Errc::OperationFailed); + + ASSERT_EQ (client.connect (_host), 0) << join::lastError.message (); + ASSERT_EQ (client.asyncWrite (_data, sizeof (_data), onReport), 0) << join::lastError.message (); + + { + ScopedLock lock (_mut); + ASSERT_TRUE (_cond.timedWait (lock, std::chrono::milliseconds (_timeout), [] () { + return _completions >= 1; + })); + ASSERT_FALSE (_code) << _code.message (); + ASSERT_EQ (_transferred, sizeof (_data)); + } + + client.close (); +} + +/** + * @brief Test asyncRead method. + */ +TEST_F (IcmpAsyncDatagramSocket, asyncRead) +{ + Icmp::AsyncSocket client; + + ASSERT_EQ (client.asyncRead (_buf, sizeof (_buf), nullptr), -1); + ASSERT_EQ (join::lastError, Errc::OperationFailed); + + ASSERT_EQ (client.connect (_host), 0) << join::lastError.message (); + ASSERT_EQ (client.asyncRead (_buf, sizeof (_buf), onReport), 0) << join::lastError.message (); + ASSERT_EQ (client.asyncWrite (_data, sizeof (_data), nullptr), 0) << join::lastError.message (); + + { + ScopedLock lock (_mut); + ASSERT_TRUE (_cond.timedWait (lock, std::chrono::milliseconds (_timeout), [] () { + return _completions >= 1; + })); + ASSERT_FALSE (_code) << _code.message (); + ASSERT_GT (_transferred, 0u); + } + + client.close (); +} + +/** + * @brief Test cancelRead method. + */ +TEST_F (IcmpAsyncDatagramSocket, cancelRead) +{ + Icmp::AsyncSocket client; + + ASSERT_EQ (client.cancelRead (), 0) << join::lastError.message (); + ASSERT_EQ (client.bind (_host), 0) << join::lastError.message (); + ASSERT_EQ (client.asyncReadFrom (_buf, sizeof (_buf), _from, onReport), 0) << join::lastError.message (); + ASSERT_EQ (client.cancelRead (), 0) << join::lastError.message (); + + { + ScopedLock lock (_mut); + ASSERT_TRUE (_cond.timedWait (lock, std::chrono::milliseconds (_timeout), [] () { + return _completions >= 1; + })); + ASSERT_EQ (_code, std::errc::operation_canceled); + } + + client.close (); +} + +/** + * @brief Test cancelWrite method. + */ +TEST_F (IcmpAsyncDatagramSocket, cancelWrite) +{ + Icmp::AsyncSocket client; + + ASSERT_EQ (client.cancelWrite (), 0) << join::lastError.message (); + ASSERT_EQ (client.open (Icmp::v4 ()), 0) << join::lastError.message (); + ASSERT_EQ (client.cancelWrite (), 0) << join::lastError.message (); + client.close (); +} + +/** + * @brief Test setOption method. + */ +TEST_F (IcmpAsyncDatagramSocket, setOption) +{ + Icmp::AsyncSocket client; + + ASSERT_EQ (client.setOption (Icmp::Socket::RcvBuffer, 1500), -1); + ASSERT_EQ (join::lastError, Errc::OperationFailed); + ASSERT_EQ (client.open (Icmp::v4 ()), 0) << join::lastError.message (); + ASSERT_EQ (client.setOption (Icmp::Socket::RcvBuffer, 1500), 0) << join::lastError.message (); + client.close (); +} + +/** + * @brief Test localEndpoint method. + */ +TEST_F (IcmpAsyncDatagramSocket, localEndpoint) +{ + Icmp::AsyncSocket client; + + ASSERT_EQ (client.localEndpoint (), Icmp::Endpoint{}); + ASSERT_EQ (client.bind (_host), 0) << join::lastError.message (); + ASSERT_EQ (client.localEndpoint ().ip (), IpAddress (_host)); + client.close (); +} + +/** + * @brief Test remoteEndpoint method. + */ +TEST_F (IcmpAsyncDatagramSocket, remoteEndpoint) +{ + Icmp::AsyncSocket client; + + ASSERT_EQ (client.connect (_host), 0) << join::lastError.message (); + ASSERT_EQ (client.remoteEndpoint (), Icmp::Endpoint (_host)); + client.close (); +} + +/** + * @brief Test opened method. + */ +TEST_F (IcmpAsyncDatagramSocket, opened) +{ + Icmp::AsyncSocket client; + + ASSERT_FALSE (client.opened ()); + ASSERT_EQ (client.open (Icmp::v4 ()), 0) << join::lastError.message (); + ASSERT_TRUE (client.opened ()); + client.close (); + ASSERT_FALSE (client.opened ()); +} + +/** + * @brief Test connected method. + */ +TEST_F (IcmpAsyncDatagramSocket, connected) +{ + Icmp::AsyncSocket client; + + ASSERT_FALSE (client.connected ()); + ASSERT_EQ (client.connect (_host), 0) << join::lastError.message (); + ASSERT_TRUE (client.connected ()); + client.close (); + ASSERT_FALSE (client.connected ()); +} + +/** + * @brief Test canRead method. + */ +TEST_F (IcmpAsyncDatagramSocket, canRead) +{ + Icmp::AsyncSocket client; + + ASSERT_EQ (client.canRead (), -1); + ASSERT_EQ (join::lastError, Errc::OperationFailed); + ASSERT_EQ (client.open (Icmp::v4 ()), 0) << join::lastError.message (); + ASSERT_EQ (client.canRead (), 0) << join::lastError.message (); + client.close (); +} + +/** + * @brief Test mtu method. + */ +TEST_F (IcmpAsyncDatagramSocket, mtu) +{ + Icmp::AsyncSocket client; + + ASSERT_EQ (client.mtu (), -1); + ASSERT_EQ (client.connect (_host), 0) << join::lastError.message (); + ASSERT_NE (client.mtu (), -1) << join::lastError.message (); + client.close (); + ASSERT_EQ (client.mtu (), -1); +} + +/** + * @brief Test ttl method. + */ +TEST_F (IcmpAsyncDatagramSocket, ttl) +{ + Icmp::AsyncSocket client; + + ASSERT_EQ (client.ttl (), 60); + + Icmp::AsyncSocket other (32); + + ASSERT_EQ (other.ttl (), 32); +} + +/** + * @brief Test family method. + */ +TEST_F (IcmpAsyncDatagramSocket, family) +{ + Icmp::AsyncSocket client; + + ASSERT_EQ (client.open (Icmp::v4 ()), 0) << join::lastError.message (); + ASSERT_EQ (client.family (), AF_INET); + client.close (); +} + +/** + * @brief Test type method. + */ +TEST_F (IcmpAsyncDatagramSocket, type) +{ + Icmp::AsyncSocket client; + + ASSERT_EQ (client.type (), SOCK_RAW); +} + +/** + * @brief Test protocol method. + */ +TEST_F (IcmpAsyncDatagramSocket, protocol) +{ + Icmp::AsyncSocket client; + + ASSERT_EQ (client.bind (IpAddress (AF_INET6)), 0) << join::lastError.message (); + ASSERT_EQ (client.protocol (), IPPROTO_ICMPV6); + client.close (); + + ASSERT_EQ (client.bind (IpAddress (AF_INET)), 0) << join::lastError.message (); + ASSERT_EQ (client.protocol (), IPPROTO_ICMP); + client.close (); +} + +/** + * @brief Test handle method. + */ +TEST_F (IcmpAsyncDatagramSocket, handle) +{ + Icmp::AsyncSocket client; + + ASSERT_EQ (client.handle (), -1); + ASSERT_EQ (client.open (Icmp::v4 ()), 0) << join::lastError.message (); + ASSERT_GT (client.handle (), -1); + client.close (); +} + +/** + * @brief main function. + */ +int main (int argc, char** argv) +{ + testing::InitGoogleTest (&argc, argv); + return RUN_ALL_TESTS (); +} diff --git a/core/tests/raw_async_socket_test.cpp b/core/tests/raw_async_socket_test.cpp index 063d73a3..907cc476 100644 --- a/core/tests/raw_async_socket_test.cpp +++ b/core/tests/raw_async_socket_test.cpp @@ -174,52 +174,6 @@ char RawAsyncSocket::_buf[2048] = {}; const std::string RawAsyncSocket::_interface = "lo"; const int RawAsyncSocket::_timeout = 1000; -/** - * @brief Test move with an operation in flight. - */ -TEST_F (RawAsyncSocket, move) -{ - Raw::AsyncSocket rawSocket; - - ASSERT_EQ (rawSocket.bind (_interface), 0) << join::lastError.message (); - ASSERT_EQ (rawSocket.asyncRead (_buf, sizeof (_buf), onCompletion), 0) << join::lastError.message (); - - ASSERT_EQ (rawSocket.asyncRead (_buf, sizeof (_buf), nullptr), -1); - ASSERT_EQ (join::lastError, Errc::InUse); - - Raw::AsyncSocket moved (std::move (rawSocket)); - - ASSERT_EQ (moved.asyncRead (_buf, sizeof (_buf), nullptr), -1); - ASSERT_EQ (join::lastError, Errc::InUse); - - ASSERT_EQ (moved.asyncWrite (reinterpret_cast (&_packet), sizeof (_packet), nullptr), 0) - << join::lastError.message (); - - ASSERT_TRUE (wait (1)); - ASSERT_FALSE (_code) << _code.message (); - ASSERT_GT (_transferred, 0u); - - Raw::AsyncSocket assigned; - assigned = std::move (moved); - - ASSERT_TRUE (assigned.opened ()); - - // the moved from socket must stay safe to use. - ASSERT_EQ (moved.cancelRead (), 0) << join::lastError.message (); - ASSERT_EQ (moved.cancelWrite (), 0) << join::lastError.message (); - ASSERT_FALSE (moved.opened ()); - moved.close (); - - Raw::AsyncSocket chained (std::move (moved)); - ASSERT_FALSE (chained.opened ()); - - Raw::AsyncSocket reassigned; - reassigned = std::move (chained); - ASSERT_FALSE (reassigned.opened ()); - - assigned.close (); -} - /** * @brief Test open method. */ diff --git a/core/tests/tcp_async_acceptor_test.cpp b/core/tests/tcp_async_acceptor_test.cpp index b41fab40..15e90eda 100644 --- a/core/tests/tcp_async_acceptor_test.cpp +++ b/core/tests/tcp_async_acceptor_test.cpp @@ -54,22 +54,40 @@ class TcpAsyncAcceptor : public ::testing::Test _code = {}; _completions = 0; - _peerConnected = false; - _peerFamily = -1; + + peer ().close (); + spare ().close (); + } + + /** + * @brief get the socket receiving the accepted connections. + * @return the socket receiving the accepted connections. + */ + static Tcp::AsyncSocket& peer () + { + static Tcp::AsyncSocket sock; + return sock; + } + + /** + * @brief get the socket receiving the connection accepted by a resubmitted acceptation. + * @return the socket receiving the connection accepted by a resubmitted acceptation. + */ + static Tcp::AsyncSocket& spare () + { + static Tcp::AsyncSocket sock; + return sock; } /** * @brief report a completion to the test thread. * @param ec error reported by the acceptor. - * @param peer accepted socket. */ - static void onReport (const std::error_code& ec, Tcp::AsyncSocket&& peer) + static void onReport (const std::error_code& ec) { ScopedLock lock (_mut); _code = ec; - _peerConnected = peer.connected (); - _peerFamily = peer.family (); ++_completions; _cond.signal (); } @@ -77,28 +95,26 @@ class TcpAsyncAcceptor : public ::testing::Test /** * @brief handler resubmitting an acceptation from within itself. * @param ec error reported by the acceptor. - * @param peer accepted socket. */ - static void onAccept (const std::error_code& ec, Tcp::AsyncSocket&& peer) + static void onAccept (const std::error_code& ec) { if (!ec) { - _current->asyncAccept (onAccept); + _current->asyncAccept (spare (), onAccept); } - onReport (ec, std::move (peer)); + onReport (ec); } /** * @brief handler closing the acceptor from within itself. * @param ec error reported by the acceptor. - * @param peer accepted socket. */ - static void onAcceptAndClose (const std::error_code& ec, Tcp::AsyncSocket&& peer) + static void onAcceptAndClose (const std::error_code& ec) { _current->close (); - onReport (ec, std::move (peer)); + onReport (ec); } /// acceptor address. @@ -122,12 +138,6 @@ class TcpAsyncAcceptor : public ::testing::Test /// number of completions reported. static int _completions; - /// state of the last accepted socket. - static bool _peerConnected; - - /// address family of the last accepted socket. - static int _peerFamily; - /// acceptor used by the resubmitting handler. static Tcp::AsyncAcceptor* _current; }; @@ -139,65 +149,8 @@ Mutex TcpAsyncAcceptor::_mut; Condition TcpAsyncAcceptor::_cond; std::error_code TcpAsyncAcceptor::_code; int TcpAsyncAcceptor::_completions = 0; -bool TcpAsyncAcceptor::_peerConnected = false; -int TcpAsyncAcceptor::_peerFamily = -1; Tcp::AsyncAcceptor* TcpAsyncAcceptor::_current = nullptr; -/** - * @brief Test move with an acceptation in flight. - */ -TEST_F (TcpAsyncAcceptor, move) -{ - Tcp::AsyncAcceptor server; - Tcp::Socket client (Tcp::Socket::Blocking); - - ASSERT_EQ (server.create ({_address, _port}), 0) << join::lastError.message (); - - ASSERT_EQ (server.asyncAccept ([] (const std::error_code& ec, Tcp::AsyncSocket&& peer) { - onReport (ec, std::move (peer)); - }), - 0) - << join::lastError.message (); - - ASSERT_EQ (server.asyncAccept (nullptr), -1); - ASSERT_EQ (join::lastError, Errc::InUse); - - Tcp::AsyncAcceptor moved (std::move (server)); - - ASSERT_EQ (moved.asyncAccept (nullptr), -1); - ASSERT_EQ (join::lastError, Errc::InUse); - - ASSERT_EQ (client.connect ({_address, _port}), 0) << join::lastError.message (); - - { - ScopedLock lock (_mut); - ASSERT_TRUE (_cond.timedWait (lock, std::chrono::milliseconds (_timeout), [] () { - return _completions >= 1; - })); - ASSERT_FALSE (_code) << _code.message (); - ASSERT_TRUE (_peerConnected); - } - - Tcp::AsyncAcceptor assigned; - assigned = std::move (moved); - - ASSERT_TRUE (assigned.opened ()); - - ASSERT_EQ (moved.cancelAccept (), 0) << join::lastError.message (); - ASSERT_FALSE (moved.opened ()); - moved.close (); - - Tcp::AsyncAcceptor chained (std::move (moved)); - ASSERT_FALSE (chained.opened ()); - - Tcp::AsyncAcceptor reassigned; - reassigned = std::move (chained); - ASSERT_FALSE (reassigned.opened ()); - - client.close (); - assigned.close (); -} - /** * @brief Test create method. */ @@ -231,18 +184,14 @@ TEST_F (TcpAsyncAcceptor, asyncAccept) Tcp::AsyncAcceptor server; Tcp::Socket client (Tcp::Socket::Blocking); - ASSERT_EQ (server.asyncAccept (nullptr), -1); + ASSERT_EQ (server.asyncAccept (peer (), nullptr), -1); ASSERT_EQ (join::lastError, Errc::OperationFailed); ASSERT_EQ (server.create ({_address, _port}), 0) << join::lastError.message (); - ASSERT_EQ (server.asyncAccept ([] (const std::error_code& ec, Tcp::AsyncSocket&& peer) { - onReport (ec, std::move (peer)); - }), - 0) - << join::lastError.message (); + ASSERT_EQ (server.asyncAccept (peer (), onReport), 0) << join::lastError.message (); - ASSERT_EQ (server.asyncAccept (nullptr), -1); + ASSERT_EQ (server.asyncAccept (peer (), nullptr), -1); ASSERT_EQ (join::lastError, Errc::InUse); ASSERT_EQ (client.connect ({_address, _port}), 0) << join::lastError.message (); @@ -253,10 +202,14 @@ TEST_F (TcpAsyncAcceptor, asyncAccept) return _completions >= 1; })); ASSERT_FALSE (_code) << _code.message (); - ASSERT_TRUE (_peerConnected); - ASSERT_EQ (_peerFamily, AF_INET6); } + ASSERT_TRUE (peer ().connected ()); + ASSERT_EQ (peer ().family (), AF_INET6); + + ASSERT_EQ (server.asyncAccept (peer (), nullptr), -1); + ASSERT_EQ (join::lastError, Errc::InUse); + client.close (); server.close (); } @@ -273,7 +226,7 @@ TEST_F (TcpAsyncAcceptor, resubmit) _current = &server; ASSERT_EQ (server.create ({_address, _port}), 0) << join::lastError.message (); - ASSERT_EQ (server.asyncAccept (onAccept), 0) << join::lastError.message (); + ASSERT_EQ (server.asyncAccept (peer (), onAccept), 0) << join::lastError.message (); ASSERT_EQ (client1.connect ({_address, _port}), 0) << join::lastError.message (); { @@ -309,8 +262,8 @@ TEST_F (TcpAsyncAcceptor, discard) Tcp::Socket client (Tcp::Socket::Blocking); ASSERT_EQ (server.create ({_address, _port}), 0) << join::lastError.message (); - ASSERT_EQ (server.asyncAccept (nullptr), 0) << join::lastError.message (); - ASSERT_EQ (server.asyncAccept (nullptr), -1); + ASSERT_EQ (server.asyncAccept (peer (), nullptr), 0) << join::lastError.message (); + ASSERT_EQ (server.asyncAccept (peer (), nullptr), -1); ASSERT_EQ (join::lastError, Errc::InUse); ASSERT_EQ (client.connect ({_address, _port}), 0) << join::lastError.message (); @@ -319,7 +272,7 @@ TEST_F (TcpAsyncAcceptor, discard) for (int i = 0; (i < 100) && (rearmed == -1); ++i) { std::this_thread::sleep_for (std::chrono::milliseconds (10)); - rearmed = server.asyncAccept (nullptr); + rearmed = server.asyncAccept (spare (), nullptr); } ASSERT_EQ (rearmed, 0) << join::lastError.message (); @@ -339,7 +292,7 @@ TEST_F (TcpAsyncAcceptor, closeFromHandler) _current = &server; ASSERT_EQ (server.create ({_address, _port}), 0) << join::lastError.message (); - ASSERT_EQ (server.asyncAccept (onAcceptAndClose), 0) << join::lastError.message (); + ASSERT_EQ (server.asyncAccept (peer (), onAcceptAndClose), 0) << join::lastError.message (); ASSERT_EQ (client.connect ({_address, _port}), 0) << join::lastError.message (); { @@ -366,11 +319,7 @@ TEST_F (TcpAsyncAcceptor, cancelAccept) ASSERT_EQ (server.cancelAccept (), 0) << join::lastError.message (); ASSERT_EQ (server.create ({_address, _port}), 0) << join::lastError.message (); - ASSERT_EQ (server.asyncAccept ([] (const std::error_code& ec, Tcp::AsyncSocket&& peer) { - onReport (ec, std::move (peer)); - }), - 0) - << join::lastError.message (); + ASSERT_EQ (server.asyncAccept (peer (), onReport), 0) << join::lastError.message (); ASSERT_EQ (server.cancelAccept (), 0) << join::lastError.message (); @@ -380,7 +329,7 @@ TEST_F (TcpAsyncAcceptor, cancelAccept) return _completions >= 1; })); ASSERT_EQ (_code, std::errc::operation_canceled); - ASSERT_FALSE (_peerConnected); + ASSERT_FALSE (peer ().connected ()); } server.close (); diff --git a/core/tests/tcp_async_stream_socket_test.cpp b/core/tests/tcp_async_stream_socket_test.cpp index c2817d6a..d80e249a 100644 --- a/core/tests/tcp_async_stream_socket_test.cpp +++ b/core/tests/tcp_async_stream_socket_test.cpp @@ -51,7 +51,7 @@ class TcpAsyncStreamSocket : public ::testing::Test void SetUp () override { ASSERT_EQ (_server.create ({IpAddress::ipv6Wildcard, _port}), 0) << join::lastError.message (); - ASSERT_EQ (_server.asyncAccept (onEchoAccept), 0) << join::lastError.message (); + ASSERT_EQ (_server.asyncAccept (peer (), onEchoAccept), 0) << join::lastError.message (); ScopedLock lock (_mut); @@ -109,13 +109,11 @@ class TcpAsyncStreamSocket : public ::testing::Test /** * @brief adopt the socket accepted by the echo server. * @param ec error reported by the acceptor. - * @param sock accepted socket. */ - static void onEchoAccept (const std::error_code& ec, Tcp::AsyncSocket&& sock) + static void onEchoAccept (const std::error_code& ec) { if (!ec) { - peer () = std::move (sock); peer ().asyncRead (_echobuf, sizeof (_echobuf), onEchoRead); } } @@ -236,83 +234,6 @@ const uint16_t TcpAsyncStreamSocket::_port = 5034; const uint16_t TcpAsyncStreamSocket::_stallport = 5035; const int TcpAsyncStreamSocket::_timeout = 1000; -/** - * @brief Test move with an operation in flight. - */ -TEST_F (TcpAsyncStreamSocket, move) -{ - Tcp::AsyncSocket client; - - ASSERT_EQ (client.asyncConnect ({_host, _port}, - [] (const std::error_code& ec) { - ScopedLock lock (_mut); - _code = ec; - ++_completions; - _cond.signal (); - }), - 0) - << join::lastError.message (); - - { - ScopedLock lock (_mut); - ASSERT_TRUE (_cond.timedWait (lock, std::chrono::milliseconds (_timeout), [] () { - return _completions >= 1; - })); - ASSERT_FALSE (_code) << _code.message (); - } - - ASSERT_EQ (client.asyncRead (_buf, sizeof (_buf), - [] (const std::error_code& ec, size_t size) { - ScopedLock lock (_mut); - _code = ec; - _transferred = size; - ++_completions; - _cond.signal (); - }), - 0) - << join::lastError.message (); - - ASSERT_EQ (client.asyncRead (_buf, sizeof (_buf), nullptr), -1); - ASSERT_EQ (join::lastError, Errc::InUse); - - Tcp::AsyncSocket moved (std::move (client)); - - ASSERT_EQ (moved.asyncRead (_buf, sizeof (_buf), nullptr), -1); - ASSERT_EQ (join::lastError, Errc::InUse); - - ASSERT_EQ (moved.asyncWrite ("one", 3, nullptr), 0) << join::lastError.message (); - - { - ScopedLock lock (_mut); - ASSERT_TRUE (_cond.timedWait (lock, std::chrono::milliseconds (_timeout), [] () { - return _completions >= 2; - })); - ASSERT_FALSE (_code) << _code.message (); - ASSERT_EQ (_transferred, 3u); - } - - Tcp::AsyncSocket assigned; - assigned = std::move (moved); - - ASSERT_TRUE (assigned.connected ()); - - ASSERT_EQ (moved.cancelRead (), 0) << join::lastError.message (); - ASSERT_EQ (moved.cancelWrite (), 0) << join::lastError.message (); - ASSERT_EQ (moved.asyncConnect ({_host, _port}, nullptr), -1); - ASSERT_EQ (join::lastError, Errc::OperationFailed); - ASSERT_FALSE (moved.opened ()); - moved.close (); - - Tcp::AsyncSocket chained (std::move (moved)); - ASSERT_FALSE (chained.opened ()); - - Tcp::AsyncSocket reassigned; - reassigned = std::move (chained); - ASSERT_FALSE (reassigned.opened ()); - - assigned.close (); -} - /** * @brief Test open method. */ @@ -766,7 +687,7 @@ TEST_F (TcpAsyncStreamSocket, cancelWrite) ASSERT_LT (filled, 4096); - ASSERT_EQ (sender.asyncWrite (_buf, sizeof (_buf), nullptr), 0) << join::lastError.message (); + ASSERT_EQ (sender.asyncWrite (_buf, sizeof (_buf), onWrite), 0) << join::lastError.message (); ASSERT_EQ (sender.asyncWrite (_buf, sizeof (_buf), nullptr), -1); ASSERT_EQ (join::lastError, Errc::InUse); @@ -776,6 +697,16 @@ TEST_F (TcpAsyncStreamSocket, cancelWrite) ASSERT_EQ (sender.cancelWrite (), 0) << join::lastError.message (); + { + ScopedLock lock (_mut); + ASSERT_TRUE (_cond.timedWait (lock, std::chrono::milliseconds (_timeout), [] () { + return _completions >= 2; + })); + ASSERT_EQ (_code, std::errc::operation_canceled); + } + + ASSERT_EQ (sender.asyncWrite (_buf, sizeof (_buf), nullptr), 0) << join::lastError.message (); + sender.close (); peer.close (); stall.close (); diff --git a/core/tests/udp_async_datagram_socket_test.cpp b/core/tests/udp_async_datagram_socket_test.cpp new file mode 100644 index 00000000..15129c49 --- /dev/null +++ b/core/tests/udp_async_datagram_socket_test.cpp @@ -0,0 +1,747 @@ +/** + * MIT License + * + * Copyright (c) 2026 Mathieu Rabine + * + * Permission is hereby granted, free of charge, to any person obtaining a copy + * of this software and associated documentation files (the "Software"), to deal + * in the Software without restriction, including without limitation the rights + * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell + * copies of the Software, and to permit persons to whom the Software is + * furnished to do so, subject to the following conditions: + * + * The above copyright notice and this permission notice shall be included in all + * copies or substantial portions of the Software. + * + * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR + * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, + * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE + * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER + * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, + * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE + * SOFTWARE. + */ + +// libjoin. +#include +#include + +// Libraries. +#include + +using join::Errc; +using join::Mutex; +using join::Condition; +using join::ScopedLock; +using join::IpAddress; +using join::Udp; + +/** + * @brief Class used to test the udp asynchronous datagram socket API. + */ +class UdpAsyncDatagramSocket : public ::testing::Test +{ +protected: + /** + * @brief Sets up the test fixture. + */ + void SetUp () override + { + ASSERT_EQ (server ().bind ({IpAddress::ipv6Wildcard, _port}), 0) << join::lastError.message (); + ASSERT_EQ (server ().asyncReadFrom (_echobuf, sizeof (_echobuf), _echofrom, onEchoRead), 0) + << join::lastError.message (); + + ScopedLock lock (_mut); + + _code = {}; + _completions = 0; + _transferred = 0; + _rearms = 0; + } + + /** + * @brief Tears down the test fixture. + */ + void TearDown () override + { + server ().close (); + } + + /** + * @brief get the echo server socket. + * @return the echo server socket. + */ + static Udp::AsyncSocket& server () + { + static Udp::AsyncSocket sock; + return sock; + } + + /** + * @brief send back the datagram received by the echo server. + * @param ec error reported by the socket. + * @param size number of bytes read. + */ + static void onEchoRead (const std::error_code& ec, size_t size) + { + if (!ec) + { + server ().asyncWriteTo (_echobuf, size, _echofrom, onEchoWrite); + } + } + + /** + * @brief wait for the next datagram to echo. + * @param ec error reported by the socket. + * @param size number of bytes written. + */ + static void onEchoWrite (const std::error_code& ec, [[maybe_unused]] size_t size) + { + if (!ec) + { + server ().asyncReadFrom (_echobuf, sizeof (_echobuf), _echofrom, onEchoRead); + } + } + + /** + * @brief report a completion to the test thread. + * @param ec error reported by the socket. + * @param size number of bytes transferred. + */ + static void onReport (const std::error_code& ec, size_t size) + { + ScopedLock lock (_mut); + + _code = ec; + _transferred = size; + ++_completions; + _cond.signal (); + } + + /** + * @brief handler resubmitting a read from within itself. + * @param ec error reported by the socket. + * @param size number of bytes read. + */ + static void onRead (const std::error_code& ec, size_t size) + { + if (!ec && (_rearms > 0)) + { + --_rearms; + _current->asyncReadFrom (_buf, sizeof (_buf), _from, onRead); + } + + onReport (ec, size); + } + + /** + * @brief handler resubmitting a write from within itself. + * @param ec error reported by the socket. + * @param size number of bytes written. + */ + static void onWrite (const std::error_code& ec, size_t size) + { + if (!ec && (_rearms > 0)) + { + --_rearms; + _current->asyncWriteTo ("two", 3, _dest, onWrite); + } + + onReport (ec, size); + } + + /** + * @brief handler closing the socket from within itself. + * @param ec error reported by the socket. + * @param size number of bytes written. + */ + static void onWriteAndClose (const std::error_code& ec, size_t size) + { + _current->close (); + + onReport (ec, size); + } + + /// condition mutex. + static Mutex _mut; + + /// condition variable. + static Condition _cond; + + /// last reported error. + static std::error_code _code; + + /// number of completions reported. + static int _completions; + + /// number of bytes reported by the last completion. + static size_t _transferred; + + /// read buffer. + static char _buf[1024]; + + /// endpoint the last datagram was received from. + static Udp::Endpoint _from; + + /// buffer used by the echo server. + static char _echobuf[1024]; + + /// endpoint the echo server received the last datagram from. + static Udp::Endpoint _echofrom; + + /// socket used by the resubmitting handler. + static Udp::AsyncSocket* _current; + + /// number of resubmissions left to perform from a handler. + static int _rearms; + + /// destination used by the resubmitting write handler. + static Udp::Endpoint _dest; + + /// host. + static const std::string _host; + + /// port. + static const uint16_t _port; + + /// timeout. + static const int _timeout; +}; + +Mutex UdpAsyncDatagramSocket::_mut; +Condition UdpAsyncDatagramSocket::_cond; +std::error_code UdpAsyncDatagramSocket::_code; +int UdpAsyncDatagramSocket::_completions = 0; +size_t UdpAsyncDatagramSocket::_transferred = 0; +char UdpAsyncDatagramSocket::_buf[1024] = {}; +Udp::Endpoint UdpAsyncDatagramSocket::_from; +char UdpAsyncDatagramSocket::_echobuf[1024] = {}; +Udp::Endpoint UdpAsyncDatagramSocket::_echofrom; +Udp::AsyncSocket* UdpAsyncDatagramSocket::_current = nullptr; +int UdpAsyncDatagramSocket::_rearms = 0; +Udp::Endpoint UdpAsyncDatagramSocket::_dest; +const std::string UdpAsyncDatagramSocket::_host = "127.0.0.1"; +const uint16_t UdpAsyncDatagramSocket::_port = 5036; +const int UdpAsyncDatagramSocket::_timeout = 1000; + +/** + * @brief Test open method. + */ +TEST_F (UdpAsyncDatagramSocket, open) +{ + Udp::AsyncSocket client; + + ASSERT_EQ (client.open (), 0) << join::lastError.message (); + ASSERT_EQ (client.open (), -1); + ASSERT_EQ (join::lastError, Errc::InUse); + client.close (); +} + +/** + * @brief Test close method. + */ +TEST_F (UdpAsyncDatagramSocket, close) +{ + Udp::AsyncSocket client; + + ASSERT_EQ (client.open (), 0) << join::lastError.message (); + ASSERT_TRUE (client.opened ()); + client.close (); + ASSERT_FALSE (client.opened ()); +} + +/** + * @brief Test bind method. + */ +TEST_F (UdpAsyncDatagramSocket, bind) +{ + Udp::AsyncSocket client; + + ASSERT_EQ (client.open (Udp::v6 ()), 0) << join::lastError.message (); + ASSERT_EQ (client.bind (IpAddress (AF_INET6)), 0) << join::lastError.message (); + client.close (); +} + +/** + * @brief Test bindToDevice method. + */ +TEST_F (UdpAsyncDatagramSocket, bindToDevice) +{ + Udp::AsyncSocket client; + + ASSERT_EQ (client.bindToDevice ("lo"), -1); + ASSERT_EQ (client.open (Udp::v6 ()), 0) << join::lastError.message (); + ASSERT_EQ (client.bindToDevice ("lo"), 0) << join::lastError.message (); + ASSERT_EQ (client.bindToDevice ("foo"), -1); + client.close (); +} + +/** + * @brief Test connect method. + */ +TEST_F (UdpAsyncDatagramSocket, connect) +{ + Udp::AsyncSocket client; + + ASSERT_EQ (client.connect ({"255.255.255.255", _port}), -1); + + ASSERT_EQ (client.connect ({_host, _port}), 0) << join::lastError.message (); + ASSERT_TRUE (client.connected ()); + ASSERT_EQ (client.connect ({_host, _port}), -1); + ASSERT_EQ (join::lastError, Errc::InUse); + client.close (); +} + +/** + * @brief Test disconnect method. + */ +TEST_F (UdpAsyncDatagramSocket, disconnect) +{ + Udp::AsyncSocket client; + + ASSERT_EQ (client.disconnect (), 0) << join::lastError.message (); + ASSERT_EQ (client.connect ({_host, _port}), 0) << join::lastError.message (); + ASSERT_TRUE (client.connected ()); + ASSERT_EQ (client.disconnect (), 0) << join::lastError.message (); + ASSERT_FALSE (client.connected ()); + client.close (); +} + +/** + * @brief Test asyncWriteTo method. + */ +TEST_F (UdpAsyncDatagramSocket, asyncWriteTo) +{ + Udp::AsyncSocket client; + Udp::Endpoint dest (_host, _port); + + ASSERT_FALSE (client.opened ()); + ASSERT_EQ (client.asyncWriteTo ("hello", 5, dest, onReport), 0) << join::lastError.message (); + ASSERT_TRUE (client.opened ()); + + ASSERT_EQ (client.asyncWriteTo ("hello", 5, dest, nullptr), -1); + ASSERT_EQ (join::lastError, Errc::InUse); + + { + ScopedLock lock (_mut); + ASSERT_TRUE (_cond.timedWait (lock, std::chrono::milliseconds (_timeout), [] () { + return _completions >= 1; + })); + ASSERT_FALSE (_code) << _code.message (); + ASSERT_EQ (_transferred, 5u); + } + + client.close (); +} + +/** + * @brief Test asyncReadFrom method. + */ +TEST_F (UdpAsyncDatagramSocket, asyncReadFrom) +{ + Udp::AsyncSocket client; + + ASSERT_EQ (client.asyncReadFrom (_buf, sizeof (_buf), _from, nullptr), -1); + ASSERT_EQ (join::lastError, Errc::OperationFailed); + + ASSERT_EQ (client.connect ({_host, _port}), 0) << join::lastError.message (); + ASSERT_EQ (client.asyncReadFrom (_buf, sizeof (_buf), _from, onReport), 0) << join::lastError.message (); + + ASSERT_EQ (client.asyncReadFrom (_buf, sizeof (_buf), _from, nullptr), -1); + ASSERT_EQ (join::lastError, Errc::InUse); + + ASSERT_EQ (client.asyncWrite ("hello", 5, nullptr), 0) << join::lastError.message (); + + { + ScopedLock lock (_mut); + ASSERT_TRUE (_cond.timedWait (lock, std::chrono::milliseconds (_timeout), [] () { + return _completions >= 1; + })); + ASSERT_FALSE (_code) << _code.message (); + ASSERT_EQ (_transferred, 5u); + ASSERT_EQ (std::string (_buf, 5), "hello"); + } + + client.close (); + + ASSERT_EQ (_from, Udp::Endpoint (_host, _port)); +} + +/** + * @brief Test asyncWrite method. + */ +TEST_F (UdpAsyncDatagramSocket, asyncWrite) +{ + Udp::AsyncSocket client; + + ASSERT_EQ (client.asyncWrite ("hello", 5, nullptr), -1); + ASSERT_EQ (join::lastError, Errc::OperationFailed); + + ASSERT_EQ (client.connect ({_host, _port}), 0) << join::lastError.message (); + ASSERT_EQ (client.asyncWrite ("hello", 5, onReport), 0) << join::lastError.message (); + + { + ScopedLock lock (_mut); + ASSERT_TRUE (_cond.timedWait (lock, std::chrono::milliseconds (_timeout), [] () { + return _completions >= 1; + })); + ASSERT_FALSE (_code) << _code.message (); + ASSERT_EQ (_transferred, 5u); + } + + client.close (); +} + +/** + * @brief Test asyncRead method. + */ +TEST_F (UdpAsyncDatagramSocket, asyncRead) +{ + Udp::AsyncSocket client; + + ASSERT_EQ (client.asyncRead (_buf, sizeof (_buf), nullptr), -1); + ASSERT_EQ (join::lastError, Errc::OperationFailed); + + ASSERT_EQ (client.connect ({_host, _port}), 0) << join::lastError.message (); + ASSERT_EQ (client.asyncRead (_buf, sizeof (_buf), onReport), 0) << join::lastError.message (); + ASSERT_EQ (client.asyncWrite ("hello", 5, nullptr), 0) << join::lastError.message (); + + { + ScopedLock lock (_mut); + ASSERT_TRUE (_cond.timedWait (lock, std::chrono::milliseconds (_timeout), [] () { + return _completions >= 1; + })); + ASSERT_FALSE (_code) << _code.message (); + ASSERT_EQ (_transferred, 5u); + ASSERT_EQ (std::string (_buf, 5), "hello"); + } + + client.close (); +} + +/** + * @brief Test asyncReadFrom method resubmitted from its own handler. + */ +TEST_F (UdpAsyncDatagramSocket, resubmit) +{ + Udp::AsyncSocket client; + Udp::Endpoint dest (_host, _port); + + _current = &client; + _rearms = 1; + + ASSERT_EQ (client.bind (Udp::Endpoint (_host, 0)), 0) << join::lastError.message (); + ASSERT_EQ (client.asyncReadFrom (_buf, sizeof (_buf), _from, onRead), 0) << join::lastError.message (); + ASSERT_EQ (client.asyncWriteTo ("one", 3, dest, nullptr), 0) << join::lastError.message (); + + { + ScopedLock lock (_mut); + ASSERT_TRUE (_cond.timedWait (lock, std::chrono::milliseconds (_timeout), [] () { + return _completions >= 1; + })); + ASSERT_FALSE (_code) << _code.message (); + } + + ASSERT_EQ (client.asyncWriteTo ("two", 3, dest, nullptr), 0) << join::lastError.message (); + + { + ScopedLock lock (_mut); + ASSERT_TRUE (_cond.timedWait (lock, std::chrono::milliseconds (_timeout), [] () { + return _completions >= 2; + })); + ASSERT_FALSE (_code) << _code.message (); + } + + _dest = Udp::Endpoint (_host, _port); + _rearms = 1; + + ASSERT_EQ (client.asyncWriteTo ("one", 3, _dest, onWrite), 0) << join::lastError.message (); + + { + ScopedLock lock (_mut); + ASSERT_TRUE (_cond.timedWait (lock, std::chrono::milliseconds (_timeout), [] () { + return _completions >= 4; + })); + ASSERT_FALSE (_code) << _code.message (); + } + + client.close (); + _current = nullptr; +} + +/** + * @brief Test close called from within a write handler. + */ +TEST_F (UdpAsyncDatagramSocket, closeFromWriteHandler) +{ + Udp::AsyncSocket client; + Udp::Endpoint dest (_host, _port); + + _current = &client; + + ASSERT_EQ (client.asyncWriteTo ("hello", 5, dest, onWriteAndClose), 0) << join::lastError.message (); + + { + ScopedLock lock (_mut); + ASSERT_TRUE (_cond.timedWait (lock, std::chrono::milliseconds (_timeout), [] () { + return _completions >= 1; + })); + ASSERT_FALSE (_code) << _code.message (); + } + + ASSERT_FALSE (client.opened ()); + _current = nullptr; +} + +/** + * @brief Test a datagram larger than the supplied buffer. + */ +TEST_F (UdpAsyncDatagramSocket, truncated) +{ + Udp::AsyncSocket client; + char small[4] = {}; + + ASSERT_EQ (client.connect ({_host, _port}), 0) << join::lastError.message (); + ASSERT_EQ (client.asyncReadFrom (small, sizeof (small), _from, onReport), 0) << join::lastError.message (); + ASSERT_EQ (client.asyncWrite ("hello world", 11, nullptr), 0) << join::lastError.message (); + + { + ScopedLock lock (_mut); + ASSERT_TRUE (_cond.timedWait (lock, std::chrono::milliseconds (_timeout), [] () { + return _completions >= 1; + })); + ASSERT_EQ (_code, Errc::MessageTooLong); + } + + client.close (); +} + +/** + * @brief Test an empty datagram. + */ +TEST_F (UdpAsyncDatagramSocket, empty) +{ + Udp::AsyncSocket client; + Udp::Socket sender; + Udp::Endpoint self (_host, uint16_t (_port + 2)); + + ASSERT_EQ (client.bind (self), 0) << join::lastError.message (); + ASSERT_EQ (client.asyncReadFrom (_buf, sizeof (_buf), _from, onReport), 0) << join::lastError.message (); + + ASSERT_EQ (sender.writeTo ("", 0, self), 0) << join::lastError.message (); + + { + ScopedLock lock (_mut); + ASSERT_TRUE (_cond.timedWait (lock, std::chrono::milliseconds (_timeout), [] () { + return _completions >= 1; + })); + ASSERT_EQ (_code, Errc::ConnectionClosed); + } + + ASSERT_FALSE (client.connected ()); + + sender.close (); + client.close (); +} + +/** + * @brief Test cancelRead method. + */ +TEST_F (UdpAsyncDatagramSocket, cancelRead) +{ + Udp::AsyncSocket client; + + ASSERT_EQ (client.cancelRead (), 0) << join::lastError.message (); + ASSERT_EQ (client.bind (Udp::Endpoint (_host, 0)), 0) << join::lastError.message (); + ASSERT_EQ (client.asyncReadFrom (_buf, sizeof (_buf), _from, onReport), 0) << join::lastError.message (); + ASSERT_EQ (client.cancelRead (), 0) << join::lastError.message (); + + { + ScopedLock lock (_mut); + ASSERT_TRUE (_cond.timedWait (lock, std::chrono::milliseconds (_timeout), [] () { + return _completions >= 1; + })); + ASSERT_EQ (_code, std::errc::operation_canceled); + } + + client.close (); +} + +/** + * @brief Test cancelWrite method. + */ +TEST_F (UdpAsyncDatagramSocket, cancelWrite) +{ + Udp::AsyncSocket client; + + ASSERT_EQ (client.cancelWrite (), 0) << join::lastError.message (); + ASSERT_EQ (client.open (), 0) << join::lastError.message (); + ASSERT_EQ (client.cancelWrite (), 0) << join::lastError.message (); + client.close (); +} + +/** + * @brief Test setOption method. + */ +TEST_F (UdpAsyncDatagramSocket, setOption) +{ + Udp::AsyncSocket client; + + ASSERT_EQ (client.setOption (Udp::Socket::RcvBuffer, 4096), -1); + ASSERT_EQ (client.open (Udp::v6 ()), 0) << join::lastError.message (); + ASSERT_EQ (client.setOption (Udp::Socket::RcvBuffer, 4096), 0) << join::lastError.message (); + client.close (); +} + +/** + * @brief Test localEndpoint method. + */ +TEST_F (UdpAsyncDatagramSocket, localEndpoint) +{ + Udp::AsyncSocket client; + + ASSERT_EQ (client.open (Udp::v6 ()), 0) << join::lastError.message (); + ASSERT_EQ (client.bind ({IpAddress::ipv6Wildcard, uint16_t (_port + 1)}), 0) << join::lastError.message (); + ASSERT_EQ (client.localEndpoint ().port (), uint16_t (_port + 1)); + client.close (); +} + +/** + * @brief Test remoteEndpoint method. + */ +TEST_F (UdpAsyncDatagramSocket, remoteEndpoint) +{ + Udp::AsyncSocket client; + + ASSERT_EQ (client.connect ({_host, _port}), 0) << join::lastError.message (); + ASSERT_EQ (client.remoteEndpoint ().ip (), _host); + ASSERT_EQ (client.remoteEndpoint ().port (), _port); + client.close (); +} + +/** + * @brief Test opened method. + */ +TEST_F (UdpAsyncDatagramSocket, opened) +{ + Udp::AsyncSocket client; + + ASSERT_FALSE (client.opened ()); + ASSERT_EQ (client.open (), 0) << join::lastError.message (); + ASSERT_TRUE (client.opened ()); + client.close (); + ASSERT_FALSE (client.opened ()); +} + +/** + * @brief Test connected method. + */ +TEST_F (UdpAsyncDatagramSocket, connected) +{ + Udp::AsyncSocket client; + + ASSERT_FALSE (client.connected ()); + ASSERT_EQ (client.connect ({_host, _port}), 0) << join::lastError.message (); + ASSERT_TRUE (client.connected ()); + client.close (); + ASSERT_FALSE (client.connected ()); +} + +/** + * @brief Test canRead method. + */ +TEST_F (UdpAsyncDatagramSocket, canRead) +{ + Udp::AsyncSocket client; + + ASSERT_EQ (client.canRead (), -1); + ASSERT_EQ (client.open (Udp::v6 ()), 0) << join::lastError.message (); + ASSERT_EQ (client.canRead (), 0) << join::lastError.message (); + client.close (); +} + +/** + * @brief Test mtu method. + */ +TEST_F (UdpAsyncDatagramSocket, mtu) +{ + Udp::AsyncSocket client; + + ASSERT_EQ (client.mtu (), -1); + ASSERT_EQ (join::lastError, Errc::OperationFailed); + ASSERT_EQ (client.connect ({_host, _port}), 0) << join::lastError.message (); + ASSERT_GT (client.mtu (), 0) << join::lastError.message (); + client.close (); +} + +/** + * @brief Test ttl method. + */ +TEST_F (UdpAsyncDatagramSocket, ttl) +{ + Udp::AsyncSocket client; + + ASSERT_EQ (client.ttl (), 60); + + Udp::AsyncSocket other (32); + + ASSERT_EQ (other.ttl (), 32); +} + +/** + * @brief Test family method. + */ +TEST_F (UdpAsyncDatagramSocket, family) +{ + Udp::AsyncSocket client; + + ASSERT_EQ (client.open (Udp::v6 ()), 0) << join::lastError.message (); + ASSERT_EQ (client.family (), AF_INET6); + client.close (); +} + +/** + * @brief Test type method. + */ +TEST_F (UdpAsyncDatagramSocket, type) +{ + Udp::AsyncSocket client; + + ASSERT_EQ (client.open (), 0) << join::lastError.message (); + ASSERT_EQ (client.type (), SOCK_DGRAM); + client.close (); +} + +/** + * @brief Test protocol method. + */ +TEST_F (UdpAsyncDatagramSocket, protocol) +{ + Udp::AsyncSocket client; + + ASSERT_EQ (client.open (), 0) << join::lastError.message (); + ASSERT_EQ (client.protocol (), IPPROTO_UDP); + client.close (); +} + +/** + * @brief Test handle method. + */ +TEST_F (UdpAsyncDatagramSocket, handle) +{ + Udp::AsyncSocket client; + + ASSERT_EQ (client.handle (), -1); + ASSERT_EQ (client.open (), 0) << join::lastError.message (); + ASSERT_GT (client.handle (), -1); + client.close (); +} + +/** + * @brief main function. + */ +int main (int argc, char** argv) +{ + testing::InitGoogleTest (&argc, argv); + return RUN_ALL_TESTS (); +} diff --git a/core/tests/udp_socket_test.cpp b/core/tests/udp_socket_test.cpp index 71277db9..f6d56505 100644 --- a/core/tests/udp_socket_test.cpp +++ b/core/tests/udp_socket_test.cpp @@ -271,6 +271,15 @@ TEST_F (UdpSocket, readFrom) ASSERT_EQ (udpSocket.write (data, sizeof (data)), sizeof (data)) << join::lastError.message (); ASSERT_TRUE (udpSocket.waitReadyRead (_timeout)) << join::lastError.message (); ASSERT_EQ (udpSocket.readFrom (data, udpSocket.canRead (), &from), sizeof (data)) << join::lastError.message (); + + char small[4]; + + ASSERT_TRUE (udpSocket.waitReadyWrite (_timeout)) << join::lastError.message (); + ASSERT_EQ (udpSocket.write (data, sizeof (data)), sizeof (data)) << join::lastError.message (); + ASSERT_TRUE (udpSocket.waitReadyRead (_timeout)) << join::lastError.message (); + ASSERT_EQ (udpSocket.readFrom (small, sizeof (small), &from), -1); + ASSERT_EQ (join::lastError, Errc::MessageTooLong); + udpSocket.close (); ASSERT_EQ (from, Udp::Endpoint (_host, _port)); } diff --git a/core/tests/unix_async_acceptor_test.cpp b/core/tests/unix_async_acceptor_test.cpp index 8c2480fe..bf219ba2 100644 --- a/core/tests/unix_async_acceptor_test.cpp +++ b/core/tests/unix_async_acceptor_test.cpp @@ -53,8 +53,9 @@ class UnixAsyncAcceptor : public ::testing::Test _code = {}; _completions = 0; - _peerConnected = false; - _peerFamily = -1; + + peer ().close (); + spare ().close (); } /** @@ -65,18 +66,35 @@ class UnixAsyncAcceptor : public ::testing::Test ::unlink (_path.c_str ()); } + /** + * @brief get the socket receiving the accepted connections. + * @return the socket receiving the accepted connections. + */ + static UnixStream::AsyncSocket& peer () + { + static UnixStream::AsyncSocket sock; + return sock; + } + + /** + * @brief get the socket receiving the connection accepted by a resubmitted acceptation. + * @return the socket receiving the connection accepted by a resubmitted acceptation. + */ + static UnixStream::AsyncSocket& spare () + { + static UnixStream::AsyncSocket sock; + return sock; + } + /** * @brief report a completion to the test thread. * @param ec error reported by the acceptor. - * @param peer accepted socket. */ - static void onReport (const std::error_code& ec, UnixStream::AsyncSocket&& peer) + static void onReport (const std::error_code& ec) { ScopedLock lock (_mut); _code = ec; - _peerConnected = peer.connected (); - _peerFamily = peer.family (); ++_completions; _cond.signal (); } @@ -84,28 +102,26 @@ class UnixAsyncAcceptor : public ::testing::Test /** * @brief handler resubmitting an acceptation from within itself. * @param ec error reported by the acceptor. - * @param peer accepted socket. */ - static void onAccept (const std::error_code& ec, UnixStream::AsyncSocket&& peer) + static void onAccept (const std::error_code& ec) { if (!ec) { - _current->asyncAccept (onAccept); + _current->asyncAccept (spare (), onAccept); } - onReport (ec, std::move (peer)); + onReport (ec); } /** * @brief handler closing the acceptor from within itself. * @param ec error reported by the acceptor. - * @param peer accepted socket. */ - static void onAcceptAndClose (const std::error_code& ec, UnixStream::AsyncSocket&& peer) + static void onAcceptAndClose (const std::error_code& ec) { _current->close (); - onReport (ec, std::move (peer)); + onReport (ec); } /// acceptor path. @@ -126,12 +142,6 @@ class UnixAsyncAcceptor : public ::testing::Test /// number of completions reported. static int _completions; - /// state of the last accepted socket. - static bool _peerConnected; - - /// address family of the last accepted socket. - static int _peerFamily; - /// acceptor used by the resubmitting handler. static UnixStream::AsyncAcceptor* _current; }; @@ -142,65 +152,8 @@ Mutex UnixAsyncAcceptor::_mut; Condition UnixAsyncAcceptor::_cond; std::error_code UnixAsyncAcceptor::_code; int UnixAsyncAcceptor::_completions = 0; -bool UnixAsyncAcceptor::_peerConnected = false; -int UnixAsyncAcceptor::_peerFamily = -1; UnixStream::AsyncAcceptor* UnixAsyncAcceptor::_current = nullptr; -/** - * @brief Test move with an acceptation in flight. - */ -TEST_F (UnixAsyncAcceptor, move) -{ - UnixStream::AsyncAcceptor server; - UnixStream::Socket client (UnixStream::Socket::Blocking); - - ASSERT_EQ (server.create (_path), 0) << join::lastError.message (); - - ASSERT_EQ (server.asyncAccept ([] (const std::error_code& ec, UnixStream::AsyncSocket&& peer) { - onReport (ec, std::move (peer)); - }), - 0) - << join::lastError.message (); - - ASSERT_EQ (server.asyncAccept (nullptr), -1); - ASSERT_EQ (join::lastError, Errc::InUse); - - UnixStream::AsyncAcceptor moved (std::move (server)); - - ASSERT_EQ (moved.asyncAccept (nullptr), -1); - ASSERT_EQ (join::lastError, Errc::InUse); - - ASSERT_EQ (client.connect (_path), 0) << join::lastError.message (); - - { - ScopedLock lock (_mut); - ASSERT_TRUE (_cond.timedWait (lock, std::chrono::milliseconds (_timeout), [] () { - return _completions >= 1; - })); - ASSERT_FALSE (_code) << _code.message (); - ASSERT_TRUE (_peerConnected); - } - - UnixStream::AsyncAcceptor assigned; - assigned = std::move (moved); - - ASSERT_TRUE (assigned.opened ()); - - ASSERT_EQ (moved.cancelAccept (), 0) << join::lastError.message (); - ASSERT_FALSE (moved.opened ()); - moved.close (); - - UnixStream::AsyncAcceptor chained (std::move (moved)); - ASSERT_FALSE (chained.opened ()); - - UnixStream::AsyncAcceptor reassigned; - reassigned = std::move (chained); - ASSERT_FALSE (reassigned.opened ()); - - client.close (); - assigned.close (); -} - /** * @brief Test create method. */ @@ -234,18 +187,14 @@ TEST_F (UnixAsyncAcceptor, asyncAccept) UnixStream::AsyncAcceptor server; UnixStream::Socket client (UnixStream::Socket::Blocking); - ASSERT_EQ (server.asyncAccept (nullptr), -1); + ASSERT_EQ (server.asyncAccept (peer (), nullptr), -1); ASSERT_EQ (join::lastError, Errc::OperationFailed); ASSERT_EQ (server.create (_path), 0) << join::lastError.message (); - ASSERT_EQ (server.asyncAccept ([] (const std::error_code& ec, UnixStream::AsyncSocket&& peer) { - onReport (ec, std::move (peer)); - }), - 0) - << join::lastError.message (); + ASSERT_EQ (server.asyncAccept (peer (), onReport), 0) << join::lastError.message (); - ASSERT_EQ (server.asyncAccept (nullptr), -1); + ASSERT_EQ (server.asyncAccept (peer (), nullptr), -1); ASSERT_EQ (join::lastError, Errc::InUse); ASSERT_EQ (client.connect (_path), 0) << join::lastError.message (); @@ -256,10 +205,14 @@ TEST_F (UnixAsyncAcceptor, asyncAccept) return _completions >= 1; })); ASSERT_FALSE (_code) << _code.message (); - ASSERT_TRUE (_peerConnected); - ASSERT_EQ (_peerFamily, AF_UNIX); } + ASSERT_TRUE (peer ().connected ()); + ASSERT_EQ (peer ().family (), AF_UNIX); + + ASSERT_EQ (server.asyncAccept (peer (), nullptr), -1); + ASSERT_EQ (join::lastError, Errc::InUse); + client.close (); server.close (); } @@ -276,7 +229,7 @@ TEST_F (UnixAsyncAcceptor, resubmit) _current = &server; ASSERT_EQ (server.create (_path), 0) << join::lastError.message (); - ASSERT_EQ (server.asyncAccept (onAccept), 0) << join::lastError.message (); + ASSERT_EQ (server.asyncAccept (peer (), onAccept), 0) << join::lastError.message (); ASSERT_EQ (client1.connect (_path), 0) << join::lastError.message (); { @@ -312,8 +265,8 @@ TEST_F (UnixAsyncAcceptor, discard) UnixStream::Socket client (UnixStream::Socket::Blocking); ASSERT_EQ (server.create (_path), 0) << join::lastError.message (); - ASSERT_EQ (server.asyncAccept (nullptr), 0) << join::lastError.message (); - ASSERT_EQ (server.asyncAccept (nullptr), -1); + ASSERT_EQ (server.asyncAccept (peer (), nullptr), 0) << join::lastError.message (); + ASSERT_EQ (server.asyncAccept (peer (), nullptr), -1); ASSERT_EQ (join::lastError, Errc::InUse); ASSERT_EQ (client.connect (_path), 0) << join::lastError.message (); @@ -322,7 +275,7 @@ TEST_F (UnixAsyncAcceptor, discard) for (int i = 0; (i < 100) && (rearmed == -1); ++i) { std::this_thread::sleep_for (std::chrono::milliseconds (10)); - rearmed = server.asyncAccept (nullptr); + rearmed = server.asyncAccept (spare (), nullptr); } ASSERT_EQ (rearmed, 0) << join::lastError.message (); @@ -342,7 +295,7 @@ TEST_F (UnixAsyncAcceptor, closeFromHandler) _current = &server; ASSERT_EQ (server.create (_path), 0) << join::lastError.message (); - ASSERT_EQ (server.asyncAccept (onAcceptAndClose), 0) << join::lastError.message (); + ASSERT_EQ (server.asyncAccept (peer (), onAcceptAndClose), 0) << join::lastError.message (); ASSERT_EQ (client.connect (_path), 0) << join::lastError.message (); { @@ -369,11 +322,7 @@ TEST_F (UnixAsyncAcceptor, cancelAccept) ASSERT_EQ (server.cancelAccept (), 0) << join::lastError.message (); ASSERT_EQ (server.create (_path), 0) << join::lastError.message (); - ASSERT_EQ (server.asyncAccept ([] (const std::error_code& ec, UnixStream::AsyncSocket&& peer) { - onReport (ec, std::move (peer)); - }), - 0) - << join::lastError.message (); + ASSERT_EQ (server.asyncAccept (peer (), onReport), 0) << join::lastError.message (); ASSERT_EQ (server.cancelAccept (), 0) << join::lastError.message (); @@ -383,7 +332,7 @@ TEST_F (UnixAsyncAcceptor, cancelAccept) return _completions >= 1; })); ASSERT_EQ (_code, std::errc::operation_canceled); - ASSERT_FALSE (_peerConnected); + ASSERT_FALSE (peer ().connected ()); } server.close (); diff --git a/core/tests/unix_async_datagram_socket_test.cpp b/core/tests/unix_async_datagram_socket_test.cpp new file mode 100644 index 00000000..917a53c8 --- /dev/null +++ b/core/tests/unix_async_datagram_socket_test.cpp @@ -0,0 +1,762 @@ +/** + * MIT License + * + * Copyright (c) 2026 Mathieu Rabine + * + * Permission is hereby granted, free of charge, to any person obtaining a copy + * of this software and associated documentation files (the "Software"), to deal + * in the Software without restriction, including without limitation the rights + * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell + * copies of the Software, and to permit persons to whom the Software is + * furnished to do so, subject to the following conditions: + * + * The above copyright notice and this permission notice shall be included in all + * copies or substantial portions of the Software. + * + * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR + * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, + * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE + * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER + * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, + * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE + * SOFTWARE. + */ + +// libjoin. +#include +#include + +// Libraries. +#include + +// C. +#include + +using join::Errc; +using join::Mutex; +using join::Condition; +using join::ScopedLock; +using join::UnixDgram; + +/** + * @brief Class used to test the unix asynchronous datagram socket API. + */ +class UnixAsyncDatagramSocket : public ::testing::Test +{ +protected: + /** + * @brief Sets up the test fixture. + */ + void SetUp () override + { + ::unlink (_serverpath.c_str ()); + ::unlink (_clientpath.c_str ()); + ::unlink (_senderpath.c_str ()); + + ASSERT_EQ (server ().bind (_serverpath), 0) << join::lastError.message (); + ASSERT_EQ (server ().asyncReadFrom (_echobuf, sizeof (_echobuf), _echofrom, onEchoRead), 0) + << join::lastError.message (); + + ScopedLock lock (_mut); + + _code = {}; + _completions = 0; + _transferred = 0; + _rearms = 0; + } + + /** + * @brief Tears down the test fixture. + */ + void TearDown () override + { + server ().close (); + + ::unlink (_serverpath.c_str ()); + ::unlink (_clientpath.c_str ()); + ::unlink (_senderpath.c_str ()); + } + + /** + * @brief get the echo server socket. + * @return the echo server socket. + */ + static UnixDgram::AsyncSocket& server () + { + static UnixDgram::AsyncSocket sock; + return sock; + } + + /** + * @brief send back the datagram received by the echo server. + * @param ec error reported by the socket. + * @param size number of bytes read. + */ + static void onEchoRead (const std::error_code& ec, size_t size) + { + if (!ec) + { + server ().asyncWriteTo (_echobuf, size, _echofrom, onEchoWrite); + } + } + + /** + * @brief wait for the next datagram to echo. + * @param ec error reported by the socket. + * @param size number of bytes written. + */ + static void onEchoWrite (const std::error_code& ec, [[maybe_unused]] size_t size) + { + if (!ec) + { + server ().asyncReadFrom (_echobuf, sizeof (_echobuf), _echofrom, onEchoRead); + } + } + + /** + * @brief report a completion to the test thread. + * @param ec error reported by the socket. + * @param size number of bytes transferred. + */ + static void onReport (const std::error_code& ec, size_t size) + { + ScopedLock lock (_mut); + + _code = ec; + _transferred = size; + ++_completions; + _cond.signal (); + } + + /** + * @brief handler resubmitting a read from within itself. + * @param ec error reported by the socket. + * @param size number of bytes read. + */ + static void onRead (const std::error_code& ec, size_t size) + { + if (!ec && (_rearms > 0)) + { + --_rearms; + _current->asyncReadFrom (_buf, sizeof (_buf), _from, onRead); + } + + onReport (ec, size); + } + + /** + * @brief handler resubmitting a write from within itself. + * @param ec error reported by the socket. + * @param size number of bytes written. + */ + static void onWrite (const std::error_code& ec, size_t size) + { + if (!ec && (_rearms > 0)) + { + --_rearms; + _current->asyncWriteTo ("two", 3, _dest, onWrite); + } + + onReport (ec, size); + } + + /** + * @brief handler closing the socket from within itself. + * @param ec error reported by the socket. + * @param size number of bytes written. + */ + static void onWriteAndClose (const std::error_code& ec, size_t size) + { + _current->close (); + + onReport (ec, size); + } + + /// condition mutex. + static Mutex _mut; + + /// condition variable. + static Condition _cond; + + /// last reported error. + static std::error_code _code; + + /// number of completions reported. + static int _completions; + + /// number of bytes reported by the last completion. + static size_t _transferred; + + /// read buffer. + static char _buf[1024]; + + /// endpoint the last datagram was received from. + static UnixDgram::Endpoint _from; + + /// buffer used by the echo server. + static char _echobuf[1024]; + + /// endpoint the echo server received the last datagram from. + static UnixDgram::Endpoint _echofrom; + + /// socket used by the resubmitting handler. + static UnixDgram::AsyncSocket* _current; + + /// number of resubmissions left to perform from a handler. + static int _rearms; + + /// destination used by the resubmitting write handler. + static UnixDgram::Endpoint _dest; + + /// echo server path. + static const std::string _serverpath; + + /// client path. + static const std::string _clientpath; + + /// path used by the synchronous sender. + static const std::string _senderpath; + + /// timeout. + static const int _timeout; +}; + +Mutex UnixAsyncDatagramSocket::_mut; +Condition UnixAsyncDatagramSocket::_cond; +std::error_code UnixAsyncDatagramSocket::_code; +int UnixAsyncDatagramSocket::_completions = 0; +size_t UnixAsyncDatagramSocket::_transferred = 0; +char UnixAsyncDatagramSocket::_buf[1024] = {}; +UnixDgram::Endpoint UnixAsyncDatagramSocket::_from; +char UnixAsyncDatagramSocket::_echobuf[1024] = {}; +UnixDgram::Endpoint UnixAsyncDatagramSocket::_echofrom; +UnixDgram::AsyncSocket* UnixAsyncDatagramSocket::_current = nullptr; +int UnixAsyncDatagramSocket::_rearms = 0; +UnixDgram::Endpoint UnixAsyncDatagramSocket::_dest; +const std::string UnixAsyncDatagramSocket::_serverpath = "/tmp/unixasyncdgramserver_test.sock"; +const std::string UnixAsyncDatagramSocket::_clientpath = "/tmp/unixasyncdgramclient_test.sock"; +const std::string UnixAsyncDatagramSocket::_senderpath = "/tmp/unixasyncdgramsender_test.sock"; +const int UnixAsyncDatagramSocket::_timeout = 1000; + +/** + * @brief Test open method. + */ +TEST_F (UnixAsyncDatagramSocket, open) +{ + UnixDgram::AsyncSocket client; + + ASSERT_EQ (client.open (), 0) << join::lastError.message (); + ASSERT_EQ (client.open (), -1); + ASSERT_EQ (join::lastError, Errc::InUse); + client.close (); +} + +/** + * @brief Test close method. + */ +TEST_F (UnixAsyncDatagramSocket, close) +{ + UnixDgram::AsyncSocket client; + + ASSERT_EQ (client.open (), 0) << join::lastError.message (); + ASSERT_TRUE (client.opened ()); + client.close (); + ASSERT_FALSE (client.opened ()); +} + +/** + * @brief Test bind method. + */ +TEST_F (UnixAsyncDatagramSocket, bind) +{ + UnixDgram::AsyncSocket client; + + ASSERT_EQ (client.bind (_clientpath), 0) << join::lastError.message (); + ASSERT_EQ (client.bind (_clientpath), -1); + client.close (); +} + +/** + * @brief Test bindToDevice method. + */ +TEST_F (UnixAsyncDatagramSocket, bindToDevice) +{ + UnixDgram::AsyncSocket client; + + ASSERT_EQ (client.bindToDevice (_clientpath), -1); + client.close (); +} + +/** + * @brief Test connect method. + */ +TEST_F (UnixAsyncDatagramSocket, connect) +{ + UnixDgram::AsyncSocket client; + + ASSERT_EQ (client.connect (""), -1); + + ASSERT_EQ (client.connect (_serverpath), 0) << join::lastError.message (); + ASSERT_TRUE (client.connected ()); + ASSERT_EQ (client.connect (_serverpath), -1); + ASSERT_EQ (join::lastError, Errc::InUse); + client.close (); +} + +/** + * @brief Test disconnect method. + */ +TEST_F (UnixAsyncDatagramSocket, disconnect) +{ + UnixDgram::AsyncSocket client; + + ASSERT_EQ (client.disconnect (), 0) << join::lastError.message (); + ASSERT_EQ (client.connect (_serverpath), 0) << join::lastError.message (); + ASSERT_TRUE (client.connected ()); + ASSERT_EQ (client.disconnect (), 0) << join::lastError.message (); + ASSERT_FALSE (client.connected ()); + client.close (); +} + +/** + * @brief Test asyncWriteTo method. + */ +TEST_F (UnixAsyncDatagramSocket, asyncWriteTo) +{ + UnixDgram::AsyncSocket client; + UnixDgram::Endpoint dest (_serverpath); + + ASSERT_FALSE (client.opened ()); + ASSERT_EQ (client.asyncWriteTo ("hello", 5, dest, onReport), 0) << join::lastError.message (); + ASSERT_TRUE (client.opened ()); + + ASSERT_EQ (client.asyncWriteTo ("hello", 5, dest, nullptr), -1); + ASSERT_EQ (join::lastError, Errc::InUse); + + { + ScopedLock lock (_mut); + ASSERT_TRUE (_cond.timedWait (lock, std::chrono::milliseconds (_timeout), [] () { + return _completions >= 1; + })); + ASSERT_FALSE (_code) << _code.message (); + ASSERT_EQ (_transferred, 5u); + } + + client.close (); +} + +/** + * @brief Test asyncReadFrom method. + */ +TEST_F (UnixAsyncDatagramSocket, asyncReadFrom) +{ + UnixDgram::AsyncSocket client; + + ASSERT_EQ (client.asyncReadFrom (_buf, sizeof (_buf), _from, nullptr), -1); + ASSERT_EQ (join::lastError, Errc::OperationFailed); + + ASSERT_EQ (client.bind (_clientpath), 0) << join::lastError.message (); + ASSERT_EQ (client.connect (_serverpath), 0) << join::lastError.message (); + ASSERT_EQ (client.asyncReadFrom (_buf, sizeof (_buf), _from, onReport), 0) << join::lastError.message (); + + ASSERT_EQ (client.asyncReadFrom (_buf, sizeof (_buf), _from, nullptr), -1); + ASSERT_EQ (join::lastError, Errc::InUse); + + ASSERT_EQ (client.asyncWrite ("hello", 5, nullptr), 0) << join::lastError.message (); + + { + ScopedLock lock (_mut); + ASSERT_TRUE (_cond.timedWait (lock, std::chrono::milliseconds (_timeout), [] () { + return _completions >= 1; + })); + ASSERT_FALSE (_code) << _code.message (); + ASSERT_EQ (_transferred, 5u); + ASSERT_EQ (std::string (_buf, 5), "hello"); + } + + client.close (); + + ASSERT_EQ (_from, UnixDgram::Endpoint (_serverpath)); +} + +/** + * @brief Test asyncWrite method. + */ +TEST_F (UnixAsyncDatagramSocket, asyncWrite) +{ + UnixDgram::AsyncSocket client; + + ASSERT_EQ (client.asyncWrite ("hello", 5, nullptr), -1); + ASSERT_EQ (join::lastError, Errc::OperationFailed); + + ASSERT_EQ (client.bind (_clientpath), 0) << join::lastError.message (); + ASSERT_EQ (client.connect (_serverpath), 0) << join::lastError.message (); + ASSERT_EQ (client.asyncWrite ("hello", 5, onReport), 0) << join::lastError.message (); + + { + ScopedLock lock (_mut); + ASSERT_TRUE (_cond.timedWait (lock, std::chrono::milliseconds (_timeout), [] () { + return _completions >= 1; + })); + ASSERT_FALSE (_code) << _code.message (); + ASSERT_EQ (_transferred, 5u); + } + + client.close (); +} + +/** + * @brief Test asyncRead method. + */ +TEST_F (UnixAsyncDatagramSocket, asyncRead) +{ + UnixDgram::AsyncSocket client; + + ASSERT_EQ (client.asyncRead (_buf, sizeof (_buf), nullptr), -1); + ASSERT_EQ (join::lastError, Errc::OperationFailed); + + ASSERT_EQ (client.bind (_clientpath), 0) << join::lastError.message (); + ASSERT_EQ (client.connect (_serverpath), 0) << join::lastError.message (); + ASSERT_EQ (client.asyncRead (_buf, sizeof (_buf), onReport), 0) << join::lastError.message (); + ASSERT_EQ (client.asyncWrite ("hello", 5, nullptr), 0) << join::lastError.message (); + + { + ScopedLock lock (_mut); + ASSERT_TRUE (_cond.timedWait (lock, std::chrono::milliseconds (_timeout), [] () { + return _completions >= 1; + })); + ASSERT_FALSE (_code) << _code.message (); + ASSERT_EQ (_transferred, 5u); + ASSERT_EQ (std::string (_buf, 5), "hello"); + } + + client.close (); +} + +/** + * @brief Test asyncReadFrom method resubmitted from its own handler. + */ +TEST_F (UnixAsyncDatagramSocket, resubmit) +{ + UnixDgram::AsyncSocket client; + UnixDgram::Endpoint dest (_serverpath); + + _current = &client; + _rearms = 1; + + ASSERT_EQ (client.bind (_clientpath), 0) << join::lastError.message (); + ASSERT_EQ (client.asyncReadFrom (_buf, sizeof (_buf), _from, onRead), 0) << join::lastError.message (); + ASSERT_EQ (client.asyncWriteTo ("one", 3, dest, nullptr), 0) << join::lastError.message (); + + { + ScopedLock lock (_mut); + ASSERT_TRUE (_cond.timedWait (lock, std::chrono::milliseconds (_timeout), [] () { + return _completions >= 1; + })); + ASSERT_FALSE (_code) << _code.message (); + } + + ASSERT_EQ (client.asyncWriteTo ("two", 3, dest, nullptr), 0) << join::lastError.message (); + + { + ScopedLock lock (_mut); + ASSERT_TRUE (_cond.timedWait (lock, std::chrono::milliseconds (_timeout), [] () { + return _completions >= 2; + })); + ASSERT_FALSE (_code) << _code.message (); + } + + _dest = UnixDgram::Endpoint (_serverpath); + _rearms = 1; + + ASSERT_EQ (client.asyncWriteTo ("one", 3, _dest, onWrite), 0) << join::lastError.message (); + + { + ScopedLock lock (_mut); + ASSERT_TRUE (_cond.timedWait (lock, std::chrono::milliseconds (_timeout), [] () { + return _completions >= 4; + })); + ASSERT_FALSE (_code) << _code.message (); + } + + client.close (); + _current = nullptr; +} + +/** + * @brief Test close called from within a write handler. + */ +TEST_F (UnixAsyncDatagramSocket, closeFromWriteHandler) +{ + UnixDgram::AsyncSocket client; + UnixDgram::Endpoint dest (_serverpath); + + _current = &client; + + ASSERT_EQ (client.asyncWriteTo ("hello", 5, dest, onWriteAndClose), 0) << join::lastError.message (); + + { + ScopedLock lock (_mut); + ASSERT_TRUE (_cond.timedWait (lock, std::chrono::milliseconds (_timeout), [] () { + return _completions >= 1; + })); + ASSERT_FALSE (_code) << _code.message (); + } + + ASSERT_FALSE (client.opened ()); + _current = nullptr; +} + +/** + * @brief Test a datagram larger than the supplied buffer. + */ +TEST_F (UnixAsyncDatagramSocket, truncated) +{ + UnixDgram::AsyncSocket client; + char small[4] = {}; + + ASSERT_EQ (client.bind (_clientpath), 0) << join::lastError.message (); + ASSERT_EQ (client.connect (_serverpath), 0) << join::lastError.message (); + ASSERT_EQ (client.asyncReadFrom (small, sizeof (small), _from, onReport), 0) << join::lastError.message (); + ASSERT_EQ (client.asyncWrite ("hello world", 11, nullptr), 0) << join::lastError.message (); + + { + ScopedLock lock (_mut); + ASSERT_TRUE (_cond.timedWait (lock, std::chrono::milliseconds (_timeout), [] () { + return _completions >= 1; + })); + ASSERT_EQ (_code, Errc::MessageTooLong); + } + + client.close (); +} + +/** + * @brief Test an empty datagram. + */ +TEST_F (UnixAsyncDatagramSocket, empty) +{ + UnixDgram::AsyncSocket client; + UnixDgram::Socket sender; + UnixDgram::Endpoint self (_clientpath); + + ASSERT_EQ (client.bind (_clientpath), 0) << join::lastError.message (); + ASSERT_EQ (client.asyncReadFrom (_buf, sizeof (_buf), _from, onReport), 0) << join::lastError.message (); + + ASSERT_EQ (sender.bind (_senderpath), 0) << join::lastError.message (); + ASSERT_EQ (sender.writeTo ("", 0, self), 0) << join::lastError.message (); + + { + ScopedLock lock (_mut); + ASSERT_TRUE (_cond.timedWait (lock, std::chrono::milliseconds (_timeout), [] () { + return _completions >= 1; + })); + ASSERT_EQ (_code, Errc::ConnectionClosed); + } + + ASSERT_FALSE (client.connected ()); + + sender.close (); + client.close (); +} + +/** + * @brief Test cancelRead method. + */ +TEST_F (UnixAsyncDatagramSocket, cancelRead) +{ + UnixDgram::AsyncSocket client; + + ASSERT_EQ (client.cancelRead (), 0) << join::lastError.message (); + ASSERT_EQ (client.bind (_clientpath), 0) << join::lastError.message (); + ASSERT_EQ (client.asyncReadFrom (_buf, sizeof (_buf), _from, onReport), 0) << join::lastError.message (); + ASSERT_EQ (client.cancelRead (), 0) << join::lastError.message (); + + { + ScopedLock lock (_mut); + ASSERT_TRUE (_cond.timedWait (lock, std::chrono::milliseconds (_timeout), [] () { + return _completions >= 1; + })); + ASSERT_EQ (_code, std::errc::operation_canceled); + } + + client.close (); +} + +/** + * @brief Test cancelWrite method. + */ +TEST_F (UnixAsyncDatagramSocket, cancelWrite) +{ + UnixDgram::AsyncSocket client; + + ASSERT_EQ (client.cancelWrite (), 0) << join::lastError.message (); + ASSERT_EQ (client.open (), 0) << join::lastError.message (); + ASSERT_EQ (client.cancelWrite (), 0) << join::lastError.message (); + client.close (); +} + +/** + * @brief Test setOption method. + */ +TEST_F (UnixAsyncDatagramSocket, setOption) +{ + UnixDgram::AsyncSocket client; + + ASSERT_EQ (client.setOption (UnixDgram::Socket::RcvBuffer, 1500), -1); + ASSERT_EQ (join::lastError, Errc::OperationFailed); + ASSERT_EQ (client.open (), 0) << join::lastError.message (); + ASSERT_EQ (client.setOption (UnixDgram::Socket::RcvBuffer, 1500), 0) << join::lastError.message (); + client.close (); +} + +/** + * @brief Test localEndpoint method. + */ +TEST_F (UnixAsyncDatagramSocket, localEndpoint) +{ + UnixDgram::AsyncSocket client; + + ASSERT_EQ (client.bind (_clientpath), 0) << join::lastError.message (); + ASSERT_EQ (client.localEndpoint (), UnixDgram::Endpoint (_clientpath)); + client.close (); +} + +/** + * @brief Test remoteEndpoint method. + */ +TEST_F (UnixAsyncDatagramSocket, remoteEndpoint) +{ + UnixDgram::AsyncSocket client; + + ASSERT_EQ (client.connect (_serverpath), 0) << join::lastError.message (); + ASSERT_EQ (client.remoteEndpoint (), UnixDgram::Endpoint (_serverpath)); + client.close (); +} + +/** + * @brief Test opened method. + */ +TEST_F (UnixAsyncDatagramSocket, opened) +{ + UnixDgram::AsyncSocket client; + + ASSERT_FALSE (client.opened ()); + ASSERT_EQ (client.open (), 0) << join::lastError.message (); + ASSERT_TRUE (client.opened ()); + client.close (); + ASSERT_FALSE (client.opened ()); +} + +/** + * @brief Test connected method. + */ +TEST_F (UnixAsyncDatagramSocket, connected) +{ + UnixDgram::AsyncSocket client; + + ASSERT_FALSE (client.connected ()); + ASSERT_EQ (client.connect (_serverpath), 0) << join::lastError.message (); + ASSERT_TRUE (client.connected ()); + client.close (); + ASSERT_FALSE (client.connected ()); +} + +/** + * @brief Test canRead method. + */ +TEST_F (UnixAsyncDatagramSocket, canRead) +{ + UnixDgram::AsyncSocket client; + + ASSERT_EQ (client.canRead (), -1); + ASSERT_EQ (join::lastError, Errc::OperationFailed); + ASSERT_EQ (client.open (), 0) << join::lastError.message (); + ASSERT_EQ (client.canRead (), 0) << join::lastError.message (); + client.close (); +} + +/** + * @brief Test mtu method. + */ +TEST_F (UnixAsyncDatagramSocket, mtu) +{ + UnixDgram::AsyncSocket client; + + ASSERT_EQ (client.mtu (), -1); + ASSERT_EQ (client.connect (_serverpath), 0) << join::lastError.message (); + ASSERT_EQ (client.mtu (), -1); + client.close (); +} + +/** + * @brief Test ttl method. + */ +TEST_F (UnixAsyncDatagramSocket, ttl) +{ + UnixDgram::AsyncSocket client; + + ASSERT_EQ (client.ttl (), 60); + + UnixDgram::AsyncSocket other (32); + + ASSERT_EQ (other.ttl (), 32); +} + +/** + * @brief Test family method. + */ +TEST_F (UnixAsyncDatagramSocket, family) +{ + UnixDgram::AsyncSocket client; + + ASSERT_EQ (client.open (), 0) << join::lastError.message (); + ASSERT_EQ (client.family (), AF_UNIX); + client.close (); +} + +/** + * @brief Test type method. + */ +TEST_F (UnixAsyncDatagramSocket, type) +{ + UnixDgram::AsyncSocket client; + + ASSERT_EQ (client.open (), 0) << join::lastError.message (); + ASSERT_EQ (client.type (), SOCK_DGRAM); + client.close (); +} + +/** + * @brief Test protocol method. + */ +TEST_F (UnixAsyncDatagramSocket, protocol) +{ + UnixDgram::AsyncSocket client; + + ASSERT_EQ (client.open (), 0) << join::lastError.message (); + ASSERT_EQ (client.protocol (), 0); + client.close (); +} + +/** + * @brief Test handle method. + */ +TEST_F (UnixAsyncDatagramSocket, handle) +{ + UnixDgram::AsyncSocket client; + + ASSERT_EQ (client.handle (), -1); + ASSERT_EQ (client.open (), 0) << join::lastError.message (); + ASSERT_GT (client.handle (), -1); + client.close (); +} + +/** + * @brief main function. + */ +int main (int argc, char** argv) +{ + testing::InitGoogleTest (&argc, argv); + return RUN_ALL_TESTS (); +} diff --git a/core/tests/unix_async_stream_socket_test.cpp b/core/tests/unix_async_stream_socket_test.cpp index 90271370..22f96bc2 100644 --- a/core/tests/unix_async_stream_socket_test.cpp +++ b/core/tests/unix_async_stream_socket_test.cpp @@ -50,7 +50,7 @@ class UnixAsyncStreamSocket : public ::testing::Test void SetUp () override { ASSERT_EQ (_server.create (_serverpath), 0) << join::lastError.message (); - ASSERT_EQ (_server.asyncAccept (onEchoAccept), 0) << join::lastError.message (); + ASSERT_EQ (_server.asyncAccept (peer (), onEchoAccept), 0) << join::lastError.message (); ScopedLock lock (_mut); @@ -118,13 +118,11 @@ class UnixAsyncStreamSocket : public ::testing::Test /** * @brief adopt the socket accepted by the echo server. * @param ec error reported by the acceptor. - * @param sock accepted socket. */ - static void onEchoAccept (const std::error_code& ec, UnixStream::AsyncSocket&& sock) + static void onEchoAccept (const std::error_code& ec) { if (!ec) { - peer () = std::move (sock); peer ().asyncRead (_echobuf, sizeof (_echobuf), onEchoRead); } } @@ -243,83 +241,6 @@ const std::string UnixAsyncStreamSocket::_clientpath = "/tmp/unixasyncclient_tes const std::string UnixAsyncStreamSocket::_stallpath = "/tmp/unixasyncstall_test.sock"; const int UnixAsyncStreamSocket::_timeout = 1000; -/** - * @brief Test move with an operation in flight. - */ -TEST_F (UnixAsyncStreamSocket, move) -{ - UnixStream::AsyncSocket client; - - ASSERT_EQ (client.asyncConnect (_serverpath, - [] (const std::error_code& ec) { - ScopedLock lock (_mut); - _code = ec; - ++_completions; - _cond.signal (); - }), - 0) - << join::lastError.message (); - - { - ScopedLock lock (_mut); - ASSERT_TRUE (_cond.timedWait (lock, std::chrono::milliseconds (_timeout), [] () { - return _completions >= 1; - })); - ASSERT_FALSE (_code) << _code.message (); - } - - ASSERT_EQ (client.asyncRead (_buf, sizeof (_buf), - [] (const std::error_code& ec, size_t size) { - ScopedLock lock (_mut); - _code = ec; - _transferred = size; - ++_completions; - _cond.signal (); - }), - 0) - << join::lastError.message (); - - ASSERT_EQ (client.asyncRead (_buf, sizeof (_buf), nullptr), -1); - ASSERT_EQ (join::lastError, Errc::InUse); - - UnixStream::AsyncSocket moved (std::move (client)); - - ASSERT_EQ (moved.asyncRead (_buf, sizeof (_buf), nullptr), -1); - ASSERT_EQ (join::lastError, Errc::InUse); - - ASSERT_EQ (moved.asyncWrite ("one", 3, nullptr), 0) << join::lastError.message (); - - { - ScopedLock lock (_mut); - ASSERT_TRUE (_cond.timedWait (lock, std::chrono::milliseconds (_timeout), [] () { - return _completions >= 2; - })); - ASSERT_FALSE (_code) << _code.message (); - ASSERT_EQ (_transferred, 3u); - } - - UnixStream::AsyncSocket assigned; - assigned = std::move (moved); - - ASSERT_TRUE (assigned.connected ()); - - ASSERT_EQ (moved.cancelRead (), 0) << join::lastError.message (); - ASSERT_EQ (moved.cancelWrite (), 0) << join::lastError.message (); - ASSERT_EQ (moved.asyncConnect ({_serverpath}, nullptr), -1); - ASSERT_EQ (join::lastError, Errc::OperationFailed); - ASSERT_FALSE (moved.opened ()); - moved.close (); - - UnixStream::AsyncSocket chained (std::move (moved)); - ASSERT_FALSE (chained.opened ()); - - UnixStream::AsyncSocket reassigned; - reassigned = std::move (chained); - ASSERT_FALSE (reassigned.opened ()); - - assigned.close (); -} - /** * @brief Test open method. */ @@ -771,7 +692,7 @@ TEST_F (UnixAsyncStreamSocket, cancelWrite) ASSERT_LT (filled, 4096); - ASSERT_EQ (sender.asyncWrite (_buf, sizeof (_buf), nullptr), 0) << join::lastError.message (); + ASSERT_EQ (sender.asyncWrite (_buf, sizeof (_buf), onWrite), 0) << join::lastError.message (); ASSERT_EQ (sender.asyncWrite (_buf, sizeof (_buf), nullptr), -1); ASSERT_EQ (join::lastError, Errc::InUse); @@ -781,6 +702,16 @@ TEST_F (UnixAsyncStreamSocket, cancelWrite) ASSERT_EQ (sender.cancelWrite (), 0) << join::lastError.message (); + { + ScopedLock lock (_mut); + ASSERT_TRUE (_cond.timedWait (lock, std::chrono::milliseconds (_timeout), [] () { + return _completions >= 2; + })); + ASSERT_EQ (_code, std::errc::operation_canceled); + } + + ASSERT_EQ (sender.asyncWrite (_buf, sizeof (_buf), nullptr), 0) << join::lastError.message (); + sender.close (); peer.close (); stall.close ();