RequestGateway.cpp

Go to the documentation of this file.

#include "RequestGateway.h"
#include "UnitTestFramework.h"
 
namespace   cmlabs{
 

// Request Queue
 
RequestQueue::RequestQueue() {
    id = 0;
}
 
RequestQueue::~RequestQueue() {
    if (mutex.enter(500, __FUNCTION__))
        requestQ.clear();
    mutex.leave();
}
 
bool RequestQueue::init(uint32 id) {
    this->id = id;
    return true;
}
 
bool RequestQueue::addRequest(RequestReply* req, double priority) {
    if (!req || !req->gatewayRef)
        return false;
    if (!mutex.enter(500, __FUNCTION__))
        return false;
    std::list<RequestReply*>::iterator i = requestQ.begin(), e = requestQ.end(), p = requestQ.end();
    while (i != e) {
        if (((*i) == req) || ((*i)->gatewayRef == req->gatewayRef)) {
            mutex.leave();
            return false;
        }
        if ((*i)->origin == req->origin)
            p = i;
        i++;
    }
    if (priority > 0) {
        if (p == e) {
            requestQ.push_front(req);
            //LogPrint(0, LOG_SYSTEM, 2, "--- Priority [%.3f] --> front\n", priority);
        }
        else {
            p++;
            if (p != e) p++;
            if ((p != e) && (priority < 0.5)) p++;
            if ((p != e) && (priority < 0.2)) p++;
            if (p == e) {
                requestQ.push_back(req);
                //LogPrint(0, LOG_SYSTEM, 2, "--- Priority [%.3f] --> end\n", priority);
            }
            else {
                requestQ.insert(p, req);
                //LogPrint(0, LOG_SYSTEM, 1, "--- Priority [%.3f] --> middle\n", priority);
            }
        }
    }
    else
        requestQ.push_back(req);
    mutex.leave();
    return true;
}
 
 
RequestReply* RequestQueue::getNextRequest() {
    if (!mutex.enter(500, __FUNCTION__))
        return NULL;
    RequestStatus status;
    RequestReply* req = NULL;
    std::list<RequestReply*>::iterator i = requestQ.begin(), e = requestQ.end();
    while (i != e) {
        req = *i;
        if ( (status = req->getStatus()) == RequestStatus::QUEUED) {
            // mark request as in use
            req->setStatus(RequestStatus::PROCESSING);
            mutex.leave();
            return req;
        }
        else if (status == RequestStatus::LOCALERROR) {
            requestQ.remove(req);
            delete req;
            mutex.leave();
            return NULL;
        }
        i++;
    }
    mutex.leave();
    return NULL;
}
 
RequestReply* RequestQueue::getNextTimedoutRequest() {
    if (!mutex.enter(500, __FUNCTION__))
        return NULL;
    RequestStatus status;
    RequestReply* req = NULL;
    std::list<RequestReply*>::iterator i = requestQ.begin(), e = requestQ.end();
    while (i != e) {
        req = *i;
        if ((status = req->getStatus()) == RequestStatus::TIMEOUT) {
            requestQ.remove(req);
            mutex.leave();
            return req;
        }
        i++;
    }
    mutex.leave();
    return NULL;
}
 
 
//RequestReply* RequestQueue::getRequest(uint64 ref) {
//  if (!mutex.enter(1000))
//      return NULL;
//  RequestReply* req = NULL;
//  std::list<RequestReply*>::iterator i = requestQ.begin(), e = requestQ.end();
//  while (i != e) {
//      req = *i;
//      if (req->gatewayRef == ref) {
//          mutex.leave();
//          return req;
//      }
//      i++;
//  }
//  mutex.leave();
//  return NULL;
//}
 
bool RequestQueue::completeRequest(RequestReply* req) {
    if (!mutex.enter(500, __FUNCTION__))
        return false;
    requestQ.remove(req);
    mutex.leave();
    return true;
}
 
std::list<RequestReply*> RequestQueue::takeQueue() {
    std::list<RequestReply*> q;
    if (!mutex.enter(1000, __FUNCTION__))
        return q;
    q.assign(requestQ.begin(), requestQ.end());
    requestQ.clear();
    mutex.leave();
    return q;
}
 
uint32 RequestQueue::removeStaleRequests(uint32 ttlMS) {
    uint32 count = 0;
    if (!mutex.enter(500, __FUNCTION__))
        return 0;
    RequestStatus status;
    RequestReply* req = NULL;
    std::list<RequestReply*>::iterator i = requestQ.begin(), e = requestQ.end();
    while (i != e) {
        req = *i;
        status = req->getStatus();
        if ((status == RequestStatus::PROCESSING) || (status == RequestStatus::SENT)) {
            if ((int64)GetTimeAgeMS(req->startTime) > (int64)ttlMS) {
                req->setStatus(RequestStatus::TIMEOUT);
                count++;
            }
        }
        i++;
    }
    mutex.leave();
    return count;
}
 
std::string RequestQueue::toXML() {
    if (!mutex.enter(1000, __FUNCTION__))
        return "";
    uint32 p = 0;
    std::string xml;
    std::list<RequestReply*>::iterator i = requestQ.begin(), e = requestQ.end();
    while (i != e) {
        p++;
        xml += utils::StringFormat("<entry pos=\"%u\" ref=\"%llu\" origin=\"%u\" systemid=\"%u\" status=\"%s\" age=\"%d\" />\n",
            p, (*i)->gatewayRef, (*i)->origin, (*i)->systemID, (*i)->getStatusText().c_str(), GetTimeAgeMS((*i)->startTime));
        i++;
    }
    mutex.leave();
    return xml;
}
 
std::string RequestQueue::toJSON() {
    if (!mutex.enter(1000, __FUNCTION__))
        return "";
    uint32 p = 0;
    std::string json;
    std::list<RequestReply*>::iterator i = requestQ.begin(), e = requestQ.end();
    while (i != e) {
        if (!p) {
            p++;
            json += utils::StringFormat("{\"pos\":%u, \"ref\":%llu, \"origin\":%llu, \"systemid\":%u, \"status\":\"%s\", \"age\":%d}\n",
                p, (*i)->gatewayRef, (*i)->origin, (*i)->systemID, (*i)->getStatusText().c_str(), GetTimeAgeMS((*i)->startTime));
        }
        else {
            p++;
            json += utils::StringFormat(",{\"pos\":%u, \"ref\":%llu, \"origin\":%llu, \"systemid\":%u, \"status\":\"%s\", \"age\":%d}\n",
                p, (*i)->gatewayRef, (*i)->origin, (*i)->systemID, (*i)->getStatusText().c_str(), GetTimeAgeMS((*i)->startTime));
        }
        i++;
    }
    mutex.leave();
    return json;
}
 
 
 
 
 
 
 
 
 
 
 
bool RequestQueue::UnitTest() {
 
    uint64 lastRefID = 0;
    uint32 conid = 1;
 
    RequestQueue reqQ;
 
    unittest::progress(5, "empty queue");
    if (reqQ.getCount() != 0) { unittest::fail("RequestQueue test: fresh queue not empty"); return false; }
    if (reqQ.getNextRequest() != NULL) { unittest::fail("RequestQueue test: getNextRequest on empty queue returned non-NULL"); return false; }
 
    // Add a single queued request and verify it can be retrieved.
    unittest::progress(20, "single add/get");
    RequestReply* reply = new RequestReply();
    reply->origin = conid;
    reply->clientRef = 15;
    reply->gatewayRef = ++lastRefID;
    reply->giveRequestMessage(new DataMessage());
    reply->setStatus(QUEUED);
    if (!reqQ.addRequest(reply)) { unittest::fail("RequestQueue test: addRequest failed for first request"); return false; }
    if (reqQ.getCount() != 1) { unittest::fail("RequestQueue test: count != 1 after first add"); return false; }
 
    // Duplicate gatewayRef must be rejected.
    if (reqQ.addRequest(reply)) { unittest::fail("RequestQueue test: duplicate addRequest unexpectedly succeeded"); return false; }
 
    // getNextRequest should hand back the queued request and mark it PROCESSING.
    RequestReply* got = reqQ.getNextRequest();
    if (got != reply) { unittest::fail("RequestQueue test: getNextRequest did not return the queued request"); return false; }
    if (got->getStatus() != PROCESSING) { unittest::fail("RequestQueue test: retrieved request not marked PROCESSING"); return false; }
    // Already PROCESSING (not QUEUED), so a second call finds nothing to dispatch.
    if (reqQ.getNextRequest() != NULL) { unittest::fail("RequestQueue test: getNextRequest returned a non-QUEUED request"); return false; }
 
    // Complete it and confirm the queue empties.
    if (!reqQ.completeRequest(got)) { unittest::fail("RequestQueue test: completeRequest failed"); return false; }
    if (reqQ.getCount() != 0) { unittest::fail("RequestQueue test: count != 0 after completeRequest"); return false; }
    delete got;
 
    // Priority ordering: enqueue a batch, ensure a high-priority insert lands ahead.
    unittest::progress(45, "priority ordering");
    const uint32 N = 2000;
    uint64 t0 = GetTimeNow();
    for (uint32 k = 0; k < N; k++) {
        RequestReply* r = new RequestReply();
        r->origin = 1000 + k;            // distinct origins so none collide
        r->gatewayRef = ++lastRefID;
        r->giveRequestMessage(new DataMessage());
        r->setStatus(QUEUED);
        if (!reqQ.addRequest(r, (k == N - 1) ? 1.0 : 0.0)) { unittest::fail("RequestQueue test: bulk addRequest failed"); return false; }
    }
    double addUs = (double)(GetTimeNow() - t0);
    if (reqQ.getCount() != N) { unittest::fail("RequestQueue test: count mismatch after bulk add"); return false; }
 
    // The high-priority (last-added) request should now be at the front.
    unittest::progress(70, "drain queue");
    RequestReply* front = reqQ.getNextRequest();
    if (!front || front->origin != (uint64)(1000 + (N - 1))) {
        unittest::fail("RequestQueue test: high-priority request not at front");
        return false;
    }
 
    // Drain everything via takeQueue and free it.
    std::list<RequestReply*> drained = reqQ.takeQueue();
    if (drained.size() != N) { unittest::fail("RequestQueue test: takeQueue returned wrong count"); return false; }
    if (reqQ.getCount() != 0) { unittest::fail("RequestQueue test: queue not empty after takeQueue"); return false; }
    for (std::list<RequestReply*>::iterator i = drained.begin(), e = drained.end(); i != e; i++)
        delete(*i);
 
    unittest::detail("RequestQueue: added %u requests in %.3f ms\n", N, addUs / 1000.0);
    unittest::metric("enqueue_throughput", (double)N / addUs * 1e6, "ops/s", true);
    unittest::metric("avg_enqueue_latency", addUs / (double)N, "us", false);
 
    unittest::progress(100, "done");
    return true;
}
 
 
 
 
 
 
 
 
 
 
 

// RequestGateway
 
RequestGateway::RequestGateway(uint32 id, const char* version) {
    systemStartTime = GetTimeNow();
    if (version)
        versionString = version;
    //httpAuth[base64_encode(utils::StringFormat("user2:password"))] = GetTimeNow();
    cacheFiles = false;
    replyXML = true;
    webServerName = "Server";
    internalAPITitle = "gw/";
    externalAPITitle = "api/";
    sslSupport = false;
    threadID = threadIDClient = 0;
    lastRefID = 0;
    clientSentCount = 0;
    clientReceivedCount = 0;
    execSentCount = 0;
    execReceivedCount = 0;
    shouldContinue = true;
    isRunning = false;
    this->id = id;
    channel = NULL;
    manager = new NetworkManager();
    longReqLimit = 0;
    executorHeartbeatTimeout = 0;
    maxExecutorRequestTimeoutMS = 0;
    maxRequestQueueSize = 100;
    maxRequestProcessingSize = 3;
    priorityThreshold = 60000;
    callLogMax = 1000;
    callLogCount = 0;
}
 
RequestGateway::~RequestGateway() {
    shouldContinue = false;
    stop();
 
    while (ThreadManager::IsThreadRunning(threadID))
        utils::Sleep(20);
    while (ThreadManager::IsThreadRunning(threadIDClient))
        utils::Sleep(20);
 
    std::map<uint64, RequestReply*>::iterator i, e;
    mutex.enter(200, __FUNCTION__);
    for (i = requestMap.begin(), e = requestMap.end(); i != e; i++)
        delete(i->second);
    requestMap.clear();
    mutex.leave();
 
    delete(manager);
}
 
bool RequestGateway::init(const char* sslCertPath, const char* sslKeyPath) {
    if (sslCertPath && sslKeyPath) {
        if (!manager->setSSLCertificate(sslCertPath, sslKeyPath)) {
            LogPrint(0, LOG_SYSTEM, 0, "SSL Certificate or Key not found in paths: %s and %s!", sslCertPath, sslKeyPath);
            sslSupport = false;
            return false;
        }
        sslSupport = true;
    }
 
    if (!mutex.enter(200, __FUNCTION__))
        return false;
 
    if (!ThreadManager::CreateThread(RequestGatewayExecRun, this, threadID, 0)) {
        shouldContinue = false;
        mutex.leave();
        return false;
    }
    if (!ThreadManager::CreateThread(RequestGatewayClientRun, this, threadIDClient, 0)) {
        shouldContinue = false;
        mutex.leave();
        return false;
    }
 
    mutex.leave();
    return true;
}
 
bool RequestGateway::setExecutorHeartbeatTimeout(uint32 timeout) {
    executorHeartbeatTimeout = timeout;
    return true;
}
 
bool RequestGateway::setMaxExecutorRequestTimeout(uint32 timeout) {
    maxExecutorRequestTimeoutMS = timeout;
    return true;
}
 
bool RequestGateway::setResponseType(const char* type) {
    if (!type || !strlen(type))
        return false;
    replyXML = (strstr(type, "xml") || strstr(type, "XML"));
    return true;
}
 
bool RequestGateway::setWebServerInfo(const char* name, const char* rootdir, const char* indexfile) {
    if (!name || !rootdir)
        return false;
    webServerName = name;
    this->rootdir = rootdir;
    if (indexfile)
        indexFilename = indexfile;
    return true;
}
 
bool RequestGateway::setExternalAPILabel(const char* label) {
    if (!label)
        return false;
    externalAPITitle = label;
    return true;
}
 
bool RequestGateway::setInternalAPILabel(const char* label) {
    if (!label)
        return false;
    internalAPITitle = label;
    return true;
}
 
bool RequestGateway::setCacheFiles(bool cache) {
    cacheFiles = cache;
    return true;
}
 
bool RequestGateway::setQueuingParameters(uint32 maxRequestQueueSize, uint32 maxRequestProcessingSize, uint32 priorityThreshold) {
    this->maxRequestQueueSize = maxRequestQueueSize;
    this->maxRequestProcessingSize = maxRequestProcessingSize;
    this->priorityThreshold = priorityThreshold;
    return true;
}
 
bool RequestGateway::addPort(uint16 port, uint8 encryption, bool enableHTTP, uint32 timeout) {
    if (!mutex.enter(200, __FUNCTION__))
        return false;
 
    if ((encryption == SSLENC) && !sslSupport) {
        LogPrint(0, LOG_SYSTEM, 0, "SSL encryption not enabled!");
        mutex.leave();
        return false;
    }
 
    if (enableHTTP) {
        if (!channel) {
            channel = manager->createListener(port, encryption, PROTOCOL_HTTP_SERVER, true, timeout, false, 0, this);
            if (!channel) {
                mutex.leave();
                return false;
            }
        }
        else {
            if (!channel->startListener(0, port, encryption, PROTOCOL_HTTP_SERVER, true, timeout, false)) {
                mutex.leave();
                return false;
            }
        }
    }
 
    if (!channel) {
        channel = manager->createListener(port, encryption, PROTOCOL_MESSAGE, true, timeout, false, 0, this);
        if (!channel) {
            mutex.leave();
            return false;
        }
    }
    else {
        if (!channel->startListener(0, port, encryption, PROTOCOL_MESSAGE, true, timeout, false)) {
            mutex.leave();
            return false;
        }
    }
 
    mutex.leave();
    return true;
}
 
 
bool RequestGateway::addAuthUser(const char* user, const char* password) {
    if (!user)
        return false;
    if (password)
        httpAuth[base64_encode(utils::StringFormat("%s:%s", user, password))] = GetTimeNow();
    else
        httpAuth[base64_encode(utils::StringFormat("%s:", user, password))] = GetTimeNow();
    return true;
}
 
bool RequestGateway::addGateway(uint32 id, std::string addr, uint16 port) {
    return true;
}
 
bool RequestGateway::receiveNetworkEvent(NetworkEvent* evt, NetworkChannel* channel, uint64 conid) {
    if (!mutex.enter(200, __FUNCTION__))
        return false;
    uint64 now = GetTimeNow();
    std::map<uint64,RequestConnection>::iterator i;
    if ( (i = executors.find(conid)) != executors.end()) {
        switch(evt->type) {
            case NETWORKEVENT_DISCONNECT:
                i->second.lastFailTime = now;
                break;
            case NETWORKEVENT_RECONNECT:
                i->second.lastFailTime = 0;
                break;
            default:
                break;
        }
    }
    else if ( (i = clients.find(conid)) != clients.end()) {
        switch(evt->type) {
            case NETWORKEVENT_DISCONNECT:
                i->second.lastFailTime = now;
                break;
            case NETWORKEVENT_RECONNECT:
                i->second.lastFailTime = 0;
                break;
            default:
                break;
        }
    }
    else if ((i = webClients.find(conid)) != webClients.end()) {
        switch (evt->type) {
        case NETWORKEVENT_DISCONNECT:
            i->second.lastFailTime = now;
            break;
        case NETWORKEVENT_RECONNECT:
            i->second.lastFailTime = 0;
            break;
        default:
            break;
        }
    }
    else if ((i = webSockets.find(conid)) != webSockets.end()) {
        switch (evt->type) {
        case NETWORKEVENT_DISCONNECT:
            i->second.lastFailTime = now;
            break;
        case NETWORKEVENT_RECONNECT:
            i->second.lastFailTime = 0;
            break;
        default:
            break;
        }
    }
    delete(evt);
    mutex.leave();
    return true;
}
 
bool RequestGateway::receiveMessage(DataMessage* msg, NetworkChannel* channel, uint64 conid) {
    if (!msg) return false;
    RequestConnection *con;
    int64 val1, val2;
    std::map<uint64,RequestConnection>::iterator i;
    uint64 now = GetTimeNow();
    const char* req = msg->getString("URI");
 
    if (!mutex.enter(1000, __FUNCTION__)) {
        LogPrint(0, LOG_SYSTEM, 0, "Error locking Gateway mutex");
        delete(msg);
        return true;
    }
 
    msg->setRecvTime(now);
 
    if ( (i=executors.find(conid)) != executors.end()) {
        if (req && (strcmp(req, "ExecutorStatus") == 0) &&
             msg->getInt("ShortQSize", val1) &&
             msg->getInt("LongQSize", val2) ) {
            i->second.lastStatusTime = now;
            i->second.reportedShortReqQSize = (uint32) val1;
            i->second.reportedLongReqQSize = (uint32) val2;
            delete(msg);
            LogPrint(0, LOG_NETWORK, 8, "Got heartbeat from Executor %llu...", i->second.conID);
        }
        else {
            LogPrint(0, LOG_NETWORK, 3, "Got reply on request %llu from Executor %llu...", msg->getReference(), i->second.conID);
            execReceivedCount++;
 
            std::map<uint64, RequestReply*>::iterator ii = requestMap.find(msg->getReference());
            if (ii != requestMap.end())
                ii->second->setStatus(SUCCESS);
            replyQ.add(msg);
        }
    }
    else if (clients.find(conid) != clients.end()) {
        LogPrint(0, LOG_NETWORK, 3, "   - Gateway received message from Client %llu (%llu - %s)", conid, msg->getReference(), req);
        if (req && (strcmp(req, "ClientStatus") == 0)) {
            // for now do nothing, just testing the connection
            delete(msg);
        }
        else {
            LogPrint(0, LOG_NETWORK, 3, "Gateway received request from client %llu: '%s'", conid, req);
            uint64 origin = channel->getRemoteAddress(conid);
 
            addToRequestQueue(msg, origin, conid, msg->getReference());
        }
    }
    else {
        if (req && strcmp(req, "ExecutorConnect") == 0) {
            std::vector<std::string> longRequests = utils::TextListSplit(msg->getString("LongRequests"), ",");
            if (longRequests.size())
                longReqNames = longRequests;
            con = &executors[conid];
            con->clear();
            con->conID = conid;
            con->lastConTime = now;
            con->location = channel->getRemoteAddress(conid);
            //executors[conid] = con;
            LogPrint(0, LOG_SYSTEM, 0, "Executor connected on con: %llu", conid);
        }
        else if (req && strcmp(req, "ClientConnect") == 0) {
            con = &clients[conid];
            con->clear();
            con->conID = conid;
            con->lastConTime = now;
            con->location = channel->getRemoteAddress(conid);
            //clients[conid] = con;
            LogPrint(0, LOG_SYSTEM, 0, "Client connected on con: %llu", conid);
        }
        delete(msg);
    }
    mutex.leave();
    return true;
}
 
bool RequestGateway::addToRequestQueue(DataMessage* msg, uint64 origin, uint64 conID, uint64 clientRef) {
 
    msg->setString("REMOTE_ADDR", utils::StringFormat("%u.%u.%u.%u:%u", GETIPADDRESSQUADPORT(origin)).c_str());
 
    RequestReply* reply = new RequestReply();
    reply->origin = conID;
    reply->gatewayRef = ++lastRefID;
    if (clientRef)
        reply->clientRef = clientRef;
    else
        reply->clientRef = reply->gatewayRef;
    msg->setReference(reply->gatewayRef);
    reply->giveRequestMessage(msg);
    reply->setStatus(QUEUED);
    clientReceivedCount++;
 
    // Add to requestMap
    requestMap[reply->gatewayRef] = reply;
    //printf("Adding request GWID: %llu\n", reply->gatewayRef);
 
    return addRequestReplyToRequestQueue(reply);
}
 
bool RequestGateway::addRequestReplyToRequestQueue(RequestReply* reply) {
    DataMessage *replyMsg;
    DataMessage* msg = reply->peekRequestMessage();
    if (!msg) {
        LogPrint(0, LOG_SYSTEM, 0, "No request message found!");
        reply->setStatus(FAILED);
        replyMsg = new DataMessage();
        //replyMsg->setReference(reply->clientRef);
        replyMsg->setReference(reply->gatewayRef);
        if (replyXML)
            replyMsg->setString("XML", "<error>No request message available</error>");
        else
            replyMsg->setString("JSON", "{\"error\": \"No request message available\"}\n");
        replyQ.add(replyMsg);
        return false;
    }
 
    // Decide if short or long request
    const char* reqStr = msg->getString("URI");
    if (!reqStr)
        reqStr = msg->getString("REQUEST");
    if (!reqStr) {
        LogPrint(0, LOG_SYSTEM, 0, "No request string found!");
        reply->setStatus(FAILED);
        replyMsg = new DataMessage();
        //replyMsg->setReference(reply->clientRef);
        replyMsg->setReference(reply->gatewayRef);
        if (replyXML)
            replyMsg->setString("XML", "<error>No request data available</error>");
        else
            replyMsg->setString("JSON", "{\"error\": \"No request data available\"}\n");
        replyQ.add(replyMsg);
        return false;
    }
 
    // Find the least busy executor
    uint64 bestID = getBestExecutorID(reqStr, msg->getSize(), reply->isLongReq);
    if (!bestID) {
        LogPrint(0, LOG_SYSTEM, 0, "No executors found!");
        reply->setStatus(FAILED);
        replyMsg = new DataMessage();
        //replyMsg->setReference(reply->clientRef);
        replyMsg->setReference(reply->gatewayRef);
        if (replyXML)
            replyMsg->setString("XML", "<error>No processors available</error>");
        else
            replyMsg->setString("JSON", "{\"error\": \"No processors available\"}\n");
        replyQ.add(replyMsg);
        return false;
    }
 
    // Add to executor queue
    RequestConnection* reqCon = &executors[bestID];
    if (!reqCon) {
        LogPrint(0, LOG_SYSTEM, 0, "Executors %llu not found!", bestID);
        reply->setStatus(FAILED);
        replyMsg = new DataMessage();
        //replyMsg->setReference(reply->clientRef);
        replyMsg->setReference(reply->gatewayRef);
        if (replyXML)
            replyMsg->setString("XML", "<error>No processors available</error>");
        else
            replyMsg->setString("JSON", "{\"error\": \"No processors available\"}\n");
        replyQ.add(replyMsg);
        return false;
    }
 
    reply->processor = bestID;
 
    double conPriority = 0;
    std::map<uint64, RequestConnection>::iterator i;
    if ( ((i = clients.find(reply->origin)) != clients.end()) || ((i = webClients.find(reply->origin)) != webClients.end())
        || ((i = webSockets.find(reply->origin)) != webSockets.end())) {
        if (i->second.lastConTime) {
            int32 conTime = GetTimeAgeMS(i->second.lastConTime);
            if (priorityThreshold && (conTime < (int64)priorityThreshold)) {
                conPriority = (((double)priorityThreshold) - conTime) / priorityThreshold;
            }
        }
    }
 
    if (reply->isLongReq) {
        if (reqCon->longReqQueue.getCount() >= maxRequestQueueSize) {
            LogPrint(0, LOG_SYSTEM, 0, "All long executors too busy (%llu), rejecting request ID %llu", bestID, reply->gatewayRef);
            reply->setStatus(TOOBUSY);
            reply->processor = 0;
            replyMsg = new DataMessage();
            //replyMsg->setReference(reply->clientRef);
            replyMsg->setReference(reply->gatewayRef);
            if (replyXML)
                replyMsg->setString("XML", "<error>All long processors too busy</error>");
            else
                replyMsg->setString("JSON", "{\"error\": \"All long processors too busy\"}\n");
            replyQ.add(replyMsg);
            return false;
        }
        reqCon->longReqQueue.addRequest(reply, conPriority);
 
        /*
        Long requests are added to the relevant queue
        - when added check if we can send it right away, if so, add to execQ
        - if not, leave it in the relevant queue
        - when reply comes back from executor to free up a space
        choose the next one in the queue and add to execQ
        - periodically check queues for missed slots
        */
 
        if ((reqCon->longReqQProcessingSize < maxRequestProcessingSize) && (reqCon->longReqQueue.getCount() < maxRequestProcessingSize)) {
            reply->setStatus(PROCESSING);
            reqCon->longReqQProcessingSize++;
            execQ.add(reply);
            LogPrint(0, LOG_SYSTEM, 3, "Sending LONG request %llu to Executor %llu immediately (%u)...", reply->gatewayRef, reply->processor, reqCon->longReqQueue.getCount());
        }
        else {
            LogPrint(0, LOG_SYSTEM, 3, "Added LONG request %llu to Executor %llu queue (%u)...", reply->gatewayRef, reply->processor, reqCon->longReqQueue.getCount());
        }
    }
    else {
        if (reqCon->shortReqQueue.getCount() >= maxRequestQueueSize) {
            LogPrint(0, LOG_SYSTEM, 0, "All short executors too busy (%llu), rejecting request!", bestID);
            reply->setStatus(TOOBUSY);
            reply->processor = 0;
            replyMsg = new DataMessage();
            //replyMsg->setReference(reply->clientRef);
            replyMsg->setReference(reply->gatewayRef);
            if (replyXML)
                replyMsg->setString("XML", "<error>All short processors too busy</error>");
            else
                replyMsg->setString("JSON", "{\"error\": \"All short processors too busy\"}\n");
            replyQ.add(replyMsg);
            return false;
        }
        reqCon->shortReqQueue.addRequest(reply, conPriority);
 
        // Short requests are added to the relevant queue and put directly into the execQ for sending
        reply->setStatus(PROCESSING);
        execQ.add(reply);
        //LogPrint(0, LOG_SYSTEM, 0, "Sending SHORT request %llu to Executor %llu immediately (%u)...", reply->gatewayRef, reply->processor, reqCon->shortReqQueue.getCount());
    }
    return true;
}
 
 
bool RequestGateway::receiveHTTPRequest(HTTPRequest* req, NetworkChannel* channel, uint64 conid) {
//  const char* requestString = req->getRequest();
    const char* requestString = req->getURI();
    if (!requestString || !strlen(requestString))
        return false;
 
    uint64 now = GetTimeNow();
    requestString++; // skip past the first /
 
    HTTPReply* replyPage = NULL;
    //char* exttype;
 
//  uint64 t1 = GetTimeNow();
 
    if (!mutex.enter(1000, __FUNCTION__))
        return false;
 
//  uint64 t2 = GetTimeNow();
 
//  uint32 contentSize = 0;
//  utils::WriteAFile("d:/requestin.dat", req->getRawContent(contentSize), contentSize, true);
 
    //uint64 start = GetTimeNow();
    if (strstr(requestString, internalAPITitle.c_str()) == requestString) {
        callInternalAPI(requestString + internalAPITitle.length(), req, channel, conid);
        //printf("*** Internal API: %.3fms\n", (double)GetTimeAge(start)/1000.0);
        delete(req);
        mutex.leave();
        //printf("\n         -----     HTTP Internal took %s       (waited: %s)\n\n",
        //  PrintTimeDifString(GetTimeAge(t2)).c_str(),
        //  PrintTimeDifString(t2-t1).c_str());
        return true;
    }
    else if (rootdir.length()) {
        //exttype = new char[1024];
        //utils::strcpyavail(exttype, "text/html", 1024, true);
        if ( (!requestString || !strlen(requestString)) && indexFilename.length()) {
            LogPrint(0, LOG_NETWORK, 3, "Serving up root index file: %s", indexFilename.c_str());
            replyPage = new HTTPReply();
            replyPage->createFromFile(GetTimeNow(), webServerName.c_str(), req->ifModifiedSince, true, cacheFiles, (rootdir+indexFilename).c_str());
 
            addToCallLog(++callLogCount, 2, 3, false, req->source, 0, 0, 0, req->endReceiveTime, GetTimeNow(), 0, GetTimeAge(now), req->type, req->getSize(), replyPage->getSize(), "Success",
                utils::StringFormat("Index: %s", indexFilename.c_str()).c_str(), req->getHeaderEntry("User-Agent"), "");
 
            //printf("*** Read default file: %.3fms\n", (double)GetTimeAge(start)/1000.0);
            //if (!(html = utils::ReadAFileString(rootdir+indexFilename)).size())
            //  html = utils::StringFormat("Could read index file...");
        }
        else if (strstr(requestString, externalAPITitle.c_str()) == requestString) {
            LogPrint(0, LOG_NETWORK, 3, "Serving up API call: %s", requestString);
            callExternalAPI(requestString+externalAPITitle.length(), req, channel, conid);
            //printf("*** External API: %.3fms\n", (double)GetTimeAge(start)/1000.0);
            delete(req);
            mutex.leave();
            //printf("\n         -----     HTTP Executor took %s       (waited: %s)\n\n",
            //  PrintTimeDifString(GetTimeAge(t2)).c_str(),
            //  PrintTimeDifString(t2 - t1).c_str());
            return true;
        }
        else { // if (strstr(requestString, "html/") == requestString) {
            LogPrint(0, LOG_NETWORK, 3, "Serving up hosted file for request: %s", requestString);
            replyPage = new HTTPReply();
            replyPage->createFromFile(GetTimeNow(), webServerName.c_str(), req->ifModifiedSince, true, cacheFiles, (rootdir+requestString).c_str());
 
            addToCallLog(++callLogCount, 2, 3, false, req->source, 0, 0, 0, req->endReceiveTime, GetTimeNow(), 0, GetTimeAge(now), req->type, req->getSize(), replyPage->getSize(), "Success",
                utils::StringFormat("File: %s", requestString).c_str(), req->getHeaderEntry("User-Agent"), "");
 
            //printf("*** Read file '%s': %.3fms\n", requestString, (double)GetTimeAge(start)/1000.0);
            //const char* filename = requestString;
            //if ( utils::TextEndsWith(filename, ".html", false) || utils::TextEndsWith(filename, ".txt", false) ) {
            //  if (!(html = utils::ReadAFileString(rootdir+filename)).size())
            //      html = utils::StringFormat("Could read file '%s'...", filename);
            //}
            //else {
            //  if ( !(data = utils::ReadAFile((rootdir+filename).c_str(), datasize, true)))
            //      html = utils::StringFormat("Could read file '%s'...", filename);
            //  else {
            //      if (!html::GetMimeType(exttype, filename, 1024))
            //          html = utils::StringFormat("Unsupported mime type for file '%s'...", filename);
            //  }
            //}
        }
        //if (!replyPage) {
        //  if (data) {
        //      replyPage = new HTTPReply();
        //      replyPage->createPage(HTTP_OK, GetTimeNow(), webServerName.c_str(), 0, true, cacheFiles, exttype, data, datasize);
        //  }
        //  else if (html.length()) {
        //      replyPage = new HTTPReply();
        //      replyPage->createPage(HTTP_OK, GetTimeNow(), webServerName.c_str(), 0, true, cacheFiles, exttype, html.c_str());
        //  }
        //}
        if (replyPage) {
            mutex.leave();
            channel->sendHTTPReply(replyPage, conid);
            delete(replyPage);
            //delete [] data;
            //delete [] exttype;
            delete(req);
            return true;
        }
        else {
            //delete [] data;
            //delete [] exttype;
        }
    }
    mutex.leave();
    delete(req);
    //printf("*** No reply: %.3fms\n", (double)GetTimeAge(start)/1000.0);
    return false;
 
}
 
 
 
 
 
 
 
 
 
bool RequestGateway::receiveWebsocketData(WebsocketData* wsData, NetworkChannel* channel, uint64 conid) {
 
    if (!wsData)
        return false;
 
 
    JSONM* jm = new JSONM(wsData->data, wsData->contentSize);
    if (!jm->isValid()) {
        LogPrint(0, LOG_NETWORK, 0, "Invalid Websocket request received (%llu):\n%s", wsData->contentSize, wsData->data);
        delete jm;
        return false;
    }
    std::string json = jm->getJSON();
 
    LogPrint(0, LOG_NETWORK, 3, "---- WebSocketData  JSON size: %u  data size: %llu   data count: %llu\n%s\n",
        (uint32)json.length(), jm->jmSize - json.length(), jm->entries.size(), json.c_str());
 
    //"operation": "POST",
    //"uri" : "/processing",
    //"authorization" : "abcdefg",
    //"requestid" : 223,
 
    jsmn_parser parser;
    jsmntok_t* tokens = new jsmntok_t[1024];
    int tokenCount = 0;
    jsmn_init(&parser);
    if ((tokenCount = jsmn_parse(&parser, json.c_str(), json.length(), tokens, 1024)) <= 0) {
        // No JSON found
        LogPrint(0, LOG_NETWORK, 0, "Invalid Websocket JSON request received (%llu):\n%s", wsData->contentSize, json.c_str());
        delete jm;
        delete[] tokens;
        return false;
    }
 
    std::string requestString = GetJSONChildValueString(tokens, tokenCount, json.c_str(), "uri", 0, JSMN_UNDEFINED);
    uint64 ref = GetJSONChildValueUint64(tokens, tokenCount, json.c_str(), "requestid", 0);
    std::string operation = GetJSONChildValueString(tokens, tokenCount, json.c_str(), "operation", 0, JSMN_UNDEFINED);
    std::string authorization = GetJSONChildValueString(tokens, tokenCount, json.c_str(), "authorization", 0, JSMN_UNDEFINED);
 
    if (!requestString.length() || !ref || !operation.length()) {
        LogPrint(0, LOG_NETWORK, 0, "Invalid Websocket JSON request received, required JSON values not found (%llu):\n%s", wsData->contentSize, wsData->data);
        delete jm;
        delete[] tokens;
        return false;
    }
 
    // convert to DataMessage
    DataMessage* msg = jm->convertToMessage();
    if (!msg) {
        LogPrint(0, LOG_NETWORK, 0, "Invalid Websocket request received, message unknown");
        delete jm;
        delete[] tokens;
        return false;
    }
 
    if (stricmp(operation.c_str(), "post") == 0)
        msg->setInt("HTTP_OPERATION", HTTP_POST);
    else if (stricmp(operation.c_str(), "get") == 0)
        msg->setInt("HTTP_OPERATION", HTTP_GET);
    else if (stricmp(operation.c_str(), "put") == 0)
        msg->setInt("HTTP_OPERATION", HTTP_PUT);
    else if (stricmp(operation.c_str(), "delete") == 0)
        msg->setInt("HTTP_OPERATION", HTTP_DELETE);
    else {
        LogPrint(0, LOG_NETWORK, 0, "Invalid Websocket operation received '%s'", operation.c_str());
        delete jm;
        delete[] tokens;
        return false;
    }
 
    msg->setString("REQUEST", requestString.c_str());
    msg->setString("URI", requestString.c_str());
    msg->setString("Authorization", authorization.c_str());
 
    RequestConnection *con;
    std::map<uint64, RequestConnection>::iterator i;
    //RequestReply* reply;
    uint64 now = GetTimeNow();
 
    if (!mutex.enter(1000, __FUNCTION__)) {
        LogPrint(0, LOG_SYSTEM, 0, "Error locking Gateway mutex");
        delete(msg);
        delete jm;
        delete[] tokens;
        return true;
    }
 
    if ((i = webSockets.find(conid)) == webSockets.end()) {
        con = &webSockets[conid];
        con->clear();
        con->conID = conid;
        con->lastConTime = now;
    }
 
    addToRequestQueue(msg, wsData->source, conid, ref);
 
    mutex.leave();
    delete jm;
    delete[] tokens;
    return true;
}
 
 
 
 
 
 
 
 
 
 
 
 
bool RequestGateway::callExternalAPI(const char* apiName, HTTPRequest* req, NetworkChannel* channel, uint64 conid) {
    RequestConnection *con;
    std::map<uint64,RequestConnection>::iterator i;
    //RequestReply* reply;
    uint64 now = GetTimeNow();
    DataMessage* msg;
    uint64 ref = utils::Ascii2Uint64(req->getParameter("Reference"));
 
    if ( (i=webClients.find(conid)) == webClients.end()) {
        con = &webClients[conid];
        con->clear();
        con->conID = conid;
        con->lastConTime = now;
        //webClients[conid] = con;
    }
 
    // convert to DataMessage
    msg = req->convertToMessage();
//  msg->setString("REQUEST", apiName);
    msg->setString("URI", apiName);
 
    addToRequestQueue(msg, req->source, conid, ref);
    return true;
}
 
bool RequestGateway::callInternalAPI(const char* apiName, HTTPRequest* req, NetworkChannel* channel, uint64 conid) {
    HTTPReply* reply = new HTTPReply();
    std::string xml, subXML, json, execJSON, clientJSON, webClientJSON, webSocketJSON;
    std::map<uint64,RequestConnection>::iterator i,e;
    std::list<CallLogEntry*>::reverse_iterator ic, ec;
    uint32 clientCount = 0, execCount = 0, webClientCount = 0, webSocketCount = 0;
    uint64 now = GetTimeNow();
    uint32 ms1 = 1000, ms2 = 10000, ms3 = 60000;
    double lvalPerSec1, lcountPerSec1, lvalPerSec2, lcountPerSec2, lvalPerSec3, lcountPerSec3;
    double svalPerSec1, scountPerSec1, svalPerSec2, scountPerSec2, svalPerSec3, scountPerSec3;
 
    if (httpAuth.size()) {
        // https://developer.mozilla.org/en-US/docs/Web/HTTP/Authentication
        const char* auth = req->getBasicAuthorization();
        if (!auth || (httpAuth.find(auth) == httpAuth.end())) {
            reply = HTTPReply::CreateAuthorizationReply("Gateway");
            // mutex.leave();
            channel->sendHTTPReply(reply, conid);
            //delete(req);
 
            addToCallLog(++callLogCount, 2, 3, false, req->source, 0, 0, 0, req->endReceiveTime, GetTimeNow(), 0, GetTimeAge(now), req->type, req->getSize(), reply->getSize(), "Auth Failed",
                utils::StringFormat("Internal: %s", apiName).c_str(), req->getHeaderEntry("User-Agent"), "");
 
            delete(reply);
            return true;
        }
    }
 
    if (((stricmp(apiName, "status") == 0)) || utils::TextStartsWith(apiName, "status?")) {
        svalPerSec1 = scountPerSec1 = svalPerSec2 = scountPerSec2 = svalPerSec3 = scountPerSec3 =
            lvalPerSec1 = lcountPerSec1 = lvalPerSec2 = lcountPerSec2 = lvalPerSec3 = lcountPerSec3 = 0;
        for (i=executors.begin(), e=executors.end(); i!=e; i++) {
            if (!i->second.lastFailTime) {
                //i->second.shortAvgStats.getThroughputMulti(ms1, ms2, ms3, svalPerSec1, scountPerSec1, svalPerSec2, scountPerSec2, svalPerSec3, scountPerSec3, now);
                //i->second.longAvgStats.getThroughputMulti(ms1, ms2, ms3, lvalPerSec1, lcountPerSec1, lvalPerSec2, lcountPerSec2, lvalPerSec3, lcountPerSec3, now);
                if (replyXML) {
                    subXML += utils::StringFormat("<executor id=\"%llu\" location=\"%u.%u.%u.%u:%u\" shortQueue=\"%u\" longQueue=\"%u\" count=\"%u\" uptime=\"%s\" statustime=\"%s\" />\n",
                        i->second.conID,
                        GETIPADDRESSQUADPORT(i->second.location),
                        i->second.shortReqQueue.getCount(),
                        i->second.longReqQueue.getCount(),
                        i->second.count,
                        PrintTimeDifString(GetTimeAge(i->second.lastConTime)).c_str(),
                        PrintTimeDifString(GetTimeAge(i->second.lastStatusTime)).c_str()
                        );
                }
                else {
                    if (execJSON.length())
                        execJSON += ",";
                    execJSON += utils::StringFormat("{\"id\":%llu, \"location\":\"%u.%u.%u.%u:%u\", \"shortQueue\":%u, \"longQueue\":%u, \"count\":%u, \"uptime\":\"%s\", \"statustime\":\"%s\" }\n",
                        i->second.conID,
                        GETIPADDRESSQUADPORT(i->second.location),
                        i->second.shortReqQueue.getCount(),
                        i->second.longReqQueue.getCount(),
                        i->second.count,
                        PrintTimeDifString(GetTimeAge(i->second.lastConTime)).c_str(),
                        PrintTimeDifString(GetTimeAge(i->second.lastStatusTime)).c_str()
                    );
                }
                execCount++;
            }
        }
        for (i=clients.begin(), e=clients.end(); i!=e; i++) {
            if (!i->second.lastFailTime) {
                //shortAvgStatsClients[i->second.conID].getThroughputMulti(ms1, ms2, ms3, svalPerSec1, scountPerSec1, svalPerSec2, scountPerSec2, svalPerSec3, scountPerSec3, now);
                //longAvgStatsClients[i->second.conID].getThroughputMulti(ms1, ms2, ms3, lvalPerSec1, lcountPerSec1, lvalPerSec2, lcountPerSec2, lvalPerSec3, lcountPerSec3, now);
                //i->second.shortAvgStats.getThroughputMulti(ms1, ms2, ms3, svalPerSec1, scountPerSec1, svalPerSec2, scountPerSec2, svalPerSec3, scountPerSec3, now);
                //i->second.longAvgStats.getThroughputMulti(ms1, ms2, ms3, lvalPerSec1, lcountPerSec1, lvalPerSec2, lcountPerSec2, lvalPerSec3, lcountPerSec3, now);
                if (replyXML) {
                    subXML += utils::StringFormat("<client id=\"%llu\" location=\"%u.%u.%u.%u:%u\" count=\"%u\" uptime=\"%s\" />\n",
                        i->second.conID,
                        GETIPADDRESSQUADPORT(i->second.location),
                        i->second.count,
                        PrintTimeDifString(GetTimeAge(i->second.lastConTime)).c_str()
                        );
                }
                else {
                    if (clientJSON.length())
                        clientJSON += ",";
                    clientJSON += utils::StringFormat("{\"id\":%llu, \"location\":\"%u.%u.%u.%u:%u\", \"count\":%u, \"uptime\":\"%s\" }\n",
                        i->second.conID,
                        GETIPADDRESSQUADPORT(i->second.location),
                        i->second.count,
                        PrintTimeDifString(GetTimeAge(i->second.lastConTime)).c_str()
                        );
                }
                clientCount++;
            }
        }
        for (i = webClients.begin(), e = webClients.end(); i != e; i++) {
            if (!i->second.lastFailTime) {
                //shortAvgStatsWeb[i->second.conID].getThroughputMulti(ms1, ms2, ms3, svalPerSec1, scountPerSec1, svalPerSec2, scountPerSec2, svalPerSec3, scountPerSec3, now);
                //longAvgStatsWeb[i->second.conID].getThroughputMulti(ms1, ms2, ms3, lvalPerSec1, lcountPerSec1, lvalPerSec2, lcountPerSec2, lvalPerSec3, lcountPerSec3, now);
                //i->second.shortAvgStats.getThroughputMulti(ms1, ms2, ms3, svalPerSec1, scountPerSec1, svalPerSec2, scountPerSec2, svalPerSec3, scountPerSec3, now);
                //i->second.longAvgStats.getThroughputMulti(ms1, ms2, ms3, lvalPerSec1, lcountPerSec1, lvalPerSec2, lcountPerSec2, lvalPerSec3, lcountPerSec3, now);
                if (replyXML) {
                    subXML += utils::StringFormat("<webclient id=\"%llu\" location=\"%u.%u.%u.%u:%u\" count=\"%u\" uptime=\"%s\" />\n",
                        i->second.conID,
                        GETIPADDRESSQUADPORT(i->second.location),
                        i->second.count,
                        PrintTimeDifString(GetTimeAge(i->second.lastConTime)).c_str()
                        );
                }
                else {
                    if (webClientJSON.length())
                        webClientJSON += ",";
                    webClientJSON += utils::StringFormat("{\"id\":%llu, \"location\":\"%u.%u.%u.%u:%u\", \"count\":%u, \"uptime\":\"%s\" }\n",
                        i->second.conID,
                        GETIPADDRESSQUADPORT(i->second.location),
                        i->second.count,
                        PrintTimeDifString(GetTimeAge(i->second.lastConTime)).c_str()
                        );
                }
                webClientCount++;
            }
        }
        for (i = webSockets.begin(), e = webSockets.end(); i != e; i++) {
            if (!i->second.lastFailTime) {
                //shortAvgStatsWeb[i->second.conID].getThroughputMulti(ms1, ms2, ms3, svalPerSec1, scountPerSec1, svalPerSec2, scountPerSec2, svalPerSec3, scountPerSec3, now);
                //longAvgStatsWeb[i->second.conID].getThroughputMulti(ms1, ms2, ms3, lvalPerSec1, lcountPerSec1, lvalPerSec2, lcountPerSec2, lvalPerSec3, lcountPerSec3, now);
                //i->second.shortAvgStats.getThroughputMulti(ms1, ms2, ms3, svalPerSec1, scountPerSec1, svalPerSec2, scountPerSec2, svalPerSec3, scountPerSec3, now);
                //i->second.longAvgStats.getThroughputMulti(ms1, ms2, ms3, lvalPerSec1, lcountPerSec1, lvalPerSec2, lcountPerSec2, lvalPerSec3, lcountPerSec3, now);
                if (replyXML) {
                    subXML += utils::StringFormat("<websocket id=\"%llu\" location=\"%u.%u.%u.%u:%u\" count=\"%u\" uptime=\"%s\" />\n",
                        i->second.conID,
                        GETIPADDRESSQUADPORT(i->second.location),
                        i->second.count,
                        PrintTimeDifString(GetTimeAge(i->second.lastConTime)).c_str()
                    );
                }
                else {
                    if (webSocketJSON.length())
                        webSocketJSON += ",";
                    webSocketJSON += utils::StringFormat("{\"id\":%llu, \"location\":\"%u.%u.%u.%u:%u\", \"count\":%u, \"uptime\":\"%s\" }\n",
                        i->second.conID,
                        GETIPADDRESSQUADPORT(i->second.location),
                        i->second.count,
                        PrintTimeDifString(GetTimeAge(i->second.lastConTime)).c_str()
                    );
                }
                webSocketCount++;
            }
        }
        shortAvgStats.getThroughputMulti(ms1, ms2, ms3, svalPerSec1, scountPerSec1, svalPerSec2, scountPerSec2, svalPerSec3, scountPerSec3, now);
        longAvgStats.getThroughputMulti(ms1, ms2, ms3, lvalPerSec1, lcountPerSec1, lvalPerSec2, lcountPerSec2, lvalPerSec3, lcountPerSec3, now);
        if (replyXML)
            xml = utils::StringFormat("<status uptime=\"%s\" version=\"%s\" executors=\"%u\" clients=\"%u\" webclients=\"%u\" websockets=\"%u\" maxqueuesize=\"%u\" minqueuesize=\"%u\" "
                "shortBytesPerSec1=\"%.3f\" shortCountPerSec1=\"%.3f\" shortBytesPerSec10=\"%.3f\" shortCountPerSec10=\"%.3f\" shortBytesPerSec60=\"%.3f\" shortCountPerSec60=\"%.3f\" "
                "longBytesPer1Sec=\"%.3f\" longCountPer1Sec=\"%.3f\" longBytesPer10Sec=\"%.3f\" longCountPer10Sec=\"%.3f\" longBytesPer60Sec=\"%.3f\" longCountPer60Sec=\"%.3f\" "
                ">\n%s</status>\n",
                PrintTimeDifString(GetTimeAge(systemStartTime)).c_str(), versionString.c_str(), execCount, clientCount, webClientCount, webSocketCount,
                maxRequestQueueSize, maxRequestProcessingSize,
                svalPerSec1, scountPerSec1, svalPerSec2, scountPerSec2, svalPerSec3, scountPerSec3,
                lvalPerSec1, lcountPerSec1, lvalPerSec2, lcountPerSec2, lvalPerSec3, lcountPerSec3,
                subXML.c_str());
        else
            json = utils::StringFormat("{\"uptime\":\"%s\", \"version\":\"%s\", \"executorcount\":%u, \"clientcount\":%u, \"webclientcount\":%u, \"websocketcount\":%u, \"maxqueuesize\":%u, \"minqueuesize\":%u, "
                "\"shortBytesPerSec1\": %.3f, \"shortCountPerSec1\": %.3f, \"shortBytesPerSec10\": %.3f, \"shortCountPerSec10\": %.3f, \"shortBytesPerSec60\": %.3f, \"shortCountPerSec60\": %.3f, "
                "\"longBytesPer1Sec\": %.3f, \"longCountPer1Sec\": %.3f, \"longBytesPer10Sec\": %.3f, \"longCountPer10Sec\": %.3f, \"longBytesPer60Sec\": %.3f, \"longCountPer60Sec\": %.3f, "
                "\"executors\":[%s], \"clients\":[%s], \"webclients\":[%s], \"websockets\":[%s]}\n",
                PrintTimeDifString(GetTimeAge(systemStartTime)).c_str(), versionString.c_str(), execCount, clientCount, webClientCount, webSocketCount,
                maxRequestQueueSize, maxRequestProcessingSize,
                svalPerSec1, scountPerSec1, svalPerSec2, scountPerSec2, svalPerSec3, scountPerSec3,
                lvalPerSec1, lcountPerSec1, lvalPerSec2, lcountPerSec2, lvalPerSec3, lcountPerSec3,
                execJSON.c_str(), clientJSON.c_str(), webClientJSON.c_str(), webSocketJSON.c_str());
    }
    else if (((stricmp(apiName, "extstatus") == 0)) || utils::TextStartsWith(apiName, "extstatus?")) {
        svalPerSec1 = scountPerSec1 = svalPerSec2 = scountPerSec2 = svalPerSec3 = scountPerSec3 =
            lvalPerSec1 = lcountPerSec1 = lvalPerSec2 = lcountPerSec2 = lvalPerSec3 = lcountPerSec3 = 0;
        for (i = executors.begin(), e = executors.end(); i != e; i++) {
            if (!i->second.lastFailTime) {
                i->second.shortAvgStats.getThroughputMulti(ms1, ms2, ms3, svalPerSec1, scountPerSec1, svalPerSec2, scountPerSec2, svalPerSec3, scountPerSec3, now);
                i->second.longAvgStats.getThroughputMulti(ms1, ms2, ms3, lvalPerSec1, lcountPerSec1, lvalPerSec2, lcountPerSec2, lvalPerSec3, lcountPerSec3, now);
                if (replyXML) {
                    subXML += utils::StringFormat("<executor id=\"%llu\" location=\"%u.%u.%u.%u:%u\" shortqueuesize=\"%u\" longqueuesize=\"%u\" count=\"%u\" uptime=\"%s\" statustime=\"%s\" "
                        "shortBytesPerSec1=\"%.3f\" shortCountPerSec1=\"%.3f\" shortBytesPerSec10=\"%.3f\" shortCountPerSec10=\"%.3f\" shortBytesPerSec60=\"%.3f\" shortCountPerSec60=\"%.3f\" "
                        "longBytesPer1Sec=\"%.3f\" longCountPer1Sec=\"%.3f\" longBytesPer10Sec=\"%.3f\" longCountPer10Sec=\"%.3f\" longBytesPer60Sec=\"%.3f\" longCountPer60Sec=\"%.3f\" ",
                        ">\n<shortqueue>\n%s</shortqueue>\n<longqueue>\n%s</longqueue></executor>\n",
                        i->second.conID,
                        GETIPADDRESSQUADPORT(i->second.location),
                        i->second.shortReqQueue.getCount(),
                        i->second.longReqQueue.getCount(),
                        i->second.count,
                        PrintTimeDifString(GetTimeAge(i->second.lastConTime)).c_str(),
                        PrintTimeDifString(GetTimeAge(i->second.lastStatusTime)).c_str(),
                        svalPerSec1, scountPerSec1, svalPerSec2, scountPerSec2, svalPerSec3, scountPerSec3,
                        lvalPerSec1, lcountPerSec1, lvalPerSec2, lcountPerSec2, lvalPerSec3, lcountPerSec3,
                        i->second.shortReqQueue.toXML().c_str(),
                        i->second.longReqQueue.toXML().c_str()
                    );
                }
                else {
                    if (execJSON.length())
                        execJSON += ",";
                    execJSON += utils::StringFormat("{\"id\":%llu, \"location\":\"%u.%u.%u.%u:%u\", \"shortqueuesize\":%u, \"longqueuesize\":%u, \"count\":%u, \"uptime\":\"%s\", \"statustime\":\"%s\", "
                        " \"shortBytesPerSec1\": %.3f, \"shortCountPerSec1\": %.3f, \"shortBytesPerSec10\": %.3f, \"shortCountPerSec10\": %.3f, \"shortBytesPerSec60\": %.3f, \"shortCountPerSec60\": %.3f, "
                        " \"longBytesPer1Sec\": %.3f, \"longCountPer1Sec\": %.3f, \"longBytesPer10Sec\": %.3f, \"longCountPer10Sec\": %.3f, \"longBytesPer60Sec\": %.3f, \"longCountPer60Sec\": %.3f, "
                        " \"shortqueue\":[%s], \"longqueue\":[%s] }\n",
                        i->second.conID,
                        GETIPADDRESSQUADPORT(i->second.location),
                        i->second.shortReqQueue.getCount(),
                        i->second.longReqQueue.getCount(),
                        i->second.count,
                        PrintTimeDifString(GetTimeAge(i->second.lastConTime)).c_str(),
                        PrintTimeDifString(GetTimeAge(i->second.lastStatusTime)).c_str(),
                        svalPerSec1, scountPerSec1, svalPerSec2, scountPerSec2, svalPerSec3, scountPerSec3,
                        lvalPerSec1, lcountPerSec1, lvalPerSec2, lcountPerSec2, lvalPerSec3, lcountPerSec3,
                        i->second.shortReqQueue.toJSON().c_str(),
                        i->second.longReqQueue.toJSON().c_str()
                    );
                }
                execCount++;
            }
        }
        for (i = clients.begin(), e = clients.end(); i != e; i++) {
            if (!i->second.lastFailTime) {
                //shortAvgStatsClients[i->second.conID].getThroughputMulti(ms1, ms2, ms3, svalPerSec1, scountPerSec1, svalPerSec2, scountPerSec2, svalPerSec3, scountPerSec3, now);
                //longAvgStatsClients[i->second.conID].getThroughputMulti(ms1, ms2, ms3, lvalPerSec1, lcountPerSec1, lvalPerSec2, lcountPerSec2, lvalPerSec3, lcountPerSec3, now);
                i->second.shortAvgStats.getThroughputMulti(ms1, ms2, ms3, svalPerSec1, scountPerSec1, svalPerSec2, scountPerSec2, svalPerSec3, scountPerSec3, now);
                i->second.longAvgStats.getThroughputMulti(ms1, ms2, ms3, lvalPerSec1, lcountPerSec1, lvalPerSec2, lcountPerSec2, lvalPerSec3, lcountPerSec3, now);
                if (replyXML) {
                    subXML += utils::StringFormat("<client id=\"%llu\" location=\"%u.%u.%u.%u:%u\" count=\"%u\" uptime=\"%s\" "
                        "shortBytesPerSec1=\"%.3f\" shortCountPerSec1=\"%.3f\" shortBytesPerSec10=\"%.3f\" shortCountPerSec10=\"%.3f\" shortBytesPerSec60=\"%.3f\" shortCountPerSec60=\"%.3f\" "
                        "longBytesPer1Sec=\"%.3f\" longCountPer1Sec=\"%.3f\" longBytesPer10Sec=\"%.3f\" longCountPer10Sec=\"%.3f\" longBytesPer60Sec=\"%.3f\" longCountPer60Sec=\"%.3f\" ",
                        "/>\n",
                        i->second.conID,
                        GETIPADDRESSQUADPORT(i->second.location),
                        i->second.count,
                        PrintTimeDifString(GetTimeAge(i->second.lastConTime)).c_str(),
                        svalPerSec1, scountPerSec1, svalPerSec2, scountPerSec2, svalPerSec3, scountPerSec3,
                        lvalPerSec1, lcountPerSec1, lvalPerSec2, lcountPerSec2, lvalPerSec3, lcountPerSec3
                        );
                }
                else {
                    if (clientJSON.length())
                        clientJSON += ",";
                    clientJSON += utils::StringFormat("{\"id\":%llu, \"location\":\"%u.%u.%u.%u:%u\", \"count\":%u, \"uptime\":\"%s\", "
                        " \"shortBytesPerSec1\": %.3f, \"shortCountPerSec1\": %.3f, \"shortBytesPerSec10\": %.3f, \"shortCountPerSec10\": %.3f, \"shortBytesPerSec60\": %.3f, \"shortCountPerSec60\": %.3f, "
                        " \"longBytesPer1Sec\": %.3f, \"longCountPer1Sec\": %.3f, \"longBytesPer10Sec\": %.3f, \"longCountPer10Sec\": %.3f, \"longBytesPer60Sec\": %.3f, \"longCountPer60Sec\": %.3f "
                        "}\n",
                        i->second.conID,
                        GETIPADDRESSQUADPORT(i->second.location),
                        i->second.count,
                        PrintTimeDifString(GetTimeAge(i->second.lastConTime)).c_str(),
                        svalPerSec1, scountPerSec1, svalPerSec2, scountPerSec2, svalPerSec3, scountPerSec3,
                        lvalPerSec1, lcountPerSec1, lvalPerSec2, lcountPerSec2, lvalPerSec3, lcountPerSec3
                        );
                }
                clientCount++;
            }
        }
        for (i = webClients.begin(), e = webClients.end(); i != e; i++) {
            if (!i->second.lastFailTime) {
                //shortAvgStatsWeb[i->second.conID].getThroughputMulti(ms1, ms2, ms3, svalPerSec1, scountPerSec1, svalPerSec2, scountPerSec2, svalPerSec3, scountPerSec3, now);
                //longAvgStatsWeb[i->second.conID].getThroughputMulti(ms1, ms2, ms3, lvalPerSec1, lcountPerSec1, lvalPerSec2, lcountPerSec2, lvalPerSec3, lcountPerSec3, now);
                i->second.shortAvgStats.getThroughputMulti(ms1, ms2, ms3, svalPerSec1, scountPerSec1, svalPerSec2, scountPerSec2, svalPerSec3, scountPerSec3, now);
                i->second.longAvgStats.getThroughputMulti(ms1, ms2, ms3, lvalPerSec1, lcountPerSec1, lvalPerSec2, lcountPerSec2, lvalPerSec3, lcountPerSec3, now);
                if (replyXML) {
                    subXML += utils::StringFormat("<webclient id=\"%llu\" location=\"%u.%u.%u.%u:%u\" count=\"%u\" uptime=\"%s\" "
                        "shortBytesPerSec1=\"%.3f\" shortCountPerSec1=\"%.3f\" shortBytesPerSec10=\"%.3f\" shortCountPerSec10=\"%.3f\" shortBytesPerSec60=\"%.3f\" shortCountPerSec60=\"%.3f\" "
                        "longBytesPer1Sec=\"%.3f\" longCountPer1Sec=\"%.3f\" longBytesPer10Sec=\"%.3f\" longCountPer10Sec=\"%.3f\" longBytesPer60Sec=\"%.3f\" longCountPer60Sec=\"%.3f\" "
                        "/>\n",
                        i->second.conID,
                        GETIPADDRESSQUADPORT(i->second.location),
                        i->second.count,
                        PrintTimeDifString(GetTimeAge(i->second.lastConTime)).c_str(),
                        svalPerSec1, scountPerSec1, svalPerSec2, scountPerSec2, svalPerSec3, scountPerSec3,
                        lvalPerSec1, lcountPerSec1, lvalPerSec2, lcountPerSec2, lvalPerSec3, lcountPerSec3
                        );
                }
                else {
                    if (webClientJSON.length())
                        webClientJSON += ",";
                    webClientJSON += utils::StringFormat("{\"id\":%llu, \"location\":\"%u.%u.%u.%u:%u\", \"count\":%u, \"uptime\":\"%s\", "
                        " \"shortBytesPerSec1\": %.3f, \"shortCountPerSec1\": %.3f, \"shortBytesPerSec10\": %.3f, \"shortCountPerSec10\": %.3f, \"shortBytesPerSec60\": %.3f, \"shortCountPerSec60\": %.3f, "
                        " \"longBytesPer1Sec\": %.3f, \"longCountPer1Sec\": %.3f, \"longBytesPer10Sec\": %.3f, \"longCountPer10Sec\": %.3f, \"longBytesPer60Sec\": %.3f, \"longCountPer60Sec\": %.3f "
                        "}\n",
                        i->second.conID,
                        GETIPADDRESSQUADPORT(i->second.location),
                        i->second.count,
                        PrintTimeDifString(GetTimeAge(i->second.lastConTime)).c_str(),
                        svalPerSec1, scountPerSec1, svalPerSec2, scountPerSec2, svalPerSec3, scountPerSec3,
                        lvalPerSec1, lcountPerSec1, lvalPerSec2, lcountPerSec2, lvalPerSec3, lcountPerSec3
                    );
                }
                webClientCount++;
            }
        }
        for (i = webSockets.begin(), e = webSockets.end(); i != e; i++) {
            if (!i->second.lastFailTime) {
                //shortAvgStatsWeb[i->second.conID].getThroughputMulti(ms1, ms2, ms3, svalPerSec1, scountPerSec1, svalPerSec2, scountPerSec2, svalPerSec3, scountPerSec3, now);
                //longAvgStatsWeb[i->second.conID].getThroughputMulti(ms1, ms2, ms3, lvalPerSec1, lcountPerSec1, lvalPerSec2, lcountPerSec2, lvalPerSec3, lcountPerSec3, now);
                i->second.shortAvgStats.getThroughputMulti(ms1, ms2, ms3, svalPerSec1, scountPerSec1, svalPerSec2, scountPerSec2, svalPerSec3, scountPerSec3, now);
                i->second.longAvgStats.getThroughputMulti(ms1, ms2, ms3, lvalPerSec1, lcountPerSec1, lvalPerSec2, lcountPerSec2, lvalPerSec3, lcountPerSec3, now);
                if (replyXML) {
                    subXML += utils::StringFormat("<websocket id=\"%llu\" location=\"%u.%u.%u.%u:%u\" count=\"%u\" uptime=\"%s\" "
                        "shortBytesPerSec1=\"%.3f\" shortCountPerSec1=\"%.3f\" shortBytesPerSec10=\"%.3f\" shortCountPerSec10=\"%.3f\" shortBytesPerSec60=\"%.3f\" shortCountPerSec60=\"%.3f\" "
                        "longBytesPer1Sec=\"%.3f\" longCountPer1Sec=\"%.3f\" longBytesPer10Sec=\"%.3f\" longCountPer10Sec=\"%.3f\" longBytesPer60Sec=\"%.3f\" longCountPer60Sec=\"%.3f\" "
                        "/>\n",
                        i->second.conID,
                        GETIPADDRESSQUADPORT(i->second.location),
                        i->second.count,
                        PrintTimeDifString(GetTimeAge(i->second.lastConTime)).c_str(),
                        svalPerSec1, scountPerSec1, svalPerSec2, scountPerSec2, svalPerSec3, scountPerSec3,
                        lvalPerSec1, lcountPerSec1, lvalPerSec2, lcountPerSec2, lvalPerSec3, lcountPerSec3
                    );
                }
                else {
                    if (webSocketJSON.length())
                        webSocketJSON += ",";
                    webSocketJSON += utils::StringFormat("{\"id\":%llu, \"location\":\"%u.%u.%u.%u:%u\", \"count\":%u, \"uptime\":\"%s\", "
                        " \"shortBytesPerSec1\": %.3f, \"shortCountPerSec1\": %.3f, \"shortBytesPerSec10\": %.3f, \"shortCountPerSec10\": %.3f, \"shortBytesPerSec60\": %.3f, \"shortCountPerSec60\": %.3f, "
                        " \"longBytesPer1Sec\": %.3f, \"longCountPer1Sec\": %.3f, \"longBytesPer10Sec\": %.3f, \"longCountPer10Sec\": %.3f, \"longBytesPer60Sec\": %.3f, \"longCountPer60Sec\": %.3f "
                        "}\n",
                        i->second.conID,
                        GETIPADDRESSQUADPORT(i->second.location),
                        i->second.count,
                        PrintTimeDifString(GetTimeAge(i->second.lastConTime)).c_str(),
                        svalPerSec1, scountPerSec1, svalPerSec2, scountPerSec2, svalPerSec3, scountPerSec3,
                        lvalPerSec1, lcountPerSec1, lvalPerSec2, lcountPerSec2, lvalPerSec3, lcountPerSec3
                    );
                }
                webSocketCount++;
            }
        }
        shortAvgStats.getThroughputMulti(ms1, ms2, ms3, svalPerSec1, scountPerSec1, svalPerSec2, scountPerSec2, svalPerSec3, scountPerSec3, now);
        longAvgStats.getThroughputMulti(ms1, ms2, ms3, lvalPerSec1, lcountPerSec1, lvalPerSec2, lcountPerSec2, lvalPerSec3, lcountPerSec3, now);
        if (replyXML)
            xml = utils::StringFormat("<status uptime=\"%s\" version=\"%s\" executors=\"%u\" clients=\"%u\" webclients=\"%u\" websockets=\"%u\" maxqueuesize=\"%u\" minqueuesize=\"%u\" "
                "shortBytesPerSec1=\"%.3f\" shortCountPerSec1=\"%.3f\" shortBytesPerSec10=\"%.3f\" shortCountPerSec10=\"%.3f\" shortBytesPerSec60=\"%.3f\" shortCountPerSec60=\"%.3f\" "
                "longBytesPer1Sec=\"%.3f\" longCountPer1Sec=\"%.3f\" longBytesPer10Sec=\"%.3f\" longCountPer10Sec=\"%.3f\" longBytesPer60Sec=\"%.3f\" longCountPer60Sec=\"%.3f\" "
                ">\n%s</status>\n",
                PrintTimeDifString(GetTimeAge(systemStartTime)).c_str(), versionString.c_str(), execCount, clientCount, webClientCount,
                maxRequestQueueSize, maxRequestProcessingSize,
                svalPerSec1, scountPerSec1, svalPerSec2, scountPerSec2, svalPerSec3, scountPerSec3,
                lvalPerSec1, lcountPerSec1, lvalPerSec2, lcountPerSec2, lvalPerSec3, lcountPerSec3,
                subXML.c_str());
        else
            json = utils::StringFormat("{\"uptime\":\"%s\", \"version\":\"%s\", \"executorcount\":%u, \"clientcount\":%u, \"webclientcount\":%u, \"websocketcount\":%u, \"maxqueuesize\":%u, \"minqueuesize\":%u, "
                "\"shortBytesPerSec1\": %.3f, \"shortCountPerSec1\": %.3f, \"shortBytesPerSec10\": %.3f, \"shortCountPerSec10\": %.3f, \"shortBytesPerSec60\": %.3f, \"shortCountPerSec60\": %.3f, "
                "\"longBytesPer1Sec\": %.3f, \"longCountPer1Sec\": %.3f, \"longBytesPer10Sec\": %.3f, \"longCountPer10Sec\": %.3f, \"longBytesPer60Sec\": %.3f, \"longCountPer60Sec\": %.3f, "
                "\"executors\":[%s], \"clients\":[%s], \"webclients\":[%s], \"websockets\":[%s]}\n",
                PrintTimeDifString(GetTimeAge(systemStartTime)).c_str(), versionString.c_str(), execCount, clientCount, webClientCount, webSocketCount,
                maxRequestQueueSize, maxRequestProcessingSize,
                svalPerSec1, scountPerSec1, svalPerSec2, scountPerSec2, svalPerSec3, scountPerSec3,
                lvalPerSec1, lcountPerSec1, lvalPerSec2, lcountPerSec2, lvalPerSec3, lcountPerSec3,
                execJSON.c_str(), clientJSON.c_str(), webClientJSON.c_str(), webSocketJSON.c_str());
    }
    else if (((stricmp(apiName, "calls") == 0)) || utils::TextStartsWith(apiName, "calls?")) {
        uint32 count = 0;
        uint32 maxCount = utils::Ascii2Uint32(req->getParameter("maxcount"));
        if (!maxCount) maxCount = 100;
        uint32 accountID = utils::Ascii2Uint32(req->getParameter("accountid"));
        uint32 clientID = utils::Ascii2Uint32(req->getParameter("clientid"));
        uint32 port = utils::Ascii2Uint32(req->getParameter("port"));
        uint32 minduration = utils::Ascii2Uint32(req->getParameter("minduration"));
        uint32 maxduration = utils::Ascii2Uint32(req->getParameter("maxduration"));
        uint32 webonly = utils::Ascii2Uint32(req->getParameter("webonly"));
        uint32 messageonly = utils::Ascii2Uint32(req->getParameter("messageonly"));
        const char* status = req->getParameter("status");
        if (!replyXML) {
            if (callLogMutex.enter(100)) {
                for (ic = callLog.rbegin(), ec = callLog.rend(); ic != ec; ic++) {
                    if (accountID && ((*ic)->accountID != accountID)) continue;
                    if (clientID && ((*ic)->clientID != clientID)) continue;
                    if (port && ((*ic)->port != port)) continue;
                    if (minduration && ( ((*ic)->endTime - (*ic)->startTime) < (minduration * 1000))) continue;
                    if (maxduration && (((*ic)->endTime - (*ic)->startTime) > (maxduration * 1000))) continue;
                    if (webonly && ((*ic)->conType != 2)) continue;
                    if (messageonly && ((*ic)->conType != 1)) continue;
                    if (status && (!utils::stristr((*ic)->status.c_str(), status))) continue;
 
                    if ((json = (*ic)->toJSON()).length()) {
                        if (execJSON.length())
                            execJSON += ",";
                        execJSON += json;
                    }
                    if (++count > maxCount) break;
                }
                callLogMutex.leave();
            }
            json = utils::StringFormat(
                "{\"uptime\":\"%s\", \"version\":\"%s\", \"executorcount\":%u, \"clientcount\":%u, \"webclientcount\":%u, \"websocketcount\":%u,\n"
                "\"count\":%u, \"calllog\": [%s]}\n",
                PrintTimeDifString(GetTimeAge(systemStartTime)).c_str(), versionString.c_str(), execCount, clientCount, webClientCount, webSocketCount,
                count, execJSON.c_str());
        }
    }
    else if (((stricmp(apiName, "throughput") == 0)) || utils::TextStartsWith(apiName, "throughput?")) {
        if (replyXML)
            xml = utils::StringFormat("<throughput><short>%s</short><long>%s</long></throughput>",
                shortAvgStats.getPerfXML().c_str(), longAvgStats.getPerfXML().c_str());
        else
            json = utils::StringFormat("{\"shortthroughput\": %s, \"longthroughput\": %s }",
                shortAvgStats.getPerfJSON().c_str(), longAvgStats.getPerfJSON().c_str());
    }
    else if (((stricmp(apiName, "available") == 0)) || utils::TextStartsWith(apiName, "available?")) {
        bool isAvailable = false;
        for (i = executors.begin(), e = executors.end(); i != e; i++) {
            if (!i->second.lastFailTime) {
                isAvailable = true;
                break;
                execCount++;
            }
        }
        if (isAvailable) {
            if (replyXML)
                xml = utils::StringFormat("<status available=\"yes\" />\n");
            else
                json = utils::StringFormat("{\"available\":\"yes\"}\n");
        }
    }
    else if (((stricmp(apiName, "stats") == 0)) || utils::TextStartsWith(apiName, "stats?")) {
        svalPerSec1 = scountPerSec1 = svalPerSec2 = scountPerSec2 = svalPerSec3 = scountPerSec3 =
            lvalPerSec1 = lcountPerSec1 = lvalPerSec2 = lcountPerSec2 = lvalPerSec3 = lcountPerSec3 = 0;
        for (i = executors.begin(), e = executors.end(); i != e; i++) {
            if (!i->second.lastFailTime) {
                //shortAvgStatsExec[i->second.conID].getThroughputMulti(ms1, ms2, ms3, svalPerSec1, scountPerSec1, svalPerSec2, scountPerSec2, svalPerSec3, scountPerSec3, now);
                //longAvgStatsExec[i->second.conID].getThroughputMulti(ms1, ms2, ms3, lvalPerSec1, lcountPerSec1, lvalPerSec2, lcountPerSec2, lvalPerSec3, lcountPerSec3, now);
                i->second.shortAvgStats.getThroughputMulti(ms1, ms2, ms3, svalPerSec1, scountPerSec1, svalPerSec2, scountPerSec2, svalPerSec3, scountPerSec3, now);
                i->second.longAvgStats.getThroughputMulti(ms1, ms2, ms3, lvalPerSec1, lcountPerSec1, lvalPerSec2, lcountPerSec2, lvalPerSec3, lcountPerSec3, now);
                if (replyXML) {
                    subXML += utils::StringFormat("<executor id=\"%llu\" location=\"%u.%u.%u.%u:%u\" shortQueue=\"%u\" longQueue=\"%u\" count=\"%u\" uptime=\"%s\" "
                        "shortBytesPerSec1=\"%.3f\" shortCountPerSec1=\"%.3f\" shortBytesPerSec10=\"%.3f\" shortCountPerSec10=\"%.3f\" shortBytesPerSec60=\"%.3f\" shortCountPerSec60=\"%.3f\" "
                        "longBytesPer1Sec=\"%.3f\" longCountPer1Sec=\"%.3f\" longBytesPer10Sec=\"%.3f\" longCountPer10Sec=\"%.3f\" longBytesPer60Sec=\"%.3f\" longCountPer60Sec=\"%.3f\" />\n",
                        i->second.conID,
                        GETIPADDRESSQUADPORT(i->second.location),
                        i->second.shortReqQueue.getCount(),
                        i->second.longReqQueue.getCount(),
                        i->second.count,
                        PrintTimeDifString(GetTimeAge(i->second.lastConTime)).c_str(),
                        svalPerSec1, scountPerSec1, svalPerSec2, scountPerSec2, svalPerSec3, scountPerSec3,
                        lvalPerSec1, lcountPerSec1, lvalPerSec2, lcountPerSec2, lvalPerSec3, lcountPerSec3
                        );
                }
                else {
                    if (execJSON.length())
                        execJSON += ",";
                    //execJSON += utils::StringFormat("{\"id\":%llu, \"location\":\"%u.%u.%u.%u:%u\", \"shortQueue\":%u, \"longQueue\":%u, \"count\":%u, \"uptime\":\"%s\" }\n",
                    //  i->second.conID,
                    //  GETIPADDRESSQUADPORT(i->second.location),
                    //  i->second.shortReqQSize,
                    //  i->second.longReqQSize,
                    //  i->second.count,
                    //  PrintTimeDifString(GetTimeAge(i->second.lastConTime)).c_str()
                    //  );
                    execJSON += utils::StringFormat("{\"id\":%llu, \"location\":\"%u.%u.%u.%u:%u\", \"shortQueue\":%u, \"longQueue\":%u, \"count\":%u, \"uptime\":\"%s\", "
                        "\"shortBytesPerSec1\": %.3f, \"shortCountPerSec1\": %.3f, \"shortBytesPerSec10\": %.3f, \"shortCountPerSec10\": %.3f, \"shortBytesPerSec60\": %.3f, \"shortCountPerSec60\": %.3f, "
                        "\"longBytesPer1Sec\": %.3f, \"longCountPer1Sec\": %.3f, \"longBytesPer10Sec\": %.3f, \"longCountPer10Sec\": %.3f, \"longBytesPer60Sec\": %.3f, \"longCountPer60Sec\": %.3f }\n",
                        i->second.conID,
                        GETIPADDRESSQUADPORT(i->second.location),
                        i->second.shortReqQueue.getCount(),
                        i->second.longReqQueue.getCount(),
                        i->second.count,
                        PrintTimeDifString(GetTimeAge(i->second.lastConTime)).c_str(),
                        svalPerSec1, scountPerSec1, svalPerSec2, scountPerSec2, svalPerSec3, scountPerSec3,
                        lvalPerSec1, lcountPerSec1, lvalPerSec2, lcountPerSec2, lvalPerSec3, lcountPerSec3
                        );
                }
                execCount++;
            }
        }
        for (i = clients.begin(), e = clients.end(); i != e; i++) {
            if (!i->second.lastFailTime) {
                //shortAvgStatsClients[i->second.conID].getThroughputMulti(ms1, ms2, ms3, svalPerSec1, scountPerSec1, svalPerSec2, scountPerSec2, svalPerSec3, scountPerSec3, now);
                //longAvgStatsClients[i->second.conID].getThroughputMulti(ms1, ms2, ms3, lvalPerSec1, lcountPerSec1, lvalPerSec2, lcountPerSec2, lvalPerSec3, lcountPerSec3, now);
                i->second.shortAvgStats.getThroughputMulti(ms1, ms2, ms3, svalPerSec1, scountPerSec1, svalPerSec2, scountPerSec2, svalPerSec3, scountPerSec3, now);
                i->second.longAvgStats.getThroughputMulti(ms1, ms2, ms3, lvalPerSec1, lcountPerSec1, lvalPerSec2, lcountPerSec2, lvalPerSec3, lcountPerSec3, now);
                if (replyXML) {
                    subXML += utils::StringFormat("<client id=\"%llu\" location=\"%u.%u.%u.%u:%u\" count=\"%u\" uptime=\"%s\" "
                        "shortBytesPerSec1=\"%.3f\" shortCountPerSec1=\"%.3f\" shortBytesPerSec10=\"%.3f\" shortCountPerSec10=\"%.3f\" shortBytesPerSec60=\"%.3f\" shortCountPerSec60=\"%.3f\" "
                        "longBytesPer1Sec=\"%.3f\" longCountPer1Sec=\"%.3f\" longBytesPer10Sec=\"%.3f\" longCountPer10Sec=\"%.3f\" longBytesPer60Sec=\"%.3f\" longCountPer60Sec=\"%.3f\" />\n",
                        i->second.conID,
                        GETIPADDRESSQUADPORT(i->second.location),
                        i->second.count,
                        PrintTimeDifString(GetTimeAge(i->second.lastConTime)).c_str(),
                        svalPerSec1, scountPerSec1, svalPerSec2, scountPerSec2, svalPerSec3, scountPerSec3,
                        lvalPerSec1, lcountPerSec1, lvalPerSec2, lcountPerSec2, lvalPerSec3, lcountPerSec3
                        );
                }
                else {
                    if (clientJSON.length())
                        clientJSON += ",";
                    clientJSON += utils::StringFormat("{\"id\":%llu, \"location\":\"%u.%u.%u.%u:%u\", \"count\":%u, \"uptime\":\"%s\", "
                        "\"shortBytesPerSec1\": %.3f, \"shortCountPerSec1\": %.3f, \"shortBytesPerSec10\": %.3f, \"shortCountPerSec10\": %.3f, \"shortBytesPerSec60\": %.3f, \"shortCountPerSec60\": %.3f, "
                        "\"longBytesPer1Sec\": %.3f, \"longCountPer1Sec\": %.3f, \"longBytesPer10Sec\": %.3f, \"longCountPer10Sec\": %.3f, \"longBytesPer60Sec\": %.3f, \"longCountPer60Sec\": %.3f }\n",
                        i->second.conID,
                        GETIPADDRESSQUADPORT(i->second.location),
                        i->second.count,
                        PrintTimeDifString(GetTimeAge(i->second.lastConTime)).c_str(),
                        svalPerSec1, scountPerSec1, svalPerSec2, scountPerSec2, svalPerSec3, scountPerSec3,
                        lvalPerSec1, lcountPerSec1, lvalPerSec2, lcountPerSec2, lvalPerSec3, lcountPerSec3
                        );
                }
                clientCount++;
            }
        }
        for (i = webClients.begin(), e = webClients.end(); i != e; i++) {
            if (!i->second.lastFailTime) {
                //shortAvgStatsWeb[i->second.conID].getThroughputMulti(ms1, ms2, ms3, svalPerSec1, scountPerSec1, svalPerSec2, scountPerSec2, svalPerSec3, scountPerSec3, now);
                //longAvgStatsWeb[i->second.conID].getThroughputMulti(ms1, ms2, ms3, lvalPerSec1, lcountPerSec1, lvalPerSec2, lcountPerSec2, lvalPerSec3, lcountPerSec3, now);
                i->second.shortAvgStats.getThroughputMulti(ms1, ms2, ms3, svalPerSec1, scountPerSec1, svalPerSec2, scountPerSec2, svalPerSec3, scountPerSec3, now);
                i->second.longAvgStats.getThroughputMulti(ms1, ms2, ms3, lvalPerSec1, lcountPerSec1, lvalPerSec2, lcountPerSec2, lvalPerSec3, lcountPerSec3, now);
                if (replyXML) {
                    subXML += utils::StringFormat("<webclient id=\"%llu\" location=\"%u.%u.%u.%u:%u\" count=\"%u\" uptime=\"%s\" "
                        "shortBytesPerSec1=\"%.3f\" shortCountPerSec1=\"%.3f\" shortBytesPerSec10=\"%.3f\" shortCountPerSec10=\"%.3f\" shortBytesPerSec60=\"%.3f\" shortCountPerSec60=\"%.3f\" "
                        "longBytesPer1Sec=\"%.3f\" longCountPer1Sec=\"%.3f\" longBytesPer10Sec=\"%.3f\" longCountPer10Sec=\"%.3f\" longBytesPer60Sec=\"%.3f\" longCountPer60Sec=\"%.3f\" />\n",
                        i->second.conID,
                        GETIPADDRESSQUADPORT(i->second.location),
                        i->second.count,
                        PrintTimeDifString(GetTimeAge(i->second.lastConTime)).c_str(),
                        svalPerSec1, scountPerSec1, svalPerSec2, scountPerSec2, svalPerSec3, scountPerSec3,
                        lvalPerSec1, lcountPerSec1, lvalPerSec2, lcountPerSec2, lvalPerSec3, lcountPerSec3
                        );
                }
                else {
                    if (webClientJSON.length())
                        webClientJSON += ",";
                    webClientJSON += utils::StringFormat("{\"id\":%llu, \"location\":\"%u.%u.%u.%u:%u\", \"count\":%u, \"uptime\":\"%s\", "
                        "\"shortBytesPerSec1\": %.3f, \"shortCountPerSec1\": %.3f, \"shortBytesPerSec10\": %.3f, \"shortCountPerSec10\": %.3f, \"shortBytesPerSec60\": %.3f, \"shortCountPerSec60\": %.3f, "
                        "\"longBytesPer1Sec\": %.3f, \"longCountPer1Sec\": %.3f, \"longBytesPer10Sec\": %.3f, \"longCountPer10Sec\": %.3f, \"longBytesPer60Sec\": %.3f, \"longCountPer60Sec\": %.3f }\n",
                        i->second.conID,
                        GETIPADDRESSQUADPORT(i->second.location),
                        i->second.count,
                        PrintTimeDifString(GetTimeAge(i->second.lastConTime)).c_str(),
                        svalPerSec1, scountPerSec1, svalPerSec2, scountPerSec2, svalPerSec3, scountPerSec3,
                        lvalPerSec1, lcountPerSec1, lvalPerSec2, lcountPerSec2, lvalPerSec3, lcountPerSec3
                        );
                }
                webClientCount++;
            }
        }
        for (i = webSockets.begin(), e = webSockets.end(); i != e; i++) {
            if (!i->second.lastFailTime) {
                //shortAvgStatsWeb[i->second.conID].getThroughputMulti(ms1, ms2, ms3, svalPerSec1, scountPerSec1, svalPerSec2, scountPerSec2, svalPerSec3, scountPerSec3, now);
                //longAvgStatsWeb[i->second.conID].getThroughputMulti(ms1, ms2, ms3, lvalPerSec1, lcountPerSec1, lvalPerSec2, lcountPerSec2, lvalPerSec3, lcountPerSec3, now);
                i->second.shortAvgStats.getThroughputMulti(ms1, ms2, ms3, svalPerSec1, scountPerSec1, svalPerSec2, scountPerSec2, svalPerSec3, scountPerSec3, now);
                i->second.longAvgStats.getThroughputMulti(ms1, ms2, ms3, lvalPerSec1, lcountPerSec1, lvalPerSec2, lcountPerSec2, lvalPerSec3, lcountPerSec3, now);
                if (replyXML) {
                    subXML += utils::StringFormat("<websocket id=\"%llu\" location=\"%u.%u.%u.%u:%u\" count=\"%u\" uptime=\"%s\" "
                        "shortBytesPerSec1=\"%.3f\" shortCountPerSec1=\"%.3f\" shortBytesPerSec10=\"%.3f\" shortCountPerSec10=\"%.3f\" shortBytesPerSec60=\"%.3f\" shortCountPerSec60=\"%.3f\" "
                        "longBytesPer1Sec=\"%.3f\" longCountPer1Sec=\"%.3f\" longBytesPer10Sec=\"%.3f\" longCountPer10Sec=\"%.3f\" longBytesPer60Sec=\"%.3f\" longCountPer60Sec=\"%.3f\" />\n",
                        i->second.conID,
                        GETIPADDRESSQUADPORT(i->second.location),
                        i->second.count,
                        PrintTimeDifString(GetTimeAge(i->second.lastConTime)).c_str(),
                        svalPerSec1, scountPerSec1, svalPerSec2, scountPerSec2, svalPerSec3, scountPerSec3,
                        lvalPerSec1, lcountPerSec1, lvalPerSec2, lcountPerSec2, lvalPerSec3, lcountPerSec3
                    );
                }
                else {
                    if (webSocketJSON.length())
                        webSocketJSON += ",";
                    webSocketJSON += utils::StringFormat("{\"id\":%llu, \"location\":\"%u.%u.%u.%u:%u\", \"count\":%u, \"uptime\":\"%s\", "
                        "\"shortBytesPerSec1\": %.3f, \"shortCountPerSec1\": %.3f, \"shortBytesPerSec10\": %.3f, \"shortCountPerSec10\": %.3f, \"shortBytesPerSec60\": %.3f, \"shortCountPerSec60\": %.3f, "
                        "\"longBytesPer1Sec\": %.3f, \"longCountPer1Sec\": %.3f, \"longBytesPer10Sec\": %.3f, \"longCountPer10Sec\": %.3f, \"longBytesPer60Sec\": %.3f, \"longCountPer60Sec\": %.3f }\n",
                        i->second.conID,
                        GETIPADDRESSQUADPORT(i->second.location),
                        i->second.count,
                        PrintTimeDifString(GetTimeAge(i->second.lastConTime)).c_str(),
                        svalPerSec1, scountPerSec1, svalPerSec2, scountPerSec2, svalPerSec3, scountPerSec3,
                        lvalPerSec1, lcountPerSec1, lvalPerSec2, lcountPerSec2, lvalPerSec3, lcountPerSec3
                    );
                }
                webSocketCount++;
            }
        }
        shortAvgStats.getThroughputMulti(ms1, ms2, ms3, svalPerSec1, scountPerSec1, svalPerSec2, scountPerSec2, svalPerSec3, scountPerSec3, now);
        longAvgStats.getThroughputMulti(ms1, ms2, ms3, lvalPerSec1, lcountPerSec1, lvalPerSec2, lcountPerSec2, lvalPerSec3, lcountPerSec3, now);
        if (replyXML)
            xml = utils::StringFormat("<status executors=\"%u\" clients=\"%u\" webclients=\"%u\" websockets=\"%u\" "
            "shortBytesPerSec1=\"%.3f\" shortCountPerSec1=\"%.3f\" shortBytesPerSec10=\"%.3f\" shortCountPerSec10=\"%.3f\" shortBytesPerSec60=\"%.3f\" shortCountPerSec60=\"%.3f\" "
            "longBytesPer1Sec=\"%.3f\" longCountPer1Sec=\"%.3f\" longBytesPer10Sec=\"%.3f\" longCountPer10Sec=\"%.3f\" longBytesPer60Sec=\"%.3f\" longCountPer60Sec=\"%.3f\" "
            ">\n%s</status>\n", execCount, clientCount, webClientCount, webSocketCount,
            svalPerSec1, scountPerSec1, svalPerSec2, scountPerSec2, svalPerSec3, scountPerSec3,
            lvalPerSec1, lcountPerSec1, lvalPerSec2, lcountPerSec2, lvalPerSec3, lcountPerSec3,
            subXML.c_str());
        else
            json = utils::StringFormat("{\"executorcount\":%u, \"clientcount\":%u, \"webclientcount\":%u, \"websocketcount\":%u, "
            "\"shortBytesPerSec1\": %.3f, \"shortCountPerSec1\": %.3f, \"shortBytesPerSec10\": %.3f, \"shortCountPerSec10\": %.3f, \"shortBytesPerSec60\": %.3f, \"shortCountPerSec60\": %.3f, "
            "\"longBytesPer1Sec\": %.3f, \"longCountPer1Sec\": %.3f, \"longBytesPer10Sec\": %.3f, \"longCountPer10Sec\": %.3f, \"longBytesPer60Sec\": %.3f, \"longCountPer60Sec\": %.3f, "
            "\"executors\":[%s], \"clients\":[%s], \"webclients\":[%s], \"websockets\":[%s]}\n",
            execCount, clientCount, webClientCount, webSocketCount,
            svalPerSec1, scountPerSec1, svalPerSec2, scountPerSec2, svalPerSec3, scountPerSec3,
            lvalPerSec1, lcountPerSec1, lvalPerSec2, lcountPerSec2, lvalPerSec3, lcountPerSec3,
            execJSON.c_str(), clientJSON.c_str(), webClientJSON.c_str(), webSocketJSON.c_str());
    }
    else if (stricmp(apiName, "sizeechotest") == 0) {
        // size = contentlength - 40 (as we add about 40 chars of XML below)
        uint32 size = (req->contentLength < 50) ? 10 : req->contentLength - 40;
        uint32 n;
        char* echocontent = new char[size+1];
        for (n=0; n<size; n++)
            echocontent[n] = 48 + (n%10);
        echocontent[n] = 0;
        if (replyXML)
            xml = utils::StringFormat("<echosizetest size=\"%u\">%s</echosizetest>", size, echocontent);
        else
            json = utils::StringFormat("{\"size\":%u, \"content\":\"%s\"}\n", size, echocontent);
        delete [] echocontent;
    }
    else if (utils::TextStartsWith(apiName, "restart?", false)) {
        uint64 idNum = 0;
        const char* id = req->getParameter("id");
        if (id && (stricmp(id, "all") == 0)) {
            uint32 successCount = 0;
            uint32 failedCount = 0;
            std::map<uint64, RequestConnection>::iterator i, e;
            for (i = executors.begin(), e = executors.end(); i != e; i++) {
                if (sendRestartToExecutor(i->first))
                    successCount++;
                else
                    failedCount++;
            }
            if (!successCount) {
                if (replyXML)
                    xml = utils::StringFormat("<error text=\"None of the %u executors were restarted\" />", failedCount);
                else
                    json = utils::StringFormat("{\"error\":\"None of the %u executors were restarted\"}\n", failedCount);
            }
            else {
                if (replyXML)
                    xml = utils::StringFormat("<success text=\"Restart sent to %u executor successfully (%u failed sending)\" />", successCount, failedCount);
                else
                    json = utils::StringFormat("{\"success\":\"Restart sent to %u executor successfully (%u failed sending)\"}\n", successCount, failedCount);
            }
        }
        else if (id && ((idNum = utils::Ascii2Uint64(id)) > 0)) {
            if (sendRestartToExecutor(idNum)) {
                if (replyXML)
                    xml = utils::StringFormat("<success text=\"Restart sent to executor %llu successfully\" />", idNum);
                else
                    json = utils::StringFormat("{\"success\":\"Restart sent to executor %llu successfully\"}\n", idNum);
            }
            else {
                if (replyXML)
                    xml = utils::StringFormat("<error text=\"Failed to send restart to executor %llu\" />", idNum);
                else
                    json = utils::StringFormat("{\"error\":\"Failed to send restart to executor %llu\"}\n", idNum);
            }
        }
        else {
            if (replyXML)
                xml = utils::StringFormat("<error text=\"No valid id provided\" />");
            else
                json = utils::StringFormat("{\"error\":\"No valid id provided\"}\n");
        }
    }
    else if (strlen(apiName)) {
        if (rootdir.length()) {
            if (reply->createFromFile(GetTimeNow(), webServerName.c_str(), req->ifModifiedSince, true, cacheFiles, (rootdir + apiName).c_str())) {
                addToCallLog(++callLogCount, 2, 3, false, req->source, 0, 0, 0, req->endReceiveTime, GetTimeNow(), 0, GetTimeAge(now), req->type, req->getSize(), reply->getSize(), "Success",
                    utils::StringFormat("Internal: %s", apiName).c_str(), req->getHeaderEntry("User-Agent"), "");
                channel->sendHTTPReply(reply, conid);
                delete(reply);
                return true;
            }
            else {
                if (replyXML)
                    xml = utils::StringFormat("<error code=\"-1\" text=\"File %s not found in rootdir\" />", apiName);
                else
                    json = utils::StringFormat("{\"errorcode\":-1, \"text\":\"File %s not found in rootdir\"}\n", apiName);
            }
        }
        else {
            if (replyXML)
                xml = utils::StringFormat("<error code=\"-1\" text=\"Cannot generate %s, no rootdir configured\" />", apiName);
            else
                json = utils::StringFormat("{\"errorcode\":-1, \"text\":\"Cannot generate %s, no rootdir configured\"}\n", apiName);
        }
    }
    else {
        if (rootdir.length()) {
            if (reply->createFromFile(GetTimeNow(), webServerName.c_str(), req->ifModifiedSince, true, cacheFiles, (rootdir + indexFilename).c_str())) {
                channel->sendHTTPReply(reply, conid);
                addToCallLog(++callLogCount, 2, 3, false, req->source, 0, 0, 0, req->endReceiveTime, GetTimeNow(), 0, GetTimeAge(now), req->type, req->getSize(), reply->getSize(), "Success",
                    utils::StringFormat("Internal: %s", apiName).c_str(), req->getHeaderEntry("User-Agent"), "");
                delete(reply);
                return true;
            }
            else {
                if (replyXML)
                    xml = utils::StringFormat("<error code=\"-1\" text=\"No default file found in rootdir\" />");
                else
                    json = utils::StringFormat("{\"errorcode\":-1, \"text\":\"No default file found in rootdir\"}\n");
            }
        }
        else {
            if (replyXML)
                xml = utils::StringFormat("<error code=\"-1\" text=\"Cannot generate %s, no rootdir configured\" />", apiName);
            else
                json = utils::StringFormat("{\"errorcode\":-1, \"text\":\"Cannot generate %s, no rootdir configured\"}\n", apiName);
        }
    }
 
    if (!xml.length() && !json.length()) {
        reply->createPage(HTTP_SERVICE_NOT_AVAILABLE, GetTimeNow(), webServerName.c_str(), 0, true, false, "text/plain", "");
        addToCallLog(++callLogCount, 2, 3, false, req->source, 0, 0, 0, req->endReceiveTime, GetTimeNow(), 0, GetTimeAge(now), req->type, req->getSize(), reply->getSize(), "Internal Service Not Available",
            utils::StringFormat("Internal: %s", apiName).c_str(), req->getHeaderEntry("User-Agent"), "");
    }
    else if (replyXML) {
        reply->createPage(HTTP_OK, GetTimeNow(), webServerName.c_str(), 0, true, false, "text/xml", xml.c_str());
        addToCallLog(++callLogCount, 2, 3, false, req->source, 0, 0, 0, req->endReceiveTime, GetTimeNow(), 0, GetTimeAge(now), req->type, req->getSize(), reply->getSize(), "Success",
            utils::StringFormat("Internal: %s", apiName).c_str(), req->getHeaderEntry("User-Agent"), "");
    }
    else {
        reply->createPage(HTTP_OK, GetTimeNow(), webServerName.c_str(), 0, true, false, "application/json", json.c_str());
        addToCallLog(++callLogCount, 2, 3, false, req->source, 0, 0, 0, req->endReceiveTime, GetTimeNow(), 0, GetTimeAge(now), req->type, req->getSize(), reply->getSize(), "Success",
            utils::StringFormat("Internal: %s", apiName).c_str(), req->getHeaderEntry("User-Agent"), "");
    }
 
    channel->sendHTTPReply(reply, conid);
 
    delete(reply);
    return true;
}
 
bool RequestGateway::setLongRequestLimit(uint32 limit) {
    longReqLimit = limit;
    return true;
}
 
//bool shortQIsSmaller(const RequestConnection& r1, const RequestConnection& r2) {
//   return (r1.shortReqQueue.getCount() < r2.shortReqQueue.getCount());
//}
//
//bool longQIsSmaller(const RequestConnection& r1, const RequestConnection& r2) {
//   return (r1.longReqQueue.getCount() < r2.longReqQueue.getCount());
//}
 
 
uint64 RequestGateway::getBestExecutorID(std::string requestString, uint32 reqSize, bool& isLongReq) {
    // Load-balancing pick. First classify the request: "long" when the raw
    // request size exceeds longReqLimit (if set) or when the request string
    // contains any registered long-request name (case-insensitive substring).
    // Then choose among executors that are currently connected (lastConTime
    // set, no recorded failure) the one with the shortest queue. Note: the
    // comparison always uses longReqQueue.getCount(), even for short requests
    // — short-queue depth and the latency MovingAverages tracked per
    // connection do not currently influence the choice. Returns 0 when no
    // healthy executor exists (callers then fail the request as TOOBUSY /
    // server-unavailable).
    isLongReq = false;
    if (longReqLimit)
        isLongReq = (reqSize >= longReqLimit);
    if (!isLongReq) {
        std::vector<std::string>::iterator il = longReqNames.begin(), el = longReqNames.end();
        while (il != el) {
            if (utils::stristr(requestString.c_str(), il->c_str())) {
                isLongReq = true;
                break;
            }
            il++;
        }
    }
 
    uint64 bestID = 0;
    uint32 bestQSize = MAXVALUINT32;
    std::map<uint64, RequestConnection>::iterator i, e;
    for (i = executors.begin(), e = executors.end(); i != e; i++) {
        if (i->second.lastConTime && !i->second.lastFailTime) {
            if (i->second.longReqQueue.getCount() < bestQSize) {
                bestQSize = i->second.longReqQueue.getCount();
                bestID = i->first;
            }
        }
    }
//  printf("\n         -----     Best Executor is ID: %llu (Q: %u)\n\n", bestID, bestQSize);
    return bestID;
}
 
 
bool RequestGateway::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
) {
    if (!callLogMutex.enter(100)) return false;
 
    // NULL-safe: callers pass getString()/getHeaderEntry() results which can be
    // NULL (e.g. a DataMessage request has no "User-Agent"). Constructing a
    // std::string from NULL would crash, so coerce NULLs to empty here.
    std::string statusS      = status      ? status      : "";
    std::string requestPathS = requestPath ? requestPath : "";
    std::string agentS       = agent       ? agent       : "";
    std::string outputS      = output      ? output      : "";
 
    std::basic_string<char>::size_type t = requestPathS.find("?_=");
    if (t != std::string::npos)
        requestPathS.erase(t);
    t = requestPathS.find("&_=");
    if (t != std::string::npos)
        requestPathS.erase(t);
 
    CallLogEntry* entry = new CallLogEntry(id);
    entry->conType = conType;
    entry->messageType = messageType;
    entry->longRequest = longRequest;
    entry->source = source;
    entry->port = port;
    entry->executorID = executorID;
    entry->clientID = clientID;
    entry->startTime = startTime;
    entry->endTime = endTime;
    entry->queueTime = queueTime;
    entry->processTime = processTime;
    entry->requestType = requestType;
    entry->requestSize = requestSize;
    entry->replySize = replySize;
    entry->status = statusS;
    entry->requestPath = requestPathS;
    entry->agent = agentS;
    entry->output = outputS;
 
    callLog.push_back(entry);
    while (callLog.size() > callLogMax) {
        delete callLog.front();
        callLog.pop_front();
    }
 
    callLogMutex.leave();
    return true;
}
 
 
 
 
 
 
 
 
 
 
 
//std::list<RequestConnection> RequestGateway::getTopExecutors(std::string requestString, uint32 reqSize, bool& isLongReq) {
//  isLongReq = false;
//  if (longReqLimit)
//      isLongReq = (reqSize >= longReqLimit);
//  if (!isLongReq) {
//      std::vector<std::string>::iterator il = longReqNames.begin(), el = longReqNames.end();
//      while (il != el) {
//          if (utils::stristr(requestString.c_str(), il->c_str())) {
//              isLongReq = true;
//              break;
//          }
//          il++;
//      }
//  }
//  std::list<RequestConnection> executorsList;
//  std::map<uint64,RequestConnection>::iterator i,e;
//  for (i=executors.begin(), e=executors.end(); i!=e; i++)
//      executorsList.push_back(i->second);
//  if (isLongReq) {
//      executorsList.sort(longQIsSmaller);
//      //printf("--- Long EXELIST [%u] ---\n", executorsList.size());
//  }
//  else {
//      executorsList.sort(shortQIsSmaller);
//      //std::sort(executorsList.begin(), executorsList.end(), shortQIsSmaller);
//      //printf("--- Short EXELIST [%u] ---\n", executorsList.size());
//  }
//
//  //uint32 n = 0;
//  //std::list<RequestConnection>::iterator ii,ee;
//  //for (ii=executorsList.begin(), ee=executorsList.end(); ii!=ee; ii++) {
//  //  printf("    [%u] [%llu]  SQ: %u  LQ: %u\n", n, (*ii).conID, (*ii).shortReqQSize, (*ii).longReqQSize);
//  //  fflush(stdout);
//  //}
//
//  return executorsList;
//}
 
bool RequestGateway::sendRestartToExecutor(uint64 id) {
    std::map<uint64, RequestConnection>::iterator i = executors.find(id);
    if (i == executors.end())
        return false;
 
    DataMessage* msg = new DataMessage();
    msg->setString("REQUEST", "__RESTART__");
 
    //printf("Sending restart message to con id %llu...\n\n", i->second.conID);
    bool res = channel->sendMessage(msg, i->second.conID);
    delete msg;
    return res;
}
 
 
//bool RequestGateway::sendToExecutor(RequestReply* reply) {
    //std::list<RequestConnection> topExecutors;
    //std::list<RequestConnection>::iterator i, e;
    //requestMap[reply->gatewayRef] = reply;
    //DataMessage *replyMsg;
 
    //DataMessage* reqMsg = reply->peekRequestMessage();
    //if (!reqMsg) {
    //  replyMsg = new DataMessage();
    //  replyMsg->setReference(reply->clientRef);
    //  if (replyXML)
    //      replyMsg->setString("XML", "<error>No input data available</error>");
    //  else
    //      replyMsg->setString("JSON", "{\"error\": \"No input data available\"}\n");
    //  replyQ.add(replyMsg);
    //  return false;
    //}
    //const char* reqStr = reqMsg->getString("URI");
    //if (!reqStr)
    //  reqStr = reqMsg->getString("REQUEST");
    //if (!reqStr) {
    //  replyMsg = new DataMessage();
    //  replyMsg->setReference(reply->clientRef);
    //  if (replyXML)
    //      replyMsg->setString("XML", "<error>No request data available</error>");
    //  else
    //      replyMsg->setString("JSON", "{\"error\": \"No request data available\"}\n");
    //  replyQ.add(replyMsg);
    //  return false;
    //}
 
    //topExecutors = getTopExecutors(reqStr, reqMsg->getSize(), reply->isLongReq);
 
    //if (!topExecutors.size()) {
    //  LogPrint(0, LOG_SYSTEM, 0, "No executors found!");
    //  reply->setStatus(FAILED);
    //  replyMsg = new DataMessage();
    //  replyMsg->setReference(reply->clientRef);
    //  if (replyXML)
    //      replyMsg->setString("XML", "<error>No processors available</error>");
    //  else
    //      replyMsg->setString("JSON", "{\"error\": \"No processors available\"}\n");
    //  replyQ.add(replyMsg);
    //  return false;
    //}
 
    //bool wasSent = false;
    //for (i=topExecutors.begin(), e=topExecutors.end(); i!=e; i++) {
    //  uint64 t = GetTimeNow();
    //  if (channel->sendMessage(reply->peekRequestMessage(), (*i).conID)) {
    //      //printf("[[[%.3fms]]]\n", GetTimeAge(t)/1000.0);
    //      executors[(*i).conID].count++;
    //      reply->processor = (*i).conID;
    //      //printf("--- GW -> EX[%llu] ---\n", (*i).conID);
    //      //fflush(stdout);
    //      //LogPrint(0, LOG_SYSTEM, 0, "Gateway sent request to Executor");
    //      wasSent = true;
    //      break;
    //  }
    //}
    //if (wasSent) {
    //  reply->setStatus(SENT);
    //  execSentCount++;
    //}
    //else {
    //  reply->setStatus(SERVERERROR);
    //  // #########
    //  // reply to client: server error
    //  return false;
    //}
    //return true;
//}
 
bool RequestGateway::replyToClient(DataMessage* msg) {
    uint64 gwRef = msg->getReference();
    uint64 now = GetTimeNow();
 
    if (!mutex.enter(1000, __FUNCTION__)) {
        LogPrint(0, LOG_SYSTEM, 0, "Error locking Gateway Client Reply mutex");
        delete(msg);
        return true;
    }
 
    std::map<uint64,RequestReply*>::iterator i = requestMap.find(gwRef), e;
    std::map<uint64,RequestConnection>::iterator con;
 
    if (i == requestMap.end()) {
        // #############
        LogPrint(0, LOG_SYSTEM, 0, "Gateway couldn't find request to reply to, ref: %llu", gwRef);
        delete(msg);
        //for (i = requestMap.begin(), e = requestMap.end(); i != e; i++) {
        //  printf("[%llu] (%llu->%llu) %s\n", i->second->gatewayRef, i->second->origin, i->second->processor, i->second->getStatusText().c_str());
        //}
        mutex.leave();
        return false;
    }
    RequestReply* reply = i->second;
 
    // Add Gateway stats
    //MovingAverage* avg;
    uint32 reqSize = reply->peekRequestMessage()->getSize();
    if (reply->isLongReq) {
        longAvgStats.add(reqSize);
        //avg = &(longAvgStatsExec[reply->processor]);
        //avg->add(reqSize);
        if (reply->processor) {
            executors[reply->processor].longReqQProcessingSize--;
            executors[reply->processor].longReqQueue.completeRequest(reply);
            executors[reply->processor].longAvgStats.add(reqSize);
        }
    }
    else {
        shortAvgStats.add(reqSize);
        //avg = &(shortAvgStatsExec[reply->processor]);
        //avg->add(reqSize);
        if (reply->processor) {
            executors[reply->processor].shortReqQueue.completeRequest(reply);
            executors[reply->processor].shortAvgStats.add(reqSize);
        }
    }
 
    if ( (con=webClients.find(reply->origin)) != webClients.end()) {
        if (reply->isLongReq)
            con->second.longAvgStats.add(reqSize);
        else
            con->second.shortAvgStats.add(reqSize);
        //  avg = &(longAvgStatsWeb[reply->origin]);
        //  avg = &(shortAvgStatsWeb[reply->origin]);
        //avg->add(reqSize);
        HTTPReply* httpReply = new HTTPReply();
        const char* xml, *html, *text, *data, *mime, *json, *cache, *options, *origin;
        int64 resultCode = 0;
        msg->getInt("ResultCode", resultCode);
        if (!resultCode) resultCode = HTTP_OK;
        cache = msg->getString("CACHE");
        bool cacheable = (cache && (utils::stristr(cache, "yes") == cache));
        uint32 size;
        // ########## Add reference and reply to HTTP reply #######
        if (xml = msg->getString("XML"))
            httpReply->createPage((uint8)resultCode, GetTimeNow(), webServerName.c_str(), 0, true, cacheable, "text/xml", xml);
        else if (json = msg->getString("JSON"))
            httpReply->createPage((uint8)resultCode, GetTimeNow(), webServerName.c_str(), 0, true, cacheable, "application/json", json);
        else if (html = msg->getString("HTML"))
            httpReply->createPage((uint8)resultCode, GetTimeNow(), webServerName.c_str(), 0, true, cacheable, "text/html", html);
        else if (text = msg->getString("TEXT"))
            httpReply->createPage((uint8)resultCode, GetTimeNow(), webServerName.c_str(), 0, true, cacheable, "text/text", html);
        else if ( (data = msg->getData("BINARY", size)) && size && (mime = msg->getString("MIMETYPE")) )
            httpReply->createPage((uint8)resultCode, GetTimeNow(), webServerName.c_str(), 0, true, cacheable, mime, data, size);
        else if ( (options = msg->getString("OPTIONS")) && (origin = msg->getString("ORIGIN")) )
            httpReply->createOptionsResponse((uint8)resultCode, GetTimeNow(), webServerName.c_str(), 0, origin, options);
        else
            httpReply->createErrorPage(HTTP_SERVER_NOREPLY, GetTimeNow(), webServerName.c_str(), true);
 
        if (channel->sendHTTPReply(httpReply, reply->origin)) {
 
            addToCallLog(++callLogCount, 2, 3, reply->isLongReq, reply->peekRequestMessage()->getTime("SOURCE"), 0, reply->processor, 0, reply->startTime, now, 0, GetTimeAge(reply->startTime), (uint8)reply->peekRequestMessage()->getInt("HTTP_OPERATION"), reply->peekRequestMessage()->getSize(), msg->getSize(), "Success",
                reply->peekRequestMessage()->getString("URI"), reply->peekRequestMessage()->getString("User-Agent"), "");
 
            LogPrint(0, LOG_SYSTEM, 3, "Replied to %s request %llu from Executor %llu via HTTP...", reply->isLongReq ? "LONG" : "SHORT", reply->gatewayRef, reply->processor);
            con->second.count++;
            delete(httpReply);
            delete(msg);
            requestMap.erase(i);
            delete(reply);
            mutex.leave();
            return true;
        }
        else {
 
            addToCallLog(++callLogCount, 2, 3, reply->isLongReq, reply->peekRequestMessage()->getTime("SOURCE"), 0, reply->processor, 0, reply->startTime, now, 0, GetTimeAge(reply->startTime), (uint8)reply->peekRequestMessage()->getInt("HTTP_OPERATION"), reply->peekRequestMessage()->getSize(), msg->getSize(), "Failed to send reply",
                reply->peekRequestMessage()->getString("URI"), reply->peekRequestMessage()->getString("User-Agent"), "");
 
            LogPrint(0, LOG_SYSTEM, 0, "Failed to reply to %s request %llu from Executor %llu via HTTP...", reply->isLongReq ? "LONG" : "SHORT", reply->gatewayRef, reply->processor);
            delete(httpReply);
            delete(msg);
            requestMap.erase(i);
            delete(reply);
            mutex.leave();
            return false;
        }
    }
    else if ((con = webSockets.find(reply->origin)) != webSockets.end()) {
        if (reply->isLongReq)
            con->second.longAvgStats.add(reqSize);
        else
            con->second.shortAvgStats.add(reqSize);
        //  avg = &(longAvgStatsWeb[reply->origin]);
        //  avg = &(shortAvgStatsWeb[reply->origin]);
        //avg->add(reqSize);
 
        WebsocketData* wsData = new WebsocketData();
        JSONM* jm = new JSONM();
        int64 resultCode = 0;
        msg->getInt("ResultCode", resultCode);
        if (!resultCode) resultCode = HTTP_OK;
 
        uint64 ref = reply->clientRef;
 
        std::string subJSON = utils::StringFormat("\"resultcode\": \"%s\", \"requestid\": %llu",
            HTTP_Status[resultCode], ref);
        std::string json = AddToJSONRoot(msg->getString("JSON"), subJSON.c_str());
        jm->setJSON(json.c_str());
 
        const char* data;
        const char* mime;
        uint32 size;
        if ((data = msg->getData("BINARY", size)) && size && (mime = msg->getString("MIMETYPE")))
            jm->addData("binary", data, size, mime);
        wsData->setData(wsData->BINARY, false, jm->jmData, jm->jmSize);
 
        if (channel->sendWebsocketData(wsData, reply->origin)) {
 
            addToCallLog(++callLogCount, 3, 3, reply->isLongReq, reply->peekRequestMessage()->getTime("SOURCE"), 0, reply->processor, 0, reply->startTime, now, 0, GetTimeAge(reply->startTime), (uint8)reply->peekRequestMessage()->getInt("HTTP_OPERATION"), reply->peekRequestMessage()->getSize(), msg->getSize(), "Success",
                reply->peekRequestMessage()->getString("URI"), reply->peekRequestMessage()->getString("User-Agent"), "");
 
            LogPrint(0, LOG_SYSTEM, 3, "Replied to %s request %llu from Executor %llu via WebSocket...", reply->isLongReq ? "LONG" : "SHORT", reply->gatewayRef, reply->processor);
            con->second.count++;
            delete(wsData);
            delete(msg);
            requestMap.erase(i);
            delete(reply);
            mutex.leave();
            return true;
        }
        else {
 
            addToCallLog(++callLogCount, 3, 3, reply->isLongReq, reply->peekRequestMessage()->getTime("SOURCE"), 0, reply->processor, 0, reply->startTime, now, 0, GetTimeAge(reply->startTime), (uint8)reply->peekRequestMessage()->getInt("HTTP_OPERATION"), reply->peekRequestMessage()->getSize(), msg->getSize(), "Failed to send reply",
                reply->peekRequestMessage()->getString("URI"), reply->peekRequestMessage()->getString("User-Agent"), "");
 
            LogPrint(0, LOG_SYSTEM, 0, "Failed to reply to %s request %llu from Executor %llu via WebSocket...", reply->isLongReq ? "LONG" : "SHORT", reply->gatewayRef, reply->processor);
            delete(wsData);
            delete(msg);
            requestMap.erase(i);
            delete(reply);
            mutex.leave();
            return false;
        }
    }
    else if ( (con=clients.find(reply->origin)) != clients.end()) {
        //LogPrint(0, LOG_SYSTEM, 0, "Gateway received reply for client %llu", reply->origin);
        if (reply->isLongReq)
            con->second.longAvgStats.add(reqSize);
        else
            con->second.shortAvgStats.add(reqSize);
        //  avg = &(longAvgStatsClients[reply->origin]);
        //  avg = &(shortAvgStatsClients[reply->origin]);
        //avg->add(reqSize);
        msg->setReference(reply->clientRef);
        msg->setFrom((uint32)reply->processor);
        msg->setStatus((uint16)reply->getStatus());
        if (channel->sendMessage(msg, reply->origin)) {
 
            addToCallLog(++callLogCount, 1, 3, reply->isLongReq, reply->peekRequestMessage()->getTime("SOURCE"), 0, reply->processor, 0, reply->startTime, now, 0, GetTimeAge(reply->startTime), (uint8)reply->peekRequestMessage()->getInt("HTTP_OPERATION"), reply->peekRequestMessage()->getSize(), msg->getSize(), "Success",
                reply->peekRequestMessage()->getString("URI"), reply->peekRequestMessage()->getString("User-Agent"), "");
 
            LogPrint(0, LOG_SYSTEM, 3, "Replied to %s request %llu from Executor %llu via DataMessage...", reply->isLongReq ? "LONG" : "SHORT", reply->gatewayRef, reply->processor);
            con->second.count++;
            clientSentCount++;
            // success
            delete(msg);
            requestMap.erase(i);
            delete(reply);
            mutex.leave();
            return true;
        }
        else {
 
            addToCallLog(++callLogCount, 1, 3, reply->isLongReq, reply->peekRequestMessage()->getTime("SOURCE"), 0, reply->processor, 0, reply->startTime, now, 0, GetTimeAge(reply->startTime), (uint8)reply->peekRequestMessage()->getInt("HTTP_OPERATION"), reply->peekRequestMessage()->getSize(), msg->getSize(), "Failed to send reply",
                reply->peekRequestMessage()->getString("URI"), reply->peekRequestMessage()->getString("User-Agent"), "");
 
            LogPrint(0, LOG_SYSTEM, 0, "Failed to reply to %s request %llu from Executor %llu via DataMessage...", reply->isLongReq ? "LONG" : "SHORT", reply->gatewayRef, reply->processor);
            delete(msg);
            requestMap.erase(i);
            delete(reply);
            mutex.leave();
            return false;
        }
    }
 
    // else it failed
    // ########
 
    addToCallLog(++callLogCount, 1, 3, reply->isLongReq, reply->peekRequestMessage()->getTime("SOURCE"), 0, reply->processor, 0, reply->startTime, now, 0, GetTimeAge(reply->startTime), (uint8)reply->peekRequestMessage()->getInt("HTTP_OPERATION"), reply->peekRequestMessage()->getSize(), msg->getSize(), "Failed to find connection to reply to",
        reply->peekRequestMessage()->getString("URI"), reply->peekRequestMessage()->getString("User-Agent"), "");
 
    LogPrint(0, LOG_SYSTEM, 0, "Failed to find anyone to reply to for %s request %llu from Executor %llu via DataMessage...", reply->isLongReq ? "LONG" : "SHORT", reply->gatewayRef, reply->processor);
    //printf("Failed request GWID: %llu\n", reply->gatewayRef);
    delete(msg);
    requestMap.erase(i);
    delete(reply);
    mutex.leave();
    return false;
}
 
 
bool RequestGateway::runExec() {
    DataMessage* replyMsg;
    RequestReply* reply;
    isRunning = true;
    uint64 t = GetTimeNow();
    uint64 lastHeartbeatTest = GetTimeNow();
    std::map<uint64, RequestConnection>::iterator i, e = executors.end(), ii;
 
    uint32 count = 0;
    while (shouldContinue) {
 
        /*
            Short requests are added to the relevant queue and put directly into the execQ for sending
            Long requests are added to the relevant queue
            - when added check if we can send it right away, if so, add to execQ
            - if not, leave it in the relevant queue
            - when reply comes back from executor to free up a space
              choose the next one in the queue and add to execQ
            - periodically check queues for missed slots
        */
 
        while (execQ.waitForAndTakeFirst(reply, 50)) {
            if (!mutex.enter(1000, __FUNCTION__)) {
                LogPrint(0, LOG_SYSTEM, 0, "Error locking Gateway Exec mutex");
                continue;
            }
            count++;
            t = GetTimeNow();
            if (
                (((ii = webClients.find(reply->origin)) != webClients.end()) && !ii->second.lastFailTime) ||
                (((ii = webSockets.find(reply->origin)) != webSockets.end()) && !ii->second.lastFailTime) ||
                (((ii = clients.find(reply->origin)) != clients.end()) && !ii->second.lastFailTime) ) {
 
                if (channel->sendMessage(reply->peekRequestMessage(), reply->processor)) {
                    //printf("[[[%.3fms]]]\n", GetTimeAge(t)/1000.0);
                    executors[reply->processor].count++;
                    //printf("--- GW -> EX[%llu] ---\n", (*i).conID);
                    //fflush(stdout);
                    //LogPrint(0, LOG_SYSTEM, 0, "Gateway sent request to Executor");
                    reply->setStatus(SENT);
                    execSentCount++;
                    //LogPrint(0, LOG_SYSTEM, 0, "Sent %s request %llu to Executor %llu via network...", reply->isLongReq ? "LONG" : "SHORT", reply->gatewayRef, reply->processor);
                }
                else {
                    reply->setStatus(SERVERERROR);
                    // #########
                    // reply to client: server error
                    LogPrint(0, LOG_SYSTEM, 0, "Error sending %s request %llu to Executor %llu via network, giving up", reply->isLongReq ? "LONG" : "SHORT", reply->gatewayRef, reply->processor);
                }
            }
            else {
                LogPrint(0, LOG_SYSTEM, 0, "Cancelling %s request %llu bound for Executor %llu from disconnected Client %llu...", reply->isLongReq ? "LONG" : "SHORT", reply->gatewayRef, reply->processor, reply->origin);
                if (reply->isLongReq)
                    executors[reply->processor].longReqQueue.completeRequest(reply);
                else
                    executors[reply->processor].shortReqQueue.completeRequest(reply);
                requestMap.erase(reply->gatewayRef);
                delete(reply);
            }
            mutex.leave();
        }
 
        if (mutex.enter(1000, __FUNCTION__)) {
            i = executors.begin();
            while (i != e) {
                while (i->second.longReqQProcessingSize < maxRequestProcessingSize) {
                    if (reply = i->second.longReqQueue.getNextRequest()) {
                        // is now marked as processing by the getNextRequest function
                        // check that the client is still connected
                        if (
                            (((ii = webClients.find(reply->origin)) != webClients.end()) && !ii->second.lastFailTime) ||
                            (((ii = webSockets.find(reply->origin)) != webSockets.end()) && !ii->second.lastFailTime) ||
                            (((ii = clients.find(reply->origin)) != clients.end()) && !ii->second.lastFailTime)) {
 
                            i->second.longReqQProcessingSize++;
                            execQ.add(reply);
                            //LogPrint(0, LOG_SYSTEM, 0, "Scheduling %s request %llu to be sent to Executor %llu shortly for Client %llu...", reply->isLongReq ? "LONG" : "SHORT", reply->gatewayRef, reply->processor, reply->origin);
                        }
                        else {
                            LogPrint(0, LOG_SYSTEM, 0, "Cancelling %s request %llu bound for Executor %llu from disconnected Client %llu...", reply->isLongReq ? "LONG" : "SHORT", reply->gatewayRef, reply->processor, reply->origin);
                            if (reply->isLongReq)
                                executors[reply->processor].longReqQueue.completeRequest(reply);
                            else
                                executors[reply->processor].shortReqQueue.completeRequest(reply);
                            requestMap.erase(reply->gatewayRef);
                            delete(reply);
                        }
                    }
                    else
                        break;
                }
 
                if (maxExecutorRequestTimeoutMS) {
                    if (i->second.removeStaleRequests(maxExecutorRequestTimeoutMS)) {
                        while ( (reply = i->second.shortReqQueue.getNextTimedoutRequest()) ||
                                (reply = i->second.longReqQueue.getNextTimedoutRequest()) ) {
                            LogPrint(0, LOG_SYSTEM, 0, "Cancelling %s request %llu bound for Executor %llu due to timeout (%s)...",
                                reply->isLongReq ? "LONG" : "SHORT", reply->gatewayRef, reply->processor,
                                PrintTimeDifString(maxExecutorRequestTimeoutMS * 1000, false, false).c_str());
                            replyMsg = new DataMessage();
                            replyMsg->setReference(reply->gatewayRef);
                            if (replyXML)
                                replyMsg->setString("XML", "<error>Timeout processing request</error>");
                            else
                                replyMsg->setString("JSON", "{\"error\": \"Timeout processing request\"}\n");
                            replyQ.add(replyMsg);
                            requestMap.erase(reply->gatewayRef);
                            delete(reply);
                        }
                    }
                }
                i++;
            }
 
            // Check executors for heartbeat timeout
            if (executorHeartbeatTimeout && (GetTimeAgeMS(lastHeartbeatTest) > 5000)) {
                i = executors.begin();
                while (i != e) {
                    if (i->second.conID && (GetTimeAgeMS(i->second.lastStatusTime) > (int32)executorHeartbeatTimeout)) {
                        LogPrint(0, LOG_SYSTEM, 0, "Gateway found frozen executor ID %u (%s), current queue size short: %u long: %u, sending restart message",
                            i->second.conID, PrintTimeDifString(GetTimeAge(i->second.lastStatusTime), false, true).c_str(),
                            i->second.longReqQueue.getCount(), i->second.shortReqQueue.getCount());
 
                        // Redistribute queued requests to other executors
                        std::list<RequestReply*> shortQueue = i->second.shortReqQueue.takeQueue();
                        std::list<RequestReply*> longQueue = i->second.longReqQueue.takeQueue();
 
                        if (!sendRestartToExecutor(i->second.conID))
                            LogPrint(0, LOG_SYSTEM, 0, "Gateway was unable to send restart message to frozen executor ID %u", i->second.conID);
                        i = executors.erase(i);
 
                        distributeDeadExecutorRequests(shortQueue, longQueue);
 
                    }
                    else
                        i++;
                }
                lastHeartbeatTest = GetTimeNow();
            }
        }
        mutex.leave();
    }
    isRunning = false;
    return true;
}
 
 
bool RequestGateway::distributeDeadExecutorRequests(std::list<RequestReply*>& shortQueue, std::list<RequestReply*>& longQueue) {
 
    RequestReply* reply;
    std::list<RequestReply*>::iterator rI, rE;
 
    std::map<uint64, RequestConnection>::iterator i, e = executors.end();
 
    while (shortQueue.size() || longQueue.size()) {
        if (shortQueue.size()) {
            reply = shortQueue.front();
            shortQueue.pop_front();
            if (reply) {
                reply->setStatus(QUEUED);
                addRequestReplyToRequestQueue(reply);
            }
        }
        if (longQueue.size()) {
            reply = longQueue.front();
            longQueue.pop_front();
            if (reply) {
                reply->setStatus(QUEUED);
                addRequestReplyToRequestQueue(reply);
            }
        }
    }
    return true;
}
 
bool RequestGateway::runClient() {
    DataMessage* msg;
    isRunning = true;
 
    uint64 lastMaintenance = GetTimeNow();
    std::map<uint64, RequestConnection>::iterator i, e = clients.end();
 
    while (shouldContinue) {
        while (msg = replyQ.waitForAndTakeFirst(50))
            replyToClient(msg);
 
        if (GetTimeAgeMS(lastMaintenance) > 5000) {
            i = clients.begin();
            while (i != e) {
                if (i->second.lastFailTime && (GetTimeAgeMS(i->second.lastFailTime) > 5000))
                    i = clients.erase(i);
                else
                    i++;
            }
        }
    }
    isRunning = false;
    return true;
}
 
THREAD_RET THREAD_FUNCTION_CALL RequestGatewayExecRun(THREAD_ARG arg) {
    if (arg == NULL) thread_ret_val(1);
    thread_ret_val((int)(((RequestGateway*)arg)->runExec() ? 0 : 1));
}
 
THREAD_RET THREAD_FUNCTION_CALL RequestGatewayClientRun(THREAD_ARG arg) {
    if (arg == NULL) thread_ret_val(1);
    thread_ret_val((int)(((RequestGateway*)arg)->runClient() ? 0 : 1));
}
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
bool RequestGateway::UnitTest(uint16 port, uint32 executorCount, uint32 gatewayCount, uint32 clientCount, uint32 webCount, uint32 webSocketCount, double sendfreq, int32 reconnect, uint32 payload, uint32 processTime, uint32 runtime) {
    uint16 n, m;
 
    //uint8 encryption = SSLENC;
    uint8 encryption = NOENC;
 
    //DataMessage *msg;
    //RequestReply* reply;
    std::list<RequestReply*> replyList;
    //RequestStatus status;
 
    RequestGateway* gateway;
    TestRequestExecutor* exec;
    //RequestClient* client;
    TestRequestClient* testClient;
    TestWebRequestClient* testWebClient;
    TestWebSocketRequestClient* testWebSocketClient;
 
    std::vector<RequestGateway*> gateways;
    std::vector<TestRequestExecutor*> executors;
    std::vector<RequestClient*> clients;
    std::vector<TestRequestClient*> testClients;
    std::vector<TestWebRequestClient*> testWebClients;
    std::vector<TestWebSocketRequestClient*> testWebSocketClients;
 
    std::string sslCertPath = "../cert";
    std::string sslKeyPath = "../pkey";
 
    // Start N gateways
    unittest::progress(5, "start gateways");
    for (n=0; n<gatewayCount; n++) {
        gateway = new RequestGateway(n);
        if (encryption == SSLENC) {
            if (gateway->init(sslCertPath.c_str(), sslKeyPath.c_str()))
                gateways.push_back(gateway);
            else {
                unittest::fail("RequestGateway test: couldn't initialise gateway %u", n);
                return false;
            }
        }
        else {
            if (gateway->init())
                gateways.push_back(gateway);
            else {
                unittest::fail("RequestGateway test: couldn't initialise gateway %u", n);
                return false;
            }
        }
        if (!gateway->addPort(port+n, encryption, true)) {
            unittest::fail("RequestGateway test: couldn't bind gateway %u public port %u", n, port + n);
            return false;
        }
        if (!gateway->addPort(port-100+n, encryption)) {
            unittest::fail("RequestGateway test: couldn't bind gateway %u port %u", n, port-100+n);
            return false;
        }
        gateway->setResponseType("JSON");
        gateway->setWebServerInfo("Psyclone", "../html/", "index.html");
    }
 
    // Start N executors
    unittest::progress(20, "start executors");
    for (n=0; n<executorCount; n++) {
        exec = new TestRequestExecutor(n+1, processTime);
        exec->addLongRequestName("longtest");
        for (m=0; m<gatewayCount; m++) {
            if (!exec->addGateway(m, "localhost", port-100+m, encryption)) {
                unittest::fail("RequestGateway test: couldn't add gateway %u to executor %u", m, n);
                return false;
            }
        }
        //if (!exec->addGateway(m, "localhost", port - 100 + m, encryption)) {
        //  LogPrint(0, LOG_SYSTEM, 0, "Couldn't add gateway %u to executor %u...", m, n);
        //  return false;
        //}
        if (!gatewayCount) {
            if (!exec->addGateway(m, "localhost", port-100, encryption)) {
                unittest::fail("RequestGateway test: couldn't add gateway 0 to executor %u", n);
                return false;
            }
            //if (!exec->addGateway(m+1, "localhost", port - 99, encryption)) {
            //  LogPrint(0, LOG_SYSTEM, 0, "Couldn't add gateway 1 to executor %u...", n);
            //  return false;
            //}
        }
        executors.push_back(exec);
    }
 
    unittest::detail("Gateway(s) and Executor(s) set up and started\n");
    utils::Sleep(1000);
 
    // Start N clients
    unittest::progress(35, "start clients");
    for (n = 0; n < clientCount; n++) {
        testClient = new TestRequestClient(n+1);
        if (!testClient->init(port, gatewayCount, sendfreq, reconnect, payload, "longtest")) {
            unittest::fail("RequestGateway test: couldn't initialise test client %u", n);
            return false;
        }
        testClients.push_back(testClient);
    }
 
    // Start N web clients
    for (n = 0; n < webCount; n++) {
        testWebClient = new TestWebRequestClient(n + 1);
        if (!testWebClient->init(port, gatewayCount, sendfreq, reconnect, payload, "longtest")) {
            unittest::fail("RequestGateway test: couldn't initialise test web client %u", n);
            return false;
        }
        testWebClients.push_back(testWebClient);
    }
 
    // Start N websocket clients
    for (n = 0; n < webSocketCount; n++) {
        testWebSocketClient = new TestWebSocketRequestClient(n + 1);
        if (!testWebSocketClient->init(port, gatewayCount, sendfreq, reconnect, payload, "longtest")) {
            unittest::fail("RequestGateway test: couldn't initialise test websocket client %u", n);
            return false;
        }
        testWebSocketClients.push_back(testWebSocketClient);
    }
 
    // Drive a bounded, OBSERVABLE batch of synchronous requests through a real
    // RequestClient so we can produce a concrete pass/fail and latency metric.
    // (The Test* clients above exercise the harness but do not expose counters.)
    unittest::progress(45, "synchronous request batch");
    uint32 reqSent = 0, reqOK = 0;
    double totalLatencyUs = 0.0;
    {
        RequestClient probe;
        bool added = false;
        for (m = 0; m < (gatewayCount ? gatewayCount : 1); m++) {
            if (probe.addGateway(m, "localhost", port + m, encryption))
                added = true;
        }
        if (!added) {
            unittest::fail("RequestGateway test: probe client couldn't add any gateway");
            return false;
        }
        probe.waitForConnection(3000);
 
        // Warm up the full client->gateway->executor pipeline before measuring.
        // The executor<->gateway registration can lag the client connection, so a
        // cold request can stall for its entire timeout. Poll with short-timeout
        // single requests until one round-trips (bounded), so the measured batch
        // below always runs against a hot pipeline and completes quickly. Without
        // this, cold requests occasionally burned 3s each and pushed the whole
        // test past its harness timeout.
        uint64 warmupStart = GetTimeNow();
        bool warm = false;
        while (!warm && GetTimeAgeMS(warmupStart) < 15000) {
            DataMessage* wmsg = new DataMessage();
            wmsg->setString("URI", "shorttest");
            RequestReply* wreply = probe.postRequest(wmsg);
            if (!wreply) { delete wmsg; utils::Sleep(100); continue; }
            if (wreply->waitForResult(1000) == SUCCESS)
                warm = true;
            probe.finishRequest(wreply);
        }
        if (!warm) {
            unittest::fail("RequestGateway test: pipeline never became ready (no request round-tripped within 15s)");
            return false;
        }
 
        // Drive a bounded batch of synchronous requests end-to-end through the
        // full client->gateway->executor->reply->client pipeline and confirm
        // every one comes back successfully.
        const uint32 probeCount = 20;
        for (uint32 r = 0; r < probeCount; r++) {
            DataMessage* msg = new DataMessage();
            msg->setString("URI", "shorttest");
            RequestReply* reply = probe.postRequest(msg);
            if (!reply) { delete msg; continue; }
            reqSent++;
            if (reply->waitForResult(3000) == SUCCESS) {
                reqOK++;
                totalLatencyUs += (double)reply->getRequestDuration();
            }
            probe.finishRequest(reply);
        }
    }
    unittest::detail("Synchronous probe: %u/%u requests succeeded\n", reqOK, reqSent);
 
    uint64 runtimeStart = GetTimeNow();
 
    // runtime==0 in the original means "run forever"; the harness must terminate,
    // so callers always pass a small positive runtime (see Test_RequestGateway).
    if (!runtime) {
        unittest::fail("RequestGateway test: runtime==0 would run forever; refusing");
        return false;
    }
 
    unittest::detail("Running background harness for %s...\n", PrintTimeDifString((uint64)runtime*1000000).c_str());
    unittest::progress(70, "run harness");
    while (GetTimeAgeMS(runtimeStart) < ((int32)runtime * 1000))
        utils::Sleep(100);
 
    // Shut down the clients...
    unittest::progress(85, "teardown");
    std::vector<TestRequestClient*>::iterator ci, ce;
    for (ci = testClients.begin(), ce = testClients.end(); ci != ce; ci++) {
        (*ci)->finishUp();
        while ((*ci)->isStillRunning())
            utils::Sleep(100);
        delete(*ci);
    }
 
    std::vector<TestWebRequestClient*>::iterator wi, we;
    for (wi = testWebClients.begin(), we = testWebClients.end(); wi != we; wi++) {
        (*wi)->finishUp();
        while ((*wi)->isStillRunning())
            utils::Sleep(100);
        delete(*wi);
    }
 
    std::vector<TestWebSocketRequestClient*>::iterator si, se;
    for (si = testWebSocketClients.begin(), se = testWebSocketClients.end(); si != se; si++) {
        (*si)->finishUp();
        while ((*si)->isStillRunning())
            utils::Sleep(100);
        delete(*si);
    }
 
    utils::Sleep(200);
    std::vector<TestRequestExecutor*>::iterator ei, ee;
    for (ei = executors.begin(), ee = executors.end(); ei != ee; ei++)
        delete(*ei);
 
    utils::Sleep(200);
    std::vector<RequestGateway*>::iterator gi, ge;
    for (gi = gateways.begin(), ge = gateways.end(); gi != ge; gi++)
        delete(*gi);
 
    // A real pass requires that the bounded synchronous batch actually completed.
    if (reqSent == 0) {
        unittest::fail("RequestGateway test: probe client posted no requests");
        return false;
    }
    if (reqOK < reqSent) {
        unittest::fail("RequestGateway test: only %u of %u requests completed successfully", reqOK, reqSent);
        return false;
    }
 
    double avgLatencyUs = totalLatencyUs / (double)reqOK;
    unittest::metric("requests_completed", (double)reqOK, "", true);
    unittest::metric("request_success_rate", 100.0 * (double)reqOK / (double)reqSent, "%", true);
    unittest::metric("avg_request_latency", avgLatencyUs, "us", false);
 
    unittest::progress(100, "done");
    return true;
 
/*  utils::Sleep(1000000000);
 
    for (n=0; n<clientCount; n++) {
        client = new RequestClient();
        for (m=0; m<gatewayCount; m++) {
            if (!client->addGateway(m, "localhost", port+m, encryption)) {
                LogPrint(0, LOG_SYSTEM, 0, "Couldn't add gateway %u to client %u...\n", m, n);
                return false;
            }
        }
        clients.push_back(client);
    }
 
    char* bigdata = new char[payload];
    memset(bigdata, 1, payload);
 
    uint32 requestCount = 300;
    // Send asynchronous binary requests
    for (n=0; n<clientCount; n++) {
        client = clients[n];
        for (m=0; m<requestCount/2; m++) {
            msg = new DataMessage();
            msg->setString("URI", "shorttest");
            if (reply = client->postRequest(msg)) {
                reply->customRef = 1;
                replyList.push_back(reply);
            }
            else {
                LogPrint(0, LOG_SYSTEM, 0, "Couldn't post short query %u via client %u...\n", m, n);
                delete(msg);
                return false;
            }
            //printf("Posted short query %u via client %u...\n", m, n);
            msg = new DataMessage();
            msg->setString("URI", "longtest");
            msg->setData("BigData", bigdata, payload);
            if (reply = client->postRequest(msg)) {
                reply->customRef = 2;
                replyList.push_back(reply);
            }
            else {
                LogPrint(0, LOG_SYSTEM, 0, "Couldn't post long query %u via client %u...\n", m, n);
                delete(msg);
                utils::Sleep(50000);
                return false;
            }
            //printf("Posted long query %u via client %u...\n", m, n);
        }
        utils::Sleep(200);
    }
    //printf("Finished posting, now waiting...\n");
 
    // Wait for results of asynchronous requests
 
    std::list<RequestReply*>::iterator i,e;
 
    uint64 startTime = GetTimeNow();
    uint32 stillWaiting, failed, timeout, errors, success, total;
    stillWaiting = failed = timeout = errors = success = total = 0;
    while (true) {
        stillWaiting = failed = timeout = errors = success = total = 0;
        for (i=replyList.begin(),e=replyList.end(); i!=e; i++) {
            reply = *i;
            status = reply->getStatus();
            total++;
            switch(status) {
                case SUCCESS:
                    //if (reply->customRef == 2) {
                    //  longStats.add((double)reply->getRequestDuration());
                    //  //printf("--- [%u] --- Long request %.3fms \n", reply->peekReplyMessage()->getFrom(), reply->getRequestDuration()/1000.0);
                    //}
                    //else if (reply->customRef == 1) {
                    //  shortStats.add((double)reply->getRequestDuration());
                    //  //printf("--- [%u] --- Short request %.3fms \n", reply->peekReplyMessage()->getFrom(), reply->getRequestDuration()/1000.0);
                    //}
                    success++;
                    break;
                case FAILED:
                    failed++;
                    break;
                case LOCALERROR:
                case NETWORKERROR:
                case SERVERERROR:
                case TOOBUSY:
                    errors++;
                    break;
                case TIMEOUT:
                    timeout++;
                    break;
                default:
                    stillWaiting++;
                    break;
            }
        }
        if (!stillWaiting || GetTimeAgeMS(startTime) > 60000)
            break;
        //if (stillWaiting != total) {
            LogPrint(0, LOG_SYSTEM, 0, "----- [%u/%u] success: %u  failed: %u  timeout: %u  errors: %u\n",
                stillWaiting, total, success, failed, timeout, errors);
            //printf("--- [%u] --- Long request %.3fms \n", reply->peekReplyMessage()->getFrom(), reply->getRequestDuration()/1000.0);
        //}
        utils::Sleep(200);
    }
    if (total == success) {
        Stats shortStats, longStats;
        for (i=replyList.begin(),e=replyList.end(); i!=e; i++) {
            reply = *i;
            if (reply->customRef == 2) {
                longStats.add((double)reply->getRequestDuration());
                LogPrint(0, LOG_SYSTEM, 0, "--- [%u] --- Long request %.3fms \n", reply->peekReplyMessage()->getFrom(), reply->getRequestDuration()/1000.0);
            }
            else if (reply->customRef == 1) {
                shortStats.add((double)reply->getRequestDuration());
                LogPrint(0, LOG_SYSTEM, 0, "--- [%u] --- Short request %.3fms \n", reply->peekReplyMessage()->getFrom(), reply->getRequestDuration()/1000.0);
            }
        }
        LogPrint(0, LOG_SYSTEM, 0, "[%u] success - short stats: %.3f (%.3f)  long stats: %.3f (%.3f)\n",
            success, shortStats.getAverage()/1000.0, shortStats.getStdDev()/1000.0, longStats.getAverage()/1000.0, longStats.getStdDev()/1000.0);
    }
    else {
        LogPrint(0, LOG_SYSTEM, 0, "[%u/%u] success: %u  failed: %u  timeout: %u  errors: %u\n",
            stillWaiting, total, success, failed, timeout, errors);
    }
 
    utils::Sleep(500000);
    // Send synchronous HTTP requests
    // #########################
 
    std::vector<RequestClient*>::iterator ci, ce;
    for (ci = clients.begin(), ce = clients.end(); ci != ce; ci++)
        delete(*ci);
    std::vector<TestRequestExecutor*>::iterator ei, ee;
    for (ei = executors.begin(), ee = executors.end(); ei != ee; ei++)
        delete(*ei);
    std::vector<RequestGateway*>::iterator gi, ge;
    for (gi = gateways.begin(), ge = gateways.end(); gi != ge; gi++)
        delete(*gi);
 
    return true;*/
}
 
 
// Adapter for the multi-argument RequestGateway::UnitTest. The framework can only
// register zero-arg bool() tests, so this spins up a small, fully self-contained
// localhost harness with a SHORT bounded runtime and returns a real pass/fail.
static bool Test_RequestGateway() {
    // port 38200 public, 38100 internal (port-100); 1 gateway, 1 executor,
    // 1 client, no web/websocket; modest send freq; no reconnect; 4 KB payload;
    // small processing time; 3-second bounded runtime so it terminates quickly.
    return RequestGateway::UnitTest(
        /*port*/          38200,
        /*executorCount*/ 1,
        /*gatewayCount*/  1,
        /*clientCount*/   1,
        /*webCount*/      0,
        /*webSocketCount*/0,
        /*sendfreq*/      20.0,
        /*reconnect*/     0,
        /*payload*/       4096,
        /*processTime*/   5,
        /*runtime*/       3);
}
 
void Register_RequestGateway_Tests() {
    UnitTestRunner::instance().registerTest("requestqueue", RequestQueue::UnitTest,
        "Request queue add/get/priority/drain", "network");
    UnitTestRunner::instance().registerTest("requestgateway", Test_RequestGateway,
        "Gateway/executor/client request round-trip over localhost", "network");
}
 
 
}