NPLConnection.h

#pragma once
#include "NPLMessage.h"
#include "NPLCommon.h"
#include "NPLMsgIn.h"
#include "NPLMsgOut.h"
#include "NPLMsgIn_parser.h"
#include "NPLMessageQueue.h"
#include "WebSocket/WebSocketReader.h"
#include "WebSocket/WebSocketWriter.h"

#include <boost/asio.hpp>
#include <boost/array.hpp>
#include <boost/noncopyable.hpp>
#include <boost/shared_ptr.hpp>
#include <boost/enable_shared_from_this.hpp>

namespace NPL
{
    class CNPLDispatcher;

    class CNPLConnection : 
        public boost::enable_shared_from_this<CNPLConnection>,
        private boost::noncopyable
    {
    public:
        enum ProtocolType
        {
            NPL = 0,
            WEBSOCKET = 1,
            TCP_CUSTOM = 2, // any custom protocol, like google protocol buffer
        };
        friend class CNPLDispatcher;
        typedef concurrent_ptr_queue<NPLMsgOut_ptr, dummy_condition_variable> RingBuffer_Type;
        typedef std::map<std::string, int>  StringMap_Type;

        ~CNPLConnection();
        
        explicit CNPLConnection(boost::asio::io_service& io_service,
            CNPLConnectionManager& manager, CNPLDispatcher& msg_dispatcher);

        boost::asio::ip::tcp::socket& socket();

        void start();

        void stop(bool bRemoveConnection = true, int nReason = 0);

        void CloseAfterSend();

        void connect();

        CNPLConnectionManager& GetConnectionManager() {return m_connection_manager;};

        NPLReturnCode SendMessage(const NPLFileName& file_name, const char * code = NULL, int nLength=0, int priority=0);

        NPLReturnCode SendMessage(const NPLMessage& msg);

        NPLReturnCode SendMessage(const char* sCommandName, const char* sCommandData);

        NPLReturnCode SendMessage(NPLMsgOut_ptr& msg);

        void SetNPLRuntimeAddress(NPLRuntimeAddress_ptr runtime_address);
    
    public:
        virtual void GetStatistics( int &totalIn, int &totalOut );
    
        NPLConnectionState GetState() const { return m_state; }

        const string& GetNID() const;

        string GetIP();

        string GetPort();

        bool SetNID(const char* sNID);

        void SetAuthenticated(bool bAuthenticated);

        bool IsAuthenticated() const;

        bool IsConnected() const;

        void SetUseCompression(bool bUseCompression);

        bool IsUseCompression();

        void SetCompressionLevel(int nLevel);
        int GetCompressionLevel();

        
        void SetCompressionThreshold(int nThreshold);
        int GetCompressionThreshold();


        void SetTCPKeepAlive(bool bEnable);

        void SetKeepAlive(bool bEnable);
        bool IsKeepAliveEnabled();

        void EnableIdleTimeout(bool bEnable);
        bool IsIdleTimeoutEnabled();

        void SetIdleTimeoutPeriod(int nMilliseconds);
        int GetIdleTimeoutPeriod();

        unsigned int GetLastActiveTime();

        void TickSend();

        void TickReceive();

        int CheckIdleTimeout(unsigned int nCurTime);

        bool HasUnsentData();

        int GetLogLevel();

        void SetProtocol(ProtocolType protocolType = ProtocolType::NPL);
    public:
        //
        // In case, one wants to use a different connection data handler,  the following interface are provided. 
        //

        virtual bool handleReceivedData(int bytes_transferred);

        virtual void handleConnect();

        virtual void handleDisconnect(int reason);

        virtual bool handleMessageIn();

    private:
        bool handle_websocket_data(int bytes_transferred);
        bool handle_tcp_custom_data(int bytes_transferred);
        //
        // boost io service call backs. 
        //

        void handle_read(const boost::system::error_code& e,std::size_t bytes_transferred);

        void handle_write(const boost::system::error_code& e);

        void handle_stop();

        void handle_resolve(const boost::system::error_code& err, boost::asio::ip::tcp::resolver::iterator endpoint_iterator);

        void handle_connect(const boost::system::error_code& error, boost::asio::ip::tcp::resolver::iterator endpoint_iterator);

        boost::asio::ip::tcp::socket m_socket;

    private:
        typedef boost::array<char, 8192> Buffer_Type;
        boost::shared_ptr< boost::asio::ip::tcp::resolver > m_resolver;

        CNPLConnectionManager& m_connection_manager;

        CNPLDispatcher& m_msg_dispatcher;

        NPLRuntimeAddress_ptr m_address;

        Buffer_Type m_buffer;

        NPLMsgIn m_input_msg;

        NPLMsgIn_parser m_parser;

        RingBuffer_Type m_queueOutput;

        NPLConnectionState m_state;

        ParaEngine::mutex m_mutex;

        StringMap_Type m_filename_id_map;

        uint32 m_totalBytesIn;
        uint32 m_totalBytesOut;
        
        bool m_bDebugConnection;

        int32 m_nCompressionLevel;

        int32 m_nCompressionThreshold;

        bool m_bKeepAlive;

        bool m_bEnableIdleTimeout;

        uint32 m_nSendCount;
        uint32 m_nFinishedCount;

        bool m_bCloseAfterSend;

        uint32 m_nIdleTimeoutMS;

        uint32 m_nLastActiveTime;

        uint32 m_nStartTime;

        int32 m_nStopReason;

        WebSocket::WebSocketReader m_websocket_reader;
        WebSocket::WebSocketWriter m_websocket_writer;
        std::vector<byte> m_websocket_input_data;
        std::vector<byte> m_websocket_out_data;

        ProtocolType m_protocolType;
    };
    

    struct NPLConnection_PtrOps
    {   // functor for operator<
        bool operator()(const NPLConnection_ptr& _Left, const NPLConnection_ptr& _Right) const
        {   // apply operator< to operands
            return (_Left.get()<_Right.get());
        }
    };
}