jaus++/2.0/src/jaus/core/transport/jtcp.cpp

Go to the documentation of this file.

#include "jaus/core/transport/jtcp.h"
#include "jaus/core/transport/transportmanager.h"

using namespace JAUS;


JTCP::JTCP() : mDelayTimeMs(0), 
               mpSocket(0),
               mpManager(0),
               mReceivedValidDataFlag(false),
               mLocalConnectionFlag(true)
{

}


JTCP::~JTCP()
{
    Shutdown();
}


bool JTCP::Initialize(TransportManager* manager,
                      CxUtils::Socket* socket)
{
    Shutdown();
    if(manager && 
       socket && 
       socket->IsValid() && 
       (socket->GetType() == CxUtils::Socket::TCPServer || socket->GetType() == CxUtils::Socket::TCPClient))
    {
        mpManager = manager;
        mpSocket = socket;
        mUpdateTime.SetCurrentTime();
        if(socket->GetType() == CxUtils::Socket::TCPServer)
        {
            CxUtils::TcpServer* tcp = (CxUtils::TcpServer*)socket;
            if(socket->IsHostAddress(tcp->GetClientAddress()))
            {
                mLocalConnectionFlag = true;
            }
            else
            {
                mLocalConnectionFlag = false;
            }
        }
        if(socket->GetType() == CxUtils::Socket::TCPClient)
        {
            CxUtils::TcpClient* tcp = (CxUtils::TcpClient*)socket;
            if(socket->IsHostAddress(tcp->GetServerAddress()))
            {
                mLocalConnectionFlag = true;
            }
            else
            {
                mLocalConnectionFlag = false;
            }
        }
        return mRecvThread.CreateThread(JTCP::ReceiveThread, this) > 0 ? true : false;
    }
    return false;
}


void JTCP::Shutdown()
{
    if(mpSocket)
    {
        mpSocket->Shutdown();
    }
    mRecvThread.StopThread();
    if(mpSocket)
    {
        delete mpSocket;
    }
    mpSocket = NULL;
    mpManager = NULL;
    mUpdateTime.Clear();
}


void JTCP::SetPacketPollingDelayMs(const unsigned int delayTimeMs)
{
    if(mRecvThread.IsThreadActive() == false)
    {
        mDelayTimeMs = delayTimeMs;
    }
}


Address JTCP::GetLastSourceID() const
{
    Mutex::ScopedLock lock(&mDataMutex);
    return mLastSourceID;
}


Time JTCP::GetUpdateTime() const
{
    Mutex::ScopedLock lock(&mDataMutex);
    return mUpdateTime;
}


void JTCP::ProcessPacket(Packet& packet,
                         Header& header,
                         const IP4Address& ipAddress,
                         const unsigned short sourcePort)
{
    if(mpManager)
    {
        mpManager->ProcessPacket(packet, header, ipAddress, sourcePort, TransportManager::TCP);
    }
}

void JTCP::ReceiveThread(void* args)
{
    JTCP* transport = (JTCP*)args;
    Packet smallBuffer;
    Packet largeBuffer;
    long int timeoutMs = 100;

    largeBuffer.Reserve(4*(JAUS::Header::MaxPacketSize + OverheadSizeBytes));
    smallBuffer.Reserve(JAUS::Header::MaxPacketSize + OverheadSizeBytes);

#ifdef WIN32
    int loopCounter = 0;
#endif
    unsigned int failureCount = 0;
    while(transport &&
          transport->mRecvThread.QuitThreadFlag() == false &&
          transport->ExitThread() == false)
    {
        // Detect shutdown of socket from server.
        if(transport->mpSocket->IsValid() == false || failureCount == 10)
        {
            //  Socket closed on server side, exit.
            smallBuffer.Destroy();
            largeBuffer.Destroy();
            transport->mpSocket->Shutdown();
            break;
        }
        // Try receive some data.
        smallBuffer.Clear();
        Time::Stamp startTimeMs = Time::GetUtcTimeMs();
        IP4Address ipAddress;
        unsigned short sourcePort = 0;

        if(transport->mpSocket->Recv(smallBuffer, 
                                     5000,
                                     timeoutMs,
                                     &ipAddress,
                                     &sourcePort) > 0)
        {
            failureCount = 0;
            largeBuffer += smallBuffer;

            // Try parse out a JAUS message.
            unsigned char* ptr = largeBuffer.Ptr();
            unsigned int pos = 0;
            unsigned int endPos = 0;
            while(pos < largeBuffer.Length() - JAUS::Header::MinSize)
            {
                // Check for JTCP Header
                if(ptr[pos] == Version)
                {
                    // Try read message data.
                    JAUS::Header generalHeader;
                    Packet::Wrapper subPacket(&ptr[pos + sizeof(Version)], largeBuffer.Length() - pos);
                    std::string errorMessage;
                    int bytesRead = 0;
                    if( (bytesRead = generalHeader.Read(*subPacket.GetData())) > 0 &&
                        generalHeader.IsValid(&errorMessage) &&
                        subPacket->Length() >= generalHeader.mSize)
                    {
                        // Read header data, now pull payload data.
                        Packet::Wrapper finalPacket(&ptr[pos], 
                                                    (unsigned int)generalHeader.mSize + sizeof(Version));
                        // Process data.
                        transport->ProcessPacket(*finalPacket.GetData(), generalHeader, ipAddress, sourcePort);
                        {
                            Mutex::ScopedLock lock(&transport->mDataMutex);
                            transport->mUpdateTime.SetCurrentTime();
                            transport->mReceivedValidDataFlag = true;
                            transport->mLastSourceID = generalHeader.mSourceID;
                        }
                        // Advance position in stream buffer.
                        pos += finalPacket->Length();
                        endPos = pos;
                        continue;
                    }
                    if(errorMessage.empty() == false)
                    {
                        //std::cout << "JTCP::Error [Header] - " << errorMessage << std::endl;
                    }
                }
                pos++;
            }

            if(endPos > 0)
            {
                // Delete data we have read from the buffer.
                largeBuffer.Delete(endPos, 0);
            }

            if(largeBuffer.Size() >= largeBuffer.Reserved() - JAUS::Header::MinSize + 1)
            {
                // Getting bad data, flush out buffers.
                // Hopefully we never get to this point in code.
                std::cout << "JTCP::ERROR - Buffer Overflow.\n";
                largeBuffer.Clear();
                smallBuffer.Clear();
            }
        }
        else if(Time::GetUtcTimeMs() - startTimeMs <= (Time::Stamp)(timeoutMs/2.0))
        {
            failureCount++;
        }

        if(transport->mDelayTimeMs == 0)
        {
#ifdef WIN32
            // Only sleep every N loops
            if( loopCounter++ == 250)
            {
                loopCounter = 0;
                CxUtils::SleepMs(1);
            }
#else
            usleep(500);
#endif
        }
        else
        {
            CxUtils::SleepMs(transport->mDelayTimeMs);
        }
    }
}

/*  End of File */