ostinato/rpc/rpcserver.cpp

//#include "pbhelper.h"
#include "rpcserver.h"

RpcServer::RpcServer()
{
    server = NULL;
    clientSock = NULL;

    service = NULL; 

    isPending = false;
    pendingMethodId = -1; // don't care as long as isPending is false
    controller_.Reset();
}

RpcServer::~RpcServer()
{ 
    if (server) 
        delete server; 
}

bool RpcServer::registerService(::google::protobuf::Service *service, 
    quint16 tcpPortNum)
{
    this->service = service;

    server = new QTcpServer();
    connect(server, SIGNAL(newConnection()), this, SLOT(when_newConnection()));
    if (!server->listen(QHostAddress::Any, tcpPortNum)) 
    {
        qDebug("Unable to start the server: %s", 
                server->errorString().toAscii().constData());
        errorString_ = QString("Error starting Ostinato server: %1").arg(
                server->errorString());
        return false;
    }

    qDebug("The server is running on %s: %d", 
            server->serverAddress().toString().toAscii().constData(),
            server->serverPort());
    errorString_ = QString();
    return true;
}

QString RpcServer::errorString()
{
    return errorString_;
}

void RpcServer::done(::google::protobuf::Message *request, 
        ::google::protobuf::Message *response)
{
    QIODevice *blob;
    char msg[MSGBUF_SIZE];
    int len;

    //qDebug("In RpcServer::done");

    if (controller_.Failed())
    {
        qDebug("rpc failed");
        goto _exit;
    }

    blob = controller_.binaryBlob();
    if (blob)
    {
        len = blob->size();
        qDebug("is binary blob of len %d", len);

        *((quint16*)(&msg[0])) = HTONS(PB_MSG_TYPE_BINBLOB); // type
        *((quint16*)(&msg[2])) = HTONS(pendingMethodId); // method
        (*(quint32*)(&msg[4])) = HTONL(len); // len

        clientSock->write(msg, PB_HDR_SIZE);

        blob->seek(0);
        while (!blob->atEnd())
        {    
            int l;

            len = blob->read(msg, sizeof(msg));
            l = clientSock->write(msg, len);
            Q_ASSERT(l == len);
        }

        goto _exit;
    }

    if (!response->IsInitialized())
    {
        qWarning("response missing required fields!!");
        qDebug(response->InitializationErrorString().c_str());
        qFatal("exiting");
        goto _exit;
    }

    response->SerializeToArray((void*) &msg[PB_HDR_SIZE], sizeof(msg));

    len = response->ByteSize();

    *((quint16*)(&msg[0])) = HTONS(PB_MSG_TYPE_RESPONSE); // type
    *((quint16*)(&msg[2])) = HTONS(pendingMethodId); // method
    *((quint32*)(&msg[4])) = HTONL(len); // len

    // Avoid printing stats since it happens once every couple of seconds
    if (pendingMethodId != 12)
    {
        qDebug("Server(%s): sending %d bytes to client encoding <%s>", 
            __FUNCTION__, len + PB_HDR_SIZE, response->DebugString().c_str());
        //BUFDUMP(msg, len + 8);
    }

    clientSock->write(msg, PB_HDR_SIZE + len);

_exit:
    delete request;
    delete response;
    isPending = false;
}

void RpcServer::when_newConnection()
{
    if (clientSock)
    {
        QTcpSocket    *sock;

        qDebug("already connected, no new connections will be accepted");

        // Accept and close connection
        //! \todo (MED) Send reason msg to client
        sock = server->nextPendingConnection();
        sock->disconnectFromHost();
        sock->deleteLater();
        goto _exit;
    }

    clientSock = server->nextPendingConnection();
    qDebug("accepting new connection from %s: %d", 
            clientSock->peerAddress().toString().toAscii().constData(),
            clientSock->peerPort());

    connect(clientSock, SIGNAL(readyRead()), 
        this, SLOT(when_dataAvail()));
    connect(clientSock, SIGNAL(disconnected()), 
        this, SLOT(when_disconnected()));
    connect(clientSock, SIGNAL(error(QAbstractSocket::SocketError)), 
        this, SLOT(when_error(QAbstractSocket::SocketError)));

_exit:
    return;
}

void RpcServer::when_disconnected()
{
    qDebug("connection closed from %s: %d",
            clientSock->peerAddress().toString().toAscii().constData(),
            clientSock->peerPort());

    clientSock->deleteLater();
    clientSock = NULL;
}

void RpcServer::when_error(QAbstractSocket::SocketError socketError)
{
    qDebug("%s (%d)", clientSock->errorString().toAscii().constData(), 
            socketError);
}

void RpcServer::when_dataAvail()
{
    char    msg[MSGBUF_SIZE];
    int        msgLen;
    static bool parsing = false;
    static quint16    type, method;
    static quint32 len;
    const ::google::protobuf::MethodDescriptor    *methodDesc;
    ::google::protobuf::Message    *req, *resp;

    if (!parsing)
    {
        if (clientSock->bytesAvailable() < PB_HDR_SIZE)
            return;

        msgLen = clientSock->read(msg, PB_HDR_SIZE);

        Q_ASSERT(msgLen == PB_HDR_SIZE);

        type = NTOHS(GET16(&msg[0]));
        method = NTOHS(GET16(&msg[2]));
        len = NTOHL(GET32(&msg[4]));
        //qDebug("type = %d, method = %d, len = %d", type, method, len);

        parsing = true;
    }

    if (clientSock->bytesAvailable() < len)
        return;

    msgLen = clientSock->read(msg, sizeof(msg));
    Q_ASSERT((unsigned) msgLen == len);

    if (type != PB_MSG_TYPE_REQUEST)
    {
        qDebug("server(%s): unexpected msg type %d (expected %d)", __FUNCTION__,
            type, PB_MSG_TYPE_REQUEST);
        goto _error_exit;
    }

    methodDesc = service->GetDescriptor()->method(method);
    if (!methodDesc)
    {
        qDebug("server(%s): invalid method id %d", __FUNCTION__, method);
        goto _error_exit; //! \todo Return Error to client
    }

    if (isPending)
    {
        qDebug("server(%s): rpc pending, try again", __FUNCTION__);
        goto _error_exit; //! \todo Return Error to client
    }

    pendingMethodId = method;
    isPending = true;

    req = service->GetRequestPrototype(methodDesc).New();
    resp = service->GetResponsePrototype(methodDesc).New();

    req->ParseFromArray((void*)msg, len);
    if (!req->IsInitialized())
    {
        qWarning("Missing required fields in request");
        qDebug(req->InitializationErrorString().c_str());
        qFatal("exiting");
        delete req;
        delete resp;

        goto _error_exit;
    }
    //qDebug("Server(%s): successfully parsed as <%s>", __FUNCTION__, 
        //resp->DebugString().c_str());

    controller_.Reset();

    //qDebug("before service->callmethod()");

    service->CallMethod(methodDesc, &controller_, req, resp,
        NewCallback(this, &RpcServer::done, req, resp));

    parsing = false;

    return;

_error_exit:
    parsing = false;
    qDebug("server(%s): discarding msg from client", __FUNCTION__);
    return;
}