RequestExecutor.h

Go to the documentation of this file.

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

// Request Executor

class RequestExecutor : public Runnable, public NetworkReceiver {
public:
    friend THREAD_RET THREAD_FUNCTION_CALL RequestExecutorRun(THREAD_ARG arg);

    RequestExecutor(uint32 id = 0, const char* name = NULL);
    virtual ~RequestExecutor();
    bool shutdownNetwork();

    bool addLongRequestName(const char* name);
    bool addGateway(uint32 id, std::string addr, uint16 port, uint8 encryption = NOENC);
    bool setLongRequestLimit(uint32 limit);

    bool receiveNetworkEvent(NetworkEvent* evt, NetworkChannel* channel, uint64 conid);
    bool receiveMessage(DataMessage* msg, NetworkChannel* channel, uint64 conid);
    bool receiveHTTPRequest(HTTPRequest* req, NetworkChannel* channel, uint64 conid) {return false;}

    DataMessage* waitForLongRequest(uint32 timeoutMS);
    DataMessage* waitForShortRequest(uint32 timeoutMS);
    bool replyToQuery(DataMessage* msg);
 
protected:
    bool run();
    uint32 sendStatusNow();
    bool sendMessageToGateway(DataMessage* msg, RequestGatewayConnection& con);
 
    std::list<RequestGatewayConnection> connections;    
    //std::map<uint64, RequestGatewayConnection> gateways;

    bool replyToGateway(DataMessage* msg);
 
    uint32 executorID;          
    std::string executorName;   
    NetworkManager* manager;    
    NetworkChannel* channel;    
    utils::WaitQueuePointer<DataMessage*> shortExecQ;   
    utils::WaitQueuePointer<DataMessage*> longExecQ;    
    utils::WaitQueuePointer<DataMessage*> replyQ;       
    std::map<uint64,RequestReply*> requestMap;          
    std::vector<std::string> longReqNames;              
    utils::Mutex mutex;         
    utils::Mutex conMutex;      
    utils::Semaphore shortExecQSemaphore;   
    utils::Semaphore longExecQSemaphore;    
    uint32 threadID;            
    uint64 lastRefID;           
    uint64 repliedCount;        
    uint64 sentCount;           
    uint64 shortReceivedCount;  
    uint64 longReceivedCount;   
    int32 heartbeatIntervalMS;  
    uint64 lastHeartbeat;       
    uint32 longReqLimit;        
    //MovingAverage shortAvgStats;
    //MovingAverage longAvgStats;
};

THREAD_RET THREAD_FUNCTION_CALL RequestExecutorRun(THREAD_ARG arg);

class TestRequestExecutor : public RequestExecutor {
public:
    friend THREAD_RET THREAD_FUNCTION_CALL ShortExecutorRun(THREAD_ARG arg);
    friend THREAD_RET THREAD_FUNCTION_CALL LongExecutorRun(THREAD_ARG arg);
    TestRequestExecutor(uint32 id, uint32 processTimeMS);
protected:
    bool shortRequestRun();
    bool longRequestRun();
    uint32 id;
    uint32 processTimeMS;
};
 
THREAD_RET THREAD_FUNCTION_CALL ShortExecutorRun(THREAD_ARG arg);
THREAD_RET THREAD_FUNCTION_CALL LongExecutorRun(THREAD_ARG arg);
 
}
 
#endif // _REQUESTEXECUTOR_H_