NetMessageProtocol.cpp

Go to the documentation of this file.

/***************************** License **********************************

 Copyright (C) 2008 by Communicative Machines
 http://www.cmlabs.com   All rights reserved.

 This library is free software; you can redistribute it and/or
 modify it under the terms of the GNU Lesser General Public
 License as published by the Free Software Foundation; either
 version 2.1 of the License, or (at your option) any later version.

 This library is distributed in the hope that it will be useful,
 but WITHOUT ANY WARRANTY; without even the implied warranty of
 MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
 Lesser General Public License for more details.

 You should have received a copy of the GNU Lesser General Public
 License along with this library; if not, write to the Free Software
 Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA

************************************************************************/
// NetMessageProtocol.cpp: implementation of the NetMessageProtocol class.
//
//////////////////////////////////////////////////////////////////////

#include "NetMessageProtocol.h"

namespace cmlabs {

//////////////////////////////////////////////////////////////////////
// Construction/Destruction
//////////////////////////////////////////////////////////////////////

NetMessageProtocol::NetMessageProtocol():NetProtocol("Message")
{
        netTimeout = 60000;
}

NetMessageProtocol::~NetMessageProtocol()
{

}

Object* NetMessageProtocol::clone() const {
        NetMessageProtocol* nmp = new NetMessageProtocol();
        nmp->isLocalCallback = isLocalCallback;
        nmp->isRemoteCallback = isRemoteCallback;
        return nmp;
}

bool NetMessageProtocol::checkBufferForCompatibility(char* buffer, int length) {

        isRemoteCallback = false;
        if (length < 7) return false;

        JString str = buffer;
        if (str.startsWith("Message"))
                return true;
        else if (str.startsWith("Receive")) {
                isRemoteCallback = true;
                isLocalCallback = false;
                return true;
        }
        
        return false;
}

bool NetMessageProtocol::initializeConversation(JSocket* socket) {
        if (socket == NULL)
                return false;

        return true;
}

bool NetMessageProtocol::initializeAsReceiver(JSocket* socket, JString from) {

        int res;

        Message msg = Message(from, "", "RECEIVE_NAME_NOTIFICATION");
        JString str = msg.toXML();
        long size = str.length();

        char* buffer = new char[12 + size + 1];

        strcpy(buffer, "Receive");
//      *((long*) (buffer+8)) = size;
        writeLongToBytes(buffer+8, size);

        memcpy(buffer+12, str.charpoint(), size);

        if ((res=socket->write(buffer, 12+size) != (12+size))) {
                delete [] buffer;
                return false;
        }

        isLocalCallback = true;
        isRemoteCallback = false;

        /*      if ((res=socket->write(buffer, 12) != 12)) {
                delete [] buffer;
                return false;
        }

        if ((res=socket->writeString(str) != size)) {
                delete [] buffer;
                return false;
                }*/

//      wait(0);
/*      if ((res=socket->read(buffer, 3, 200)) < 3) {
                delete [] buffer;
                return false;
        }

        str = buffer;

        if (!str.equals("OK")) {
                wait(10);
                int fc = socket->flush();
                if (DEBUGLEVEL(STATUS)) {
                        printf("######### [NetMessageProtocol] Read '%s' [%c %c %c] instead of OK - flushed %d\n", (char*)str.getFirst(10), buffer[0], buffer[1], buffer[2], fc);
                }
//              wait(20000);
                delete [] buffer;
                return false;
        }
*/
        if (DEBUGLEVEL(DEBUG)) {
                printf("######### [NetMessageProtocol] InitAsRec success!\n");
        }

        delete [] buffer;
        return true;
}


bool NetMessageProtocol::sendObject(JSocket* socket, Message* msg, bool isReply) {

//      if (DEBUGLEVEL(DEBUG)) {
//              printf("s");fflush(stdout);
//      }
        currentUnsentBufferSize = 0;
        currentSentBufferSize = 0;
//      JTime t1;

        socket->readIntoBuffer();
        if (socket->charBufferCount > 0) {
                if (DEBUGLEVEL(STATUS)) {
                        printf("######### [NetMessageProtocol] Something already in input buffer\n");
                }
                return false;
        }

        int res;

        //JTime t;
        //printf("--- Send Start: %s\n", (char*)t.printTime());

//      JTime t2;

        bool binaryOnly = false;

        DataSample* sample;
        int binChunkCount = msg->getBinaryChunkCount();
        int binSize = 0;
        Collection bufSizes;
        if ( (binChunkCount == 1) && (msg->getType().equalsIgnoreCase("BinaryOnly") ) ) {
                sample = (DataSample*) msg->getObject();
                if (sample == NULL)
                        return false;
                if (msg->noreply.length() > 0)
                        sample->setParam("---MsgNoReply---", msg->noreply);
                binSize = msg->getBinarySize(0);
                binaryOnly = true;
        }
        else if (binChunkCount > 0) {
                int subBinSize = 0;
                JString bufString;
                for (int i=0; i<binChunkCount; i++) {
                        subBinSize = msg->getBinarySize(i);
                        binSize += subBinSize;
                        bufSizes.add(JString(subBinSize));
                }
                if (bufSizes.getCount() > 0) {
                        bufString = bufSizes.printListLine(",");
                        msg->set("BinarySizes", bufString);
                        msg->set("TotalBinarySize", JString(binSize));
                }
                else {
                        binChunkCount = 0;
                        binSize = 0;
                }
        }

        JString str;
        long size = 0;
        
        if (!binaryOnly) {
                str = msg->toXML();
                size = str.length();
                // printf("Message [%s]->[%s] sent at %s...\n", (char*) msg->type, (char*) msg->to, (char*) msg->getTimestamp().printTime());
        }

        long totalBytes = size + 12 + binSize;
        currentUnsentBufferSize = totalBytes;
        char* buffer = new char[totalBytes+1];
        long bytesSoFar = 0;

        if (binChunkCount > 0) {

                if (!binaryOnly) {
                        strcpy(buffer, "Message");
                        writeLongToBytes(buffer+8, size);
                        memcpy(buffer+12, str.charpoint(), size);

                        JString bufString;
                        long bufsize;
                //      char* buf;
                        for (int j=0; j<binChunkCount; j++) {
                                bufsize = bufSizes.get(j).toLong();
                                msg->toBinaryBuffer(j, buffer+12+size+bytesSoFar, bufsize);
        //                      printf("BinSend [%d]: [%c%c%c%c%c%c] [%d]\n", j, buf[0], buf[1], buf[2], buf[3], buf[4], buf[5], bufsize);
                                bytesSoFar += bufsize;
                        }
                        // printf("BinarySend: %d chunks %ld bytes\n", binChunkCount, totalBytes);
                }
                else {
                        // We now have to assume that the binary object is a DataSample
                        sample = (DataSample*) msg->getObject();
                        if (sample == NULL) {
                                return false;
                        }
                        if (!sample->toBinaryBuffer(0, buffer+12, binSize)) {
                                // Error...
                        }
                        strcpy(buffer, "BinaryM");
                        writeLongToBytes(buffer+8, binSize);
                        // printf("BinaryOnlySend: %d chunks %ld bytes\n", binChunkCount, totalBytes-12);
                }

                if (totalBytes <= socket->getSendBufferSize()) {
                        if ((res=socket->write(buffer, totalBytes) != (totalBytes))) {
                                currentUnsentBufferSize = 0;
                                currentSentBufferSize = 0;
                                delete [] buffer;
                                return false;
                        }
                        currentSentBufferSize = totalBytes;
                }
                else {
                        long breakmax = socket->getSendBufferSize();
                        res = 0;
                        int subres = 0;
                        char* lpbuffer = buffer;
                        int l = totalBytes;
                        int n = l / breakmax;
                        for (int i=0; i < n; i++) {
                                subres = socket->write(&lpbuffer[i*breakmax], breakmax);
                                if (subres == breakmax) {
                                        res += subres;
                                        currentSentBufferSize = res;
                                }
                                else {
                                        printf("*** Only partial messasge sent - %d out of %d ***\n", res, totalBytes);
                                        currentUnsentBufferSize = 0;
                                        currentSentBufferSize = 0;
                                        delete [] buffer;
                                        return false;
                                }
                                // socket->waitForWriteability(5000);
                        }
                        // wait(10);
                        n = n * breakmax;
                        //printf("                           last writing %d bytes to socket...\n", l-n);
                        subres = socket->write(&lpbuffer[n], l-n);
                        if (subres == l-n) {
                                res += subres;
                                currentSentBufferSize = res;
                        }
                        else {
                                printf("*** Only partial messasge sent - %d out of %d ***\n", res, totalBytes);
                                currentUnsentBufferSize = 0;
                                currentSentBufferSize = 0;
                                delete [] buffer;
                                return false;
                        }
                }

        }
        else {
                strcpy(buffer, "Message");

                writeLongToBytes(buffer+8, size);
                memcpy(buffer+12, str.charpoint(), size);
        //      JTime t3;

                if ((res=socket->write(buffer, 12+size) != (12+size))) {
                        currentUnsentBufferSize = 0;
                        currentSentBufferSize = 0;
                        delete [] buffer;
                        return false;
                }
                currentSentBufferSize = size+12;
        }

//      JTime t4;

//      JString from = msg->getFrom();
//      JString to = msg->getTo();
//      JString type = msg->getType();

        delete [] buffer;

        // We now put it to the calling function to delete the message upon successful transmission
        //      delete(msg);

        /*      JTime t5;
        long l1 = t2.microDifference(t1);
        long l2 = t3.microDifference(t2);
        long l3 = t4.microDifference(t3);
        long l4 = t5.microDifference(t4);
        printf("SendObject [%s] (%s) : Clear: %.3f   XML: %.3f  Write: %.3f  Clean: %.3f \n", (char*) type, (char*) t4.printTime(), l1/1000.0, l2/1000.0, l3/1000.0, l4/1000.0);*/

        return true;
//      return (str.equals("OK"));
}

Message* NetMessageProtocol::receiveObject(JSocket* socket, int timeout) {
        Message* msg = NULL;
        JTime start;

//      if (DEBUGLEVEL(DEBUG))
//              printf("r");fflush(stdout);

        currentUnreceivedBufferSize = 0;
        currentReceivedBufferSize = 0;

//      JTime t1;
        socket->readIntoBuffer();
        if (socket->charBufferCount < 12) {
                // JTime t1;
                socket->waitForReadability(timeout);
                socket->readIntoBuffer();
                if (socket->charBufferCount < 12) {
                        if (DEBUGLEVEL(KITCHENSINK)) {
                          //    printf("[NetMsg] Received Nothing [%d]!\n", socket->charBufferCount);
                        }
                        if (socket->charBufferCount > 0) {
                                // if (DEBUGLEVEL(DEBUG))
                                        printf("Incomplete Buffer: {%c %c %c %c %c %c %c %c %c %c %c %c}", 
                                                socket->charBuffer[0],
                                                socket->charBuffer[1],
                                                socket->charBuffer[2],
                                                socket->charBuffer[3],
                                                socket->charBuffer[4],
                                                socket->charBuffer[5],
                                                socket->charBuffer[6],
                                                socket->charBuffer[7],
                                                socket->charBuffer[8],
                                                socket->charBuffer[9],
                                                socket->charBuffer[10],
                                                socket->charBuffer[11]);fflush(stdout);
                        }
                //      if (DEBUGLEVEL(DEBUG))
                //              printf("{%d/%d/%d}", (int)t1.getAge(), timeout, socket->charBufferCount);fflush(stdout);
                        return NULL;
                }
        }
//      JTime t2;

        char* buffer = new char[12];
        int count = 0;

        count = socket->read(buffer, 12, timeout);

//      printf("[NetMsg] Received Something: %d chars!\n", count);

//      JTime t3;

        if (count <= 0) {
                delete [] buffer;
                if (DEBUGLEVEL(KITCHENSINK)) {
                  //    printf("[NetMsg] Received Nothing!\n");
                }
                //if (DEBUGLEVEL(DEBUG))
                //      printf("{%d}", count);fflush(stdout);
                return NULL;
        }
        else if (count < 12) {
                if (DEBUGLEVEL(STATUS)) {
                        printf("[NetMsg] Received Something: %d chars!\n", count);
                }
                delete [] buffer;
                return NULL;
        }

/*        printf("Message: ");
        for (int i=0; i<12; i++)
          printf("[%d] ", (unsigned int) buffer[i]);
        printf("\n");
*/
        JString str = buffer;
//      if (str.equals("Receive")) {
//              strcpy(buffer, "OK");
//      }

        if ( (!str.equals("Message")) && (!str.equals("Receive")) && (!str.equals("BinaryM")) ) {
                wait(10);
                int fc = socket->flush();
                if (DEBUGLEVEL(STATUS)) {
                        printf("[NetMsg] Received Something not Message - Read '%s' [%c %c %c] - flushed %d!\n", (char*)str.getFirst(10), buffer[0], buffer[1], buffer[2], fc);
                }
                delete [] buffer;
                return NULL;
        }

//      long size = *((long*) (buffer+8));
        long size = getLongFromBytes(buffer+8);
        delete [] buffer;       

        if (size < 0) {
                wait(10);
                int fc = socket->flush();
                if (DEBUGLEVEL(STATUS)) {
                        printf("[NetMsg] Received wrong length %d!\n", size);
                }
                return NULL;
        }

        currentUnreceivedBufferSize = 0;
        currentReceivedBufferSize = 0;

//      JTime t4;
//      printf("[NetMsg] Received length %d!\n", size);
        buffer = new char[size+1];
        //now = JTime();
        int tm = timeout - start.getAge();
        if (tm <= 0) tm = 10;
        count = socket->read(buffer, size, MSGTIMEOUT);

        if (count != size) {
                currentUnreceivedBufferSize = 0;
                currentReceivedBufferSize = 0;
                wait(10);
                int fc = socket->flush();
                if (DEBUGLEVEL(STATUS)) {
                        printf("[NetMsg] Received wrong amount %d ! %d!\n", count, size);
                }
                delete [] buffer;
                return NULL;
        }

//      JTime t5;

        buffer[count] = 0;

        if (!str.equals("BinaryM")) {
                msg = new Message(buffer);
                // printf("Received [%s]<-[%s] sent at %s...\n", (char*) msg->type, (char*) msg->from, (char*) msg->getTimestamp().printTime());
                // printf("Request [%s] recd at %s...\n", (char*) msg->getType(), (char*) msg->getTimestamp().printTime());
        }
        else {
                DataSample* sample = new DataSample();
                // We have pure binary data in this buffer
                if (!sample->fromBinaryBuffer(0, buffer, size)) {
                        delete(sample);
                        // if (DEBUGLEVEL(DEBUG))
                        //      printf("B");fflush(stdout);
                        return new Message("" ,"", "BINARY_RECEIVE_FAILED");
                }
                msg = new Message("" ,"", "BinaryOnly", sample);
                if (sample->getParam("---MsgNoReply---").length() > 0) {
                        msg->noreply = sample->getParam("---MsgNoReply---");
                        sample->params.remove("---MsgNoReply---");
                }
                // printf("BinaryOnlyRec: %ld bytes\n", size);
        }

        // Expensive block?
        JString hostname = socket->getRemoteIPAddress();
        if (hostname.equals("localhost"))
                hostname = socket->getLocalIPAddress();
        if (msg->getOrigin().length() == 0)
                msg->setOrigin(hostname);
        msg->set("netprotocol", name);

        delete [] buffer;

        JString binString = msg->get("BinarySizes");
        if (binString.length() == 0) {
        //      printf("Receiving Normal Message [%s] from %s...\n", (char*) msg->getType(), (char*) msg->getFrom());
                currentUnreceivedBufferSize = size+12;
                currentReceivedBufferSize = size+12;
                return msg;
        }

        Collection binSizes = binString.split(",");

        // It would seem we have binary data to read...
        long totalBinSize = msg->get("TotalBinarySize").toLong();
        // printf("BinaryRec: %d chunks %ld bytes\n", binSizes.getCount(), totalBinSize);
        currentUnreceivedBufferSize = totalBinSize + size + 12;
        currentReceivedBufferSize = size+12;

//      printf("Receiving Binary Attachements [%d bytes] [%s] from %s...\n", totalBinSize, (char*) msg->getType(), (char*) msg->getFrom());

        long subBinSize;
        for (int n=0; n<binSizes.getCount(); n++) {
        
                subBinSize = binSizes.get(n).toLong();
                if (subBinSize <= 0) {
                        // Error...
                }
                else {
                        buffer = new char[subBinSize];
                        count = socket->read(buffer, subBinSize, 10000);
                        if (count != subBinSize) {
                                delete [] buffer;
                                // Error...
                        }
                        else {
                        //      printf("BinRec [%d]: [%c%c%c%c%c%c] [%d]\n", n, buffer[0], buffer[1], buffer[2], buffer[3], buffer[4], buffer[5], subBinSize);
                                if (!msg->fromBinaryBuffer(n, buffer, subBinSize)) {
                                        // Error...
                                }
                                delete [] buffer;
                        }
                        currentReceivedBufferSize += subBinSize;
                }
        }
        
        //JTime t = JTime();
        //printf("--- Receive End: %s\n", (char*)t.printTime());


/*      JString from = msg->getFrom();
        JString to = msg->getTo();
        JString type = msg->getType();

        JTime t7;
        long l1 = t2.microDifference(t1);
        long l2 = t3.microDifference(t2);
        long l3 = t4.microDifference(t3);
        long l4 = t5.microDifference(t4);
        long l5 = t6.microDifference(t5);
        long l6 = t7.microDifference(t6);
        printf("RecObject [%s] (%s - %s): Read: %.3f   Read12: %.3f  Parse12: %.3f  Read: %.3f  XML: %.3f  Clean: %.3f \n", (char*) type, (char*) t1.printTime(), (char*) t2.printTime(), l1/1000.0, l2/1000.0, l3/1000.0, l4/1000.0, l5/1000.0, l6/1000.0);
*/
        return msg;
}


long NetMessageProtocol::getLongFromBytes(char* buffer) {
        long val = ( (unsigned char) buffer[0] + ( (unsigned char) buffer[1] << 8) + ( (unsigned char) buffer[2] << 16) + ( (unsigned char) buffer[3] << 24) );
//      long val2 = *((long*) (buffer));

        return val;
}

bool NetMessageProtocol::writeLongToBytes(char* buffer, long val) {
        buffer[0] = (unsigned char) val;
        buffer[1] = (unsigned char) (val >> 8);
        buffer[2] = (unsigned char) (val >> 16);
        buffer[3] = (unsigned char) (val >> 24);
        return true;
}






} // namespace cmlabs

---