stream_socket.hpp

Go to the documentation of this file.

 
#ifndef JOIN_CORE_STREAM_SOCKET_HPP
#define JOIN_CORE_STREAM_SOCKET_HPP
 
// libjoin.
#include <join/raw_socket.hpp>
 
// C++.
#include <chrono>
 
// C.
#include <sys/socket.h>
 
namespace join
{
    template <class Protocol>
    class BasicStreamSocket final : public BasicRawSocket<Protocol>
    {
        template <class P, class E, size_t N>
        friend class BasicAsyncStreamSocket;
 
    public:
        using Ptr = std::unique_ptr<BasicStreamSocket<Protocol>>;
        using Mode = typename BasicRawSocket<Protocol>::Mode;
        using Option = typename BasicRawSocket<Protocol>::Option;
        using State = typename BasicRawSocket<Protocol>::State;
        using Endpoint = typename Protocol::Endpoint;
        using TimePoint = typename BasicRawSocket<Protocol>::TimePoint;
 
        BasicStreamSocket () noexcept
        : BasicStreamSocket (Mode::NonBlocking)
        {
        }
 
        explicit BasicStreamSocket (Mode mode) noexcept
        : BasicRawSocket<Protocol> (mode)
        {
        }
 
        explicit BasicStreamSocket (int fd, const Endpoint& remote, Mode mode = Mode::NonBlocking)
        : BasicRawSocket<Protocol> (mode)
        {
            this->_handle = fd;
            this->_state = State::Connected;
            this->_protocol = remote.protocol ();
            _remote = remote;
 
            if (this->protocol () == IPPROTO_TCP)
            {
                setOption (Option::NoDelay, 1);
            }
        }
 
        BasicStreamSocket (const BasicStreamSocket& other) = delete;
 
        BasicStreamSocket& operator= (const BasicStreamSocket& other) = delete;
 
        BasicStreamSocket (BasicStreamSocket&& other) noexcept
        : BasicRawSocket<Protocol> (std::move (other))
        , _remote (std::move (other._remote))
        {
        }
 
        BasicStreamSocket& operator= (BasicStreamSocket&& other) noexcept
        {
            BasicRawSocket<Protocol>::operator= (std::move (other));
 
            _remote = std::move (other._remote);
 
            return *this;
        }
 
        ~BasicStreamSocket () = default;
 
        int connect (const Endpoint& endpoint) noexcept
        {
            if ((this->_state != State::Closed) && (this->_state != State::Disconnected))
            {
                lastError = make_error_code (Errc::InUse);
                return -1;
            }
 
            if ((this->_state == State::Closed) && (this->open (endpoint.protocol ()) == -1))
            {
                return -1;
            }
 
            int result = ::connect (this->_handle, endpoint.addr (), endpoint.length ());
 
            this->_state = State::Connecting;
            _remote = endpoint;
 
            if (result == -1)
            {
                lastError = std::error_code (errno, std::generic_category ());
                if (lastError != std::errc::operation_in_progress)
                {
                    close ();
                }
                return -1;
            }
 
            this->_state = State::Connected;
 
            return 0;
        }
 
        bool waitConnected ()
        {
            return waitConnected (TimePoint::max ());
        }
 
        bool waitConnected (std::chrono::nanoseconds timeout)
        {
            return waitConnected (std::chrono::steady_clock::now () + timeout);
        }
 
        bool waitConnected (TimePoint deadline)
        {
            if (this->_state != State::Connected)
            {
                if (this->_state != State::Connecting)
                {
                    lastError = make_error_code (Errc::OperationFailed);
                    return false;
                }
 
                if (this->waitUntil (false, true, deadline) == -1)
                {
                    return false;
                }
 
                return connected ();
            }
 
            return true;
        }
 
        int disconnect () noexcept
        {
            if (this->_state == State::Connected)
            {
                ::shutdown (this->_handle, SHUT_WR);
                this->_state = State::Disconnecting;
            }
 
            if (this->_state == State::Disconnecting)
            {
                char buffer[4096];
                // closing before reading can make the client
                // not see all of our output.
                // we have to do a "lingering close"
                for (;;)
                {
                    int result = this->read (buffer, sizeof (buffer));
                    if (result <= 0)
                    {
                        if ((result == -1) && (lastError == Errc::TemporaryError))
                        {
                            return -1;
                        }
 
                        break;
                    }
                }
 
                ::shutdown (this->_handle, SHUT_RD);
                this->_state = State::Disconnected;
            }
 
            return 0;
        }
 
        bool waitDisconnected ()
        {
            return waitDisconnected (TimePoint::max ());
        }
 
        bool waitDisconnected (std::chrono::nanoseconds timeout)
        {
            return waitDisconnected (std::chrono::steady_clock::now () + timeout);
        }
 
        bool waitDisconnected (TimePoint deadline)
        {
            if (JOIN_UNLIKELY ((deadline != TimePoint::max ()) && (this->_mode == Mode::Blocking)))
            {
                lastError = make_error_code (Errc::OperationFailed);
                return false;
            }
 
            if ((this->_state != State::Disconnected) && (this->_state != State::Closed))
            {
                if (this->_state != State::Disconnecting)
                {
                    lastError = make_error_code (Errc::OperationFailed);
                    return false;
                }
 
                while (lastError == Errc::TemporaryError)
                {
                    if (this->waitUntil (true, false, deadline) == -1)
                    {
                        return false;
                    }
 
                    if (disconnect () == 0)
                    {
                        return true;
                    }
                }
 
                return false;
            }
 
            return true;
        }
 
        void close () noexcept override
        {
            BasicRawSocket<Protocol>::close ();
            _remote = {};
        }
 
        ssize_t read (char* data, size_t maxSize) noexcept
        {
            ssize_t size = BasicRawSocket<Protocol>::read (data, maxSize);
            if (size == 0)
            {
                lastError = make_error_code (Errc::ConnectionClosed);
                return -1;
            }
 
            return size;
        }
 
        int readExactly (char* data, size_t size) noexcept
        {
            return readExactly (data, size, TimePoint::max ());
        }
 
        int readExactly (char* data, size_t size, std::chrono::nanoseconds timeout) noexcept
        {
            return readExactly (data, size, std::chrono::steady_clock::now () + timeout);
        }
 
        int readExactly (char* data, size_t size, TimePoint deadline) noexcept
        {
            if (JOIN_UNLIKELY ((deadline != TimePoint::max ()) && (this->_mode == Mode::Blocking)))
            {
                lastError = make_error_code (Errc::OperationFailed);
                return -1;
            }
 
            size_t numRead = 0;
 
            while (numRead < size)
            {
                ssize_t result = this->read (data + numRead, size - numRead);
                if (result == -1)
                {
                    if (lastError == Errc::TemporaryError)
                    {
                        if (this->waitUntil (true, false, deadline) == 0)
                        {
                            continue;
                        }
                    }
 
                    return -1;
                }
 
                numRead += result;
            }
 
            return 0;
        }
 
        int writeExactly (const char* data, size_t size) noexcept
        {
            return writeExactly (data, size, TimePoint::max ());
        }
 
        int writeExactly (const char* data, size_t size, std::chrono::nanoseconds timeout) noexcept
        {
            return writeExactly (data, size, std::chrono::steady_clock::now () + timeout);
        }
 
        int writeExactly (const char* data, size_t size, TimePoint deadline) noexcept
        {
            if (JOIN_UNLIKELY ((deadline != TimePoint::max ()) && (this->_mode == Mode::Blocking)))
            {
                lastError = make_error_code (Errc::OperationFailed);
                return -1;
            }
 
            size_t numWrite = 0;
 
            while (numWrite < size)
            {
                ssize_t result = this->write (data + numWrite, size - numWrite);
                if (result == -1)
                {
                    if (lastError == Errc::TemporaryError)
                    {
                        if (this->waitUntil (false, true, deadline) == 0)
                        {
                            continue;
                        }
                    }
 
                    return -1;
                }
 
                numWrite += result;
            }
 
            return 0;
        }
 
        int setOption (Option option, int value) noexcept override
        {
            if (this->_state == State::Closed)
            {
                lastError = make_error_code (Errc::OperationFailed);
                return -1;
            }
 
            int optlevel, optname;
 
            switch (option)
            {
                case Option::NoDelay:
                    optlevel = IPPROTO_TCP;
                    optname = TCP_NODELAY;
                    break;
 
                case Option::KeepIdle:
                    optlevel = IPPROTO_TCP;
                    optname = TCP_KEEPIDLE;
                    break;
 
                case Option::KeepIntvl:
                    optlevel = IPPROTO_TCP;
                    optname = TCP_KEEPINTVL;
                    break;
 
                case Option::KeepCount:
                    optlevel = IPPROTO_TCP;
                    optname = TCP_KEEPCNT;
                    break;
 
                default:
                    return BasicRawSocket<Protocol>::setOption (option, value);
            }
 
            int result = ::setsockopt (this->_handle, optlevel, optname, &value, sizeof (value));
            if (result == -1)
            {
                lastError = std::error_code (errno, std::generic_category ());
                return -1;
            }
 
            return 0;
        }
 
        const Endpoint& remoteEndpoint () const noexcept
        {
            if (_remote == Endpoint ())
            {
                struct sockaddr_storage sa;
                socklen_t sa_len = sizeof (struct sockaddr_storage);
 
                if (::getpeername (this->_handle, reinterpret_cast<struct sockaddr*> (&sa), &sa_len) != -1)
                {
                    _remote = Endpoint (reinterpret_cast<struct sockaddr*> (&sa), sa_len);
                }
            }
 
            return _remote;
        }
 
        bool connecting () const noexcept
        {
            return (this->_state == State::Connecting);
        }
 
        bool connected () noexcept
        {
            if (this->_state == State::Connected)
            {
                return true;
            }
            else if (this->_state != State::Connecting)
            {
                return false;
            }
 
            int optval;
            socklen_t optlen = sizeof (optval);
 
            int result = ::getsockopt (this->_handle, SOL_SOCKET, SO_ERROR, &optval, &optlen);
            if ((result == -1) || (optval != 0))
            {
                return false;
            }
 
            this->_state = State::Connected;
 
            return true;
        }
 
        int mtu () const noexcept
        {
            if (this->_state == State::Closed)
            {
                lastError = make_error_code (Errc::OperationFailed);
                return -1;
            }
 
            int result = -1, value = -1;
            socklen_t valueLen = sizeof (value);
 
            if (this->family () == AF_INET6)
            {
                result = ::getsockopt (this->_handle, IPPROTO_IPV6, IPV6_MTU, &value, &valueLen);
            }
            else if (this->family () == AF_INET)
            {
                result = ::getsockopt (this->_handle, IPPROTO_IP, IP_MTU, &value, &valueLen);
            }
            else
            {
                lastError = make_error_code (Errc::OperationFailed);
                return -1;
            }
 
            if (result == -1)
            {
                lastError = std::error_code (errno, std::generic_category ());
                return -1;
            }
 
            return value;
        }
 
    protected:
        mutable Endpoint _remote;
    };
 
    template <class Protocol>
    inline bool operator< (const BasicStreamSocket<Protocol>& a, const BasicStreamSocket<Protocol>& b) noexcept
    {
        return a.handle () < b.handle ();
    }
}
 
#endif