Socket< T > Class Template Reference

#include "Socket.h"

Inheritance diagram for Socket< T >:

Public Member Functions
	Socket (tcp::socket &&socket)
virtual	~Socket ()
virtual void	Start ()=0
virtual bool	Update ()
boost::asio::ip::address	GetRemoteIpAddress () const
uint16	GetRemotePort () const
void	AsyncRead ()
void	AsyncReadProxyHeader ()
void	AsyncReadWithCallback (void(T::*callback)(boost::system::error_code, std::size_t))
void	QueuePacket (MessageBuffer &&buffer)
ProxyHeaderReadingState	GetProxyHeaderReadingState () const
bool	IsOpen () const
void	CloseSocket ()
void	DelayedCloseSocket ()
	Marks the socket for closing after write buffer becomes empty. More...
MessageBuffer &	GetReadBuffer ()

Protected Member Functions
virtual void	OnClose ()
virtual void	ReadHandler ()=0
bool	AsyncProcessQueue ()
void	SetNoDelay (bool enable)

Private Member Functions
void	ReadHandlerInternal (boost::system::error_code error, std::size_t transferredBytes)
void	ProxyReadHeaderHandler (boost::system::error_code error, std::size_t transferredBytes)
void	WriteHandlerWrapper (boost::system::error_code, std::size_t)
bool	HandleQueue ()

Private Attributes
tcp::socket	_socket
boost::asio::ip::address	_remoteAddress
uint16	_remotePort
MessageBuffer	_readBuffer
std::queue< MessageBuffer >	_writeQueue
std::atomic< bool >	_closed
std::atomic< bool >	_closing
bool	_isWritingAsync
ProxyHeaderReadingState	_proxyHeaderReadingState

Detailed Description

template<class T>
class Socket< T >

Constructor & Destructor Documentation

Socket()

template<class T >

Socket< T >::Socket ( tcp::socket &&  socket )

inlineexplicit

                                        : _socket(std::move(socket)), _remoteAddress(_socket.remote_endpoint().address()),
        _remotePort(_socket.remote_endpoint().port()), _readBuffer(), _closed(false), _closing(false), _isWritingAsync(false),
        _proxyHeaderReadingState(PROXY_HEADER_READING_STATE_NOT_STARTED)
    {
        _readBuffer.Resize(READ_BLOCK_SIZE);
    }

References Socket< T >::_readBuffer, READ_BLOCK_SIZE, and MessageBuffer::Resize().

~Socket()

template<class T >

virtual Socket< T >::~Socket ( )

inlinevirtual

    {
        _closed = true;
        boost::system::error_code error;
        _socket.close(error);
    }

References Socket< T >::_closed, and Socket< T >::_socket.

Member Function Documentation

AsyncProcessQueue()

template<class T >

bool Socket< T >::AsyncProcessQueue ( )

inlineprotected

    {
        if (_isWritingAsync)
            return false;
 
        _isWritingAsync = true;
 
#ifdef AC_SOCKET_USE_IOCP
        MessageBuffer& buffer = _writeQueue.front();
        _socket.async_write_some(boost::asio::buffer(buffer.GetReadPointer(), buffer.GetActiveSize()), std::bind(&Socket<T>::WriteHandler,
            this->shared_from_this(), std::placeholders::_1, std::placeholders::_2));
#else
        _socket.async_write_some(boost::asio::null_buffers(), std::bind(&Socket<T>::WriteHandlerWrapper,
            this->shared_from_this(), std::placeholders::_1, std::placeholders::_2));
#endif
        return false;
    }

References Socket< T >::_isWritingAsync, Socket< T >::_socket, Socket< T >::_writeQueue, MessageBuffer::GetActiveSize(), and MessageBuffer::GetReadPointer().

Referenced by Socket< T >::HandleQueue(), and Socket< T >::QueuePacket().

AsyncRead()

template<class T >

void Socket< T >::AsyncRead ( )

inline

    {
        if (!IsOpen())
        {
            return;
        }
 
        _readBuffer.Normalize();
        _readBuffer.EnsureFreeSpace();
        _socket.async_read_some(boost::asio::buffer(_readBuffer.GetWritePointer(), _readBuffer.GetRemainingSpace()),
            std::bind(&Socket<T>::ReadHandlerInternal, this->shared_from_this(), std::placeholders::_1, std::placeholders::_2));
    }

References Socket< T >::_readBuffer, Socket< T >::_socket, MessageBuffer::EnsureFreeSpace(), MessageBuffer::GetRemainingSpace(), MessageBuffer::GetWritePointer(), Socket< T >::IsOpen(), and MessageBuffer::Normalize().

AsyncReadProxyHeader()

template<class T >

void Socket< T >::AsyncReadProxyHeader ( )

inline

    {
        if (!IsOpen())
        {
            return;
        }
 
        _proxyHeaderReadingState = PROXY_HEADER_READING_STATE_STARTED;
 
        _readBuffer.Normalize();
        _readBuffer.EnsureFreeSpace();
        _socket.async_read_some(boost::asio::buffer(_readBuffer.GetWritePointer(), _readBuffer.GetRemainingSpace()),
            std::bind(&Socket<T>::ProxyReadHeaderHandler, this->shared_from_this(), std::placeholders::_1, std::placeholders::_2));
    }

References Socket< T >::_proxyHeaderReadingState, Socket< T >::_readBuffer, Socket< T >::_socket, MessageBuffer::EnsureFreeSpace(), MessageBuffer::GetRemainingSpace(), MessageBuffer::GetWritePointer(), Socket< T >::IsOpen(), MessageBuffer::Normalize(), and PROXY_HEADER_READING_STATE_STARTED.

Referenced by Socket< T >::ProxyReadHeaderHandler().

AsyncReadWithCallback()

template<class T >

void Socket< T >::AsyncReadWithCallback ( void(T::*)(boost::system::error_code, std::size_t)  callback )

inline

    {
        if (!IsOpen())
        {
            return;
        }
 
        _readBuffer.Normalize();
        _readBuffer.EnsureFreeSpace();
 
        _socket.async_read_some(boost::asio::buffer(_readBuffer.GetWritePointer(), _readBuffer.GetRemainingSpace()),
            std::bind(callback, this->shared_from_this(), std::placeholders::_1, std::placeholders::_2));
    }

References Socket< T >::_readBuffer, Socket< T >::_socket, MessageBuffer::EnsureFreeSpace(), MessageBuffer::GetRemainingSpace(), MessageBuffer::GetWritePointer(), Socket< T >::IsOpen(), and MessageBuffer::Normalize().

CloseSocket()

template<class T >

void Socket< T >::CloseSocket ( )

inline

    {
        if (_closed.exchange(true))
            return;
 
        boost::system::error_code shutdownError;
        _socket.shutdown(boost::asio::socket_base::shutdown_send, shutdownError);
 
        if (shutdownError)
            LOG_DEBUG("network", "Socket::CloseSocket: {} errored when shutting down socket: {} ({})", GetRemoteIpAddress().to_string(),
                shutdownError.value(), shutdownError.message());
 
        OnClose();
    }

References Socket< T >::_closed, Socket< T >::_socket, Socket< T >::GetRemoteIpAddress(), LOG_DEBUG, and Socket< T >::OnClose().

Referenced by Socket< T >::HandleQueue(), Socket< T >::ProxyReadHeaderHandler(), and Socket< T >::ReadHandlerInternal().

DelayedCloseSocket()

template<class T >

void Socket< T >::DelayedCloseSocket ( )

inline

Marks the socket for closing after write buffer becomes empty.

{ _closing = true; }

References Socket< T >::_closing.

GetProxyHeaderReadingState()

template<class T >

ProxyHeaderReadingState Socket< T >::GetProxyHeaderReadingState ( ) const

inline

{ return _proxyHeaderReadingState; }

References Socket< T >::_proxyHeaderReadingState.

GetReadBuffer()

template<class T >

MessageBuffer & Socket< T >::GetReadBuffer ( )

inline

{ return _readBuffer; }

References Socket< T >::_readBuffer.

Referenced by Socket< T >::ProxyReadHeaderHandler().

GetRemoteIpAddress()

template<class T >

boost::asio::ip::address Socket< T >::GetRemoteIpAddress ( ) const

inline

    {
        return _remoteAddress;
    }

References Socket< T >::_remoteAddress.

Referenced by Socket< T >::CloseSocket(), Socket< T >::ProxyReadHeaderHandler(), and Socket< T >::SetNoDelay().

GetRemotePort()

template<class T >

uint16 Socket< T >::GetRemotePort ( ) const

inline

    {
        return _remotePort;
    }

References Socket< T >::_remotePort.

HandleQueue()

template<class T >

bool Socket< T >::HandleQueue ( )

inlineprivate

    {
        if (_writeQueue.empty())
            return false;
 
        MessageBuffer& queuedMessage = _writeQueue.front();
 
        std::size_t bytesToSend = queuedMessage.GetActiveSize();
 
        boost::system::error_code error;
        std::size_t bytesSent = _socket.write_some(boost::asio::buffer(queuedMessage.GetReadPointer(), bytesToSend), error);
 
        if (error)
        {
            if (error == boost::asio::error::would_block || error == boost::asio::error::try_again)
            {
                return AsyncProcessQueue();
            }
 
            _writeQueue.pop();
 
            if (_closing && _writeQueue.empty())
            {
                CloseSocket();
            }
 
            return false;
        }
        else if (bytesSent == 0)
        {
            _writeQueue.pop();
 
            if (_closing && _writeQueue.empty())
            {
                CloseSocket();
            }
 
            return false;
        }
        else if (bytesSent < bytesToSend) // now n > 0
        {
            queuedMessage.ReadCompleted(bytesSent);
            return AsyncProcessQueue();
        }
 
        _writeQueue.pop();
 
        if (_closing && _writeQueue.empty())
        {
            CloseSocket();
        }
 
        return !_writeQueue.empty();
    }

References Socket< T >::_closing, Socket< T >::_socket, Socket< T >::_writeQueue, Socket< T >::AsyncProcessQueue(), Socket< T >::CloseSocket(), MessageBuffer::GetActiveSize(), MessageBuffer::GetReadPointer(), and MessageBuffer::ReadCompleted().

Referenced by Socket< T >::Update(), and Socket< T >::WriteHandlerWrapper().

IsOpen()

template<class T >

bool Socket< T >::IsOpen ( ) const

inline

{ return !_closed && !_closing; }

References Socket< T >::_closed, and Socket< T >::_closing.

Referenced by Socket< T >::AsyncRead(), Socket< T >::AsyncReadProxyHeader(), and Socket< T >::AsyncReadWithCallback().

OnClose()

template<class T >

virtual void Socket< T >::OnClose ( )

inlineprotectedvirtual

Reimplemented in WorldSocket.

{ }

Referenced by Socket< T >::CloseSocket().

ProxyReadHeaderHandler()

template<class T >

void Socket< T >::ProxyReadHeaderHandler ( boost::system::error_code  error, std::size_t  transferredBytes  )

inlineprivate

    {
        if (error)
        {
            CloseSocket();
            return;
        }
 
        _readBuffer.WriteCompleted(transferredBytes);
 
        MessageBuffer& packet = GetReadBuffer();
 
        const int minimumProxyProtocolV2Size = 28;
        if (packet.GetActiveSize() < minimumProxyProtocolV2Size)
        {
            AsyncReadProxyHeader();
            return;
        }
 
        uint8* readPointer = packet.GetReadPointer();
 
        const uint8 signatureSize = 12;
        const uint8 expectedSignature[signatureSize] = {0x0D, 0x0A, 0x0D, 0x0A, 0x00, 0x0D, 0x0A, 0x51, 0x55, 0x49, 0x54, 0x0A};
        if (memcmp(packet.GetReadPointer(), expectedSignature, signatureSize) != 0)
        {
            _proxyHeaderReadingState = PROXY_HEADER_READING_STATE_FAILED;
            LOG_ERROR("network", "Socket::ProxyReadHeaderHandler: received bad PROXY Protocol v2 signature for {}", GetRemoteIpAddress().to_string());
            return;
        }
 
        const uint8 version = (readPointer[signatureSize] & 0xF0) >> 4;
        const uint8 command = (readPointer[signatureSize] & 0xF);
 
        if (version != 2)
        {
            _proxyHeaderReadingState = PROXY_HEADER_READING_STATE_FAILED;
            LOG_ERROR("network", "Socket::ProxyReadHeaderHandler: received bad PROXY Protocol v2 signature for {}", GetRemoteIpAddress().to_string());
            return;
        }
 
        const uint8 addressFamily = readPointer[13];
        const uint16 len = (readPointer[14] << 8) | readPointer[15];
        if (static_cast<size_t>(len+16) > packet.GetActiveSize())
        {
            AsyncReadProxyHeader();
            return;
        }
 
        // Connection created by a proxy itself (health checks?), ignore and do nothing.
        if (command == 0)
        {
            packet.ReadCompleted(len+16);
            _proxyHeaderReadingState = PROXY_HEADER_READING_STATE_FINISHED;
            return;
        }
 
        auto remainingLen = packet.GetActiveSize() - 16;
        readPointer += 16; // Skip strait to address.
 
        switch (addressFamily) {
            case PROXY_HEADER_ADDRESS_FAMILY_AND_PROTOCOL_TCP_V4:
            {
                if (remainingLen < 12)
                {
                    AsyncReadProxyHeader();
                    return;
                }
 
                boost::asio::ip::address_v4::bytes_type b;
                auto addressSize = sizeof(b);
 
                std::copy(readPointer, readPointer+addressSize, b.begin());
                _remoteAddress = boost::asio::ip::address_v4(b);
 
                readPointer += 2 * addressSize; // Skip server address.
                _remotePort = (readPointer[0] << 8) | readPointer[1];
 
                break;
            }
 
            case PROXY_HEADER_ADDRESS_FAMILY_AND_PROTOCOL_TCP_V6:
            {
                if (remainingLen < 36)
                {
                    AsyncReadProxyHeader();
                    return;
                }
 
                boost::asio::ip::address_v6::bytes_type b;
                auto addressSize = sizeof(b);
 
                std::copy(readPointer, readPointer+addressSize, b.begin());
                _remoteAddress = boost::asio::ip::address_v6(b);
 
                readPointer += 2 * addressSize; // Skip server address.
                _remotePort = (readPointer[0] << 8) | readPointer[1];
 
                break;
            }
 
            default:
                _proxyHeaderReadingState = PROXY_HEADER_READING_STATE_FAILED;
                LOG_ERROR("network", "Socket::ProxyReadHeaderHandler: unsupported address family type {}", GetRemoteIpAddress().to_string());
                return;
        }
 
        packet.ReadCompleted(len+16);
        _proxyHeaderReadingState = PROXY_HEADER_READING_STATE_FINISHED;
    }

References Socket< T >::_proxyHeaderReadingState, Socket< T >::_readBuffer, Socket< T >::_remoteAddress, Socket< T >::_remotePort, Socket< T >::AsyncReadProxyHeader(), Socket< T >::CloseSocket(), MessageBuffer::GetActiveSize(), Socket< T >::GetReadBuffer(), MessageBuffer::GetReadPointer(), Socket< T >::GetRemoteIpAddress(), LOG_ERROR, PROXY_HEADER_ADDRESS_FAMILY_AND_PROTOCOL_TCP_V4, PROXY_HEADER_ADDRESS_FAMILY_AND_PROTOCOL_TCP_V6, PROXY_HEADER_READING_STATE_FAILED, PROXY_HEADER_READING_STATE_FINISHED, MessageBuffer::ReadCompleted(), and MessageBuffer::WriteCompleted().

QueuePacket()

template<class T >

void Socket< T >::QueuePacket ( MessageBuffer &&  buffer )

inline

    {
        _writeQueue.push(std::move(buffer));
 
#ifdef AC_SOCKET_USE_IOCP
        AsyncProcessQueue();
#endif
    }

References Socket< T >::_writeQueue, and Socket< T >::AsyncProcessQueue().

ReadHandler()

template<class T >

virtual void Socket< T >::ReadHandler ( )

protectedpure virtual

Implemented in AuthSession, and WorldSocket.

Referenced by Socket< T >::ReadHandlerInternal().

ReadHandlerInternal()

template<class T >

void Socket< T >::ReadHandlerInternal ( boost::system::error_code  error, std::size_t  transferredBytes  )

inlineprivate

    {
        if (error)
        {
            CloseSocket();
            return;
        }
 
        _readBuffer.WriteCompleted(transferredBytes);
        ReadHandler();
    }

References Socket< T >::_readBuffer, Socket< T >::CloseSocket(), Socket< T >::ReadHandler(), and MessageBuffer::WriteCompleted().

SetNoDelay()

template<class T >

void Socket< T >::SetNoDelay ( bool  enable )

inlineprotected

    {
        boost::system::error_code err;
        _socket.set_option(tcp::no_delay(enable), err);
 
        if (err)
            LOG_DEBUG("network", "Socket::SetNoDelay: failed to set_option(boost::asio::ip::tcp::no_delay) for {} - {} ({})",
                GetRemoteIpAddress().to_string(), err.value(), err.message());
    }

References Socket< T >::_socket, Socket< T >::GetRemoteIpAddress(), and LOG_DEBUG.

Start()

template<class T >

virtual void Socket< T >::Start ( )

pure virtual

Implemented in AuthSession, and WorldSocket.

Update()

template<class T >

virtual bool Socket< T >::Update ( )

inlinevirtual

Reimplemented in AuthSession, and WorldSocket.

    {
        if (_closed)
        {
            return false;
        }
 
#ifndef AC_SOCKET_USE_IOCP
        if (_isWritingAsync || (_writeQueue.empty() && !_closing))
        {
            return true;
        }
 
        for (; HandleQueue();)
            ;
#endif
 
        return true;
    }

References Socket< T >::_closed, Socket< T >::_closing, Socket< T >::_isWritingAsync, Socket< T >::_writeQueue, and Socket< T >::HandleQueue().

WriteHandlerWrapper()

template<class T >

void Socket< T >::WriteHandlerWrapper ( boost::system::error_code  , std::size_t    )

inlineprivate

    {
        _isWritingAsync = false;
        HandleQueue();
    }

References Socket< T >::_isWritingAsync, and Socket< T >::HandleQueue().

Member Data Documentation

_closed

template<class T >

std::atomic<bool> Socket< T >::_closed

private

Referenced by Socket< T >::CloseSocket(), Socket< T >::IsOpen(), Socket< T >::Update(), and Socket< T >::~Socket().

_closing

template<class T >

std::atomic<bool> Socket< T >::_closing

private

Referenced by Socket< T >::DelayedCloseSocket(), Socket< T >::HandleQueue(), Socket< T >::IsOpen(), and Socket< T >::Update().

_isWritingAsync

template<class T >

bool Socket< T >::_isWritingAsync

private

Referenced by Socket< T >::AsyncProcessQueue(), Socket< T >::Update(), and Socket< T >::WriteHandlerWrapper().

_proxyHeaderReadingState

template<class T >

ProxyHeaderReadingState Socket< T >::_proxyHeaderReadingState

private

Referenced by Socket< T >::AsyncReadProxyHeader(), Socket< T >::GetProxyHeaderReadingState(), and Socket< T >::ProxyReadHeaderHandler().

_readBuffer

template<class T >

MessageBuffer Socket< T >::_readBuffer

private

Referenced by Socket< T >::AsyncRead(), Socket< T >::AsyncReadProxyHeader(), Socket< T >::AsyncReadWithCallback(), Socket< T >::GetReadBuffer(), Socket< T >::ProxyReadHeaderHandler(), Socket< T >::ReadHandlerInternal(), and Socket< T >::Socket().

_remoteAddress

template<class T >

boost::asio::ip::address Socket< T >::_remoteAddress

private

Referenced by Socket< T >::GetRemoteIpAddress(), and Socket< T >::ProxyReadHeaderHandler().

_remotePort

template<class T >

uint16 Socket< T >::_remotePort

private

Referenced by Socket< T >::GetRemotePort(), and Socket< T >::ProxyReadHeaderHandler().

_socket

template<class T >

tcp::socket Socket< T >::_socket

private

Referenced by Socket< T >::AsyncProcessQueue(), Socket< T >::AsyncRead(), Socket< T >::AsyncReadProxyHeader(), Socket< T >::AsyncReadWithCallback(), Socket< T >::CloseSocket(), Socket< T >::HandleQueue(), Socket< T >::SetNoDelay(), and Socket< T >::~Socket().

_writeQueue

template<class T >

std::queue<MessageBuffer> Socket< T >::_writeQueue

private

Referenced by Socket< T >::AsyncProcessQueue(), Socket< T >::HandleQueue(), Socket< T >::QueuePacket(), and Socket< T >::Update().