RequestClient.h

#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* getRequest(uint64 ref);
    bool completeRequest(RequestReply* req);
    //uint32 getActiveRequestCount();

    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; }
};

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 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 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);
};

THREAD_RET THREAD_FUNCTION_CALL RequestClientRun(THREAD_ARG arg);

class TestRequestClient : public RequestClient {
public:
    friend THREAD_RET THREAD_FUNCTION_CALL TestRequestClientRun(THREAD_ARG arg);
    TestRequestClient(bool isLong);
protected:
    bool testRun();
};

THREAD_RET THREAD_FUNCTION_CALL TestRequestClientRun(THREAD_ARG arg);

}

#endif // _REQUESTCLIENT_H_