RequestClient.h

Go to the documentation of this file.

#if !defined(_REQUESTCLIENT_H_)
#define _REQUESTCLIENT_H_
 
#pragma once
 
#include "NetworkManager.h"
#include "MovingAverage.h"
#include "jsmn.h"
 
namespace   cmlabs{

// Request Reply
 
class RequestReply;
typedef void (* RequestCallbackFunction)(RequestReply& reply);

enum RequestStatus { NONE = 0, IDLE = 1, QUEUED = 2, PROCESSING = 3, SENT = 4, SUCCESS = 5, FAILED = 6, TIMEOUT = 7, TOOBUSY = 8, LOCALERROR = 9, NETWORKERROR = 10, SERVERERROR = 11};

class RequestReply {
protected:
    // stats?
    RequestCallbackFunction callback;
    RequestStatus status;
    DataMessage* replyMsg;
    DataMessage* requestMsg;
    utils::Mutex mutex;
    utils::Semaphore semaphore;
 
public:
    RequestReply() { requestMsg = replyMsg = NULL; callback = NULL; status = IDLE; systemID = 0; startTime = GetTimeNow(); origin = processor = clientRef = gatewayRef = execRef = finishTime = customRef = 0; isLongReq = false; isInUse = true; }
    virtual ~RequestReply() { delete(requestMsg); delete(replyMsg); callback = NULL; status = IDLE; }
 
    uint64 startTime;   
    uint64 finishTime;  
    uint64 origin;      
    uint64 processor;   
    uint64 clientRef;   
    uint64 execRef;     
    uint64 gatewayRef;  
    uint64 customRef;   
    uint32 systemID;    
    bool isLongReq;     
    bool isInUse;       

    bool isComplete();
    bool setStatus(RequestStatus status);
    RequestStatus getStatus();
    std::string getStatusText();
    bool setCallback(RequestCallbackFunction callback);
    RequestCallbackFunction getCallback();

    bool replyToRequest(DataMessage* msg, RequestStatus status);

    bool giveRequestMessage(DataMessage* msg);
    bool setRequestMessageCopy(DataMessage* msg);
    DataMessage* getRequestMessageCopy();
    DataMessage* peekRequestMessage();
    bool giveReplyMessage(DataMessage* msg);
    bool setReplyMessageCopy(DataMessage* msg);
    DataMessage* getReplyMessageCopy();
    DataMessage* peekReplyMessage();

    RequestStatus waitForResult(uint32 timeoutMS);
    DataMessage* waitForMessage(uint32 timeoutMS, bool takeMessage = false);

    uint64 getRequestDuration();
    uint32 getRequestDurationMS();
};
 

// Request Queue  (implemented in RequestGateway.cpp)

class RequestQueue {
public:
    static bool UnitTest();
 
    RequestQueue();
    virtual ~RequestQueue();

    bool init(uint32 id);

    uint32 getCount() { return (uint32)requestQ.size(); }

    bool addRequest(RequestReply* req, double priority = 0);
    RequestReply* getNextRequest();
    RequestReply* getNextTimedoutRequest();
    // RequestReply* getRequest(uint64 ref);
    bool completeRequest(RequestReply* req);
    //uint32 getActiveRequestCount();

    uint32 removeStaleRequests(uint32 ttlMS);
    std::list<RequestReply*> takeQueue();

    std::string toXML();
    std::string toJSON();
 
protected:
    std::list<RequestReply*> requestQ;  
    utils::Mutex mutex;                 
    uint32 id;                          
};
 

// Request Connection

struct RequestConnection {
    uint64 lastConTime;             
    uint64 lastFailTime;            
    uint64 conID;                   
    uint64 location;                
    uint64 lastStatusTime;          
    uint32 reportedShortReqQSize;   
    uint32 reportedLongReqQSize;    
    uint32 longReqQProcessingSize;  
    uint32 count;                   
    RequestQueue shortReqQueue;     
    RequestQueue longReqQueue;      
    MovingAverage shortAvgStats;    
    MovingAverage longAvgStats;     
    void clear() { lastConTime=lastFailTime=conID=location=lastStatusTime=0; reportedShortReqQSize = reportedLongReqQSize = longReqQProcessingSize = count = 0; }
    uint32 removeStaleRequests(uint32 ttlMS);
};

struct RequestGatewayConnection {
    uint32 id;              
    uint64 lastConTime;     
    uint64 lastFailTime;    
    std::string addr;       
    uint16 port;            
    uint64 conID;           
    uint64 location;        
    uint8 encryption;       
    void clear() { lastConTime=lastFailTime=conID=0; port=0; id=0; addr.clear(); conID=location=0; encryption = NOENC; }
};

class RequestClient : public Runnable, public NetworkReceiver {
public:
    friend THREAD_RET THREAD_FUNCTION_CALL RequestClientRun(THREAD_ARG arg);

    static bool TestServerLogin(const char* address, uint16 port, const char* username, const char* password, const char* reqAfterLogin);

    static std::string SendRequestAndWaitForJSON(RequestClient* client, DataMessage* msg, int timeoutMS);
    static DataMessage* SendRequestAndWaitForReply(RequestClient* client, DataMessage* msg, int timeoutMS);
    static RequestReply* SendRequestAndWaitForReplyObject(RequestClient* client, DataMessage* msg, int timeoutMS);
    static std::string GetJSONReplyParameter(const char* json, const char* name);
    static DataMessage* CreateRequestMessage(uint8 operation, const char* req,
        const char* key1 = NULL, const char* val1 = NULL,
        const char* key2 = NULL, const char* val2 = NULL,
        const char* key3 = NULL, const char* val3 = NULL,
        const char* key4 = NULL, const char* val4 = NULL,
        const char* key5 = NULL, const char* val5 = NULL,
        const char* key6 = NULL, const char* val6 = NULL,
        const char* key7 = NULL, const char* val7 = NULL,
        const char* key8 = NULL, const char* val8 = NULL,
        const char* key9 = NULL, const char* val9 = NULL,
        const char* key10 = NULL, const char* val10 = NULL,
        const char* key11 = NULL, const char* val11 = NULL,
        const char* key12 = NULL, const char* val12 = NULL,
        const char* key13 = NULL, const char* val13 = NULL,
        const char* key14 = NULL, const char* val14 = NULL,
        const char* key15 = NULL, const char* val15 = NULL,
        const char* key16 = NULL, const char* val16 = NULL,
        const char* key17 = NULL, const char* val17 = NULL,
        const char* key18 = NULL, const char* val18 = NULL,
        const char* key19 = NULL, const char* val19 = NULL,
        const char* key20 = NULL, const char* val20 = NULL
        );
 
    RequestClient();
    virtual ~RequestClient();

    bool isConnected();
    bool waitForConnection(uint32 timeoutMS);
    bool reconnect();
    bool addGateway(uint32 id, std::string addr, uint16 port, uint8 encryption = NOENC);

    RequestReply* postRequest(DataMessage *msg);
    bool postRequest(DataMessage *msg, RequestCallbackFunction callback, uint32 timeoutMS);
    bool finishRequest(RequestReply* reply);

    bool receiveNetworkEvent(NetworkEvent* evt, NetworkChannel* channel, uint64 conid);
    bool receiveMessage(DataMessage* msg, NetworkChannel* channel, uint64 conid);
    bool receiveHTTPReply(HTTPReply* reply, HTTPRequest* req, NetworkChannel* channel, uint64 conid);
 
protected:
    NetworkManager* manager;        
    NetworkChannel* channel;        
    std::list<RequestGatewayConnection> connections;    
    utils::Mutex mutex;             
    utils::Mutex conMutex;          
    utils::WaitQueuePointer<RequestReply*> outQ;    
    std::map<uint64,RequestReply*> requestMap;      
    uint32 threadID;                
    uint64 lastRefID;               
    uint64 sentCount;               
    uint64 receivedCount;           
    MovingAverage avgStats;         

    bool sendRequest(RequestReply* reply);
    bool run();
    bool sendMessageToGateway(DataMessage* msg, RequestGatewayConnection& con);
    uint32 sendStatusNow();
};

THREAD_RET THREAD_FUNCTION_CALL RequestClientRun(THREAD_ARG arg);

class TestRequestClient {
public:
    friend THREAD_RET THREAD_FUNCTION_CALL TestRequestClientRun(THREAD_ARG arg);
    TestRequestClient(uint32 id);
    bool init(uint16 port, uint32 gatewayCount, double sendfreq, int32 reconnectMS, uint32 payload, const char* longTestName);
    bool finishUp();
    bool isStillRunning() { return isRunning; }
protected:
    bool run();
 
    uint32 id;
    uint16 port;
    uint32 gatewayCount;
    double sendfreq;
    int32 reconnectMS;
    uint32 payload;
    std::string longTestName;
    bool shouldContinue;
    bool shouldFinishUp;
    bool isRunning;
};
 
THREAD_RET THREAD_FUNCTION_CALL TestRequestClientRun(THREAD_ARG arg);

class TestWebRequestClient {
public:
    friend THREAD_RET THREAD_FUNCTION_CALL TestWebRequestClientRun(THREAD_ARG arg);
    TestWebRequestClient(uint32 id);
    bool init(uint16 port, uint32 gatewayCount, double sendfreq, int32 reconnectMS, uint32 payload, const char* longTestName);
    bool finishUp();
    bool isStillRunning() { return isRunning; }
protected:
    bool run();
 
    uint32 id;
    uint16 port;
    uint32 gatewayCount;
    double sendfreq;
    int32 reconnectMS;
    uint32 payload;
    std::string longTestName;
    bool shouldContinue;
    bool shouldFinishUp;
    bool isRunning;
};
 
THREAD_RET THREAD_FUNCTION_CALL TestWebRequestClientRun(THREAD_ARG arg);
 
 

class TestWebSocketRequestClient {
public:
    friend THREAD_RET THREAD_FUNCTION_CALL TestWebSocketRequestClientRun(THREAD_ARG arg);
    TestWebSocketRequestClient(uint32 id);
    bool init(uint16 port, uint32 gatewayCount, double sendfreq, int32 reconnectMS, uint32 payload, const char* longTestName);
    bool finishUp();
    bool isStillRunning() { return isRunning; }
protected:
    bool run();
 
    uint32 id;
    uint16 port;
    uint32 gatewayCount;
    double sendfreq;
    int32 reconnectMS;
    uint32 payload;
    std::string longTestName;
    bool shouldContinue;
    bool shouldFinishUp;
    bool isRunning;
};
 
THREAD_RET THREAD_FUNCTION_CALL TestWebSocketRequestClientRun(THREAD_ARG arg);
 
 
}
 
#endif // _REQUESTCLIENT_H_