RequestClient.cpp

Go to the documentation of this file.

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

// Request Connection
 
uint32 RequestConnection::removeStaleRequests(uint32 ttlMS) {
    uint32 count = longReqQueue.removeStaleRequests(ttlMS);
    count += shortReqQueue.removeStaleRequests(ttlMS);
    return count;
}
 

// Request Reply
 
bool RequestReply::isComplete() {
    switch (this->status) {
        case SUCCESS:
        case FAILED:
        case TIMEOUT:
        case TOOBUSY:
        case LOCALERROR:
        case NETWORKERROR:
        case SERVERERROR:
            return true;
        default:
            return false;
    }
    return false;
}
 
bool RequestReply::setStatus(RequestStatus status) {
    if (!mutex.enter(1000))
        return false;
    this->status = status;
    mutex.leave();
    return true;
}
 
RequestStatus RequestReply::getStatus() {
    RequestStatus res = LOCALERROR;
    if (!mutex.enter(1000))
        return res;
    res = this->status;
    mutex.leave();
    return res;
}
 
 
std::string RequestReply::getStatusText() {
    RequestStatus res = LOCALERROR;
    if (!mutex.enter(1000))
        return "Locking error";
    res = this->status;
    mutex.leave();
    switch (res) {
    case IDLE:
        return "Request idle";
    case QUEUED:
        return "Request queued";
    case SENT:
        return "Request sent";
    case SUCCESS:
        return "Request success";
    case FAILED:
        return "Request failed";
    case TIMEOUT:
        return "Request timed out";
    case LOCALERROR:
        return "Request local error";
    case NETWORKERROR:
        return "Request network error";
    case SERVERERROR:
        return "Request server error";
    case TOOBUSY:
        return "Request server too busy";
    default:
        return "Error getting status";
    }
}
 
RequestCallbackFunction RequestReply::getCallback() {
    RequestCallbackFunction res = NULL;
    if (!mutex.enter(1000))
        return res;
    res = this->callback;
    mutex.leave();
    return res;
}
 
bool RequestReply::setCallback(RequestCallbackFunction callback) {
    if (!mutex.enter(1000))
        return false;
    this->callback = callback;
    mutex.leave();
    return true;
}
 
bool RequestReply::giveRequestMessage(DataMessage* msg) {
    if (!mutex.enter(1000))
        return false;
    if (requestMsg)
        delete(requestMsg);
    requestMsg = msg;
    systemID = msg->getSystemID();
    mutex.leave();
    return true;
}
 
bool RequestReply::setRequestMessageCopy(DataMessage* msg) {
    if (!mutex.enter(1000))
        return false;
    if (requestMsg)
        delete(requestMsg);
    if (msg)
        requestMsg = new DataMessage(*msg);
    else
        requestMsg = NULL;
    systemID = msg->getSystemID();
    mutex.leave();
    return true;
}
 
DataMessage* RequestReply::getRequestMessageCopy() {
    DataMessage* msg = NULL;
    if (!mutex.enter(1000))
        return NULL;
    msg = new DataMessage(*requestMsg);
    mutex.leave();
    return msg;
}
 
DataMessage* RequestReply::peekRequestMessage() {
    return requestMsg;
}
 
bool RequestReply::giveReplyMessage(DataMessage* msg) {
    if (!mutex.enter(1000))
        return false;
    if (replyMsg)
        delete(replyMsg);
    replyMsg = msg;
    mutex.leave();
    return true;
}
 
bool RequestReply::setReplyMessageCopy(DataMessage* msg) {
    if (!mutex.enter(1000))
        return false;
    if (replyMsg)
        delete(replyMsg);
    if (msg)
        replyMsg = new DataMessage(*msg);
    else
        replyMsg = NULL;
    mutex.leave();
    return true;
}
 
DataMessage* RequestReply::getReplyMessageCopy() {
    DataMessage* msg = NULL;
    if (!mutex.enter(1000))
        return NULL;
    msg = new DataMessage(*replyMsg);
    mutex.leave();
    return msg;
}
 
DataMessage* RequestReply::peekReplyMessage() {
    return replyMsg;
}
 
bool RequestReply::replyToRequest(DataMessage* msg, RequestStatus status) {
    //if (!msg) return false;
    if (!mutex.enter(1000))
        return false;
    if (replyMsg)
        delete(replyMsg);
    replyMsg = msg;
    this->status = status;
    finishTime = GetTimeNow();
    semaphore.signal();
 
    //if (customRef == 2) {
    //  printf("--- [%u] --- Long request %.3fms \n", msg->getFrom(), ((double)finishTime-startTime)/1000.0);
    //  fflush(stdout);
    //}
    //else if (customRef == 1) {
    //  printf("--- [%u] --- Short request %.3fms \n", msg->getFrom(), ((double)finishTime-startTime)/1000.0);
    //  fflush(stdout);
    //}
 
    mutex.leave();
    return true;
}
 
uint64 RequestReply::getRequestDuration() {
    if (finishTime)
        return finishTime-startTime;
    else
        return 0;
}
 
uint32 RequestReply::getRequestDurationMS() {
    if (finishTime)
        return (uint32)((finishTime-startTime)/1000);
    else
        return 0;
}
 
RequestStatus RequestReply::waitForResult(uint32 timeoutMS) {
    RequestStatus result;
    if (!mutex.enter(1000, __FUNCTION__))
        return LOCALERROR;
 
    //if ( (status == SUCCESS) || (status == FAILED) || (status == TIMEOUT) || (status == LOCALERROR) || (status == NETWORKERROR) || (status == SERVERERROR) ) {
    if (isComplete()) {
        result = status;
        mutex.leave();
        return result;
    }
 
    mutex.leave();
    if (semaphore.wait(timeoutMS)) {
        if (!mutex.enter(1000, __FUNCTION__))
            return LOCALERROR;
        //if ( (status == SUCCESS) || (status == FAILED) || (status == TIMEOUT) || (status == LOCALERROR) || (status == NETWORKERROR) || (status == SERVERERROR) ) {
        if (isComplete()) {
            result = status;
            mutex.leave();
            return result;
        }
        mutex.leave();
    }
    return TIMEOUT;
}
 
DataMessage* RequestReply::waitForMessage(uint32 timeoutMS, bool takeMessage) {
    // Fast path: if the reply (or a terminal status) is already in, return it
    // under the mutex. Otherwise block on the semaphore that replyToRequest()
    // signals, then re-check under the mutex — the double-check is needed
    // because the semaphore can also be signalled for failure statuses where
    // replyMsg stays NULL. With takeMessage the internal pointer is cleared so
    // ownership moves to the caller (and finishRequest() won't double-delete).
    DataMessage* msg;
    if (!mutex.enter(1000, __FUNCTION__))
        return NULL;
 
    if (replyMsg || finishTime) {
        msg = replyMsg;
        if (takeMessage)
            replyMsg = NULL;
        mutex.leave();
        return msg;
    }
 
    mutex.leave();
    if (semaphore.wait(timeoutMS)) {
        if (!mutex.enter(1000, __FUNCTION__))
            return NULL;
        if (replyMsg) {
            msg = replyMsg;
            if (takeMessage)
                replyMsg = NULL;
            mutex.leave();
            return msg;
        }
        mutex.leave();
    }
    return NULL;
}
 

// Request Client
 
RequestClient::RequestClient() {
    lastRefID = 0;
    sentCount = 0;
    receivedCount = 0;
    shouldContinue = true;
    isRunning = false;
    channel = NULL;
    manager = new NetworkManager();
 
    if (!ThreadManager::CreateThread(RequestClientRun, this, threadID, 0)) {
        shouldContinue = false;
    }
}
 
RequestClient::~RequestClient() {
    conMutex.enter(3000, __FUNCTION__);
    stop();
 
    std::map<uint64, RequestReply*>::iterator i, e;
    mutex.enter(1000, __FUNCTION__);
    for (i = requestMap.begin(), e = requestMap.end(); i != e; i++)
        delete(i->second);
    requestMap.clear();
    mutex.leave();
    conMutex.leave();
 
    delete(manager);
}
 
bool RequestClient::addGateway(uint32 id, std::string addr, uint16 port, uint8 encryption) {
    
    std::list<RequestGatewayConnection>::iterator i, e = connections.end();
    if (!mutex.enter(1000, __FUNCTION__))
        return false;
    // Check to see if this gateway is already in the list
    i = connections.begin();
    while (i != e) {
        if ( (stricmp((*i).addr.c_str(), addr.c_str()) == 0) && ((*i).port == port) ) {
            mutex.leave();
            return true;
        }
        i++;
    }
 
    RequestGatewayConnection con;
    con.clear();
    con.id = id;
    con.addr = addr;
    con.port = port;
    con.encryption = encryption;
 
//  if (!mutex.enter(1000, __FUNCTION__))
//      return false;
 
    DataMessage* msgConnect = new DataMessage();
    msgConnect->setString("URI", "ClientConnect");
 
    if (!channel)
        channel = manager->addTCPConnection(con.addr.c_str(), con.port, con.encryption, PROTOCOL_MESSAGE, true, 0, this, con.conID, con.location, 1000, (const char*)msgConnect->data, msgConnect->getSize());
    else
        con.conID = channel->addTCPConnection(con.addr.c_str(), con.port, con.encryption, PROTOCOL_MESSAGE, true, con.location, 1000, (const char*)msgConnect->data, msgConnect->getSize());
 
    delete msgConnect;
 
    connections.push_back(con);
    mutex.leave();
    return true;
}
 
RequestReply* RequestClient::postRequest(DataMessage *msg) {
    // Correlation scheme: each outgoing request gets the next monotonically
    // increasing clientRef, stamped into the message's reference field. The
    // gateway echoes it back in the reply, and receiveMessage() uses it to
    // find this RequestReply in requestMap. The message itself is owned by
    // the reply object from here on; the worker thread (run()) picks the
    // reply up from outQ and dispatches it to a connected gateway.
    if (!msg)
        return NULL;
    if (!mutex.enter(1000, __FUNCTION__))
        return NULL;
    msg->setReference(++lastRefID);
    
    RequestReply* reply = new RequestReply();
    reply->setStatus(QUEUED);
    reply->giveRequestMessage(msg);
    reply->clientRef = lastRefID;
    requestMap[lastRefID] = reply;
    outQ.add(reply);
    mutex.leave();
    return reply;
}
 
bool RequestClient::postRequest(DataMessage *msg, RequestCallbackFunction callback, uint32 timeoutMS) {
    // ##############
    return false;
}
 
bool RequestClient::finishRequest(RequestReply* reply) {
    // Deletes the RequestReply found in requestMap — and with it both the
    // request and (unless taken via waitForMessage(..., true)) the reply
    // DataMessage. Despite the isInUse pooling flag, objects are currently
    // deleted outright rather than recycled (see the commented-out branch).
    if (!reply || !reply->clientRef) return false;
    if (!mutex.enter(1000, __FUNCTION__))
        return false;
 
    std::map<uint64, RequestReply*>::iterator i = requestMap.find(reply->clientRef);
    if (i == requestMap.end())
        return false;
    if (i->second) {
        delete i->second;
        requestMap.erase(i);
        //if (i->second->isComplete()) {
        //  delete i->second;
        //  requestMap.erase(i);
        //}
        //else
        //  i->second->isInUse = false;
    }
    mutex.leave();
    return true;
}
 
 
bool RequestClient::receiveNetworkEvent(NetworkEvent* evt, NetworkChannel* channel, uint64 conid) {
    uint64 now = GetTimeNow();
    std::list<RequestGatewayConnection>::iterator i, e = connections.end();
    if (!mutex.enter(1000, __FUNCTION__))
        return false;
    i = connections.begin();
    while (i != e) {
        if ((*i).conID == conid) {
            switch (evt->type) {
            case NETWORKEVENT_DISCONNECT:
            case NETWORKEVENT_DISCONNECT_RETRYING:
                LogPrint(0, LOG_SYSTEM, 0, "Disconnected from gateway %u on %s:%u", conid, (*i).addr.c_str(), (*i).port);
                (*i).lastFailTime = now;
                (*i).lastConTime = 0;
                //(*i).conID = 0;
                break;
            case NETWORKEVENT_CONNECT:
            case NETWORKEVENT_RECONNECT:
                LogPrint(0, LOG_SYSTEM, 0, "Connected to gateway %u on %s:%u", conid, (*i).addr.c_str(), (*i).port);
                (*i).lastFailTime = 0;
                (*i).lastConTime = now;
                break;
            default:
                break;
            }
        }
        i++;
    }
    delete evt;
    mutex.leave();
    return true;
}
 
bool RequestClient::receiveMessage(DataMessage* msg, NetworkChannel* channel, uint64 conid) {
    // Network-thread reply path: look up the pending RequestReply by the
    // echoed reference id and complete it via replyToRequest(), which takes
    // ownership of msg, wakes any waitForResult()/waitForMessage() waiter and
    // fires the callback. The terminal status comes from the message's own
    // status field (set by the gateway: SUCCESS, TIMEOUT, TOOBUSY, ...).
    // Unmatched replies (already finished/timed out locally) are dropped.
    if (!msg) return false;
    if (!mutex.enter(1000, __FUNCTION__)) {
        LogPrint(0, LOG_SYSTEM, 0, "Error locking Client mutex");
        delete(msg);
        return true;
    }
    uint64 ref = msg->getReference();
    std::map<uint64,RequestReply*>::iterator i = requestMap.find(ref);
 
    if (i == requestMap.end()) {
        // #############
        LogPrint(0, LOG_SYSTEM, 0, "Couldn't find reply ID %llu", ref);
        mutex.leave();
        delete(msg);
        return true;
    }
    RequestReply* reply = i->second;
    //reply->replyToRequest(msg, SUCCESS);
    reply->replyToRequest(msg, (RequestStatus)msg->getStatus());
    if (!reply->isInUse) {
        LogPrint(0, LOG_SYSTEM, 0, "Reply ID %llu no longer in use", ref);
        requestMap.erase(i);
        delete reply;
    }
    //else
    //  LogPrint(0, LOG_SYSTEM, 0, "Got reply ID %llu - status %u", ref, msg->getStatus());
    receivedCount++;
    mutex.leave();
    return true;
}
 
bool RequestClient::receiveHTTPReply(HTTPReply* reply, HTTPRequest* req, NetworkChannel* channel, uint64 conid) {
    return false;
}
 
 
bool RequestClient::sendMessageToGateway(DataMessage* msg, RequestGatewayConnection& con) {
 
    uint64 now = GetTimeNow();
    if (!channel || !con.lastConTime)
        return false;
    LogPrint(0, LOG_NETWORK, 3, "Sending data to gateway %u on %s:%u...", con.id, con.addr.c_str(), con.port);
 
    if (!channel->sendMessage(msg, con.conID)) {
        if (!con.lastFailTime)
            LogPrint(0, LOG_SYSTEM, 0, "Unable to send data to gateway %u on %s:%u, disconnecting...", con.id, con.addr.c_str(), con.port);
        con.lastFailTime = GetTimeNow();
        // clearly not working, disconnect and try again later
        //channel->endConnection(con.conID);
        //con.conID = 0;
        return false;
    }
    return true;
 
    //if (!con.conID || !channel) {
    //  if (!channel)
    //      channel = manager->createTCPConnection(con.addr.c_str(), con.port, con.encryption, PROTOCOL_MESSAGE, true, true, con.id, this, con.conID, con.location);
    //  else
    //      con.conID = channel->createTCPConnection(con.addr.c_str(), con.port, con.encryption, PROTOCOL_MESSAGE, true, true, con.location);
    //  if (!channel || !con.conID) {
    //      con.lastFailTime = GetTimeNow();
    //      return false;
    //  }
    //  DataMessage* msg = new DataMessage();
    //  msg->setString("URI", "ClientConnect");
    //  if (!channel->sendMessage(msg, con.conID)) {
    //      // clearly not working, disconnect and try again later
    //      channel->endConnection(con.conID);
    //      con.conID = 0;
    //      con.lastFailTime = GetTimeNow();
    //      return false;
    //  }
    //  con.lastConTime = GetTimeNow();
    //}
    //if (!channel->sendMessage(msg, con.conID)) {
    //  // clearly not working, disconnect and try again later
    //  channel->endConnection(con.conID);
    //  con.conID = 0;
    //  con.lastFailTime = GetTimeNow();
    //  return false;
    //}
    //con.lastFailTime = 0;
    //return true;
}
 
 
bool RequestClient::run() {
    // Worker loop: (1) every ~3s push a ClientStatus heartbeat to all gateways
    // (also doubles as a liveness probe that triggers reconnection inside
    // sendMessageToGateway); (2) drain outQ with a 50ms blocking wait and hand
    // each pending RequestReply to sendRequest(), which picks a connected
    // gateway. Reply matching happens on the network thread in
    // receiveMessage(), not here.
    RequestReply* reply;
    isRunning = true;
 
    // should regularly test for old replies to be deleted...
    // ###############
 
    int32 heartbeatInterval = 3000;
    uint64 lastHeartbeat = 0;
 
    while (shouldContinue) {
        if (!lastHeartbeat || GetTimeAgeMS(lastHeartbeat) > heartbeatInterval) {
            sendStatusNow();
            lastHeartbeat = GetTimeNow();
        }

        //if (!channel && connections.size()) {
        //  conMutex.enter(3000, __FUNCTION__);
        //  DataMessage* msg = new DataMessage();
        //  msg->setString("URI", "ClientStatus");
        //  std::list<RequestGatewayConnection>::iterator iCon, eCon = connections.end();
        //  iCon = connections.begin();
        //  while (iCon != eCon) {
        //      if (sendMessageToGateway(msg, (*iCon)))
        //          break;
        //      iCon++;
        //  }
        //  delete msg;
        //  conMutex.leave();
        //}
 
        if (reply = outQ.waitForAndTakeFirst(50))
            sendRequest(reply);

    }
    isRunning = false;
    return true;
}
 
uint32 RequestClient::sendStatusNow() {
 
    if (!conMutex.enter(1000, __FUNCTION__)) {
        LogPrint(0, LOG_SYSTEM, 0, "Mutex error sending heartbeat from Client...");
        return 0;
    }
    if (!connections.size()) {
        //LogPrint(0, LOG_SYSTEM, 0, "No connections sending heartbeat from Executor %u...", executorID);
        conMutex.leave();
        return 0;
    }
 
    DataMessage* msg = new DataMessage();
    msg->setString("URI", "ClientStatus");
 
    uint32 success = 0, failed = 0, notConnected = 0;
 
    std::list<RequestGatewayConnection>::iterator iCon, eCon = connections.end();
    iCon = connections.begin();
    while (iCon != eCon) {
        if (sendMessageToGateway(msg, (*iCon)))
            success++;
        else
            failed++;
        iCon++;
    }
    conMutex.leave();
 
    delete(msg);
    if (failed)
        LogPrint(0, LOG_SYSTEM, 2, "Failed sending heartbeat from Client to %u gateways (%u succeeded)...", failed, success);
    //  else
    //      LogPrint(0, LOG_SYSTEM, 0, "Sent heartbeat from Executor %u to %u gateways...", executorID, success);
    return success;
}
 
 
 
bool RequestClient::waitForConnection(uint32 timeoutMS) {
    bool isCon = false;
    uint64 start = GetTimeNow();
    while (!(isCon = isConnected()) && (GetTimeAgeMS(start) < (int32)timeoutMS))
        utils::Sleep(50);
    return isCon;
}
 
bool RequestClient::isConnected() {
    if (!conMutex.enter(3000, __FUNCTION__))
        return false;
    std::list<RequestGatewayConnection>::iterator iCon, eCon = connections.end();
    iCon = connections.begin();
    while (iCon != eCon) {
        //printf("RUN0\n");fflush(stdout);
        //if ((*iCon).conID && channel->isConnected((*iCon).conID)) {
        if ((*iCon).conID && (*iCon).lastConTime) {
            conMutex.leave();
            return true;
        }
        iCon++;
    }
    conMutex.leave();
    return false;
    //bool res = false;
    //DataMessage* msg = new DataMessage();
    //msg->setString("URI", "ClientStatus");
    //iCon = connections.begin();
    //while (iCon != eCon) {
    //  if (sendMessageToGateway(msg, (*iCon)))
    //      break;
    //  iCon++;
    //}
    //conMutex.leave();
    //return res;
}
 
bool RequestClient::reconnect() {
    if (!conMutex.enter(3000, __FUNCTION__))
        return false;
    std::list<RequestGatewayConnection>::iterator iCon, eCon = connections.end();
    iCon = connections.begin();
    while (iCon != eCon) {
        if ((*iCon).conID && channel->isConnected((*iCon).conID)) {
            channel->endConnection((*iCon).conID);
        }
        iCon++;
    }
    conMutex.leave();
    return true;
}
 
 
bool RequestClient::sendRequest(RequestReply* reply) {
    if (!reply)
        return false;
 
    uint64 now = GetTimeNow();
    conMutex.enter(3000, __FUNCTION__);
    std::list<RequestGatewayConnection>::iterator iCon, eCon = connections.end();
    iCon = connections.begin();
    while (iCon != eCon) {
        if (sendMessageToGateway(reply->peekRequestMessage(), (*iCon))) {
            reply->setStatus(SENT);
            sentCount++;
            conMutex.leave();
            return true;
        }
        iCon++;
    }
    // We didn't manage to send it, mark it as failed
    reply->replyToRequest(NULL, NETWORKERROR);
    //reply->setStatus(NETWORKERROR);
    conMutex.leave();
    return false;
}
 
THREAD_RET THREAD_FUNCTION_CALL RequestClientRun(THREAD_ARG arg) {
    if (arg == NULL) thread_ret_val(1);
    thread_ret_val((int)(((RequestClient*)arg)->run() ? 0 : 1));
}
 
 
 
 
 
 
std::string RequestClient::GetJSONReplyParameter(const char* json, const char* name) {
 
    jsmn_parser parser;
    jsmntok_t* tokens = new jsmntok_t[1024];
    int tokenCount = 0;
    const char* val;
    int valSize;
    jsmn_init(&parser);
    if ((tokenCount = jsmn_parse(&parser, json, strlen(json), tokens, 1024)) <= 0) {
        // No JSON found
        delete[] tokens;
        return "";
    }
    if (!(val = GetJSONValue(tokens, tokenCount, json, name, JSMN_UNDEFINED, valSize))) {
        // JSON key not found
        delete[] tokens;
        return "";
    }
    delete[] tokens;
    std::string value = std::string(val, valSize);
    return value.c_str();
}
 
DataMessage* RequestClient::SendRequestAndWaitForReply(RequestClient* client, DataMessage* msg, int timeoutMS) {
    //if (!client->isConnected())
    //  return NULL;
 
    DataMessage *replyMsg;
    RequestReply* reply;
    if (reply = client->postRequest(msg)) {
        if (replyMsg = reply->waitForMessage(timeoutMS, true)) {
            client->finishRequest(reply);
            //delete(reply);
            return replyMsg;
        }
        client->finishRequest(reply);
        //delete(reply);
    }
    return NULL;
}
 
RequestReply* RequestClient::SendRequestAndWaitForReplyObject(RequestClient* client, DataMessage* msg, int timeoutMS) {
    //DataMessage *replyMsg;
    RequestReply* reply;
    if (reply = client->postRequest(msg)) {
        reply->waitForResult(timeoutMS);
        return reply;
        //client->finishRequest(reply);
    }
    return NULL;
}
 
 
std::string RequestClient::SendRequestAndWaitForJSON(RequestClient* client, DataMessage* msg, int timeoutMS) {
    //if (!client->isConnected())
    //  return "";
 
    DataMessage *replyMsg = SendRequestAndWaitForReply(client, msg, timeoutMS);
    if (!replyMsg)
        return "";
    const char* json = replyMsg->getString("JSON");
    if (!json) {
        delete(replyMsg);
        return "";
    }
 
    std::string jsonReply = json;
    delete(replyMsg);
    return jsonReply;
}
 
DataMessage* RequestClient::CreateRequestMessage(uint8 operation, const char* req,
    const char* key1, const char* val1,
    const char* key2, const char* val2,
    const char* key3, const char* val3,
    const char* key4, const char* val4,
    const char* key5, const char* val5,
    const char* key6, const char* val6,
    const char* key7, const char* val7,
    const char* key8, const char* val8,
    const char* key9, const char* val9,
    const char* key10, const char* val10,
    const char* key11, const char* val11,
    const char* key12, const char* val12,
    const char* key13, const char* val13,
    const char* key14, const char* val14,
    const char* key15, const char* val15,
    const char* key16, const char* val16,
    const char* key17, const char* val17,
    const char* key18, const char* val18,
    const char* key19, const char* val19,
    const char* key20, const char* val20
    ) {
    DataMessage* msg = new DataMessage();
    msg->setInt("HTTP_OPERATION", operation);
    msg->setString("URI", req);
    if (key1) msg->setString(key1, val1);
    if (key2) msg->setString(key2, val2);
    if (key3) msg->setString(key3, val3);
    if (key4) msg->setString(key4, val4);
    if (key5) msg->setString(key5, val5);
    if (key6) msg->setString(key6, val6);
    if (key7) msg->setString(key7, val7);
    if (key8) msg->setString(key8, val8);
    if (key9) msg->setString(key9, val9);
    if (key10) msg->setString(key10, val10);
    if (key11) msg->setString(key11, val11);
    if (key12) msg->setString(key12, val12);
    if (key13) msg->setString(key13, val13);
    if (key14) msg->setString(key14, val14);
    if (key15) msg->setString(key15, val15);
    if (key16) msg->setString(key16, val16);
    if (key17) msg->setString(key17, val17);
    if (key18) msg->setString(key18, val18);
    if (key19) msg->setString(key19, val19);
    if (key20) msg->setString(key20, val20);
    return msg;
}
 
 
 
 
 
bool RequestClient::TestServerLogin(const char* address, uint16 port, const char* username, const char* password, const char* reqAfterLogin) {
 
    RequestClient* reqClient = new RequestClient();
    //  api->logPrint(1, "Connecting to Face Server on %s:%u...", addr.c_str(), port, username.c_str());
    if (!reqClient->addGateway(1, address, port, NOENC)) {
        LogPrint(0, 0, 1, "Could not connect to Face Server on %s:%u...", address, port);
        delete(reqClient);
        return false;
    }
 
    std::string json;
    if (!(json = RequestClient::SendRequestAndWaitForJSON(
        reqClient,
        RequestClient::CreateRequestMessage(HTTP_POST, "authenticate", "email", username, "password", password, "clientid", "1", "locationid", "1"),
        5000)).length()) {
        LogPrint(0, 0, 1, "Couldn't login to server account...");
        return false;
    }
    std::string token = RequestClient::GetJSONReplyParameter(json.c_str(), "token");
    if (!token.length()) {
        LogPrint(0, 0, 1, "Client login failed: %s...", RequestClient::GetJSONReplyParameter(json.c_str(), "text").c_str());
        return false;
    }
 
    LogPrint(0, 0, 1, "Successfully logged in with token: %s...", token.c_str());
 
    if (!reqClient->isConnected()) {
        LogPrint(0, 0, 1, "Couldn't connect to server...");
        return false;
    }
 
    return true;
}
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
TestRequestClient::TestRequestClient(uint32 id) {
    this->id = id;
    port = 10000;
    gatewayCount = 1;
    sendfreq = 1;
    reconnectMS = 0;
    payload = 1024;
    shouldContinue = true;
    isRunning = true;
    shouldFinishUp = false;
}
 
bool TestRequestClient::init(uint16 port, uint32 gatewayCount, double sendfreq, int32 reconnectMS, uint32 payload, const char* longTestName) {
    uint32 myThreadID;
    this->port = port;
    if (gatewayCount)
        this->gatewayCount = gatewayCount;
    else
        this->gatewayCount = 1;
    this->sendfreq = sendfreq;
    this->reconnectMS = reconnectMS;
    this->payload = payload;
    this->longTestName = longTestName;
    return ThreadManager::CreateThread(TestRequestClientRun, this, myThreadID, 0);
}
 
bool TestRequestClient::finishUp() {
    LogPrint(0, LOG_SYSTEM, 0, "Client %u has been asked to finish up...", id);
    shouldFinishUp = true;
    return true;
}
 
bool TestRequestClient::run() {
    RequestClient* client = NULL;
    //DataMessage* replyMsg;
    RequestReply* reply;
    std::list<RequestReply*> replyList;
    std::list<RequestReply*>::iterator i, e = replyList.end();
    RequestStatus status;
 
    int32 intervalMS = 1000;
    if (sendfreq > 0)
        intervalMS = (int32)(1000.0 / sendfreq);
 
    uint64 lastPost = 0;
    uint64 lastReconnect = 0;
    uint64 lastPrint = GetTimeNow();
    int32 printIntervalMS = 3000;
 
    uint32 stillWaiting, failed, timeout, errors, success, total, tooBusy, serverError;
    stillWaiting = failed = timeout = errors = success = total = tooBusy = serverError = 0;
 
    char* bigdata = new char[payload];
    memset(bigdata, 1, payload);
 
    DataMessage* msgTemplate = new DataMessage();
    msgTemplate->setString("URI", "longtest");
    msgTemplate->setData("BigData", bigdata, payload);
 
    delete[] bigdata;
 
    Stats responseTimes;
 
    while (shouldContinue) {
 
        if (!client) {
            client = new RequestClient();
            for (uint16 m = 0; m<gatewayCount; m++) {
                if (!client->addGateway(m, "localhost", port + m, NOENC)) {
                    LogPrint(0, LOG_SYSTEM, 0, "Couldn't add gateway %u to client %u...\n", m);
                    delete msgTemplate;
                    return false;
                }
            }
            lastReconnect = GetTimeNow();
            utils::Sleep(100);
        }
 
        // is it time to send the next message?
        if (!shouldFinishUp && (!lastPost || (GetTimeAgeMS(lastPost) >= intervalMS))) {
 
            if (reconnectMS < 0) {
                reply = RequestClient::SendRequestAndWaitForReplyObject(client, msgTemplate->copy(), 5000);
                if (!reply) {
                    LogPrint(0, 0, 1, "Couldn't send request to server...");
                    //delete msgTemplate;
                    //return false;
                    failed++;
                }
 
                status = reply->getStatus();
                switch (status) {
                case SUCCESS:
                    responseTimes.add((double)reply->getRequestDuration());
                    success++;
                    break;
                case FAILED:
                    failed++;
                    break;
                case TOOBUSY:
                    tooBusy++;
                    break;
                case LOCALERROR:
                case NETWORKERROR:
                case SERVERERROR:
                    serverError++;
                    break;
                case TIMEOUT:
                    timeout++;
                    break;
                default:
                    stillWaiting++;
                    break;
                }
                client->finishRequest(reply);
                delete client;
                client = NULL;
                lastPost = GetTimeNow();
                continue;
            }
 
            // post request
            reply = client->postRequest(msgTemplate->copy());
            if (!reply) {
                LogPrint(0, 0, 1, "Couldn't post request to server...");
                //delete msgTemplate;
                //return false;
            }
            replyList.push_back(reply);
            lastPost = GetTimeNow();
            total++;
        }
 
        stillWaiting = 0;
        i = replyList.begin();
        while (i != e) {
            reply = *i;
            status = reply->getStatus();
            switch (status) {
            case SUCCESS:
                responseTimes.add((double)reply->getRequestDuration());
                success++;
                break;
            case FAILED:
                failed++;
                break;
            case TOOBUSY:
                tooBusy++;
                break;
            case LOCALERROR:
            case NETWORKERROR:
            case SERVERERROR:
                serverError++;
                break;
            case TIMEOUT:
                timeout++;
                break;
            default:
                stillWaiting++;
                break;
            }
            if (reply->isComplete()) {
                client->finishRequest(reply);
                i = replyList.erase(i);
            }
            else
                i++;
 
            /*
            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 (shouldFinishUp && !stillWaiting) {
            break;
        }
        
        if (GetTimeAgeMS(lastPrint) > printIntervalMS) {
            if (tooBusy || serverError || failed || timeout || errors)
                LogPrint(0, LOG_SYSTEM, 0, "%s[%u] success: %u of %u (w: %u): %.3fms - busy: %u  servererror: %u  failed: %u  timeout: %u  errors: %u\n",
                    shouldFinishUp ? "Finishing " : "",
                    id, success, total, stillWaiting, responseTimes.getAverage() / 1000.0, tooBusy, serverError, failed, timeout, errors);
            else
                LogPrint(0, LOG_SYSTEM, 0, "%s[%u] success: %u of %u (w: %u): %.3fms\n",
                    shouldFinishUp ? "Finishing " : "",
                    id, success, total, stillWaiting, responseTimes.getAverage() / 1000.0);
            lastPrint = GetTimeNow();
        }
 
 
        if ((reconnectMS > 0) && (GetTimeAgeMS(lastReconnect) >= reconnectMS)) {
            LogPrint(0, LOG_SYSTEM, 0, "Client %u forcefully reconnecting...");
            delete client;
            client = NULL;
            // delete all unprocessed replies
            replyList.clear();
            continue;
        }
 
        utils::Sleep(20);
    }
 
    if (tooBusy || serverError || failed || timeout || errors)
        LogPrint(0, LOG_SYSTEM, 0, "Completed [%u] success: %u of %u (w: %u): %.3fms - busy: %u  servererror: %u  failed: %u  timeout: %u  errors: %u\n",
            id, success, total, stillWaiting, responseTimes.getAverage() / 1000.0, tooBusy, serverError, failed, timeout, errors);
    else
        LogPrint(0, LOG_SYSTEM, 0, "Completed [%u] success: %u of %u (w: %u): %.3fms\n",
            id, success, total, stillWaiting, responseTimes.getAverage() / 1000.0);
 
    delete client;
    client = NULL;
    isRunning = false;
    delete msgTemplate;
    return true;
}
 
THREAD_RET THREAD_FUNCTION_CALL TestRequestClientRun(THREAD_ARG arg) {
    if (arg == NULL) thread_ret_val(1);
    thread_ret_val((int)(((TestRequestClient*)arg)->run() ? 0 : 1));
}
 
 
 
 
 
 
 
 
 
 
 
 
 
TestWebRequestClient::TestWebRequestClient(uint32 id) {
    this->id = id;
    port = 10000;
    gatewayCount = 1;
    sendfreq = 1;
    reconnectMS = 0;
    payload = 1024;
    shouldContinue = true;
    isRunning = true;
    shouldFinishUp = false;
}
 
bool TestWebRequestClient::init(uint16 port, uint32 gatewayCount, double sendfreq, int32 reconnectMS, uint32 payload, const char* longTestName) {
    uint32 myThreadID;
    this->port = port;
    if (gatewayCount)
        this->gatewayCount = gatewayCount;
    else
        this->gatewayCount = 1;
    this->sendfreq = sendfreq;
    this->reconnectMS = reconnectMS;
    this->payload = payload;
    this->longTestName = longTestName;
    return ThreadManager::CreateThread(TestWebRequestClientRun, this, myThreadID, 0);
}
 
bool TestWebRequestClient::finishUp() {
    LogPrint(0, LOG_SYSTEM, 0, "Client %u has been asked to finish up...", id);
    shouldFinishUp = true;
    return true;
}
 
bool TestWebRequestClient::run() {
 
    int32 intervalMS = 1000;
    if (sendfreq > 0)
        intervalMS = (int32)(1000.0 / sendfreq);
    uint64 lastPost = 0;
    uint64 lastPrint = GetTimeNow();
    int32 printIntervalMS = 3000;
 
    uint32 stillWaiting, failed, timeout, errors, success, total, tooBusy, serverError;
    stillWaiting = failed = timeout = errors = success = total = tooBusy = serverError = 0;
 
    NetworkManager* manager = new NetworkManager();
 
    int8 encryption = NOENC;
 
    char* bigdata = new char[payload];
    memset(bigdata, 1, payload);
 
    std::string uriString = utils::StringFormat("/api/longtest");
 
    HTTPRequest* reqTemplate = new HTTPRequest();
    reqTemplate->createRequest(HTTP_POST, "", uriString.c_str(), bigdata, payload, false, 0);
    delete[] bigdata;
 
    HTTPRequest* req;
    HTTPReply* reply;
 
    while (shouldContinue) {
        // is it time to send the next message?
        if (!shouldFinishUp && (!lastPost || (GetTimeAgeMS(lastPost) >= intervalMS))) {
            req = new HTTPRequest(reqTemplate);
            reply = manager->makeHTTPRequest(req, "localhost", port, encryption, 5000);
            delete(req);
            if (!reply) {
                LogPrint(0, LOG_SYSTEM, 0, "WebClient %u got no reply to HTTP request", id);
                serverError++;
            }
            else {
                //LogPrint(0, LOG_SYSTEM, 0, "WebClient %u got a reply to HTTP request: %s", id, HTTP_Status[reply->type]);
                switch (reply->type) {
                case HTTP_OK:
                case HTTP_USE_LOCAL_COPY:
                    success++;
                    break;
                case HTTP_MOVED_PERMANENTLY:
                case HTTP_MALFORMED_URL:
                case HTTP_UNAUTHORIZED:
                case HTTP_ACCESS_DENIED:
                case HTTP_FILE_NOT_FOUND:
                case HTTP_INTERNAL_ERROR:
                case HTTP_NOT_IMPLEMENTED:
                case HTTP_SERVICE_NOT_AVAILABLE:
                    errors++;
                    break;
                case HTTP_GATEWAY_TIMEOUT:
                    timeout++;
                    break;
                case HTTP_SERVER_UNAVAILABLE:
                case HTTP_SERVER_NOREPLY:
                case HTTP_SERVER_MALFORMED_REPLY:
                default:
                    serverError++;
                    break;
                }
            }
            delete(reply);
            lastPost = GetTimeNow();
            total++;
        }
 
        if (shouldFinishUp) {
            LogPrint(0, LOG_SYSTEM, 0, "[%u] finished total: %u  success: %u  busy: %u  servererror: %u  failed: %u  timeout: %u  errors: %u\n",
                id, total, success, tooBusy, serverError, failed, timeout, errors);
            break;
        }
 
        if (GetTimeAgeMS(lastPrint) > printIntervalMS) {
            if (shouldFinishUp)
                LogPrint(0, LOG_SYSTEM, 0, "Client %u finishing up... (%u/%u) success: %u  busy: %u  servererror: %u  failed: %u  timeout: %u  errors: %u\n",
                    id, stillWaiting, total, success, tooBusy, serverError, failed, timeout, errors);
            else
                LogPrint(0, LOG_SYSTEM, 0, "[%u] (%u/%u) success: %u  busy: %u  servererror: %u  failed: %u  timeout: %u  errors: %u\n",
                    id, stillWaiting, total, success, tooBusy, serverError, failed, timeout, errors);
            lastPrint = GetTimeNow();
        }
 
        utils::Sleep(20);
    }
 
    delete manager;
    isRunning = false;
    return true;
}
 
THREAD_RET THREAD_FUNCTION_CALL TestWebRequestClientRun(THREAD_ARG arg) {
    if (arg == NULL) thread_ret_val(1);
    thread_ret_val((int)(((TestWebRequestClient*)arg)->run() ? 0 : 1));
}
 
 
 
 
 
 
 
 
 
 
 
 
 
TestWebSocketRequestClient::TestWebSocketRequestClient(uint32 id) {
    this->id = id;
    port = 10000;
    gatewayCount = 1;
    sendfreq = 1;
    reconnectMS = 0;
    payload = 1024;
    shouldContinue = true;
    isRunning = true;
    shouldFinishUp = false;
}
 
bool TestWebSocketRequestClient::init(uint16 port, uint32 gatewayCount, double sendfreq, int32 reconnectMS, uint32 payload, const char* longTestName) {
    uint32 myThreadID;
    this->port = port;
    if (gatewayCount)
        this->gatewayCount = gatewayCount;
    else
        this->gatewayCount = 1;
    this->sendfreq = sendfreq;
    this->reconnectMS = reconnectMS;
    this->payload = payload;
    this->longTestName = longTestName;
    return ThreadManager::CreateThread(TestWebSocketRequestClientRun, this, myThreadID, 0);
}
 
bool TestWebSocketRequestClient::finishUp() {
    LogPrint(0, LOG_SYSTEM, 0, "Client %u has been asked to finish up...", id);
    shouldFinishUp = true;
    return true;
}
 
bool TestWebSocketRequestClient::run() {
 
    int32 intervalMS = 1000;
    if (sendfreq > 0)
        intervalMS = (int32)(1000.0 / sendfreq);
    uint64 lastPost = 0;
    uint64 lastPrint = GetTimeNow();
    int32 printIntervalMS = 3000;
 
    uint32 stillWaiting, failed, timeout, errors, success, total, tooBusy, serverError;
    stillWaiting = failed = timeout = errors = success = total = tooBusy = serverError = 0;
 
    NetworkManager* manager = new NetworkManager();
 
    int8 encryption = NOENC;
    //payload = 40;
    char* bigdata = new char[payload];
    memset(bigdata, 1, payload);
 
    std::string uriString = utils::StringFormat("/longtest");
 
    WebsocketData* wsDataReply;
    JSONM* jmReply;
    WebsocketData* wsData = new WebsocketData();
    JSONM* jm = new JSONM();
//  jm->setJSON("{\"requestid\": 1}");
    jm->addData("Image1", bigdata, payload, "raw");
    //wsData->maskingKey = 2112143814;
    //wsData->setData(wsData->BINARY, true, jm->jmData, jm->getSize());
    //reqTemplate->createRequest(HTTP_POST, "", uriString.c_str(), bigdata, payload, false, 0);
    delete[] bigdata;
    //delete jm;
 
    NetworkChannel* channel = NULL;
    uint64 conID = 0;
    uint64 c = 0;
 
    while (shouldContinue) {
        // is it time to send the next message?
        if (!shouldFinishUp && (!lastPost || (GetTimeAgeMS(lastPost) >= intervalMS))) {
 
            std::string url = utils::StringFormat("http://localhost:%u/", port);
 
            if (!channel)
                channel = manager->createWebsocketConnection(url.c_str(), 0, NULL, conID, "JSONM");
            else if (!conID)
                conID = channel->createWebsocketConnection(url.c_str(), "JSONM");
 
            if (!conID) {
                LogPrint(0, 0, 1, "Couldn't connect to Websocket server on: %s", url.c_str());
                lastPost = GetTimeNow();
                continue;
            }
 
            jm->setJSON(utils::StringFormat("{\"requestid\": %llu, \"uri\": \"%s\", \"operation\": \"get\"}",
                ++c, uriString.c_str()).c_str());
            wsData->setData(wsData->BINARY, true, jm->jmData, jm->getSize());
 
            //req = new HTTPRequest(reqTemplate);
            if (!channel->sendWebsocketData(wsData, conID)) {
                channel->endConnection(conID);
                LogPrint(0, 0, 1, "Lost connection to server...");
                lastPost = GetTimeNow();
                continue;
            }
            lastPost = GetTimeNow();
            total++;
        }
 
        if (conID) {
            if (wsDataReply = channel->waitForWebsocketData(conID, 20)) {
                jmReply = new JSONM(wsDataReply->data, wsDataReply->getPayloadSize());
                if (jmReply->isValid()) {
                    success++;
                }
                else {
                    failed++;
                }
                delete jmReply;
                delete wsDataReply;
            }
        }
        else
            utils::Sleep(20);
 
        if (shouldFinishUp) {
            LogPrint(0, LOG_SYSTEM, 0, "[%u] finished total: %u  success: %u  busy: %u  servererror: %u  failed: %u  timeout: %u  errors: %u\n",
                id, total, success, tooBusy, serverError, failed, timeout, errors);
            break;
        }
 
        if (GetTimeAgeMS(lastPrint) > printIntervalMS) {
            if (shouldFinishUp)
                LogPrint(0, LOG_SYSTEM, 0, "Client %u finishing up... (%u/%u) success: %u  busy: %u  servererror: %u  failed: %u  timeout: %u  errors: %u\n",
                    id, stillWaiting, total, success, tooBusy, serverError, failed, timeout, errors);
            else
                LogPrint(0, LOG_SYSTEM, 0, "[%u] (%u/%u) success: %u  busy: %u  servererror: %u  failed: %u  timeout: %u  errors: %u\n",
                    id, stillWaiting, total, success, tooBusy, serverError, failed, timeout, errors);
            lastPrint = GetTimeNow();
        }
 
        //utils::Sleep(20);
    }
 
    delete jm;
    delete wsData;
    delete manager;
    isRunning = false;
    return true;
}
 
THREAD_RET THREAD_FUNCTION_CALL TestWebSocketRequestClientRun(THREAD_ARG arg) {
    if (arg == NULL) thread_ret_val(1);
    thread_ret_val((int)(((TestWebSocketRequestClient*)arg)->run() ? 0 : 1));
}
 

// Unit Test
//
// Self-contained: stands up a local PROTOCOL_MESSAGE listener (the "gateway"
// server) on a high localhost port, points a RequestClient at it, exchanges a
// bounded set of request/reply messages, then shuts everything down. No
// external server, no infinite waits.
//
// The server side is driven from this same thread using the listener channel's
// polling API (waitForMessage / sendMessage): for every request the client
// posts, the server receives it, stamps it SUCCESS and echoes it straight back
// on the same connection. The client matches the reply to its outstanding
// request by reference, exactly as it would against a real gateway.
 
static bool Test_RequestClient() {
    const uint16 port = 38050;
    const uint32 reqCount = 20;
    const uint32 payload = 1024;
 
    unittest::progress(0, "starting server");
 
    NetworkManager* server = new NetworkManager();
    NetworkChannel* listen = server->createListener(port, NOENC, PROTOCOL_MESSAGE, true, 0, true, 0, NULL);
    if (!listen) {
        unittest::fail("could not start message listener on localhost:%u", port);
        delete server;
        return false;
    }
 
    unittest::progress(15, "connecting client");
 
    RequestClient* client = new RequestClient();
    if (!client->addGateway(1, "localhost", port, NOENC)) {
        unittest::fail("could not add gateway localhost:%u to client", port);
        delete client;
        delete server;
        return false;
    }
 
    // The client's addGateway sends a "ClientConnect" greeting on connect; the
    // server must drain it (it carries no request reference).
    uint64 recvid = 0;
    DataMessage* greeting = listen->waitForMessage(recvid, 3000);
    if (!greeting) {
        unittest::fail("server never received the client connect greeting");
        delete client;
        delete server;
        return false;
    }
    unittest::detail("received client greeting URI='%s' on conid %llu",
        greeting->getString("URI") ? greeting->getString("URI") : "", recvid);
    delete greeting;
 
    // The client only forwards requests once its connect event has landed
    // (lastConTime set); wait for that so the first postRequest is not dropped.
    if (!client->waitForConnection(3000)) {
        unittest::fail("client did not register as connected to localhost:%u", port);
        delete client;
        delete server;
        return false;
    }
 
    unittest::progress(40, "exchanging requests");
 
    char* dat = new char[payload];
    memset(dat, 7, payload);
 
    uint32 ok = 0;
    uint64 t0 = GetTimeNow();
    for (uint32 n = 0; n < reqCount; n++) {
        // Build and post a request; postRequest takes ownership of the message
        // and stamps it with a unique reference.
        DataMessage* req = new DataMessage();
        req->setString("URI", "unittest");
        req->setData("BigData", dat, payload);
        RequestReply* reply = client->postRequest(req);
        if (!reply) {
            unittest::fail("postRequest returned NULL on iteration %u", n);
            break;
        }
 
        // Server side: receive the forwarded request and echo it back as a
        // successful reply on the same connection (reference is preserved).
        uint64 conid = 0;
        DataMessage* got = listen->waitForMessage(conid, 3000);
        if (!got) {
            unittest::fail("server did not receive request %u", n);
            break;
        }
        got->setStatus(SUCCESS);
        if (!listen->sendMessage(got, conid)) {
            unittest::fail("server failed to send reply for request %u", n);
            delete got;
            break;
        }
        // sendMessage does not take ownership here.
        delete got;
 
        // Client side: wait for the matched reply.
        DataMessage* replyMsg = reply->waitForMessage(3000, true);
        if (!replyMsg) {
            unittest::fail("client got no reply for request %u", n);
            client->finishRequest(reply);
            break;
        }
        if (replyMsg->getStatus() != SUCCESS) {
            unittest::fail("reply %u had unexpected status %u", n, replyMsg->getStatus());
            delete replyMsg;
            client->finishRequest(reply);
            break;
        }
        delete replyMsg;
        client->finishRequest(reply);
        ok++;
        unittest::detail("request %u round-tripped ok", n);
        unittest::progress(40 + (n * 50) / reqCount, "exchanging requests");
    }
    double us = (double)(GetTimeNow() - t0);
 
    delete[] dat;
 
    unittest::progress(95, "shutting down");
    delete client;
    delete server;
 
    if (ok != reqCount)
        return false;
 
    if (us > 0)
        unittest::metric("request_throughput", (double)reqCount / us * 1e6, "req/s", true);
    unittest::metric("avg_request_latency", us / reqCount, "us", false);
 
    unittest::progress(100, "done");
    return true;
}
 
void Register_RequestClient_Tests() {
    UnitTestRunner::instance().registerTest(
        "requestclient", Test_RequestClient,
        "Request client/gateway round-trip over localhost", "network");
}
 
}