ostinato/rpc/pbrpcchannel.cpp

#include "pbrpcchannel.h"

PbRpcChannel::PbRpcChannel(QHostAddress ip, quint16 port)
{
	isPending = false;
	pendingMethodId = -1;	// don't care as long as isPending is false

	controller = NULL;
	done = NULL;
	response = NULL;

	mServerAddress = ip;
	mServerPort = port;
	mpSocket = new QTcpSocket(this);

	// FIXME: Not quite sure why this ain't working!
	// QMetaObject::connectSlotsByName(this);

	connect(mpSocket, SIGNAL(connected()),
		this, SLOT(on_mpSocket_connected()));
	connect(mpSocket, SIGNAL(disconnected()),
		this, SLOT(on_mpSocket_disconnected()));
	connect(mpSocket, SIGNAL(stateChanged(QAbstractSocket::SocketState)),
		this, SLOT(on_mpSocket_stateChanged(QAbstractSocket::SocketState)));
	connect(mpSocket, SIGNAL(error(QAbstractSocket::SocketError)),
		this, SLOT(on_mpSocket_error(QAbstractSocket::SocketError)));

	connect(mpSocket, SIGNAL(readyRead()),
		this, SLOT(on_mpSocket_readyRead()));

}

PbRpcChannel::~PbRpcChannel()
{
	delete mpSocket;
}

void PbRpcChannel::establish()
{
	qDebug("In %s", __FUNCTION__);

	mpSocket->connectToHost(mServerAddress, mServerPort);
}

void PbRpcChannel::establish(QHostAddress ip, quint16 port)
{
	mServerAddress = ip;
	mServerPort = port;
	establish();
}

void PbRpcChannel::tearDown()
{
	qDebug("In %s", __FUNCTION__);

	mpSocket->disconnectFromHost();
}

void PbRpcChannel::CallMethod(
	const ::google::protobuf::MethodDescriptor *method,
	::google::protobuf::RpcController *controller,
	const ::google::protobuf::Message *req,
	::google::protobuf::Message *response,
	::google::protobuf::Closure* done)
{
	char	msg[MSGBUF_SIZE];
	int		len;
	bool	ret;
  
	if (isPending)
	{
		RpcCall call;
		qDebug("RpcChannel: queueing method %d since %d is pending", 
			method->index(), pendingMethodId);

		call.method = method;
		call.controller = controller;
		call.request = req;
		call.response = response;
		call.done = done;

		pendingCallList.append(call);

		Q_ASSERT(pendingCallList.size() < 100);

		return;
	}

	if (!req->IsInitialized())
	{
		qWarning("RpcChannel: missing required fields in request");
		qDebug(req->InitializationErrorString().c_str());

		qFatal("exiting");

		controller->SetFailed("Required fields missing");
		done->Run();
		return;
	}

	pendingMethodId = method->index();
	this->controller=controller;
	this->done=done;
	this->response=response;
	isPending = true;

	ret = req->SerializeToArray((void*) (&msg[PB_HDR_SIZE]), sizeof(msg));
	Q_ASSERT(ret == true);

	len = req->ByteSize();
	*((quint16*)(&msg[0])) = HTONS(PB_MSG_TYPE_REQUEST); // type
	*((quint16*)(&msg[2])) = HTONS(method->index()); // method id
	*((quint32*)(&msg[4])) = HTONL(len); // len

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

	mpSocket->write(msg, PB_HDR_SIZE + len);
}

void PbRpcChannel::on_mpSocket_readyRead()
{
	char	msg[MSGBUF_SIZE];
	char	*p = (char*)&msg;
	int		msgLen;
	static bool parsing = false;
	static quint16	type, method;
	static quint32	len;

	//qDebug("%s: bytesAvail = %d", __FUNCTION__, mpSocket->bytesAvailable());

	if (!parsing)
	{
		// Do we have an entire header? If not, we'll wait ...
		if (mpSocket->bytesAvailable() < PB_HDR_SIZE)
		{
			qDebug("client: not enough data available for a complete header");
			return;
		}

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

		Q_ASSERT(msgLen == PB_HDR_SIZE);

		type = NTOHS(GET16(p+0));
		method = NTOHS(GET16(p+2));
		len = NTOHL(GET32(p+4));

		//BUFDUMP(msg, PB_HDR_SIZE);
		//qDebug("type = %hu, method = %hu, len = %u", type, method, len);

		parsing = true;
	}

	switch (type)
	{
		case PB_MSG_TYPE_BINBLOB:
		{
			static quint32 cumLen = 0;
			QIODevice *blob;

			blob = static_cast<PbRpcController*>(controller)->binaryBlob();
			Q_ASSERT(blob != NULL);

			while ((cumLen < len) && mpSocket->bytesAvailable())
			{
				int l;

				l = mpSocket->read(msg, sizeof(msg));
				blob->write(msg, l);
				cumLen += l;
			}

			qDebug("%s: bin blob rcvd %d/%d", __PRETTY_FUNCTION__, cumLen, len);

			if (cumLen < len)
				return;

			cumLen = 0;

			if (!isPending)
			{
				qDebug("not waiting for response");
				goto _error_exit;
			}

			if (pendingMethodId != method)
			{
				qDebug("invalid method id %d (expected = %d)", method, 
					pendingMethodId);
				goto _error_exit;
			}

			break;
		}

		case PB_MSG_TYPE_RESPONSE:
			// Wait till we have the entire message
			if (mpSocket->bytesAvailable() < len)
			{
				qDebug("client: not enough data available for a complete msg");
				return;
			}
			
			msgLen = mpSocket->read(msg, sizeof(msg));

			Q_ASSERT((unsigned) msgLen == len);

			//qDebug("client(%s) rcvd %d bytes", __FUNCTION__, msgLen);
			//BUFDUMP(msg, msgLen);

			if (!isPending)
			{
				qDebug("not waiting for response");
				goto _error_exit;
			}

			if (pendingMethodId != method)
			{
				qDebug("invalid method id %d (expected = %d)", method, 
					pendingMethodId);
				goto _error_exit;
			}

			response->ParseFromArray((void*) msg, len);

			// Avoid printing stats
			if (method != 12)
			{
				qDebug("client(%s): Parsed as %s", __FUNCTION__,
					response->DebugString().c_str());
			}

			if (!response->IsInitialized())
			{
				qWarning("RpcChannel: missing required fields in response");
				qDebug(response->InitializationErrorString().c_str());

				controller->SetFailed("Required fields missing");
			}
			break;

		default:
			qFatal("%s: unexpected type %d", __PRETTY_FUNCTION__, type);
			goto _error_exit;
				
	}

	pendingMethodId = -1;
	controller = NULL;
	response = NULL;
	isPending = false;
	parsing = false;

	done->Run();

	if (pendingCallList.size())
	{
		RpcCall call = pendingCallList.takeFirst();
		CallMethod(call.method, call.controller, call.request, call.response,
				call.done);
	}

	return;

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

void PbRpcChannel::on_mpSocket_stateChanged(
	QAbstractSocket::SocketState socketState)
{
	qDebug("In %s", __FUNCTION__);
	emit stateChanged(socketState);
}

void PbRpcChannel::on_mpSocket_connected()
{
	qDebug("In %s", __FUNCTION__);
	emit connected(); 
}

void PbRpcChannel::on_mpSocket_disconnected()
{
	qDebug("In %s", __FUNCTION__);
	emit disconnected();
}

void PbRpcChannel::on_mpSocket_error(QAbstractSocket::SocketError socketError)
{
	qDebug("In %s", __FUNCTION__);
	emit error(socketError);
}