RequestGateway.h

Go to the documentation of this file.

#if !defined(_REQUESTGATEWAY_H_)
#define _REQUESTGATEWAY_H_
 
#pragma once
 
#include "Stats.h"
#include "NetworkManager.h"
#include "RequestExecutor.h"
#include "RequestClient.h"
 
 
namespace   cmlabs{

class CallLogEntry {
public:
    CallLogEntry(uint64 id = 0) {
        this->id = id;
        conType = messageType = requestType = 0;
        longRequest = false;
        port = 0;
        source = executorID = clientID = startTime = endTime = queueTime = processTime = 0;
        requestSize = replySize = accountID = 0;
    }
    virtual ~CallLogEntry() {};

    std::string toJSON() {
        if (!id || !startTime || !source || !conType || !messageType)
            return "";
        return utils::StringFormat(
            "{\"id\":%llu, \"port\":%u, \"source\":\"%u.%u.%u.%u:%u\", \"connectiontype\":\"%s\", \"messagetype\":\"%s\", "
            "\"requesttype\":\"%s\", \"longrequest\": \"%s\", \"status\": \"%s\", \"request\": \"%s\", \"output\": \"%s\", "
            "\"executorid\":%llu, \"clientid\":%llu, \"accountid\":%u, "
            "\"requestsize\":%u, \"replysize\":%u, "
            "\"requesttime\": \"%s\", \"duration\": \"%.3f\", \"queuetime\": \"%.3f\", \"processtime\": \"%.3f\"}\n",
            id, port, GETIPADDRESSQUADPORT(source),
            ((conType == 1) ? "Client" : (conType == 2) ? "Web" : "Websocket"),
            ((messageType == 1) ? "Binary" : (messageType == 2) ? "HTTP" : "HTTPS"),
            HTTP_Type[requestType],
            (longRequest ? "true" : "false"),
            status.c_str(), requestPath.c_str(), output.c_str(),
            executorID, clientID, accountID,
            requestSize, replySize,
            PrintTimeString(startTime).c_str(),
            (double)(endTime - startTime)/1000.0,
            (double)queueTime/1000.0,
            (double)processTime/1000.0
            );
    }
 
    uint64 id;          
    uint8 conType;      
    uint8 messageType;  
    bool longRequest;   
    uint64 source;      
    uint16 port;        
    uint64 executorID;  
    uint64 clientID;    
    uint32 accountID;   
    uint64 startTime;   
    uint64 endTime;     
    uint64 queueTime;   
    uint64 processTime; 
    uint8 requestType;  
    uint32 requestSize; 
    uint32 replySize;   
    std::string status;         
    std::string requestPath;    
    std::string agent;          
    std::string output;         
};
 
 
 

// RequestGateway

class RequestGateway : public Runnable, public NetworkReceiver {
public:
    static bool UnitTest(uint16 port, uint32 executorCount, uint32 gatewayCount, uint32 clientCount, uint32 webCount, uint32 webSocketCount, double sendfreq, int32 reconnect, uint32 payload, uint32 processTime, uint32 runtime);
 
    friend THREAD_RET THREAD_FUNCTION_CALL RequestGatewayExecRun(THREAD_ARG arg);
    friend THREAD_RET THREAD_FUNCTION_CALL RequestGatewayClientRun(THREAD_ARG arg);

    RequestGateway(uint32 id, const char* version = NULL);
    virtual ~RequestGateway();

    bool init(const char* sslCertPath = NULL, const char* sslKeyPath = NULL);

    bool setWebServerInfo(const char* name, const char* rootdir, const char* indexfile);
    bool setExternalAPILabel(const char* label);
    bool setInternalAPILabel(const char* label);
    bool setCacheFiles(bool cache);
    bool setResponseType(const char* type);
    bool setLongRequestLimit(uint32 limit);
    bool setExecutorHeartbeatTimeout(uint32 timeout);
    bool setMaxExecutorRequestTimeout(uint32 timeout);
    bool setQueuingParameters(uint32 maxRequestQueueSize, uint32 maxRequestProcessingSize, uint32 priorityThreshold);

    bool addPort(uint16 port, uint8 encryption, bool enableHTTP = false, uint32 timeout = 3000);
    bool addGateway(uint32 id, std::string addr, uint16 port);
    bool addAuthUser(const char* user, const char* password);

    bool receiveNetworkEvent(NetworkEvent* evt, NetworkChannel* channel, uint64 conid);
    bool receiveMessage(DataMessage* msg, NetworkChannel* channel, uint64 conid);
    bool receiveHTTPRequest(HTTPRequest* req, NetworkChannel* channel, uint64 conid);
    bool receiveWebsocketData(WebsocketData* wsData, NetworkChannel* channel, uint64 conid);
 
protected:
    bool runExec();
    bool runClient();

    bool callInternalAPI(const char* apiName, HTTPRequest* req, NetworkChannel* channel, uint64 conid);
    bool callExternalAPI(const char* apiName, HTTPRequest* req, NetworkChannel* channel, uint64 conid);

    bool sendRestartToExecutor(uint64 id);
    bool distributeDeadExecutorRequests(std::list<RequestReply*>& shortQueue, std::list<RequestReply*>& longQueue);

    uint64 getBestExecutorID(std::string requestString, uint32 reqSize, bool& isLongReq);
    //std::list<RequestConnection> getTopExecutors(std::string requestString, uint32 reqSize, bool& isLongReq);
    //bool sendToExecutor(RequestReply* reply);
    bool replyToClient(DataMessage* msg);
    bool addToRequestQueue(DataMessage* msg, uint64 origin, uint64 conID, uint64 clientRef);
    bool addRequestReplyToRequestQueue(RequestReply* reply);

    bool addToCallLog(
        uint64 id,
        uint8 conType,
        uint8 messageType,
        bool longRequest,
        uint64 source,
        uint16 port,
        uint64 executorID,
        uint64 clientID,
        uint64 startTime,
        uint64 endTime,
        uint64 queueTime,
        uint64 processTime,
        uint8 requestType,
        uint32 requestSize,
        uint32 replySize,
        const char* status,
        const char* requestPath,
        const char* agent,
        const char* output
    );
    utils::Mutex callLogMutex;          
    std::list<CallLogEntry*> callLog;   
    uint32 callLogMax;                  
    uint64 callLogCount;                
 
    NetworkManager* manager;    
    NetworkChannel* channel;    
    std::map<uint64,RequestConnection> executors;   
    std::map<uint64,RequestConnection> clients;     
    std::map<uint64,RequestConnection> webClients;  
    std::map<uint64,RequestConnection> webSockets;  
    utils::WaitQueue<RequestReply*> execQ;          
    utils::WaitQueuePointer<DataMessage*> replyQ;   
    std::map<uint64,RequestReply*> requestMap;      
    std::vector<std::string> longReqNames;          
    std::map<std::string, uint64> httpAuth;         
 
    utils::Mutex mutex;         
    //uint32 threadID;
    uint64 systemStartTime;     
    uint32 threadIDClient;      
    uint64 lastRefID;           
    uint64 clientSentCount;     
    uint64 clientReceivedCount; 
    uint64 execSentCount;       
    uint64 execReceivedCount;   
 
    uint32 id;                          
    uint16 port;                        
    uint32 longReqLimit;                
    uint32 executorHeartbeatTimeout;    
    uint32 maxRequestQueueSize;         
    uint32 maxRequestProcessingSize;    
    uint32 priorityThreshold;           
    uint32 maxExecutorRequestTimeoutMS; 
    bool sslSupport;                    
    bool cacheFiles;                    
    bool replyXML;                      
 
    std::string webServerName;  
    std::string rootdir;        
    std::string indexFilename;  
    std::string versionString;  
 
    std::string internalAPITitle;   
    std::string externalAPITitle;   
 
    MovingAverage shortAvgStats;    
    MovingAverage longAvgStats;     
    //std::map<uint64, MovingAverage> shortAvgStatsExec;
    //std::map<uint64, MovingAverage> shortAvgStatsClients;
    //std::map<uint64, MovingAverage> shortAvgStatsWeb;
    //std::map<uint64, MovingAverage> longAvgStatsExec;
    //std::map<uint64, MovingAverage> longAvgStatsClients;
    //std::map<uint64, MovingAverage> longAvgStatsWeb;
};

THREAD_RET THREAD_FUNCTION_CALL RequestGatewayExecRun(THREAD_ARG arg);
THREAD_RET THREAD_FUNCTION_CALL RequestGatewayClientRun(THREAD_ARG arg);
 
}
 
#endif // _REQUESTGATEWAY_H_