NetworkManager.h

Go to the documentation of this file.

#if !defined(_NETWORKMANAGER_H_)
#define _NETWORKMANAGER_H_
 
#pragma once
 
#include "NetworkConnections.h"
#include "NetworkProtocols.h"
 
 
namespace   cmlabs{

struct NetworkEvent {
    uint64 cid;         
    uint64 conid;       
    uint64 time;        
    uint8 type;         
    uint8 protocol;     
};

// Receiver Interface
 
class NetworkChannel;
class NetworkReceiver {
public:
    NetworkReceiver() {}
    virtual bool receiveNetworkEvent(NetworkEvent* evt, NetworkChannel* channel, uint64 conid) {return false;}
    virtual bool receiveHTTPRequest(HTTPRequest* req, NetworkChannel* channel, uint64 conid) {return false;}
    virtual bool receiveMessage(DataMessage* msg, NetworkChannel* channel, uint64 conid) {return false;}
    virtual bool receiveTelnetLine(TelnetLine* line, NetworkChannel* channel, uint64 conid) {return false;}
    virtual bool receiveHTTPReply(HTTPReply* reply, HTTPRequest* req, NetworkChannel* channel, uint64 conid) {return false;}
    virtual bool receiveWebsocketData(WebsocketData* wsData, NetworkChannel* channel, uint64 conid) { return false; }
};

class HTTPTestServer : public NetworkReceiver {
public:
    HTTPTestServer();
    ~HTTPTestServer();
    virtual bool receiveHTTPRequest(HTTPRequest* req, NetworkChannel* channel, uint64 conid);
};

class WebsocketTestServer : public NetworkReceiver {
public:
    WebsocketTestServer();
    ~WebsocketTestServer();
    virtual bool receiveHTTPRequest(HTTPRequest* req, NetworkChannel* channel, uint64 conid);
    virtual bool receiveWebsocketData(WebsocketData* wsData, NetworkChannel* channel, uint64 conid);
};

// Network Manager
 
class NetworkChannel;
class NetworkManager : public Runnable {
    friend class NetworkChannel;
public:
    static bool UnitTest();
    static bool UnitTestDelayedConnect();
    static bool UnitTestHTTP();
#ifdef _USE_SSL_
    static bool UnitTestHTTPS();
    static bool UnitTestSSLVerify();
    static bool UnitTestSSLHostCA();
#endif // _USE_SSL_
    static bool TestHTTP(const char* host, uint32 port, std::vector<std::string> &urls);
    static bool WebsocketTest();

    THREAD_RET THREAD_FUNCTION_CALL NetworkManagerRun(THREAD_ARG arg);
 
    NetworkManager();
    virtual ~NetworkManager();

    bool setSSLCertificate(const char* sslCertPath, const char* sslKeyPath);
 
    // SSL client verification policy for connections made through this manager:
    // -1 = use the process-wide default (SSLConnection::GetDefaultAllowSelfSigned),
    // 0 = verify peer certificates (secure), 1 = allow self-signed/untrusted peers.
    bool setSSLAllowSelfSigned(bool allow) { sslAllowSelfSigned = allow ? 1 : 0; return true; }
    int8 sslAllowSelfSigned;
    // Custom CA file/dir for SSL client peer verification on connections made
    // through this manager (empty = inherit process default / OS trust store)
    bool setSSLCALocation(const char* caFile, const char* caPath) { sslCAFile = caFile ? caFile : ""; sslCAPath = caPath ? caPath : ""; sslCASet = true; return true; }
    std::string sslCAFile;
    std::string sslCAPath;
    bool sslCASet;

    NetworkChannel* createListener(uint16 port, uint8 encryption, uint8 protocol, bool isAsync, uint32 protocolTimeout, bool isDefaultProtocol, uint32 channelID, NetworkReceiver* recv);
    NetworkChannel* createTCPConnection(const char* addr, uint16 port, uint8 encryption, uint8 protocol, bool isAsync, bool autoreconnect, uint32 channelID, NetworkReceiver* recv, uint64& conid, uint64& location, uint32 timeoutMS = 5000);
    NetworkChannel* createTCPConnection(uint64 location, uint8 encryption, uint8 protocol, bool isAsync, bool autoreconnect, uint32 channelID, NetworkReceiver* recv, uint64& conid, uint32 timeoutMS = 5000);
    NetworkChannel* createTCPConnection(const uint32* addresses, uint16 addressCount, uint16 port, uint8 encryption, uint8 protocol, bool isAsync, bool autoreconnect, uint32 channelID, NetworkReceiver* recv, uint64& conid, uint64& location, uint32 timeoutMS = 5000);
    NetworkChannel* createUDPConnection(uint16 port, uint8 protocol, bool isAsync, bool autoreconnect, uint32 channelID, NetworkReceiver* recv, uint64& conid);
    NetworkChannel* addTCPConnection(const char* addr, uint16 port, uint8 encryption, uint8 protocol, bool isAsync, uint32 channelID, NetworkReceiver* recv, uint64& conid, uint64& location, uint32 timeoutMS = 5000, const char* greetingData = NULL, uint32 greetingSize = 0);
    NetworkChannel* addTCPConnection(uint64 location, uint8 encryption, uint8 protocol, bool isAsync, uint32 channelID, NetworkReceiver* recv, uint64& conid, uint32 timeoutMS = 5000, const char* greetingData = NULL, uint32 greetingSize = 0);
    NetworkChannel* createWebsocketConnection(const char* url, uint32 channelID, NetworkReceiver* recv, uint64& conid, const char* protocolName = NULL, const char* origin = NULL, uint32 timeoutMS = 5000);
    NetworkChannel* createWebsocketConnection(const char* uri, const char* addr, uint16 port, uint8 encryption, uint32 channelID, NetworkReceiver* recv, uint64& conid, const char* protocolName = NULL, const char* origin = NULL, uint32 timeoutMS = 5000);
 

    bool stopListener(uint16 port, uint8 protocol);
    bool endConnection(uint64 conid);
    bool endUDPConnection(uint16 port);
    bool removeConnection(uint64 conid);

    uint8 getConnectionType(uint64 conid);
    uint64 getRemoteAddress(uint64 conid);
    NetworkChannel* getConnection(uint64 conid);
    NetworkChannel* getTCPConnectionByPort(uint16 port);
    NetworkChannel* getUDPConnectionByPort(uint16 port);
    uint64 addConnection(NetworkChannel* channel);

    bool sendUDPMessage(DataMessage* msg, uint64 destination);

    HTTPReply* makeHTTPRequest(const char* url, uint32 timeout, const char* content = NULL, uint32 contentSize = 0);
    HTTPReply* makeHTTPRequest(HTTPRequest* req, const char* addr, uint16 port, uint8 encryption, uint32 timeout);

    HTTPReply* makeHTTPRequest(uint8 ops, std::string addr, uint32 timeout,
        std::map<std::string, std::string>& headerEntries, const char* content, const char* contentType, uint32 contentSize,
        bool keepAlive = false, uint64 ifModifiedSince = 0);

    HTTPReply* makeHTTPRequest(uint8 ops, std::string addr, uint32 timeout,
        std::map<std::string,std::string>& headerEntries, std::map<std::string, HTTPPostEntry*>& bodyEntries,
        bool keepAlive = false, uint64 ifModifiedSince = 0);
 
protected:
    std::map<uint32, NetworkChannel*> channels;             
    std::map<uint64, NetworkChannel*> channelsByConnection; 
    std::map<uint16, NetworkChannel*> listeners;            
    std::map<uint16, NetworkChannel*> udpListeners;         
 
    std::string sslCertPath;    
    std::string sslKeyPath;     
 
    UDPConnection* udpOutputCon;        
    utils::Mutex udpOutputConMutex;     
 
    uint32 lastChannelID;       
    uint64 lastConnectionID;    
    bool run();
};

// Connections

#define NETWORKEVENT_CONNECT                1   
#define NETWORKEVENT_RECONNECT              2   
#define NETWORKEVENT_DISCONNECT             3   
#define NETWORKEVENT_DISCONNECT_RETRYING    4   
#define NETWORKEVENT_BUFFER_LOW             5   
#define NETWORKEVENT_BUFFER_FULL            6   
#define NETWORKEVENT_UNPROCESSED_DATA       7   
#define NETWORKEVENT_PROTOCOL_ERROR         8   

class NetworkThread {
public:
    NetworkThread(NetworkChannel* parent, uint64 id);
    ~NetworkThread();
 
    uint64 id;                  
    uint32 threadID;            
    uint16 port;                
    TCPListener* listener;      
    NetworkConnection* con;     
    uint32 defaultProtocol;     
    uint32 autoProtocols;       
    uint32 autoProtocolTimeout; 
    bool autoreconnect;         
    bool shouldContinue;        
    bool isRunning;             
    bool isAsync;               
    HTTPRequest* lastRequest;   
    NetworkChannel* parent;     
};

class NetworkChannel {
    friend class NetworkManager;
public:
    NetworkChannel(NetworkManager* manager);
    virtual ~NetworkChannel();
    uint32 cid; 

    bool isConnected(uint64 conid);
    // getStats();

    bool shutdown();

    bool startListener(uint64 cid, uint16 port, uint8 encryption, uint8 protocol, bool isAsync, uint32 protocolTimeout = 3000, bool isDefaultProtocol = false);
    bool stopListener(uint16 port, uint8 protocol);

    uint64 createTCPConnection(const char* addr, uint16 port, uint8 encryption, uint8 protocol, bool isAsync, bool autoreconnect, uint64& location, uint32 timeoutMS = 5000);
    uint64 createTCPConnection(uint64 location, uint8 encryption, uint8 protocol, bool isAsync, bool autoreconnect, uint32 timeoutMS = 5000);
    uint64 createTCPConnection(const uint32* addresses, uint16 addressCount, uint16 port, uint8 encryption, uint8 protocol, bool isAsync, bool autoreconnect, uint64& location, uint32 timeoutMS = 5000);
    uint64 createUDPConnection(uint16 port, uint8 protocol, bool isAsync, bool autoreconnect);
    uint64 addTCPConnection(const char* addr, uint16 port, uint8 encryption, uint8 protocol, bool isAsync, uint64& location, uint32 timeoutMS = 5000, const char* greetingData = NULL, uint32 greetingSize = 0);
    uint64 addTCPConnection(uint64 location, uint8 encryption, uint8 protocol, bool isAsync, uint32 timeoutMS = 5000, const char* greetingData = NULL, uint32 greetingSize = 0);
    
    uint64 createWebsocketConnection(const char* url, const char* protocolName, const char* origin = NULL, uint32 timeoutMS = 5000);
    uint64 createWebsocketConnection(const char* uri, const char* addr, uint16 port, uint8 encryption, const char* protocolName, const char* origin = NULL, uint32 timeoutMS = 5000);

    bool endConnection(uint64 conid);
    bool endUDPConnection(uint16 port);
    uint8 getConnectionType(uint64 conid);
    uint64 getRemoteAddress(uint64 conid);

    uint64 autoDetectConnection(NetworkConnection* con, uint16 port, uint32 autoProtocols, uint32 autoProtocolTimeout, uint32 defaultProtocol, bool isAsync, bool autoreconnect);

    uint64 startConnection(NetworkConnection* con, uint8 protocol, bool isAsync, bool autoreconnect, uint32 timeoutMS = 5000);
 
    // Per-channel SSL client verification override (-1 = inherit from manager /
    // process default). E.g. set per <interface allowselfsigned=".."> in the spec.
    bool setSSLAllowSelfSigned(bool allow) { sslAllowSelfSigned = allow ? 1 : 0; return true; }
    int8 sslAllowSelfSigned;
    // Per-channel custom CA file/dir override for SSL client peer verification
    // (e.g. set per <interface cafile=".." capath=".."> in the spec)
    bool setSSLCALocation(const char* caFile, const char* caPath) { sslCAFile = caFile ? caFile : ""; sslCAPath = caPath ? caPath : ""; sslCASet = true; return true; }
    std::string sslCAFile;
    std::string sslCAPath;
    bool sslCASet;
    // Apply the effective allow-self-signed policy to a new SSL client connection
    void applySSLClientPolicy(SSLConnection* con);

    bool setNewReceiver(NetworkReceiver* recv);

    NetworkEvent* waitForNetworkEvent(uint32 ms);

    HTTPRequest* waitForHTTPRequest(uint64& conid, uint32 ms);
    TelnetLine* waitForTelnetLine(uint64& conid, uint32 ms);
    DataMessage* waitForMessage(uint64& conid, uint32 ms);
    HTTPReply* waitForHTTPReply(uint64& conid, uint32 ms);
    WebsocketData* waitForWebsocketData(uint64& conid, uint32 ms);

    bool sendHTTPReply(HTTPReply* reply, uint64 conid);
    bool sendTelnetLine(TelnetLine* line, uint64 conid);
    bool sendMessage(DataMessage* msg, uint64 conid);
    bool sendHTTPRequest(HTTPRequest* req, uint64 conid);
    HTTPReply* sendReceiveHTTPRequest(HTTPRequest* req, uint64 conid, uint32 timeout);
    TelnetLine* sendReceiveTelnetLine(TelnetLine* line, uint64 conid, uint32 timeout, uint32 size = 0);
    bool sendWebsocketData(WebsocketData* wsData, uint64 conid);

    uint32 getOutputSpeed(uint64 conid);
    uint32 getInputSpeed(uint64 conid);
 
protected:
    NetworkManager* manager;        
    NetworkReceiver* receiver;      
    std::map<uint64, NetworkThread*> connectionThreads; 
    std::map<uint16, NetworkThread*> listeners;         
    std::map<uint16, NetworkThread*> udpListeners;      
 
    std::queue<NetworkEvent*> eventQueue;
    utils::Semaphore eventQueueSemaphore;
    utils::Mutex eventQueueMutex;

    bool enterNetworkEvent(uint8 type, uint8 protocol, uint64 conid);

    bool enterHTTPRequest(HTTPRequest* req, uint64 conid);
    bool enterHTTPReply(HTTPReply* reply, HTTPRequest* req, uint64 conid);
    bool enterMessage(DataMessage* msg, uint64 conid);
    bool enterTelnetLine(TelnetLine* line, uint64 conid);
    bool enterWebsocketData(WebsocketData* wsData, uint64 conid);
 
    std::queue<HTTPRequest*> queueHTTPRequests;
    utils::Semaphore queueHTTPRequestsSemaphore;
    utils::Mutex queueHTTPRequestsMutex;
    std::queue<HTTPReply*> queueHTTPReplies;
    utils::Semaphore queueHTTPRepliesSemaphore;
    utils::Mutex queueHTTPRepliesMutex;
    std::queue<DataMessage*> queueMessages;
    std::queue<uint64> queueMessageConIDs;   // source connection id per queued message (parallel to queueMessages)
    utils::Semaphore queueMessagesSemaphore;
    utils::Mutex queueMessagesMutex;
    std::queue<TelnetLine*> queueTelnetLines;
    utils::Semaphore queueTelnetLinesSemaphore;
    utils::Mutex queueTelnetLinesMutex;
    std::queue<WebsocketData*> queueWebsocketData;
    utils::Semaphore queueWebsocketDataSemaphore;
    utils::Mutex queueWebsocketDataMutex;
 
    utils::Mutex channelMutex;

    static THREAD_RET THREAD_FUNCTION_CALL NetworkListenerRun(THREAD_ARG arg);
    static THREAD_RET THREAD_FUNCTION_CALL ConnectionAutodetectRun(THREAD_ARG arg);

    static THREAD_RET THREAD_FUNCTION_CALL HTTPClientRun(THREAD_ARG arg);
    static THREAD_RET THREAD_FUNCTION_CALL HTTPServerRun(THREAD_ARG arg);
    static THREAD_RET THREAD_FUNCTION_CALL MessageConnectionRun(THREAD_ARG arg);
    static THREAD_RET THREAD_FUNCTION_CALL TelnetServerRun(THREAD_ARG arg);
};
 
 
}
 
#endif // _NETWORKMANAGER_H_