NetworkThread< SocketType > Class Template Reference

#include "NetworkThread.h"

Public Member Functions
	NetworkThread ()
virtual	~NetworkThread ()
void	Stop ()
bool	Start ()
void	Wait ()
int32	GetConnectionCount () const
virtual void	AddSocket (std::shared_ptr< SocketType > sock)
tcp::socket *	GetSocketForAccept ()
void	EnableProxyProtocol ()

Protected Member Functions
virtual void	SocketAdded (std::shared_ptr< SocketType >)
virtual void	SocketRemoved (std::shared_ptr< SocketType >)
void	AddNewSockets ()
void	HandleNewSocketsProxyReadingOnConnect ()
void	Run ()
void	Update ()

Private Types
using	SocketContainer = std::vector< std::shared_ptr< SocketType > >

Private Attributes
std::atomic< int32 >	_connections {}
std::atomic< bool >	_stopped {}
std::unique_ptr< std::thread >	_thread
SocketContainer	_sockets
std::mutex	_newSocketsLock
SocketContainer	_newSockets
Acore::Asio::IoContext	_ioContext
tcp::socket	_acceptSocket
Acore::Asio::DeadlineTimer	_updateTimer
bool	_proxyHeaderReadingEnabled

Detailed Description

template<class SocketType>
class NetworkThread< SocketType >

Member Typedef Documentation

SocketContainer

template<class SocketType >

using NetworkThread< SocketType >::SocketContainer = std::vector<std::shared_ptr<SocketType> >

private

Constructor & Destructor Documentation

NetworkThread()

template<class SocketType >

NetworkThread< SocketType >::NetworkThread ( )

inline

                    :
        _ioContext(1), _acceptSocket(_ioContext), _updateTimer(_ioContext), _proxyHeaderReadingEnabled(false) { }

~NetworkThread()

template<class SocketType >

virtual NetworkThread< SocketType >::~NetworkThread ( )

inlinevirtual

    {
        Stop();
 
        if (_thread)
        {
            Wait();
        }
    }

References NetworkThread< SocketType >::_thread, NetworkThread< SocketType >::Stop(), and NetworkThread< SocketType >::Wait().

Member Function Documentation

AddNewSockets()

template<class SocketType >

void NetworkThread< SocketType >::AddNewSockets ( )

inlineprotected

    {
        std::lock_guard<std::mutex> lock(_newSocketsLock);
 
        if (_newSockets.empty())
            return;
 
        if (!_proxyHeaderReadingEnabled)
        {
            for (std::shared_ptr<SocketType> sock : _newSockets)
            {
                if (!sock->IsOpen())
                {
                    SocketRemoved(sock);
                    --_connections;
                    continue;
                }
 
                _sockets.emplace_back(sock);
 
                sock->Start();
            }
 
            _newSockets.clear();
        }
        else
        {
            HandleNewSocketsProxyReadingOnConnect();
        }
    }

References NetworkThread< SocketType >::_connections, NetworkThread< SocketType >::_newSockets, NetworkThread< SocketType >::_newSocketsLock, NetworkThread< SocketType >::_proxyHeaderReadingEnabled, NetworkThread< SocketType >::_sockets, NetworkThread< SocketType >::HandleNewSocketsProxyReadingOnConnect(), and NetworkThread< SocketType >::SocketRemoved().

Referenced by NetworkThread< SocketType >::Update().

AddSocket()

template<class SocketType >

virtual void NetworkThread< SocketType >::AddSocket ( std::shared_ptr< SocketType >  sock )

inlinevirtual

    {
        std::lock_guard<std::mutex> lock(_newSocketsLock);
 
        ++_connections;
        _newSockets.emplace_back(sock);
        SocketAdded(sock);
    }

References NetworkThread< SocketType >::_connections, NetworkThread< SocketType >::_newSockets, NetworkThread< SocketType >::_newSocketsLock, and NetworkThread< SocketType >::SocketAdded().

EnableProxyProtocol()

template<class SocketType >

void NetworkThread< SocketType >::EnableProxyProtocol ( )

inline

{ _proxyHeaderReadingEnabled = true; }

References NetworkThread< SocketType >::_proxyHeaderReadingEnabled.

Referenced by AuthSocketMgr::CreateThreads().

GetConnectionCount()

template<class SocketType >

int32 NetworkThread< SocketType >::GetConnectionCount ( ) const

inline

    {
        return _connections;
    }

References NetworkThread< SocketType >::_connections.

GetSocketForAccept()

template<class SocketType >

tcp::socket * NetworkThread< SocketType >::GetSocketForAccept ( )

inline

{ return &_acceptSocket; }

References NetworkThread< SocketType >::_acceptSocket.

HandleNewSocketsProxyReadingOnConnect()

template<class SocketType >

void NetworkThread< SocketType >::HandleNewSocketsProxyReadingOnConnect ( )

inlineprotected

    {
        std::size_t index = 0;
        std::vector<int> newSocketsToRemoveIndexes;
        for (auto sock_iter = _newSockets.begin(); sock_iter != _newSockets.end(); ++sock_iter, ++index)
        {
            std::shared_ptr<SocketType> sock = *sock_iter;
 
            if (!sock->IsOpen())
            {
                newSocketsToRemoveIndexes.emplace_back(index);
                SocketRemoved(sock);
                --_connections;
                continue;
            }
 
            const auto proxyHeaderReadingState = sock->GetProxyHeaderReadingState();
            if (proxyHeaderReadingState == PROXY_HEADER_READING_STATE_STARTED)
                continue;
 
            switch (proxyHeaderReadingState) {
                case PROXY_HEADER_READING_STATE_NOT_STARTED:
                    sock->AsyncReadProxyHeader();
                    break;
 
                case PROXY_HEADER_READING_STATE_FINISHED:
                    newSocketsToRemoveIndexes.emplace_back(index);
                    _sockets.emplace_back(sock);
 
                    sock->Start();
 
                    break;
 
                default:
                    newSocketsToRemoveIndexes.emplace_back(index);
                    SocketRemoved(sock);
                    --_connections;
                    break;
            }
        }
 
        for (auto it = newSocketsToRemoveIndexes.rbegin(); it != newSocketsToRemoveIndexes.rend(); ++it)
            _newSockets.erase(_newSockets.begin() + *it);
    }

References NetworkThread< SocketType >::_connections, NetworkThread< SocketType >::_newSockets, NetworkThread< SocketType >::_sockets, PROXY_HEADER_READING_STATE_FINISHED, PROXY_HEADER_READING_STATE_NOT_STARTED, PROXY_HEADER_READING_STATE_STARTED, and NetworkThread< SocketType >::SocketRemoved().

Referenced by NetworkThread< SocketType >::AddNewSockets().

Run()

template<class SocketType >

void NetworkThread< SocketType >::Run ( )

inlineprotected

    {
        LOG_DEBUG("misc", "Network Thread Starting");
 
        _updateTimer.expires_from_now(boost::posix_time::milliseconds(1));
        _updateTimer.async_wait([this](boost::system::error_code const&) { Update(); });
        _ioContext.run();
 
        LOG_DEBUG("misc", "Network Thread exits");
        _newSockets.clear();
        _sockets.clear();
    }

References NetworkThread< SocketType >::_ioContext, NetworkThread< SocketType >::_newSockets, NetworkThread< SocketType >::_sockets, NetworkThread< SocketType >::_updateTimer, LOG_DEBUG, Acore::Asio::IoContext::run(), and NetworkThread< SocketType >::Update().

Referenced by NetworkThread< SocketType >::Start().

SocketAdded()

template<class SocketType >

virtual void NetworkThread< SocketType >::SocketAdded ( std::shared_ptr< SocketType >  )

inlineprotectedvirtual

Reimplemented in WorldSocketThread.

{ }

Referenced by NetworkThread< SocketType >::AddSocket().

SocketRemoved()

template<class SocketType >

virtual void NetworkThread< SocketType >::SocketRemoved ( std::shared_ptr< SocketType >  )

inlineprotectedvirtual

Reimplemented in WorldSocketThread.

{ }

Referenced by NetworkThread< SocketType >::AddNewSockets(), and NetworkThread< SocketType >::HandleNewSocketsProxyReadingOnConnect().

Start()

template<class SocketType >

bool NetworkThread< SocketType >::Start ( )

inline

    {
        if (_thread)
            return false;
 
        _thread = std::make_unique<std::thread>([this]() { NetworkThread::Run(); });
        return true;
    }

References NetworkThread< SocketType >::_thread, and NetworkThread< SocketType >::Run().

Stop()

template<class SocketType >

void NetworkThread< SocketType >::Stop ( )

inline

    {
        _stopped = true;
        _ioContext.stop();
    }

References NetworkThread< SocketType >::_ioContext, NetworkThread< SocketType >::_stopped, and Acore::Asio::IoContext::stop().

Referenced by NetworkThread< SocketType >::~NetworkThread().

Update()

template<class SocketType >

void NetworkThread< SocketType >::Update ( )

inlineprotected

    {
        if (_stopped)
            return;
 
        _updateTimer.expires_from_now(boost::posix_time::milliseconds(1));
        _updateTimer.async_wait([this](boost::system::error_code const&) { Update(); });
 
        AddNewSockets();
 
        _sockets.erase(std::remove_if(_sockets.begin(), _sockets.end(), [this](std::shared_ptr<SocketType> sock)
        {
            if (!sock->Update())
            {
                if (sock->IsOpen())
                    sock->CloseSocket();
 
                this->SocketRemoved(sock);
 
                --this->_connections;
                return true;
            }
 
            return false;
        }), _sockets.end());
    }

References NetworkThread< SocketType >::_sockets, NetworkThread< SocketType >::_stopped, NetworkThread< SocketType >::_updateTimer, NetworkThread< SocketType >::AddNewSockets(), and NetworkThread< SocketType >::Update().

Referenced by NetworkThread< SocketType >::Run(), and NetworkThread< SocketType >::Update().

Wait()

template<class SocketType >

void NetworkThread< SocketType >::Wait ( )

inline

    {
        ASSERT(_thread);
 
        if (_thread->joinable())
        {
            _thread->join();
        }
 
        _thread.reset();
    }

References NetworkThread< SocketType >::_thread, and ASSERT.

Referenced by NetworkThread< SocketType >::~NetworkThread().

Member Data Documentation

_acceptSocket

template<class SocketType >

tcp::socket NetworkThread< SocketType >::_acceptSocket

private

Referenced by NetworkThread< SocketType >::GetSocketForAccept().

_connections

template<class SocketType >

std::atomic<int32> NetworkThread< SocketType >::_connections {}

private

Referenced by NetworkThread< SocketType >::AddNewSockets(), NetworkThread< SocketType >::AddSocket(), NetworkThread< SocketType >::GetConnectionCount(), and NetworkThread< SocketType >::HandleNewSocketsProxyReadingOnConnect().

_ioContext

template<class SocketType >

Acore::Asio::IoContext NetworkThread< SocketType >::_ioContext

private

Referenced by NetworkThread< SocketType >::Run(), and NetworkThread< SocketType >::Stop().

_newSockets

template<class SocketType >

SocketContainer NetworkThread< SocketType >::_newSockets

private

Referenced by NetworkThread< SocketType >::AddNewSockets(), NetworkThread< SocketType >::AddSocket(), NetworkThread< SocketType >::HandleNewSocketsProxyReadingOnConnect(), and NetworkThread< SocketType >::Run().

_newSocketsLock

template<class SocketType >

std::mutex NetworkThread< SocketType >::_newSocketsLock

private

Referenced by NetworkThread< SocketType >::AddNewSockets(), and NetworkThread< SocketType >::AddSocket().

_proxyHeaderReadingEnabled

template<class SocketType >

bool NetworkThread< SocketType >::_proxyHeaderReadingEnabled

private

Referenced by NetworkThread< SocketType >::AddNewSockets(), and NetworkThread< SocketType >::EnableProxyProtocol().

_sockets

template<class SocketType >

SocketContainer NetworkThread< SocketType >::_sockets

private

Referenced by NetworkThread< SocketType >::AddNewSockets(), NetworkThread< SocketType >::HandleNewSocketsProxyReadingOnConnect(), NetworkThread< SocketType >::Run(), and NetworkThread< SocketType >::Update().

_stopped

template<class SocketType >

std::atomic<bool> NetworkThread< SocketType >::_stopped {}

private

Referenced by NetworkThread< SocketType >::Stop(), and NetworkThread< SocketType >::Update().

_thread

template<class SocketType >

std::unique_ptr<std::thread> NetworkThread< SocketType >::_thread

private

Referenced by NetworkThread< SocketType >::Start(), NetworkThread< SocketType >::Wait(), and NetworkThread< SocketType >::~NetworkThread().

_updateTimer

template<class SocketType >

Acore::Asio::DeadlineTimer NetworkThread< SocketType >::_updateTimer

private

Referenced by NetworkThread< SocketType >::Run(), and NetworkThread< SocketType >::Update().