//
// Copyright 2005 The Android Open Source Project
//
// Message stream abstraction.
//
#include "MessageStream.h"
#include "LogBundle.h"
#include "utils/Log.h"
#include <string.h>
#include <assert.h>
using namespace android;
/*
* ===========================================================================
* Message
* ===========================================================================
*/
/*
* Send a blob of raw data.
*/
void Message::setRaw(const unsigned char* data, int len, Cleanup cleanup)
{
reset();
mData = const_cast<unsigned char*>(data);
mLength = len;
mCleanup = cleanup;
mType = kTypeRaw;
}
/*
* Send a "name=value" config pair.
*/
void Message::setConfig(const char* name, const char* value)
{
reset();
assert(name != NULL && value != NULL);
int nlen = strlen(name) +1;
int vlen = strlen(value) +1;
mData = new unsigned char[nlen+vlen];
mCleanup = kCleanupDelete;
mLength = nlen + vlen;
mType = kTypeConfig;
memcpy(mData, name, nlen);
memcpy(mData + nlen, value, vlen);
}
/*
* Try to return the contents of the message as if it were a name/value pair.
*/
bool Message::getConfig(const char** pName, const char** pValue)
{
if (mLength < 2)
return false;
assert(mData != NULL);
*pName = (const char*) mData;
*pValue = (const char*) (mData + strlen((char*)mData) +1);
return true;
}
/*
* Send a command/arg pair.
*/
void Message::setCommand(int cmd, int arg)
{
reset();
mData = new unsigned char[sizeof(int) * 2];
mCleanup = kCleanupDelete;
mLength = sizeof(int) * 2;
mType = kTypeCommand;
int* pInt = (int*) mData;
pInt[0] = cmd;
pInt[1] = arg;
}
/*
* Send a command with 3 args instead of just one.
*/
void Message::setCommandExt(int cmd, int arg0, int arg1, int arg2)
{
reset();
mData = new unsigned char[sizeof(int) * 4];
mCleanup = kCleanupDelete;
mLength = sizeof(int) * 4;
mType = kTypeCommandExt;
int* pInt = (int*) mData;
pInt[0] = cmd;
pInt[1] = arg0;
pInt[2] = arg1;
pInt[3] = arg2;
}
/*
* Try to return the contents of the message as if it were a "command".
*/
bool Message::getCommand(int* pCmd, int* pArg)
{
if (mLength != sizeof(int) * 2) {
LOG(LOG_WARN, "", "type is %d, len is %d\n", mType, mLength);
return false;
}
assert(mData != NULL);
const int* pInt = (const int*) mData;
*pCmd = pInt[0];
*pArg = pInt[1];
return true;
}
/*
* Serialize a log message.
*
* DO NOT call LOG() from here.
*/
void Message::setLogBundle(const android_LogBundle* pBundle)
{
reset();
/* get string lengths; we add one here to include the '\0' */
int tagLen, msgLen;
tagLen = strlen(pBundle->tag) + 1;
size_t i;
msgLen = 0;
for (i=0; i<pBundle->msgCount; i++) msgLen += pBundle->msgVec[i].iov_len;
msgLen += 1;
/* set up the structure */
mCleanup = kCleanupDelete;
mLength = sizeof(pBundle->when) +
sizeof(pBundle->priority) +
sizeof(pBundle->pid) +
tagLen +
msgLen;
mData = new unsigned char[mLength];
mType = kTypeLogBundle;
unsigned char* pCur = mData;
/* copy the stuff over */
*((time_t*)pCur) = pBundle->when;
pCur += sizeof(pBundle->when);
*((android_LogPriority*)pCur) = pBundle->priority;
pCur += sizeof(pBundle->priority);
*((pid_t*)pCur) = pBundle->pid;
pCur += sizeof(pBundle->pid);
memcpy(pCur, pBundle->tag, tagLen);
pCur += tagLen;
for (i=0; i<pBundle->msgCount; i++) {
memcpy(pCur, pBundle->msgVec[i].iov_base, pBundle->msgVec[i].iov_len);
pCur += pBundle->msgVec[i].iov_len;
}
*pCur++ = 0;
assert(pCur - mData == mLength);
}
/*
* Extract the components of a log bundle.
*
* We're just returning points inside the message buffer, so the caller
* will need to copy them out before the next reset().
*/
bool Message::getLogBundle(android_LogBundle* pBundle)
{
if (mLength < (int)(sizeof(time_t) + sizeof(int)*2 + 4)) {
LOG(LOG_WARN, "", "type is %d, len is %d, too small\n",
mType, mLength);
return false;
}
assert(mData != NULL);
unsigned char* pCur = mData;
pBundle->when = *((time_t*) pCur);
pCur += sizeof(pBundle->when);
pBundle->priority = *((android_LogPriority*) pCur);
pCur += sizeof(pBundle->priority);
pBundle->pid = *((pid_t*) pCur);
pCur += sizeof(pBundle->pid);
pBundle->tag = (const char*) pCur;
pCur += strlen((const char*) pCur) +1;
mVec.iov_base = (char*) pCur;
mVec.iov_len = strlen((const char*) pCur);
pBundle->msgVec = &mVec;
pBundle->msgCount = 1;
pCur += mVec.iov_len +1;
if (pCur - mData != mLength) {
LOG(LOG_WARN, "", "log bundle rcvd %d, used %d\n", mLength,
(int) (pCur - mData));
return false;
}
return true;
}
/*
* Read the next event from the pipe.
*
* This is not expected to work well when multiple threads are reading.
*/
bool Message::read(Pipe* pPipe, bool wait)
{
if (pPipe == NULL)
return false;
assert(pPipe->isCreated());
if (!wait) {
if (!pPipe->readReady())
return false;
}
reset();
unsigned char header[4];
if (pPipe->read(header, 4) != 4)
return false;
mType = (MessageType) header[2];
mLength = header[0] | header[1] << 8;
mLength -= 2; // we already read two of them in the header
if (mLength > 0) {
int actual;
mData = new unsigned char[mLength];
if (mData == NULL) {
LOG(LOG_ERROR, "", "alloc failed\n");
return false;
}
mCleanup = kCleanupDelete;
actual = pPipe->read(mData, mLength);
if (actual != mLength) {
LOG(LOG_WARN, "", "failed reading message body (%d of %d bytes)\n",
actual, mLength);
return false;
}
}
return true;
}
/*
* Write this event to a pipe.
*
* It would be easiest to write the header and message body with two
* separate calls, but that will occasionally fail on multithreaded
* systems when the writes are interleaved. We have to allocate a
* temporary buffer, copy the data, and write it all at once. This
* would be easier with writev(), but we can't rely on having that.
*
* DO NOT call LOG() from here, as we could be in the process of sending
* a log message.
*/
bool Message::write(Pipe* pPipe) const
{
char tmpBuf[128];
char* writeBuf = tmpBuf;
bool result = false;
int kHeaderLen = 4;
if (pPipe == NULL)
return false;
assert(pPipe->isCreated());
if (mData == NULL || mLength < 0)
return false;
/* if it doesn't fit in stack buffer, allocate space */
if (mLength + kHeaderLen > (int) sizeof(tmpBuf)) {
writeBuf = new char[mLength + kHeaderLen];
if (writeBuf == NULL)
goto bail;
}
/*
* The current value of "mLength" does not include the 4-byte header.
* Two of the 4 header bytes are included in the length we output
* (the type byte and the pad byte), so we adjust mLength.
*/
writeBuf[0] = (unsigned char) (mLength + kHeaderLen -2);
writeBuf[1] = (unsigned char) ((mLength + kHeaderLen -2) >> 8);
writeBuf[2] = (unsigned char) mType;
writeBuf[3] = 0;
if (mLength > 0)
memcpy(writeBuf + kHeaderLen, mData, mLength);
int actual;
actual = pPipe->write(writeBuf, mLength + kHeaderLen);
if (actual != mLength + kHeaderLen) {
fprintf(stderr,
"Message::write failed writing message body (%d of %d bytes)\n",
actual, mLength + kHeaderLen);
goto bail;
}
result = true;
bail:
if (writeBuf != tmpBuf)
delete[] writeBuf;
return result;
}
/*
* ===========================================================================
* MessageStream
* ===========================================================================
*/
/*
* Get ready to go.
*/
bool MessageStream::init(Pipe* readPipe, Pipe* writePipe, bool initiateHello)
{
assert(mReadPipe == NULL && mWritePipe == NULL); // only once
/*
* Swap "hello" messages.
*
* In a more robust implementation, this would include version numbers
* and capability flags.
*/
if (initiateHello) {
long data = kHelloMsg;
Message msg;
/* send hello */
msg.setRaw((unsigned char*) &data, sizeof(data),
Message::kCleanupNoDelete);
if (!msg.write(writePipe)) {
LOG(LOG_WARN, "", "hello write failed in stream init\n");
return false;
}
LOG(LOG_DEBUG, "", "waiting for peer to ack my hello\n");
/* wait for the ack */
if (!msg.read(readPipe, true)) {
LOG(LOG_WARN, "", "hello ack read failed in stream init\n");
return false;
}
const long* pAck;
pAck = (const long*) msg.getData();
if (pAck == NULL || *pAck != kHelloAckMsg) {
LOG(LOG_WARN, "", "hello ack was bad\n");
return false;
}
} else {
long data = kHelloAckMsg;
Message msg;
LOG(LOG_DEBUG, "", "waiting for hello from peer\n");
/* wait for the hello */
if (!msg.read(readPipe, true)) {
LOG(LOG_WARN, "", "hello read failed in stream init\n");
return false;
}
const long* pAck;
pAck = (const long*) msg.getData();
if (pAck == NULL || *pAck != kHelloMsg) {
LOG(LOG_WARN, "", "hello was bad\n");
return false;
}
/* send hello ack */
msg.setRaw((unsigned char*) &data, sizeof(data),
Message::kCleanupNoDelete);
if (!msg.write(writePipe)) {
LOG(LOG_WARN, "", "hello ack write failed in stream init\n");
return false;
}
}
/* success, set up our local stuff */
mReadPipe = readPipe;
mWritePipe = writePipe;
//LOG(LOG_DEBUG, "", "init success\n");
return true;
}