async_stream_socket.hpp

Go to the documentation of this file.

 
#ifndef JOIN_CORE_ASYNC_STREAM_SOCKET_HPP
#define JOIN_CORE_ASYNC_STREAM_SOCKET_HPP
 
// libjoin.
#include <join/async_raw_socket.hpp>
#include <join/stream_socket.hpp>
 
// C++.
#include <system_error>
#include <utility>
 
namespace join
{
    template <class Protocol, class ProactorType, size_t OpCount>
    class BasicAsyncStreamSocket : public BasicAsyncRawSocket<Protocol, ProactorType, OpCount>
    {
    public:
        using Socket = typename Protocol::Socket;
        using Endpoint = typename Protocol::Endpoint;
        using AsyncWrite = BasicAsyncWrite<Protocol, ProactorType>;
        using ConnectHandler = typename AsyncWrite::Connect;
 
        explicit BasicAsyncStreamSocket (ProactorType& proactor = ProactorThread::proactor ())
        : BasicAsyncRawSocket<Protocol, ProactorType, OpCount> (proactor)
        {
        }
 
        explicit BasicAsyncStreamSocket (Socket&& sock, ProactorType& proactor = ProactorThread::proactor ())
        : BasicAsyncRawSocket<Protocol, ProactorType, OpCount> (std::move (sock), proactor)
        {
        }
 
        BasicAsyncStreamSocket (const BasicAsyncStreamSocket& other) = delete;
 
        BasicAsyncStreamSocket& operator= (const BasicAsyncStreamSocket& other) = delete;
 
        BasicAsyncStreamSocket (BasicAsyncStreamSocket&& other) noexcept = default;
 
        BasicAsyncStreamSocket& operator= (BasicAsyncStreamSocket&& other) noexcept = default;
 
        int asyncConnect (ConnectHandler handler, const Endpoint& endpoint) noexcept
        {
            if (JOIN_UNLIKELY (!this->_allocator.hasBackend ()))
            {
                lastError = make_error_code (Errc::OperationFailed);
                return -1;
            }
 
            if (this->_socket.connected () || this->_socket.connecting ())
            {
                lastError = make_error_code (Errc::InUse);
                return -1;
            }
 
            if (!this->_socket.opened () && (this->_socket.open (endpoint.protocol ()) == -1))
            {
                return -1;  // LCOV_EXCL_LINE
            }
 
            AsyncWrite* connect = this->allocateWrite ();
            if (JOIN_UNLIKELY (connect == nullptr))
            {
                // LCOV_EXCL_START
                lastError = make_error_code (Errc::OutOfMemory);
                return -1;
                // LCOV_EXCL_STOP
            }
 
            this->_socket._state = Socket::Connecting;
            this->_socket._remote = endpoint;
            connect->connectHandler = std::move (handler);
            connect->op = IoOperation::makeConnect (this->_socket.handle (), this->_socket._remote.addr (),
                                                    this->_socket._remote.length (), this);
            connect->op.state.store (IoOperation::State::Submitted, std::memory_order_release);
 
            if (this->_proactor->submit (connect->op, true, false) == -1)
            {
                // LCOV_EXCL_START
                this->releaseOperation (connect);
                this->_socket.close ();
                return -1;
                // LCOV_EXCL_STOP
            }
 
            return 0;
        }
 
        const Endpoint& remoteEndpoint () const noexcept
        {
            return this->_socket.remoteEndpoint ();
        }
 
        bool connected () noexcept
        {
            return this->_socket.connected ();
        }
 
        bool connecting () const noexcept
        {
            return this->_socket.connecting ();
        }
 
        int mtu () const noexcept
        {
            return this->_socket.mtu ();
        }
    };
}
 
#endif