resolver.hpp

Go to the documentation of this file.

 
#ifndef JOIN_FABRIC_RESOLVER_HPP
#define JOIN_FABRIC_RESOLVER_HPP
 
// libjoin.
#include <join/datagram_socket.hpp>
#include <join/dns_protocol.hpp>
#include <join/dot_protocol.hpp>
#include <join/tls_wrapper.hpp>
#include <join/dns_message.hpp>
#include <join/condition.hpp>
#include <join/reactor.hpp>
 
// C++.
#include <unordered_map>
#include <chrono>
#include <memory>
 
// C.
#include <arpa/nameser.h>
#include <netinet/in.h>
#include <resolv.h>
#include <netdb.h>
 
namespace join
{
    template <class Protocol>
    class BasicDatagramResolver : public EventHandler
    {
    public:
        using Socket = typename Protocol::Socket;
        using Endpoint = typename Protocol::Endpoint;
 
        using DnsNotify = std::function<void (const DnsPacket&)>;
 
        DnsNotify onSuccess;
 
        DnsNotify onFailure;
 
        explicit BasicDatagramResolver (const std::string& server = {}, uint16_t port = Protocol::defaultPort,
                                        Reactor& reactor = ReactorThread::reactor ())
        : BasicDatagramResolver (Socket{}, server, port, reactor)
        {
        }
 
        explicit BasicDatagramResolver (Socket&& socket, const std::string& server = {},
                                        uint16_t port = Protocol::defaultPort,
                                        Reactor& reactor = ReactorThread::reactor ())
#ifdef DEBUG
        : onSuccess (defaultOnSuccess)
        , onFailure (defaultOnFailure)
#else
        : onSuccess (nullptr)
        , onFailure (nullptr)
#endif
        , _socket (std::move (socket))
        , _server (server)
        , _port (port)
        , _reactor (reactor)
        , _buffer (std::make_unique<char[]> (Protocol::maxMsgSize))
        {
        }
 
        BasicDatagramResolver (const BasicDatagramResolver& other) = delete;
 
        BasicDatagramResolver& operator= (const BasicDatagramResolver& other) = delete;
 
        BasicDatagramResolver (BasicDatagramResolver&& other) = delete;
 
        BasicDatagramResolver& operator= (BasicDatagramResolver&& other) = delete;
 
        virtual ~BasicDatagramResolver () noexcept
        {
            disconnect ();
        }
 
        IpAddressList resolveAllAddress (const std::string& host, int family,
                                         std::chrono::milliseconds timeout = std::chrono::seconds (5))
        {
            if (host.empty ())
            {
                return {};
            }
 
            DnsPacket packet{};
            packet.id = join::randomize<uint16_t> ();
            packet.flags = 1 << 8;
 
            QuestionRecord question;
            question.host = host;
            question.type = (family == AF_INET6) ? DnsMessage::RecordType::AAAA : DnsMessage::RecordType::A;
            question.dnsclass = DnsMessage::RecordClass::IN;
            packet.questions.push_back (question);
 
            if (query (packet, timeout) == -1)
            {
                return {};
            }
 
            IpAddressList addresses;
 
            for (auto const& answer : packet.answers)
            {
                if (!answer.addr.isWildcard () && (answer.type == question.type))
                {
                    addresses.push_back (answer.addr);
                }
            }
 
            return addresses;
        }
 
        static IpAddressList lookupAllAddress (const std::string& host, int family)
        {
            for (auto const& server : nameServers ())
            {
                IpAddressList addresses =
                    BasicDatagramResolver<Protocol> (server.toString ()).resolveAllAddress (host, family);
                if (!addresses.empty ())
                {
                    return addresses;
                }
            }
 
            return {};
        }
 
        IpAddressList resolveAllAddress (const std::string& host,
                                         std::chrono::milliseconds timeout = std::chrono::seconds (5))
        {
            IpAddressList addresses;
 
            for (auto const& family : {AF_INET, AF_INET6})
            {
                IpAddressList tmp = resolveAllAddress (host, family, timeout);
                addresses.insert (addresses.end (), tmp.begin (), tmp.end ());
            }
 
            return addresses;
        }
 
        static IpAddressList lookupAllAddress (const std::string& host)
        {
            for (auto const& server : nameServers ())
            {
                IpAddressList addresses = BasicDatagramResolver<Protocol> (server.toString ()).resolveAllAddress (host);
                if (!addresses.empty ())
                {
                    return addresses;
                }
            }
 
            return {};
        }
 
        IpAddress resolveAddress (const std::string& host, int family,
                                  std::chrono::milliseconds timeout = std::chrono::seconds (5))
        {
            for (auto const& address : resolveAllAddress (host, family, timeout))
            {
                return address;
            }
 
            return IpAddress (family);
        }
 
        static IpAddress lookupAddress (const std::string& host, int family)
        {
            for (auto const& address : lookupAllAddress (host, family))
            {
                return address;
            }
 
            return IpAddress (family);
        }
 
        IpAddress resolveAddress (const std::string& host, std::chrono::milliseconds timeout = std::chrono::seconds (5))
        {
            for (auto const& address : resolveAllAddress (host, timeout))
            {
                return address;
            }
 
            return {};
        }
 
        static IpAddress lookupAddress (const std::string& host)
        {
            for (auto const& address : lookupAllAddress (host))
            {
                return address;
            }
 
            return {};
        }
 
        AliasList resolveAllName (const IpAddress& address,
                                  std::chrono::milliseconds timeout = std::chrono::seconds (5))
        {
            if (address.isWildcard ())
            {
                return {};
            }
 
            DnsPacket packet{};
            packet.id = join::randomize<uint16_t> ();
            packet.flags = 1 << 8;
 
            QuestionRecord question;
            question.host = address.toArpa ();
            question.type = DnsMessage::RecordType::PTR;
            question.dnsclass = DnsMessage::RecordClass::IN;
            packet.questions.push_back (question);
 
            if (query (packet, timeout) == -1)
            {
                return {};
            }
 
            AliasList aliases;
 
            for (auto const& answer : packet.answers)
            {
                if (!answer.name.empty () && (answer.type == DnsMessage::RecordType::PTR))
                {
                    aliases.insert (answer.name);
                }
            }
 
            return aliases;
        }
 
        static AliasList lookupAllName (const IpAddress& address)
        {
            for (auto const& server : nameServers ())
            {
                AliasList aliases = BasicDatagramResolver<Protocol> (server.toString ()).resolveAllName (address);
                if (!aliases.empty ())
                {
                    return aliases;
                }
            }
 
            return {};
        }
 
        std::string resolveName (const IpAddress& address, std::chrono::milliseconds timeout = std::chrono::seconds (5))
        {
            for (auto const& alias : resolveAllName (address, timeout))
            {
                return alias;
            }
 
            return {};
        }
 
        static std::string lookupName (const IpAddress& address)
        {
            for (auto const& alias : lookupAllName (address))
            {
                return alias;
            }
 
            return {};
        }
 
        ServerList resolveAllNameServer (const std::string& host,
                                         std::chrono::milliseconds timeout = std::chrono::seconds (5))
        {
            if (host.empty ())
            {
                return {};
            }
 
            DnsPacket packet{};
            packet.id = join::randomize<uint16_t> ();
            packet.flags = 1 << 8;
 
            QuestionRecord question;
            question.host = host;
            question.type = DnsMessage::RecordType::NS;
            question.dnsclass = DnsMessage::RecordClass::IN;
            packet.questions.push_back (question);
 
            if (query (packet, timeout) == -1)
            {
                return {};
            }
 
            ServerList servers;
 
            for (auto const& answer : packet.answers)
            {
                if (!answer.name.empty () && (answer.type == DnsMessage::RecordType::NS))
                {
                    servers.insert (answer.name);
                }
            }
 
            return servers;
        }
 
        static ServerList lookupAllNameServer (const std::string& host)
        {
            for (auto const& server : nameServers ())
            {
                ServerList servers = BasicDatagramResolver<Protocol> (server.toString ()).resolveAllNameServer (host);
                if (!servers.empty ())
                {
                    return servers;
                }
            }
 
            return {};
        }
 
        std::string resolveNameServer (const std::string& host,
                                       std::chrono::milliseconds timeout = std::chrono::seconds (5))
        {
            for (auto const& server : resolveAllNameServer (host, timeout))
            {
                return server;
            }
 
            return {};
        }
 
        static std::string lookupNameServer (const std::string& host)
        {
            for (auto const& server : lookupAllNameServer (host))
            {
                return server;
            }
 
            return {};
        }
 
        std::string resolveAuthority (const std::string& host,
                                      std::chrono::milliseconds timeout = std::chrono::seconds (5))
        {
            if (host.empty ())
            {
                return {};
            }
 
            DnsPacket packet{};
            packet.id = join::randomize<uint16_t> ();
            packet.flags = 1 << 8;
 
            QuestionRecord question;
            question.host = host;
            question.type = DnsMessage::RecordType::SOA;
            question.dnsclass = DnsMessage::RecordClass::IN;
            packet.questions.push_back (question);
 
            if (query (packet, timeout) == -1)
            {
                return {};
            }
 
            for (auto const& answer : packet.answers)
            {
                if (!answer.name.empty () && (answer.type == DnsMessage::RecordType::SOA))
                {
                    return answer.name;
                }
            }
 
            return {};
        }
 
        static std::string lookupAuthority (const std::string& host)
        {
            for (auto const& server : nameServers ())
            {
                std::string authority = BasicDatagramResolver<Protocol> (server.toString ()).resolveAuthority (host);
                if (!authority.empty ())
                {
                    return authority;
                }
            }
 
            return {};
        }
 
        ExchangerList resolveAllMailExchanger (const std::string& host,
                                               std::chrono::milliseconds timeout = std::chrono::seconds (5))
        {
            if (host.empty ())
            {
                return {};
            }
 
            DnsPacket packet{};
            packet.id = join::randomize<uint16_t> ();
            packet.flags = 1 << 8;
 
            QuestionRecord question;
            question.host = host;
            question.type = DnsMessage::RecordType::MX;
            question.dnsclass = DnsMessage::RecordClass::IN;
            packet.questions.push_back (question);
 
            if (query (packet, timeout) == -1)
            {
                return {};
            }
 
            ExchangerList exchangers;
            for (auto const& answer : packet.answers)
            {
                if (!answer.name.empty () && (answer.type == DnsMessage::RecordType::MX))
                {
                    exchangers.insert (answer.name);
                }
            }
 
            return exchangers;
        }
 
        static ExchangerList lookupAllMailExchanger (const std::string& host)
        {
            for (auto const& server : nameServers ())
            {
                ExchangerList exchangers =
                    BasicDatagramResolver<Protocol> (server.toString ()).resolveAllMailExchanger (host);
                if (!exchangers.empty ())
                {
                    return exchangers;
                }
            }
 
            return {};
        }
 
        std::string resolveMailExchanger (const std::string& host,
                                          std::chrono::milliseconds timeout = std::chrono::seconds (5))
        {
            for (auto const& exchanger : resolveAllMailExchanger (host, timeout))
            {
                return exchanger;
            }
 
            return {};
        }
 
        static std::string lookupMailExchanger (const std::string& host)
        {
            for (auto const& exchanger : lookupAllMailExchanger (host))
            {
                return exchanger;
            }
 
            return {};
        }
 
        static IpAddressList nameServers () noexcept
        {
            IpAddressList addressList;
 
            struct __res_state res;
            if (res_ninit (&res) == 0)
            {
                for (int i = 0; i < res.nscount; ++i)
                {
                    if (res.nsaddr_list[i].sin_family == AF_INET)
                    {
                        addressList.emplace_back (&res.nsaddr_list[i].sin_addr, sizeof (struct in_addr));
                    }
                    // LCOV_EXCL_START: requires specific host IPv6 configuration.
                    else if (res._u._ext.nsaddrs[i] != nullptr && res._u._ext.nsaddrs[i]->sin6_family == AF_INET6)
                    {
                        addressList.emplace_back (&res._u._ext.nsaddrs[i]->sin6_addr, sizeof (struct in6_addr));
                    }
                    // LCOV_EXCL_STOP
                }
                res_nclose (&res);
            }
 
            return addressList;
        }
 
        static uint16_t resolveService (const std::string& service) noexcept
        {
            struct servent entry, *res;
            char buffer[1024];
 
            int status = getservbyname_r (service.c_str (), nullptr, &entry, buffer, sizeof buffer, &res);
            if ((status == 0) && (res != nullptr))
            {
                return ntohs (entry.s_port);
            }
 
            return 0;
        }
 
    protected:
        virtual int connect (const Endpoint& endpoint,
                             [[maybe_unused]] std::chrono::milliseconds timeout = std::chrono::seconds (5)) noexcept
        {
            if (_socket.connect (endpoint) == -1)
            {
                _socket.close ();
                return -1;
            }
 
            _server = endpoint.hostname ().empty () ? endpoint.ip ().toString () : endpoint.hostname ();
            _port = endpoint.port ();
 
            _reactor.addHandler (_socket.handle (), this);
 
            return 0;
        }
 
        virtual int disconnect ([[maybe_unused]] std::chrono::milliseconds timeout = std::chrono::seconds (5)) noexcept
        {
            if (this->_socket.handle () == -1)
            {
                return 0;
            }
 
            _reactor.delHandler (_socket.handle ());
 
            if (_socket.disconnect () == -1)
            {
                _socket.close ();
                return -1;
            }
 
            return 0;
        }
 
        virtual bool needReconnection () noexcept
        {
            return !_socket.connected () || _socket.remoteEndpoint ().ip ().isWildcard ();
        }
 
        int reconnect (const Endpoint& endpoint, std::chrono::milliseconds timeout = std::chrono::seconds (5))
        {
            disconnect (timeout);
            return connect (endpoint, timeout);
        }
 
        virtual ssize_t read (char* data, size_t maxSize) noexcept
        {
            return _socket.read (data, maxSize);
        }
 
        virtual ssize_t write (const char* data, size_t size) noexcept
        {
            return _socket.write (data, size);
        }
 
        int query (DnsPacket& packet, std::chrono::milliseconds timeout)
        {
            auto remote = _socket.remoteEndpoint ();
 
            if (remote.ip ().isWildcard ())
            {
                IpAddress ip = IpAddress::isIpAddress (_server) ? IpAddress (_server)
                                                                : Dns::Resolver::lookupAddress (_server, AF_INET);
 
                if (ip.isWildcard ())
                {
                    lastError = make_error_code (Errc::InvalidParam);
                    notify (onFailure, packet);
                    return -1;
                }
 
                remote.ip (ip);
                remote.port (_port);
            }
 
            packet.dest = remote.ip ();
            packet.port = remote.port ();
 
            if (needReconnection ())
            {
                Endpoint endpoint{packet.dest, packet.port};
                endpoint.hostname (_server);
 
                if (reconnect (endpoint, timeout) == -1)
                {
                    notify (onFailure, packet);
                    return -1;
                }
            }
 
            packet.src = _socket.localEndpoint ().ip ();
 
            std::stringstream data;
            if (_message.serialize (packet, data) == -1)
            {
                lastError = make_error_code (Errc::InvalidParam);
                notify (onFailure, packet);
                return -1;
            }
 
            std::string buffer = data.str ();
            if (buffer.size () > Protocol::maxMsgSize)
            {
                lastError = make_error_code (Errc::MessageTooLong);
                notify (onFailure, packet);
                return -1;
            }
 
            ScopedLock<Mutex> lock (_syncMutex);
 
            auto inserted = _pending.emplace (packet.id, std::make_unique<PendingRequest> ());
            if (!inserted.second)
            {
                // LCOV_EXCL_START
                lastError = make_error_code (Errc::OperationFailed);
                notify (onFailure, packet);
                return -1;
                // LCOV_EXCL_STOP
            }
 
            if (write (buffer.data (), buffer.size ()) == -1)
            {
                // LCOV_EXCL_START
                _pending.erase (inserted.first);
                notify (onFailure, packet);
                return -1;
                // LCOV_EXCL_STOP
            }
 
            if (!inserted.first->second->cond.timedWait (lock, timeout))
            {
                _pending.erase (inserted.first);
                lastError = make_error_code (Errc::TimedOut);
                notify (onFailure, packet);
                return -1;
            }
 
            auto pendingReq = std::move (inserted.first->second);
            _pending.erase (inserted.first);
 
            if (pendingReq->ec)
            {
                lastError = pendingReq->ec;
                notify (onFailure, packet);
                return -1;
            }
 
            packet = std::move (pendingReq->packet);
            notify (onSuccess, packet);
 
            return 0;
        }
 
        void onReadable ([[maybe_unused]] int fd) override final
        {
            ssize_t size = read (_buffer.get (), Protocol::maxMsgSize);
            if (size >= int (_headerSize))
            {
                std::stringstream data;
                data.rdbuf ()->pubsetbuf (_buffer.get (), size);
 
                DnsPacket packet;
                _message.deserialize (packet, data);
                auto local = _socket.localEndpoint ();
                auto remote = _socket.remoteEndpoint ();
                packet.src = local.ip ();
                packet.dest = remote.ip ();
                packet.port = remote.port ();
 
                if (packet.flags & 0x8000)
                {
                    ScopedLock<Mutex> lock (_syncMutex);
 
                    auto it = _pending.find (packet.id);
                    if (it != _pending.end ())
                    {
                        it->second->packet = packet;
                        it->second->ec = DnsMessage::decodeError (packet.flags & 0x000F);
                        if ((packet.flags & 0x0200) && it->second->ec == std::error_code{})
                        {
                            it->second->ec = make_error_code (Errc::MessageTooLong);
                        }
                        it->second->cond.signal ();
                    }
                }
            }
        }
 
        void onClose ([[maybe_unused]] int fd) override final
        {
            disconnect ();
        }
 
#ifdef DEBUG
        /*
         * @brief default callback called when a lookup sequence succeed.
         * @param packet DNS packet.
         */
        static void defaultOnSuccess (const DnsPacket& packet)
        {
            std::cout << std::endl;
            std::cout << "SERVER: " << packet.dest << "#" << packet.port << std::endl;
 
            std::cout << std::endl;
            std::cout << ";; QUESTION SECTION: " << std::endl;
            for (auto const& question : packet.questions)
            {
                std::cout << question.host;
                std::cout << "  " << DnsMessage::typeName (question.type);
                std::cout << "  " << DnsMessage::className (question.dnsclass);
                std::cout << std::endl;
            }
 
            std::cout << std::endl;
            std::cout << ";; ANSWER SECTION: " << std::endl;
            for (auto const& answer : packet.answers)
            {
                std::cout << answer.host;
                std::cout << "  " << DnsMessage::typeName (answer.type);
                std::cout << "  " << DnsMessage::className (answer.dnsclass);
                std::cout << "  " << answer.ttl;
                if (answer.type == DnsMessage::RecordType::A)
                {
                    std::cout << "  " << answer.addr;
                }
                else if (answer.type == DnsMessage::RecordType::NS)
                {
                    std::cout << "  " << answer.name;
                }
                else if (answer.type == DnsMessage::RecordType::CNAME)
                {
                    std::cout << "  " << answer.name;
                }
                else if (answer.type == DnsMessage::RecordType::SOA)
                {
                    std::cout << "  " << answer.name;
                    std::cout << "  " << answer.mail;
                    std::cout << "  " << answer.serial;
                    std::cout << "  " << answer.refresh;
                    std::cout << "  " << answer.retry;
                    std::cout << "  " << answer.expire;
                    std::cout << "  " << answer.minimum;
                }
                else if (answer.type == DnsMessage::RecordType::PTR)
                {
                    std::cout << "  " << answer.name;
                }
                else if (answer.type == DnsMessage::RecordType::MX)
                {
                    std::cout << "  " << answer.mxpref;
                    std::cout << "  " << answer.name;
                }
                else if (answer.type == DnsMessage::RecordType::AAAA)
                {
                    std::cout << "  " << answer.addr;
                }
                std::cout << std::endl;
            }
        }
 
        /*
         * @brief default callback called when a lookup sequence failed.
         * @param packet DNS packet.
         */
        static void defaultOnFailure (const DnsPacket& packet)
        {
            std::cout << std::endl;
            std::cout << "SERVER: " << packet.dest << "#" << packet.port << std::endl;
 
            std::cout << std::endl;
            std::cout << ";; QUESTION SECTION: " << std::endl;
            for (auto const& question : packet.questions)
            {
                std::cout << question.host;
                std::cout << "  " << DnsMessage::typeName (question.type);
                std::cout << "  " << DnsMessage::className (question.dnsclass);
                std::cout << std::endl;
            }
 
            std::cout << std::endl;
            std::cout << lastError.message () << std::endl;
        }
#endif
 
        void notify (const DnsNotify& func, const DnsPacket& packet) const noexcept
        {
            if (func)
            {
                func (packet);
            }
        }
 
        static constexpr size_t _headerSize = 12;
 
        DnsMessage _message;
 
        Socket _socket;
 
        std::string _server;
 
        uint16_t _port;
 
        Reactor& _reactor;
 
        std::unique_ptr<char[]> _buffer;
 
        struct PendingRequest
        {
            Condition cond;
            DnsPacket packet;
            std::error_code ec;
        };
 
        std::unordered_map<uint16_t, std::unique_ptr<PendingRequest>> _pending;
 
        Mutex _syncMutex;
    };
 
    template <class Protocol>
    class BasicTlsResolver : public BasicDatagramResolver<Protocol>
    {
    public:
        using Socket = typename Protocol::Socket;
        using UnderlyingSocket = typename Socket::UnderlyingSocket;
        using Endpoint = typename Protocol::Endpoint;
 
        explicit BasicTlsResolver (TlsContext ctx, const std::string& server = {},
                                   uint16_t port = Protocol::defaultPort, Reactor& reactor = ReactorThread::reactor ())
        : BasicDatagramResolver<Protocol> (Socket (UnderlyingSocket{}, ctx), server, port, reactor)
        {
        }
 
        BasicTlsResolver (const BasicTlsResolver& other) = delete;
 
        BasicTlsResolver& operator= (const BasicTlsResolver& other) = delete;
 
        BasicTlsResolver (BasicTlsResolver&& other) = delete;
 
        BasicTlsResolver& operator= (BasicTlsResolver&& other) = delete;
 
        virtual ~BasicTlsResolver () noexcept
        {
            disconnect ();
        }
 
        static IpAddressList lookupAllAddress (const std::string& host, int family) = delete;
 
        static IpAddressList lookupAllAddress (const std::string& host) = delete;
 
        static IpAddress lookupAddress (const std::string& host, int family) = delete;
 
        static IpAddress lookupAddress (const std::string& host) = delete;
 
        static AliasList lookupAllName (const IpAddress& address) = delete;
 
        static std::string lookupName (const IpAddress& address) = delete;
 
        static ServerList lookupAllNameServer (const std::string& host) = delete;
 
        static std::string lookupNameServer (const std::string& host) = delete;
 
        static std::string lookupAuthority (const std::string& host) = delete;
 
        static ExchangerList lookupAllMailExchanger (const std::string& host) = delete;
 
        static std::string lookupMailExchanger (const std::string& host) = delete;
 
    private:
        int connect (const Endpoint& endpoint,
                     std::chrono::milliseconds timeout = std::chrono::seconds (5)) noexcept override final
        {
            if (this->_socket.connect (endpoint) == -1)
            {
                if (lastError != Errc::TemporaryError)
                {
                    close ();
                    return -1;
                }
 
                if (!this->_socket.waitConnected (timeout))
                {
                    close ();
                    return -1;
                }
            }
 
            this->_server = endpoint.hostname ().empty () ? endpoint.ip ().toString () : endpoint.hostname ();
            this->_port = endpoint.port ();
 
            if (this->_socket.handshake () == -1)
            {
                if (lastError != Errc::TemporaryError)
                {
                    close ();
                    return -1;
                }
 
                if (!this->_socket.waitHandshake (timeout))
                {
                    close ();
                    return -1;
                }
            }
 
            this->_reactor.addHandler (this->_socket.handle (), this);
 
            return 0;
        }
 
        int disconnect (std::chrono::milliseconds timeout = std::chrono::seconds (5)) noexcept override final
        {
            if (this->_socket.handle () == -1)
            {
                return 0;
            }
 
            this->_reactor.delHandler (this->_socket.handle ());
 
            if (this->_socket.shutdown () == -1)
            {
                if ((lastError != Errc::TemporaryError))
                {
                    close ();
                    return -1;
                }
 
                if (!this->_socket.waitShutdown (timeout))
                {
                    close ();
                    return -1;
                }
            }
 
            if (!this->_socket.waitDisconnected (timeout))
            {
                close ();
                return -1;
            }
 
            return 0;
        }
 
        void close () noexcept
        {
            this->_socket.close ();
            _offset = 0;
            _size = 0;
        }
 
        bool needReconnection () noexcept override final
        {
            return !this->_socket.encrypted () || this->_socket.remoteEndpoint ().ip ().isWildcard ();
        }
 
        ssize_t read (char* data, size_t maxSize) noexcept override final
        {
            if (_offset < _frameHeaderSize)
            {
                ssize_t nread = this->_socket.read (data + _offset, _frameHeaderSize - _offset);
                if (nread == -1)
                {
                    if (lastError != Errc::TemporaryError)
                    {
                        _offset = 0;
                        _size = 0;
                    }
                    return -1;
                }
 
                _offset += static_cast<size_t> (nread);
 
                if (_offset < _frameHeaderSize)
                {
                    lastError = make_error_code (Errc::TemporaryError);
                    return -1;
                }
 
                _size = ntohs (*reinterpret_cast<uint16_t*> (data));
 
                if (_size > maxSize)
                {
                    lastError = make_error_code (Errc::MessageTooLong);
                    _offset = 0;
                    _size = 0;
                    return -1;
                }
            }
 
            ssize_t nread =
                this->_socket.read (data + (_offset - _frameHeaderSize), _size - (_offset - _frameHeaderSize));
            if (nread == -1)
            {
                if (lastError != Errc::TemporaryError)
                {
                    _offset = 0;
                    _size = 0;
                }
                return -1;
            }
 
            _offset += static_cast<size_t> (nread);
 
            if (_offset < (_size + _frameHeaderSize))
            {
                lastError = make_error_code (Errc::TemporaryError);
                return -1;
            }
 
            int msgLen = static_cast<int> (_size);
            _offset = 0;
            _size = 0;
 
            return msgLen;
        }
 
        ssize_t write (const char* data, size_t size) noexcept override final
        {
            uint16_t msgLength = htons (static_cast<uint16_t> (size));
            const char* p = reinterpret_cast<const char*> (&msgLength);
            size_t remaining = sizeof (msgLength);
 
            while (remaining > 0)
            {
                ssize_t result = this->_socket.write (p, remaining);
                if (result == -1)
                {
                    if (lastError == Errc::TemporaryError)
                    {
                        if (this->_socket.waitReadyWrite ())
                            continue;
                    }
                    return -1;
                }
                p += result;
                remaining -= result;
            }
 
            p = data;
            remaining = size;
 
            while (remaining > 0)
            {
                ssize_t result = this->_socket.write (p, remaining);
                if (result == -1)
                {
                    if (lastError == Errc::TemporaryError)
                    {
                        if (this->_socket.waitReadyWrite ())
                            continue;
                    }
                    return -1;
                }
                p += result;
                remaining -= result;
            }
 
            return static_cast<int> (size);
        }
 
        static constexpr size_t _frameHeaderSize = 2;
 
        size_t _size = 0;
 
        size_t _offset = 0;
    };
}
 
#endif