Socket.hpp

Go to the documentation of this file.

/***************************************************
 *       _   _            _                        *
 *      | | | |          (_)                       *
 *      | |_| | ___  _ __ _ _______  _ __          *
 *      |  _  |/ _ \| '__| |_  / _ \| '_  \        *
 *      | | | | (_) | |  | |/ / (_) | | | |        *
 *      \_| |_/\___/|_|  |_/___\___/|_| |_|        *
 ***************************************************
 * This file is part of Horizon (c).
 *
 * Copyright (c) 2019 Sagun K. (sagunxp@gmail.com).
 * Copyright (c) 2019 Horizon Dev Team.
 *
 * Base Author - Sagun K. (sagunxp@gmail.com)
 *
 * This library is free software; you can redistribute it and/or modify
 * it under the terms of the GNU General Public License as published by
 * the Free Software Foundation, either version 3 of the License, or
 *  (at your option) any later version.
 * 
 * This library is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 * GNU General Public License for more details.
 * 
 * You should have received a copy of the GNU General Public License
 * along with this library.  If not, see <http://www.gnu.org/licenses/>.
 **************************************************/
 
#ifndef HORIZON_NETWORKING_SOCKET_HPP
#define HORIZON_NETWORKING_SOCKET_HPP
 
#include "Core/Logging/Logger.hpp"
#include "Buffer/ByteBuffer.hpp"
#include "Core/Multithreading/ThreadSafeQueue.hpp"
 
#include <atomic>
#include <queue>
#include <memory>
#include <functional>
#include <future>
#include <type_traits>
#include <boost/asio/ip/tcp.hpp>
#include <boost/bind/bind.hpp>
#include <boost/asio/write.hpp>
#include <boost/asio/use_future.hpp>
 
#include <iostream>
 
using boost::asio::ip::tcp;
 
#define READ_BLOCK_SIZE 0x1000
#define BOOST_BIND_GLOBAL_PLACEHOLDERS 1
 
namespace Horizon
{
namespace Networking
{
template <class SocketType>
class Socket : public std::enable_shared_from_this<SocketType>
{
public:
    explicit Socket(uint64_t socket_id)
    : _socket_id(socket_id), _socket(nullptr), _read_buffer(), _closed(false), _closing(false), _is_writing_async(false)
    {
        _read_buffer.resize(READ_BLOCK_SIZE);
    }
 
    explicit Socket(uint64_t socket_id, std::shared_ptr<tcp::socket> socket)
    : _socket_id(socket_id), _socket(socket), _remote_ip_address(_socket->remote_endpoint().address().to_string()),
    _remote_port(_socket->remote_endpoint().port()), _read_buffer(), _closed(false), _closing(false), _is_writing_async(false)
    {
        _read_buffer.resize(READ_BLOCK_SIZE);
    }
 
    virtual ~Socket()
    {
        boost::system::error_code error;
        _socket->close(error);
    }
 
    virtual void start() = 0;
 
    virtual bool update()
    {
        if (_closed)
            return false;
 
        if (_is_writing_async || (_write_queue.empty() && !_closing)) {
            return true;
        }
 
        while (handle_queue())
            ;
 
        return true;
    }
 
    /* Socket Id */
    uint64_t get_socket_id() { return _socket_id; }
 
    /* Remote IP and Port */
    std::string &remote_ip_address() { return _remote_ip_address; }
    uint16_t remote_port() const { return _remote_port; }
 
    void async_read()
    {
        if (!is_open())
            return;
 
        _read_buffer.flush();
        _read_buffer.ensure_free_space();
        
        _socket->async_read_some(boost::asio::buffer(_read_buffer.get_write_pointer(), _read_buffer.remaining_space()),
                                 boost::bind(&Socket<SocketType>::read_handler_internal, this, boost::placeholders::_1, boost::placeholders::_2));
    }
    
    void async_read_with_callback(ByteBuffer &buf, void (Socket<SocketType>::*)(boost::system::error_code, std::size_t))
    {
        if (!is_open())
            return;
 
        _read_buffer.flush();
        _read_buffer.ensure_free_space();
 
        _socket->async_read_some(boost::asio::buffer(buf.get_write_pointer(), buf.remaining_space()),
                                 boost::bind(&Socket<SocketType>::read_handler_internal, this, boost::placeholders::_1, boost::placeholders::_2));
    }
 
    virtual void queue_buffer(ByteBuffer &&buffer) { _write_queue.push(std::move(buffer)); }
 
    bool is_open() { return !_closed && !_closing; }
 
    void close_socket()
    {
        boost::system::error_code socket_error;
 
        if (_closed.exchange(true))
            return;
 
        // Finalise the child-class socket first.
        on_close();
 
        // Shutdown the socket.
        _socket->shutdown(boost::asio::socket_base::shutdown_send, socket_error);
        // Close the socket.
        _socket->close();
 
        if (socket_error) {
            HLog(error) << "Error when shutting down socket from IP " <<
                           remote_ip_address() << "(error code:" << socket_error.value() << " - " << socket_error.message().c_str() << ")";
        }
    }
 
    void delayed_close_socket() { if (_closing.exchange(true)) return; }
 
    ByteBuffer &get_read_buffer() { return _read_buffer; }
 
protected:
    virtual void on_close() = 0;
    virtual void read_handler() = 0;
    virtual void on_error() = 0;
 
    bool async_process_queue()
    {
        if (_is_writing_async)
            return false;
 
        _is_writing_async = true;
 
        _socket->async_write_some(boost::asio::null_buffers(),
                                  boost::bind(&Socket<SocketType>::write_handler_wrapper,
                                              this, boost::placeholders::_1, boost::placeholders::_2));
 
        return true;
    }
 
    void set_no_delay(bool enable)
    {
        boost::system::error_code error;
        _socket->set_option(tcp::no_delay(enable), error);
        if (error) {
            HLog(error) << "Networking: Socket::set_no_delay: failed to set_option(boost::asio::ip::tcp::no_delay) for IP " << remote_ip_address() << " (error_code: " << error.value() << " - " << error.message().c_str() << ")";
        }
    }
 
    std::size_t write_buffer_and_send(ByteBuffer &to_send, boost::system::error_code &error)
    {
        int16_t packet_id;
        std::size_t bytes_to_send = to_send.active_length();
        memcpy(&packet_id, to_send.get_read_pointer(), sizeof(int16_t));
        std::size_t bytes_sent = _socket->write_some(boost::asio::buffer(to_send.get_read_pointer(), bytes_to_send), error);
        return bytes_sent;
    }
private:
    
    void read_handler_internal(boost::system::error_code error, size_t transferredBytes)
    {
        // If there is an error on the socket, we need to close it.
        if (error) {
            // If the socket is already closed, we don't need to do anything.
            if (error.value() == boost::asio::error::eof) {
                close_socket();
            // If the connection was reset or timed out, we need to close the socket.
            } else if (error.value() == boost::system::errc::connection_reset
                       || error.value() == boost::system::errc::timed_out) {
                on_error();
                close_socket();
            // If there was any unknown error on the socket, we need to close it.
            } else {
                on_error();
                close_socket();
            }
            // Retreive the error code and handle it accordingly.
            switch (error.value())
            {
            case 2: // End of file
                break;
            default:
                HLog(debug) << "Socket::read_handler_internal: " << error.value() << " (Code: " << error.message() << ").";
            }
            return;
        }
 
        if (transferredBytes > 0) {
            // write_completed will move the write pointer forward, so we need to save the current position of the write pointer.
            _read_buffer.write_completed(transferredBytes);
            // read handler will process the data that has been written to the buffer.
            read_handler();
        }
 
        // Invoke the next read operation.
        async_read();
    }
 
    void write_handler_wrapper(boost::system::error_code /*error*/, std::size_t /*transferedBytes*/)
    {
        _is_writing_async = false;
        handle_queue();
    }
 
    bool handle_queue()
    {
        boost::system::error_code error;
 
        if (_write_queue.empty())
            return false;
 
        std::shared_ptr<ByteBuffer> to_send = _write_queue.front();
 
        //HLog(debug) << "Sent bytes: " << to_send->to_string();
        
        std::size_t bytes_sent = write_buffer_and_send(*to_send, error);
 
        if (error == boost::asio::error::would_block || error == boost::asio::error::try_again)
            return async_process_queue();
 
        if (!error && bytes_sent < to_send->active_length()) {
            to_send->read_completed(bytes_sent);
            return async_process_queue();
        }
 
        // Pop the front element.
        _write_queue.try_pop();
 
        // Close if required.
        if (_closing && _write_queue.empty())
            close_socket();
 
        // Return true if the queue is not empty. Indicating that the sending operation was successful.
        return !_write_queue.empty() && bytes_sent;
    }
 
    std::shared_ptr<tcp::socket> get_socket() { return _socket; }
 
private:
    uint64_t _socket_id;
    std::shared_ptr<tcp::socket> _socket;               
    std::string _remote_ip_address;
    uint16_t _remote_port;
    ByteBuffer _read_buffer;
    ThreadSafeQueue<ByteBuffer> _write_queue;
    std::atomic<bool> _closed;
    std::atomic<bool> _closing;
    bool _is_writing_async;
};
}
}
 
#endif //HORIZON_NETWORKING_SOCKET_HPP