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

Go to the documentation of this file.

#include "jaus/core/transport/jtcpclient.h"
#include "jaus/core/discovery/queryidentification.h"
#include "jaus/core/transport/transportmanager.h"
#include <tinyxml/tinyxml.h>
#include <iostream>
#include <iomanip>

using namespace JAUS;


JTCPClient::JTCPClient()
{
    mpTransportManager = new TransportManager();
    mpSocket = new CxUtils::TcpClient();
    mTransportHeader.Write(Version);
    mMaxPayloadSize = 1500 - JTCP::OverheadSizeBytes - USHORT_SIZE;
    mDisconnectTimeMs = 3600000;
}


JTCPClient::~JTCPClient()
{
    JTCPClient::Shutdown();
    if(mpTransportManager)
    {
        mpTransportManager->Shutdown();
        delete mpTransportManager;
    }
}


bool JTCPClient::LoadSettings(const std::string& filename)
{
    TiXmlDocument xml;

    if(xml.LoadFile(filename.c_str()) == false)
    {
        return false;
    }
    TiXmlHandle doc(&xml);
    TiXmlNode* node;

    node = doc.FirstChild("JAUS").FirstChild("Transport").FirstChild("NetAddress").FirstChild().ToNode();
    if(node && node->Value())
    {
        mHostIP.SetAddress(node->Value());
    }
    node = doc.FirstChild("JAUS").FirstChild("Transport").FirstChild("Logger").FirstChild().ToNode();
    if(node && node->Value() && atoi(node->Value()) > 0)
    {
        EnableLogging(true);
    }
    else
    {
        EnableLogging(false);
    }
    node = doc.FirstChild("JAUS").FirstChild("Transport").FirstChild("MaxPacketSizeBytes").FirstChild().ToNode();
    if(node && node->Value())
    {
        mMaxPayloadSize = (unsigned int)atoi(node->Value());
    }

    node = doc.FirstChild("JAUS").FirstChild("Transport").FirstChild("DisconnectTimeMs").FirstChild().ToNode();
    if(node && node->Value())
    {
        mDisconnectTimeMs = (Time::Stamp)atoi(node->Value());
    }

    mpTransportManager->LoadSettings(filename);

    return true;
}


bool JTCPClient::Initialize(const Address& componentID)
{
    return Initialize(componentID, "127.0.0.1");
}


bool JTCPClient::Initialize(const Address& componentID,
                            const IP4Address& host)
{
    bool result = false;

    JTCPClient::Shutdown();

    if(mpTransportManager)
    {
        mpTransportManager->Initialize();
    }

    mpSocket = new CxUtils::TcpClient();

    if(componentID.IsValid() == false || componentID.IsBroadcast())
    {
        Mutex::ScopedLock lock(&mDebugMessagesMutex);
        std::cout << "JTCP- Invalid component ID [" << componentID.ToString() << "]\n";
        return false;
    }
    mComponentID = componentID;
    IP4Address::List hostnames;
    IP4Address::List::iterator eth0;
    CxUtils::Socket::GetHostAddresses(hostnames);
    if(hostnames.size() == 0)
    {
        Mutex::ScopedLock lock(&mDebugMessagesMutex);
        std::cout << "[JTCP-" << mComponentID.ToString() << "] - No Network Interface Found.\n";
    }

    // Initialize socket.
    for(eth0 = hostnames.begin();
        eth0 != hostnames.end();
        eth0++)
    {
        // Find matching network interface
        if(mHostIP.mString == "0.0.0.0" ||
           mHostIP.mString == "127.0.0.1" ||
           mHostIP.mString.empty() ||
           mHostIP == *eth0)
        {
            mpSocket->SetNetworkInterface(*eth0);
            if(((CxUtils::TcpClient *)mpSocket)->InitializeSocket(host, Port, 0, 1000))
            {
                // Connected to TCP Server (yay!);
                SetComponentID(componentID);
                std::string threadName = std::string("JTCP ") + componentID.ToString();
                mRecvThread.SetThreadName(threadName);
                mRecvThread.CreateThread(ReceiveThread, (JTCP*)this);
                CxUtils::SleepMs(50);
                // Send a few messages to server to make sure we
                // do not get shutdown for being a duplicated JAUD ID
                // on the network
                JAUS::QueryIdentification queryIdentification(Address(Address::GlobalBroadcast, 255, 255), 
                                                              componentID);
                for(unsigned int i = 0; i < 3; i++)
                {
                    Send(&queryIdentification);
                    CxUtils::SleepMs(25);
                    if(mpSocket->IsValid() == false || mReceivedValidDataFlag == true)
                    {
                        break;
                    }
                }

                if(mpSocket->IsValid())
                {
                    result = true;
                    CxUtils::SleepMs(100);
                }
                else
                {
                    std::cout << "[JTCP-" << componentID.ToString() << "] - Component ID already in use.\n";
                }
            }
            // Bail-OUT
            break;
        }
    }

    if(result == false)
    {
        Shutdown();
    }

    return result;
}


bool JTCPClient::IsInitialized() const
{
    return mpSocket ? mpSocket->IsValid() : false;
}


void JTCPClient::Shutdown()
{
    if(mpSocket)
    {
        mpSocket->Shutdown();
    }
    for(unsigned int i = 0; i < 500; i++)
    {
        if(mRecvThread.IsThreadActive() == false)
        {
            break;
        }
        CxUtils::SleepMs(1);
    }

    // Make sure threads are stopped.
    mRecvThread.StopThread();
    mReceivedValidDataFlag = false;
    JTCP::Shutdown();
}


void JTCPClient::SetMaxPacketSize(const unsigned int maxSizeBytes)
{
    if(IsInitialized() == false && maxSizeBytes > (JTCP::OverheadSizeBytes + USHORT_SIZE))
    {
        mMaxPayloadSize = maxSizeBytes - OverheadSizeBytes - USHORT_SIZE;
    }
}


Address::List JTCPClient::GetConnections() const
{
    Mutex::ScopedLock lock(&mDataMutex);
    Address::List results;
    std::map<Address, Time::Stamp>::const_iterator client;
    for(client = mUpdateTimes.begin();
        client != mUpdateTimes.end();
        client++)
    {
        results.push_back(client->first);
    }
    return results;
}


bool JTCPClient::HaveConnection(const Address& id) const
{
    return HaveConnection(id, (unsigned int)mDisconnectTimeMs);
}


bool JTCPClient::HaveConnection(const Address& id,
                                const unsigned int timeSinceUpdateMs) const
{
    Mutex::ScopedLock lock(&mDataMutex);
    std::map<Address, Time::Stamp>::const_iterator client;
    client = mUpdateTimes.find(id);
    if(client != mUpdateTimes.end())
    {
        return true;
    }
    return false;
}


bool JTCPClient::SendPacket(const Packet& packet, 
                            const Header& header,
                            const int broadcastFlags) const
{
    bool result = false;

    if(mpSocket && mpSocket->Send(packet))
    {
        result = true;
    }

    return result;
}


bool JTCPClient::SerializeMessage(const Message* message,
                                  Packet::List& stream,
                                  Header::List& streamHeaders,
                                  const UShort startingSequenceNumber,
                                  const int broadcastFlags) const
{
    Packet packet;
    Header header;

    // Pre-allocate memory.
    packet.Reserve(JAUS::Header::MaxPacketSize + OverheadSizeBytes);

    // Clear stream/headers.
    stream.clear();
    streamHeaders.clear();

    // If the message is a large data set, create a multi-packet stream.
    if(message->IsLargeDataSet(mMaxPayloadSize))
    {
        return message->WriteLargeDataSet(stream, 
                                          streamHeaders, 
                                          mMaxPayloadSize, 
                                          &mTransportHeader,
                                          startingSequenceNumber) > 0;
    }
    // Single packet.
    else if(message->Write(packet, header, &mTransportHeader, true, startingSequenceNumber, (Byte)broadcastFlags) > 0)
    {
        stream.push_back(packet);
        streamHeaders.push_back(header);
        return true;
    }
    return false;
}


void JTCPClient::SetInterfaceIP(const IP4Address& networkIP)
{
    if(IsInitialized() == false)
    {
        mHostIP = networkIP;
    }
}


void JTCPClient::EnableDebugMessages(const bool on)
{
    if(mpTransportManager)
    {
        mpTransportManager->EnableDebugMessages(on);
    }
    Transport::EnableDebugMessages(on);
}


bool JTCPClient::AddConnection(const IP4Address& networkIP,
                               const Address& jausID)
{
    return mpTransportManager->AddConnection(networkIP, jausID);
}


void JTCPClient::CheckServiceStatus(const unsigned int timeSinceLastCheckMs)
{
    if(mpTransportManager)
    {
        mpTransportManager->CheckServiceStatus(timeSinceLastCheckMs);
    }
    // Check to see if we have lost our connection to a Transport Manager
    if(mRecvThread.IsThreadActive() == false)
    {
        // Re-connect.
        Initialize(mComponentID, "127.0.0.1");
    }

    Mutex::ScopedLock lock(&mDataMutex);
    std::map<Address, Time::Stamp>::iterator utime;
    utime = mUpdateTimes.begin();
    while(utime != mUpdateTimes.end() && mDisconnectTimeMs > 0)
    {
        CxUtils::Time::Stamp diff = Time::GetUtcTimeMs() - utime->second;
        if(diff >= mDisconnectTimeMs)
        {
            mUpdateTimes.erase(utime);
            utime = mUpdateTimes.begin();
            continue;
        }
        utime++;
    }
}


void JTCPClient::ProcessPacket(Packet& packet,
                               Header& header,
                               const IP4Address& ipAddress,
                               const unsigned short sourcePort)
{
    {
        Mutex::ScopedLock lock(&mDataMutex);
        mUpdateTimes[header.mSourceID] = Time::GetUtcTimeMs();
    }
    // Strip transport header data and process.
    unsigned int overhead = BYTE_SIZE;
    Packet::Wrapper payloadPacket(packet.Ptr() + overhead, packet.Length() - overhead);
    Transport::ProcessPacket(*payloadPacket.GetData(), header);
}


/*  End of File */