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

Go to the documentation of this file.

#include "jaus/core/transport/transportmanager.h"
#include "jaus/core/transport/judp.h"
#include <iomanip>
#include <iostream>
#include <tinyxml/tinyxml.h>

using namespace JAUS;


TransportManager::TransportManager()
{
    mDebugMessagesFlag = false;             // Turns on/off debug messages.
    mUseTcpFlag = true;                     // Try to make TCP connections first, fail to UDP
    mDisconnectTimeMs = 3600000;            // 1 Hour
    mDelayTimeMs = 0;                       // As fast as possible.
    mMulticastIP.SetAddress("239.255.0.1"); // Multicast group (default)
    mTimeToLive = 16;                       // Default mutlicast TTL
    mUseBroadcastingFlag = false;           // If true, create broadcast instead of multicast for discovery.
    mDebugMessagesMutex.Create("JAUS");     // Debug messages mutex.
}


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


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

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

    TiXmlElement* element = doc.FirstChild("JAUS").FirstChild("Transport").ToElement();
    if(element && element->Attribute("type"))
    {
        if(std::string(element->Attribute("type")) == "JUDP")
        {
            mUseTcpFlag = false;
        }
        else
        {
            mUseTcpFlag = true;
        }
    }
    TiXmlNode* node;

    node = doc.FirstChild("JAUS").FirstChild("Transport").FirstChild("BroadcastIP").FirstChild().ToNode();
    if(node && node->Value())
    {
        mUseBroadcastingFlag = true;
    }
    node = doc.FirstChild("JAUS").FirstChild("Transport").FirstChild("UseBroadcasting").FirstChild().ToNode();
    if(node && node->Value())
    {
        mUseBroadcastingFlag = atoi(node->Value()) > 0 ? true : false;
    }
    node = doc.FirstChild("JAUS").FirstChild("Transport").FirstChild("MulticastIP").FirstChild().ToNode();
    if(node && node->Value())
    {
        mMulticastIP.SetAddress(node->Value());
    }
    node = doc.FirstChild("JAUS").FirstChild("Transport").FirstChild("TTL").FirstChild().ToNode();
    if(node && node->Value())
    {
        mTimeToLive = (unsigned char )atoi(node->Value());
    }
    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("MaxPacketSizeBytes").FirstChild().ToNode();
    if(node && node->Value())
    {
        mMaxPayloadSize = (unsigned int)atoi(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("DisconnectTimeMs").FirstChild().ToNode();
    if(node && node->Value() && atoi(node->Value()) >= 0)
    {
        mDisconnectTimeMs = (unsigned int)(atoi(node->Value()));
    }
    element = doc.FirstChild("JAUS").FirstChild("Transport").FirstChild("Connection").ToElement();
    while(element)
    {
        if(element->Attribute("ip") && element->Attribute("id"))
        {
            IP4Address ip = element->Attribute("ip");
            Address id = Address::FromString(element->Attribute("id"));
            mPermanentConnections[id] = ip;
        }
        element = element->NextSiblingElement("Connection");
    }
    return true;
}


bool TransportManager::Initialize()
{
    bool result = false;
    
    Shutdown();

    IP4Address::List hostnames;
    IP4Address::List::iterator eth0;
    CxUtils::Socket::GetHostAddresses(hostnames);
    if(hostnames.size() == 0)
    {
        Mutex::ScopedLock lock(&mDebugMessagesMutex);
        std::cout << "\nTransportManager: 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)
        {
            mListenSocket.SetNetworkInterface(mHostIP);
            mUdpSocket.SetNetworkInterface(mHostIP);
            if(mListenSocket.InitializeSocket(JTCP::Port,   // JAUS Port
                                              255,          // Pending listen queue of 255
                                              0,            // Infinite wait on sending
                                              500,          // 500 ms max wait on receive
                                              1)            // Blocking socket
               &&
               mUdpSocket.InitializeMulticastSocket(mMulticastIP,           // Multicast IP
                                                    JTCP::Port,             // JAUS Port
                                                    mTimeToLive,            // Multicast TTL
                                                    mUseBroadcastingFlag,   // Turns on true broadcast
                                                    JTCP::Port,             // UDP messages come from JAUS port
                                                    false))                 // Do not allow port re-use
            {
                // Reserved JAUS Port, Yay!
                // Create threads
                mTcpListenThread.CreateThread(TransportManager::TcpListenThread, this);
                mTcpListenThread.SetThreadName("JTCP-Listen");
                mUdpReceiveThread.CreateThread(TransportManager::UdpReceiveThread, this);
                mUdpReceiveThread.SetThreadName("JUDP-RCV");
                mUdpReceiveThread.SetThreadPriority(25);
                result = true;
                // Allow time for threads to start.
                CxUtils::SleepMs(100);
            }
            // Bail-OUT
            break;
        }
    }

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

    return result;
}


void TransportManager::Shutdown()
{
    mListenSocket.Shutdown();
    mUdpSocket.Shutdown();
    mTcpListenThread.StopThread();
    mUdpReceiveThread.StopThread();

    {
        Mutex::ScopedLock lock(&mUdpMutex);

        std::map<Address, CxUtils::UdpClient*>::iterator udp;
        for(udp = mUdpConnections.begin();
            udp != mUdpConnections.end();
            udp++)
        {
            udp->second->Shutdown();
        }

        for(udp = mUdpConnections.begin();
            udp != mUdpConnections.end();
            udp++)
        {
            delete udp->second;
        }
    }
    {
        Mutex::ScopedLock lock(&mTcpMutex);
        std::map<Address, JTCP*>::iterator tcp;
        for(tcp = mTcpConnections.begin();
            tcp != mTcpConnections.end();
            tcp++)
        {
            delete tcp->second;
        }
        std::vector<JTCP*>::iterator unknown;
        for(unknown = mUnknownConnections.begin();
            unknown != mUnknownConnections.end();
            unknown++)
        {
            delete *unknown;
        }
    }

    mTcpConnections.clear();
    mUnknownConnections.clear();
    mUdpConnections.clear();
}


void TransportManager::CheckServiceStatus(const unsigned int timeSinceLastCheckMs)
{
    if(mTcpListenThread.IsThreadActive() == false ||
       mUdpReceiveThread.IsThreadActive() == false)
    {
        return;
    }
    // Check TCP data.
    {
        Mutex::ScopedLock connectionsLock(&mTcpMutex);
        // Try sort connections by JAUS ID
        std::vector<JTCP*>::iterator u = mUnknownConnections.begin();
        while(u != mUnknownConnections.end())
        {
            if((*u)->GetLastSourceID().IsValid())
            {
                // Sort and remove from list.
                std::map<Address, JTCP*>::iterator i;
                i = mTcpConnections.find((*u)->GetLastSourceID());
                if(i == mTcpConnections.end())
                {
                    mTcpConnections[(*u)->GetLastSourceID()] = (*u);
                }
                else
                {
                    // Resolve conflicting JAUS ID
                    Time::Stamp updateTimeMs = (*u)->GetUpdateTime().ToMs();
                    if(Time::GetUtcTimeMs() - updateTimeMs > mDelayTimeMs)
                    {
                        // Delete connection and remove.
                        if(mDebugMessagesFlag)
                        {
                            Mutex::ScopedLock lock(&mDebugMessagesMutex);
                            std::cout << "\nTransportManager: Dropping conflicting connection [" << i->first.ToString() << "]\n";
                        }
                        delete i->second;
                        i->second = (*u);
                    }
                }
                if(mDebugMessagesFlag)
                {
                    Mutex::ScopedLock lock(&mDebugMessagesMutex);
                    std::cout << "\nTransportManager: Identified connection [" << (*u)->GetLastSourceID().ToString() << "]\n";
                }
                // Remove from vector.
                u = mUnknownConnections.erase(u);
            }
            else
            {
                u++;
            }
        }

        // Check for dropped/closed out connections.
        std::map<Address, JTCP*>::iterator tcp = mTcpConnections.begin();
        while(tcp != mTcpConnections.end())
        {
            if(tcp->second->GetSocket()->IsValid() == false)
            {
                if(mDebugMessagesFlag)
                {
                    Mutex::ScopedLock lock(&mDebugMessagesMutex);
                    std::cout << "\nTransportManager: Dropping JTCP Connection [" << tcp->first.ToString() << "]\n";
                }
                delete tcp->second;
                mTcpConnections.erase(tcp);
                tcp = mTcpConnections.begin();
            }
            else
            {
                tcp++;
            }
        }
    }
    // Only check for drops or need for new connections every 1s
    static Time::Stamp prevCheckTime = 0;
    if(Time::GetUtcTimeMs() - prevCheckTime <= 1000)
    {
        return;
    }

    {
        Mutex::ScopedLock updLock(&mUdpMutex);
        std::map<Address, bool>::iterator perm;
        std::map<Address, Time::Stamp>::iterator utime;

        utime = mUpdateTimes.begin();
        while(utime != mUpdateTimes.end() && mDisconnectTimeMs > 0)
        {
            if(mPermanentConnections.find(utime->first) != mPermanentConnections.end())
            {
                utime++;
                continue;
            }
            Time::Stamp diff = Time::GetUtcTimeMs() - utime->second;
            if(diff >= mDisconnectTimeMs)
            {
                if(mDebugMessagesFlag)
                {
                    Mutex::ScopedLock lock(&mDebugMessagesMutex);
                    std::cout << "\nTransportManager: Connection Timeout to " << utime->first.ToString() << "\n";
                }
                std::map<Address, CxUtils::UdpClient*>::iterator judp;
                judp = mUdpConnections.find(utime->first);
                if(judp != mUdpConnections.end())
                {
                    delete judp->second;
                    mUdpConnections.erase(judp);
                }
                mUpdateTimes.erase(utime);
                utime = mUpdateTimes.begin();
            }
            else
            {
                utime++;
            }
        }
    }

    // See if we need to create any connections.
    std::map<Address, IP4Address>::iterator p;
    for(p = mPermanentConnections.begin();
        p != mPermanentConnections.end();
        p++)
    {
        std::map<Address, JTCP*>::iterator tcp;
        std::map<Address, CxUtils::UdpClient*>::iterator udp;

        {
            Mutex::ScopedLock tcpLock(&mTcpMutex);
            
            tcp = mTcpConnections.find(p->first);
            if(tcp != mTcpConnections.end())
            {
                continue;
            }
        }
        {
            Mutex::ScopedLock udpLock(&mUdpMutex);
            
            udp = mUdpConnections.find(p->first);
            if(udp != mUdpConnections.end())
            {
                continue;
            }
        }
        // Try TCP
        if(mUseTcpFlag)
        {
            CxUtils::TcpClient* sock = new CxUtils::TcpClient();
            if(sock->InitializeSocket(p->second, JTCP::Port))
            {
                JTCP* jtcp = new JTCP();
                jtcp->Initialize(this, sock);
                Mutex::ScopedLock tcpLock(&mTcpMutex);
                mTcpConnections[p->first] = jtcp;
                continue;
            }
            delete sock;
        }
        // Fall back to UDP on failure or default state.
        {
            Mutex::ScopedLock udpLock(&mUdpMutex);
            CxUtils::UdpClient* sock = mUdpSocket.CreateNewDestination(p->second, JUDP::Port);
            if(sock)
            {
                mUdpConnections[p->first] = sock;
                mUpdateTimes[p->first] = Time::GetUtcTimeMs();
            }
        }
    }

    // Update check time.
    prevCheckTime = Time::GetUtcTimeMs();
}


void TransportManager::ProcessPacket(Packet& packet,
                                     Header& header,
                                     const IP4Address& ipAddress,
                                     const unsigned short sourcePort,
                                     const int transport)
{
    // Route the packet by sending it to all destinations.
    if(header.mDestinationID.IsBroadcast())
    {
        bool fromLocalHost = false;
        if(CxUtils::Socket::IsHostAddress(ipAddress))
        {
            fromLocalHost = true;
        }
        // Check if we need to send using multicast/broadcast.
        if(transport != TransportManager::UDP &&
           header.mBroadcastFlag != Header::Broadcast::None &&
           fromLocalHost)
        {
            mUdpSocket.Send(packet);
        }
        // Send to all matching destinations.
        {
            Mutex::ScopedLock lock(&mTcpMutex);
            std::map<Address, JTCP*>::iterator tcp;
            for(tcp = mTcpConnections.begin(); 
                tcp != mTcpConnections.end(); 
                tcp++)
            {
                if(Address::DestinationMatch(header.mDestinationID, tcp->first) &&
                    tcp->first != header.mSourceID)
                {
                    if(header.mBroadcastFlag != Header::Broadcast::None)
                    {
                        // Only send global/local broadcast flagged messages
                        // to local TCP connections since they will not receive them.
                        if(tcp->second->IsLocalConnection())
                        {
                            tcp->second->GetSocket()->Send(packet);
                        }
                    }
                    // Not a local/global broadcast flagged message.  Make sure 
                    // we only send off of this host if the message originated here.
                    else if( (tcp->second->IsLocalConnection() == false && fromLocalHost) ||
                             (tcp->second->IsLocalConnection() ) )
                    {
                        tcp->second->GetSocket()->Send(packet);
                    }
                }
            }
        }
        // Let mutex unlock here so we don't block forever!
        {
            Mutex::ScopedLock lock(&mUdpMutex);
            std::map<Address, CxUtils::UdpClient*>::iterator udp;
            for(udp = mUdpConnections.begin();
                udp != mUdpConnections.end() && header.mBroadcastFlag == Header::Broadcast::None;
                udp++)
            {
                if(Address::DestinationMatch(header.mDestinationID, udp->first) &&
                    udp->first != header.mSourceID)
                {
                    udp->second->Send(packet);
                }
            }
        }
    }
    else
    {
        {   // Lookup TCP destination and send.
            Mutex::ScopedLock lock(&mTcpMutex);
            std::map<Address, JTCP*>::iterator tcp = mTcpConnections.find(header.mDestinationID);
            if(tcp != mTcpConnections.end())
            {
                if(tcp->second->GetSocket()->Send(packet) == 0)
                {
                    if(mDebugMessagesFlag)
                    {
                        Mutex::ScopedLock lock(&mDebugMessagesMutex);
                        std::cout << "\nTransportManager: ERROR Sending Over TCP - [SRC:" << header.mSourceID.ToString() << ", DST:" << header.mDestinationID.ToString() << "] Size: " << header.mSize << "\n";
                        std::cout << "\nTransportManager: Dropping JTCP Connection [" << tcp->first.ToString() << "]\n";
                    }
                    delete tcp->second;
                    mTcpConnections.erase(tcp);
                }
                return;
            }
        }
        {   // Try UDP destination and send.
            Mutex::ScopedLock lock(&mUdpMutex);
            std::map<Address, CxUtils::UdpClient*>::iterator udp = mUdpConnections.find(header.mDestinationID);
            if(udp != mUdpConnections.end())
            {
                if(udp->second->Send(packet) == 0)
                {
                    if(mDebugMessagesFlag)
                    {
                        Mutex::ScopedLock lock(&mDebugMessagesMutex);
                        std::cout << "\nTransportManager: ERROR Sending Over JUDP - [SRC:" << header.mSourceID.ToString() << ", DST:" << header.mDestinationID.ToString() << "] Size: " << header.mSize << "\n";
                    }
                }
                return;
            }
        }
        if(mDebugMessagesFlag)
        {
            Mutex::ScopedLock lock(&mDebugMessagesMutex);
            std::cout << "\nTransportManager: Unknown Destination - [SRC:" << header.mSourceID.ToString() << ", DST:" << header.mDestinationID.ToString() << "] Size: " << header.mSize << "\n";
        }
    }
}


bool TransportManager::AddConnection(const IP4Address& networkIP,
                                     const Address& jausID)
{
    if(IsInitialized() == false && jausID.IsValid())
    {
        mPermanentConnections[jausID] = networkIP;
        return true;
    }
    return false;
}


unsigned int TransportManager::GetNumLocalClients() const
{
    std::map<Address, JTCP*>::const_iterator c;
    Mutex::ScopedLock tcpLock(&mTcpMutex);
    unsigned int count = 0;
    for(c = mTcpConnections.begin();
        c != mTcpConnections.end();
        c++)
    {
        CxUtils::TcpServer* server = dynamic_cast<CxUtils::TcpServer*>(c->second->GetSocket());
        if(server)
        {
            if(CxUtils::Socket::IsHostAddress(server->GetClientAddress()))
            {
                count++;
            }
        }
    }
    return count;
}

void TransportManager::TcpListenThread(void* args)
{
    TransportManager* manager = (TransportManager*)args;
    CxUtils::TcpServer* sock = NULL;
    while(manager && manager->mTcpListenThread.QuitThreadFlag() == false)
    {
        sock = NULL;

        // This will block until a connection is made.
        sock = manager->mListenSocket.AwaitConnection();

        if(sock)
        {
            // Create a thread to process the data.
            JTCP* tcp = new JTCP();
            tcp->SetPacketPollingDelayMs(manager->mDelayTimeMs);
            if(tcp->Initialize(manager, sock))
            {
                if(manager->mDebugMessagesFlag)
                {
                    Mutex::ScopedLock lock(&manager->mDebugMessagesMutex);
                    std::cout << "\nTransportManager: New  incomming JTCP connection made.\n";
                }
                {
                    Mutex::ScopedLock lock(&manager->mTcpMutex);
                    // Add to the unknown connections since we do
                    // not yet know what JAUS ID is associated with
                    // this connection.
                    manager->mUnknownConnections.push_back(tcp);
                }
                // Updates connections.
                manager->CheckServiceStatus(0);
            }
            else
            {
                delete tcp;
                tcp = NULL;
            }
        }
        CxUtils::SleepMs(1);
    }

    if(sock)
    {
        delete sock;
    }
}


void TransportManager::UdpReceiveThread(void* args)
{
    TransportManager* manager = (TransportManager*)args;
    Packet udpMessage;
    CxUtils::IP4Address sourceAddress;
    unsigned short sourcePort = 0;
    long int timeoutMs = 100;

#ifdef WIN32
    int loopCounter = 0;
#endif

    udpMessage.Reserve(5000);

    while(manager && manager->mUdpReceiveThread.QuitThreadFlag() == false)
    {
        udpMessage.Clear(false);
        sourcePort = 0;

        if(manager->mUdpSocket.Recv(udpMessage, 
                                    5000, 
                                    timeoutMs, 
                                    &sourceAddress, 
                                    &sourcePort) > 0)
        {

            // Check for JUDP Version header.
            if(CxUtils::Socket::IsHostAddress(sourceAddress) == false &&
               udpMessage.Length() > Header::MinSize + BYTE_SIZE && *udpMessage.Ptr() == JUDP::Version)
            {
                // Strip the JUDP Version header using a wrapper packet (more efficient).
                Packet::Wrapper stripped((unsigned char *)(udpMessage.Ptr() + 1), udpMessage.Length() - 1);
                Header header;
                stripped->SetReadPos(0);
                std::string errorMessage;
                if(header.Read(*stripped.GetData()) && header.IsValid(&errorMessage))
                {
                    // Update UDP Client connections.
                    {
                        bool createConnection = false;
                        {
                            {
                                Mutex::ScopedLock udpLock(&manager->mUdpMutex);
                                manager->mUpdateTimes[header.mSourceID] = Time::GetUtcTimeMs();
                                std::map<Address, CxUtils::UdpClient*>::iterator udp;
                                udp = manager->mUdpConnections.find(header.mSourceID);
                                if(udp == manager->mUdpConnections.end())
                                {
                                    createConnection = true;
                                }
                                else if(udp->second->GetDestinationPort() != sourcePort ||
                                        udp->second->GetConnectionAddress() != sourceAddress)
                                {
                                    // Delete connection.
                                    Mutex::ScopedLock lock(&manager->mDebugMessagesMutex);
                                    std::cout << "=============================================================\n";
                                    std::cout << "=============================================================\n";
                                    std::cout << "=============================================================\n";
                                    std::cout << "=============================================================\n";
                                    std::cout << "JAUS ID DUPLICATED BY " << sourceAddress.mString << std::endl;
                                    std::cout << "=============================================================\n";
                                    std::cout << "=============================================================\n";
                                    std::cout << "=============================================================\n";
                                    std::cout << "=============================================================\n";

                                    // Reset flag.
                                    createConnection = true;
                                    delete udp->second;
                                    manager->mUdpConnections.erase(udp);
                                }
                            }
                            if(createConnection)
                            {
                                Mutex::ScopedLock tcpLock(&manager->mTcpMutex);
                                std::map<Address, JTCP*>::iterator tcp;
                                tcp = manager->mTcpConnections.find(header.mSourceID);
                                if(tcp != manager->mTcpConnections.end())
                                {
                                    createConnection = false;
                                }
                                if(manager->mUnknownConnections.size() > 0)
                                {
                                    createConnection = false;
                                }
                            }
                        }
                        if(createConnection)
                        {
                            // If we do not have a connection to this source yet
                            // make either a TCP or UDP client.
                            if(manager->mUseTcpFlag)
                            {
                                Mutex::ScopedLock tcpLock(&manager->mTcpMutex);
                                std::map<Address, JTCP*>::iterator tcp;
                                tcp = manager->mTcpConnections.find(header.mSourceID);
                                if(tcp == manager->mTcpConnections.end())
                                {
                                    CxUtils::TcpClient* sock = new CxUtils::TcpClient();
                                    if(sock->InitializeSocket(sourceAddress, sourcePort))
                                    {
                                        JTCP* jtcp = new JTCP();
                                        jtcp->Initialize(manager, sock);
                                        manager->mTcpConnections[header.mSourceID] = jtcp;
                                        createConnection = false;
                                    }
                                    else
                                    {
                                        delete sock;
                                        sock = NULL;
                                    }
                                }
                                else
                                {
                                    // Have TCP connection.
                                    createConnection = false;
                                }
                            }
                            if(createConnection == true)
                            {
                                // Create a UDP connection.
                                Mutex::ScopedLock udpLock(&manager->mUdpMutex);
                                CxUtils::UdpClient* sock = manager->mUdpSocket.CreateNewDestination(sourceAddress, sourcePort);
                                if(sock)
                                {
                                    manager->mUdpConnections[header.mSourceID] = sock;
                                }
                            }
                        }
                    }
                    manager->ProcessPacket(udpMessage, header, sourceAddress, sourcePort, TransportManager::UDP);
                }
            }
        }

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

/*  End of File */