async_datagram_socket.hpp

Go to the documentation of this file.

 
#ifndef JOIN_CORE_ASYNC_DATAGRAM_SOCKET_HPP
#define JOIN_CORE_ASYNC_DATAGRAM_SOCKET_HPP
 
// libjoin.
#include <join/async_raw_socket.hpp>
#include <join/datagram_socket.hpp>
 
// C++.
#include <system_error>
#include <utility>
 
namespace join
{
    template <class Protocol, class ProactorType, size_t OpCount>
    class BasicAsyncDatagramSocket : public BasicAsyncRawSocket<Protocol, ProactorType, OpCount>
    {
    public:
        using Socket = typename Protocol::Socket;
        using Endpoint = typename Protocol::Endpoint;
        using AsyncRead = BasicAsyncRead<Protocol, ProactorType>;
        using AsyncWrite = BasicAsyncWrite<Protocol, ProactorType>;
        using ReadFromHandler = typename AsyncRead::ReadFrom;
        using WriteHandler = typename AsyncWrite::Write;
 
        explicit BasicAsyncDatagramSocket (ProactorType& proactor = ProactorThread::proactor ())
        : BasicAsyncRawSocket<Protocol, ProactorType, OpCount> (proactor)
        {
        }
 
        explicit BasicAsyncDatagramSocket (int ttl, ProactorType& proactor = ProactorThread::proactor ())
        : BasicAsyncRawSocket<Protocol, ProactorType, OpCount> (Socket (ttl), proactor)
        {
        }
 
        explicit BasicAsyncDatagramSocket (Socket&& sock, ProactorType& proactor = ProactorThread::proactor ())
        : BasicAsyncRawSocket<Protocol, ProactorType, OpCount> (std::move (sock), proactor)
        {
        }
 
        BasicAsyncDatagramSocket (const BasicAsyncDatagramSocket& other) = delete;
 
        BasicAsyncDatagramSocket& operator= (const BasicAsyncDatagramSocket& other) = delete;
 
        BasicAsyncDatagramSocket (BasicAsyncDatagramSocket&& other) noexcept = default;
 
        BasicAsyncDatagramSocket& operator= (BasicAsyncDatagramSocket&& other) noexcept = default;
 
        int connect (const Endpoint& endpoint) noexcept
        {
            return this->_socket.connect (endpoint);
        }
 
        int disconnect () noexcept
        {
            return this->_socket.disconnect ();
        }
 
        ssize_t asyncReadFrom (ReadFromHandler handler, char* data, size_t maxSize, Endpoint& endpoint,
                               char* control = nullptr, size_t controlSize = 0, bool flush = true,
                               bool link = false) noexcept
        {
            if (JOIN_UNLIKELY (!this->_socket.opened ()))
            {
                lastError = make_error_code (Errc::OperationFailed);
                return -1;
            }
 
            AsyncRead* read = this->allocateRead ();
            if (JOIN_UNLIKELY (read == nullptr))
            {
                lastError = make_error_code (Errc::OutOfMemory);
                return -1;
            }
 
            read->readFromHandler = std::move (handler);
            read->iov.iov_base = data;
            read->iov.iov_len = maxSize;
            read->msg.msg_name = endpoint.addr ();
            read->msg.msg_namelen = sizeof (struct sockaddr_storage);
            read->msg.msg_iov = &read->iov;
            read->msg.msg_iovlen = 1;
            read->msg.msg_control = control;
            read->msg.msg_controllen = controlSize;
            read->msg.msg_flags = 0;
            read->op = IoOperation::makeRecvmsg (this->_socket.handle (), &read->msg, 0, this, link);
            read->op.state.store (IoOperation::State::Submitted, std::memory_order_release);
 
            size_t index = this->_allocator.getIndex (read);
 
            if (this->_proactor->submit (read->op, flush, false) == -1)
            {
                // LCOV_EXCL_START
                this->releaseOperation (read);
                return -1;
                // LCOV_EXCL_STOP
            }
 
            return static_cast<ssize_t> (index);
        }
 
        ssize_t asyncReadFromMulti (ReadFromHandler handler, uint16_t group, size_t controlSize = 0,
                                    bool flush = true) noexcept
        {
            if (JOIN_UNLIKELY (!this->_socket.opened ()))
            {
                lastError = make_error_code (Errc::OperationFailed);
                return -1;
            }
 
            AsyncRead* read = this->allocateRead ();
            if (JOIN_UNLIKELY (read == nullptr))
            {
                lastError = make_error_code (Errc::OutOfMemory);
                return -1;
            }
 
            read->readFromHandler = std::move (handler);
            read->msg.msg_name = nullptr;
            read->msg.msg_namelen = sizeof (struct sockaddr_storage);
            read->msg.msg_iov = &read->iov;
            read->msg.msg_iovlen = 1;
            read->msg.msg_control = nullptr;
            read->msg.msg_controllen = controlSize;
            read->msg.msg_flags = 0;
            read->op = IoOperation::makeRecvmsgMulti (this->_socket.handle (), group, &read->msg, 0, this);
            read->op.state.store (IoOperation::State::Submitted, std::memory_order_release);
 
            size_t index = this->_allocator.getIndex (read);
 
            if (this->_proactor->submit (read->op, flush, false) == -1)
            {
                // LCOV_EXCL_START
                this->releaseOperation (read);
                return -1;
                // LCOV_EXCL_STOP
            }
 
            return static_cast<ssize_t> (index);
        }
 
        ssize_t asyncWriteTo (WriteHandler handler, const char* data, size_t size, Endpoint& endpoint,
                              const char* control = nullptr, size_t controlSize = 0, bool flush = true,
                              bool link = false) noexcept
        {
            if (JOIN_UNLIKELY (!this->_allocator.hasBackend ()))
            {
                lastError = make_error_code (Errc::OperationFailed);
                return -1;
            }
 
            if (!this->_socket.opened () && (this->_socket.open (endpoint.protocol ()) == -1))
            {
                return -1;  // LCOV_EXCL_LINE
            }
 
            AsyncWrite* write = this->allocateWrite ();
            if (JOIN_UNLIKELY (write == nullptr))
            {
                lastError = make_error_code (Errc::OutOfMemory);
                return -1;
            }
 
            write->writeHandler = std::move (handler);
            write->iov.iov_base = const_cast<char*> (data);
            write->iov.iov_len = size;
            write->msg.msg_name = endpoint.addr ();
            write->msg.msg_namelen = endpoint.length ();
            write->msg.msg_iov = &write->iov;
            write->msg.msg_iovlen = 1;
            write->msg.msg_control = const_cast<char*> (control);
            write->msg.msg_controllen = controlSize;
            write->msg.msg_flags = 0;
            write->op = IoOperation::makeSendmsg (this->_socket.handle (), &write->msg, MSG_NOSIGNAL, this, link);
            write->op.state.store (IoOperation::State::Submitted, std::memory_order_release);
 
            size_t index = this->_allocator.getIndex (write);
 
            if (this->_proactor->submit (write->op, flush, false) == -1)
            {
                // LCOV_EXCL_START
                this->releaseOperation (write);
                return -1;
                // LCOV_EXCL_STOP
            }
 
            return static_cast<ssize_t> (index);
        }
 
        const Endpoint& remoteEndpoint () const noexcept
        {
            return this->_socket.remoteEndpoint ();
        }
 
        bool connected () const noexcept
        {
            return this->_socket.connected ();
        }
 
        int mtu () const noexcept
        {
            return this->_socket.mtu ();
        }
 
        int ttl () const noexcept
        {
            return this->_socket.ttl ();
        }
    };
}
 
#endif