RocServer.cpp

Go to the documentation of this file.

#include <RocServer.h>

#include <iostream>
#include <chrono>

using namespace std;

using std::chrono::high_resolution_clock;
using std::chrono::milliseconds;

#define MSG_REQUEST        (2)
#define MSG_REPLY          (3)
#define MSG_HEARTBEAT      (4)
#define MSG_DISCONNECT     (5)

#define MAX_LIVENESS       (3)

#define HB_INTERVALL       (1000)

RocServer::RocServer(zmq::context_t & _ctx,
                     const std::string & _brokerEndpoint,
                     const std::string & _serviceName):
            ctx(_ctx),
            brokerEndpoint(_brokerEndpoint),
            serviceName(_serviceName),
            timeout(HB_INTERVALL),
            keepGoing(true),
            liveness(MAX_LIVENESS) {
    serverThread = thread(&RocServer::serve, this);
}

RocServer::~RocServer() {
    cout << "stopping rocServer" << endl;
    keepGoing = false;
    if(serverThread.joinable()) {
        serverThread.join();
    }
    cout << "stopped rocServer" << endl;
}

void RocServer::startup() {

}

void RocServer::teardown() {

}

Json::Value RocServer::call(const std::string & name, const Json::Value & arguments) {
    (void) name;
    (void) arguments;
    throw RocNotImplementedError();
}

Json::Value RocServer::getattr(const std::string & name) {
    (void) name;
    throw RocIsCallable();
}

void RocServer::setattr(const std::string & name, const Json::Value & value) {
    (void) name;
    (void) value;
    throw RocNotImplementedError();
}

Json::Value RocServer::rocCall(const std::string & name, const Json::Value & arguments) {
    if(name == "getattr") {
        Json::Value ret;
        try {
            Json::Value attr = getattr(arguments[0].asString());
            ret["type"] = "value";
            ret["value"] = attr;
        } catch(RocIsCallable) {
            ret["type"] = "callable";
        }
        return ret;
    } else if (name == "setattr") {
        setattr(arguments[0].asString(), arguments[1]);
        /* if setattr throws, the next part is not reached. */
        Json::Value ret;
        ret["type"] = "ok";
        return ret;
    }
    throw RocNotImplementedError();
}

void RocServer::onRequest(std::list<zmq::message_t*>* envelope, std::list<zmq::message_t*>* data) {
    /* roc request should be in the first data element */
    auto & req = *(data->begin());
    Json::Value root;
    Json::Reader reader;
    bool parseOk = reader.parse((const char *)req->data(), ((const char *)req->data()) + req->size(), root, false);
    for(auto & arg: *data) {
        /* cout << (char*)(arg->data()) << ", "; */
        delete arg;
    }
    delete data;
    if(!parseOk) {
        replyError(envelope, ROC_JSON_PARSE_ERROR);
        return;
    }
    if(root["jsonrpc"].asString() != "2.0") {
        replyError(envelope, ROC_JSON_INVALID_REQUEST);
        return;
    }
    std::string id = root["id"].asString();
    std::string method = root["method"].asString();

    try {
        Json::Value ret;
        if(method.compare(0, 5, "_roc.") == 0) {
            ret = rocCall(method.substr(5), root["params"]);
        } else {
            ret = call(method, root["params"]);
        }
        /* cout << "return:" << ret << endl; */
        replyValue(envelope, ret, id);
        return;
    } catch (RocNotImplementedError) {
        replyError(envelope, ROC_JSON_METHOD_NOT_FOUND, id);
        return;
    } catch (exception &exc) {
        cout << "exception:" << exc.what() << endl;
        replyError(envelope, ROC_JSON_INTERNAL_ERROR, id);
        return;
    }
}

void RocServer::serve() {
    int linger = 100;
    sock = new zmq::socket_t(ctx, ZMQ_DEALER);
    sock->setsockopt(ZMQ_LINGER, &linger, sizeof(linger));
    sock->connect(brokerEndpoint.c_str());
    
    zmq::socket_t controlSock(ctx, ZMQ_SUB);
    controlSock.connect("inproc://control");
    controlSock.setsockopt(ZMQ_SUBSCRIBE, "control", 7);
    
    /* send ready */
    send(1, serviceName);
    
    startup();
    
    auto next_heartbeat = chrono::high_resolution_clock::now() + chrono::milliseconds(HB_INTERVALL);
    
    zmq::pollitem_t items [] = {
        { (void*)*sock, 0, ZMQ_POLLIN, 0},
        { (void*)controlSock, 0, ZMQ_POLLIN, 0}
    };
    
    int64_t more = 0;
    size_t more_size = sizeof(more);
    
    while(keepGoing) {
        zmq::poll(items, 2, timeout);
        if(items[0].revents & ZMQ_POLLIN) {
            /* receive data here */
            zmq::message_t empty;
            sock->recv(&empty);
            zmq::message_t header;
            sock->recv(&header);
            if(strncmp((char*)header.data(), "MDPW01", 6) == 0)  {
                liveness = MAX_LIVENESS;
                zmq::message_t command;
                sock->recv(&command);
                switch(((char*)command.data())[0]) {
                    case MSG_REQUEST: /* request */
                        /* cout << "request received!" << endl; */
                        _onRequest();
                        break;
                    case MSG_HEARTBEAT: /* heartbeat */
                        break;
                    case MSG_DISCONNECT: /* disconnect */
                        cout << "disconnect received!" << endl;
                        break;
                    default:
                        cout << "message " << (int)((char*)command.data())[0] << " received" << endl;
                }
            }
            sock->getsockopt(ZMQ_RCVMORE, &more, &more_size);
            while(more == 1) {
                cout << "recvmore" << endl;
                zmq::message_t dummy;
                sock->recv(&dummy);
                sock->getsockopt(ZMQ_RCVMORE, &more, &more_size);
            }
        }
        if(items[1].revents & ZMQ_POLLIN) {
            zmq::message_t header;
            controlSock.recv(&header);
            zmq::message_t command;
            controlSock.recv(&command);
            if(strncmp((char*)command.data(), "stop", 4) == 0) {
                cout << "stoping rocServer" << endl;
                keepGoing = false;
            }
        }
        if(next_heartbeat < chrono::high_resolution_clock::now()) {
            --liveness;
            send(MSG_HEARTBEAT);
            next_heartbeat = chrono::high_resolution_clock::now() + chrono::milliseconds(HB_INTERVALL);
        }
    }
    teardown();
    cout << "rocServer loop end" << endl;
    delete sock;
}

void RocServer::send(char _command, const std::string & _option) {
    zmq::message_t empty(0);
    sock->send(empty, ZMQ_SNDMORE);
    zmq::message_t header(6);
    memcpy(header.data(), "MDPW01", 6);
    sock->send(header, ZMQ_SNDMORE);
    zmq::message_t command(1);
    ((char*)command.data())[0] = _command;
    if(_option.length() == 0) {
        sock->send(command);
    } else {
        sock->send(command, ZMQ_SNDMORE);
        zmq::message_t data(_option.length());
        memcpy(data.data(), _option.c_str(), _option.length());
        sock->send(data);
    }
}

void RocServer::_onRequest() {
    auto requestEnvelope = new list<zmq::message_t *>;
    auto requestData = new list<zmq::message_t *>;
    
    int64_t more = 0;
    size_t more_size = sizeof(more);
    sock->getsockopt(ZMQ_RCVMORE, &more, &more_size);
    bool inEnvelope = true;
    while(more == 1) {
        auto msg = new zmq::message_t;
        sock->recv(msg);
        if(inEnvelope) {
            if(msg->size() > 0) {
                requestEnvelope->push_back(msg);
            } else {
                delete msg;
                inEnvelope = false;
            }
        } else {
            requestData->push_back(msg);
        }
        sock->getsockopt(ZMQ_RCVMORE, &more, &more_size);
    }
    onRequest(requestEnvelope, requestData);
}

void RocServer::reply(std::list<zmq::message_t*>* envelope, std::list<zmq::message_t*>* data) {
    zmq::message_t empty(0);
    sock->send(empty, ZMQ_SNDMORE);
    zmq::message_t header(6);
    memcpy(header.data(), "MDPW01", 6);
    sock->send(header, ZMQ_SNDMORE);
    zmq::message_t command(1);
    ((char*)command.data())[0] = MSG_REPLY;
    sock->send(command, ZMQ_SNDMORE);
    while(!envelope->empty()) {
        zmq::message_t * msg = envelope->front();
        envelope->pop_front();
        sock->send(*msg, ZMQ_SNDMORE);
        delete msg;
    }
    delete envelope;
    zmq::message_t delimiter(0);
    if(data == nullptr || data->empty()) {
        sock->send(delimiter);
    } else {
        sock->send(delimiter, ZMQ_SNDMORE);
        while(!data->empty()) {
            zmq::message_t * msg = data->front();
            data->pop_front();
            if(data->empty()) {
                sock->send(*msg);
            } else {
                sock->send(*msg, ZMQ_SNDMORE);
            }
            delete msg;
        }
    }
    delete data;
}

void RocServer::replyValue(std::list<zmq::message_t*>* envelope, const Json::Value & value, const std::string & id) {
    Json::Value root;
    root["jsonrpc"] = "2.0";
    root["result"] = value;
    root["id"] = id;

    std::string encodedResponse = root.toStyledString();
    auto data = new list<zmq::message_t *>;
    auto msg = new zmq::message_t(encodedResponse.length());
    memcpy(msg->data(), encodedResponse.c_str(), encodedResponse.length());
    data->push_back(msg);
    reply(envelope, data);
}

void RocServer::replyError(std::list<zmq::message_t*>* envelope, int errorNo, const std::string & id) {
    Json::Value root;
    root["jsonrpc"] = "2.0";
    Json::Value error;
    error["code"] = errorNo;
    switch (errorNo) {
        case ROC_JSON_PARSE_ERROR:
            error["message"] = "Parse error";
            break;
        case ROC_JSON_INVALID_REQUEST:
            error["message"] = "Invalid Request";
            break;
        case ROC_JSON_METHOD_NOT_FOUND:
            error["message"] = "Method not found";
            break;
        case ROC_JSON_INVALID_PARAMS:
            error["message"] = "Invalid params";
            break;
        case ROC_JSON_INTERNAL_ERROR:
            error["message"] = "Internal error";
            break;
    }
    error["data"] = "";
    root["error"] = error;
    if(id.length() > 0) {
        root["id"] = id;
    } else {
        root["id"] = Json::Value::null;
    }
    std::string encodedResponse = root.toStyledString();
    auto data = new list<zmq::message_t *>;
    auto msg = new zmq::message_t(encodedResponse.length());
    memcpy(msg->data(), encodedResponse.c_str(), encodedResponse.length());
    data->push_back(msg);
    reply(envelope, data);
}