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

Go to the documentation of this file.

#include "jaus/core/transport/largedataset.h"
#include <string.h>

using namespace JAUS;


LargeDataSet::Key::Key(const Address& src,
                       const UShort code,
                       const UInt pv)
{
    mSourceID = src;
    mMessageCode = code;
    mPresenceVector = pv;
    mIdentifier = 0;
}


LargeDataSet::Key::Key(const LargeDataSet::Key& key)
{
    *this = key;
}


LargeDataSet::Key::~Key()
{
}


void LargeDataSet::Key::Update() const
{
    Packet * p = ( (Packet *)(&mKey) );
    p->Clear();
    p->Reserve(UINT_SIZE + USHORT_SIZE + ULONG_SIZE + BYTE_SIZE);
    p->Write(mSourceID.ToUInt());
    p->Write(mMessageCode);
    p->Write(mPresenceVector);
    p->Write(mIdentifier);
}


bool LargeDataSet::Key::operator<(const LargeDataSet::Key& key) const
{
    Update();
    key.Update();

    if(memcmp(mKey.Ptr(), key.mKey.Ptr(), mKey.Length()) < 0)
    {
        return true;
    }
    return false;
}


bool LargeDataSet::Key::operator<=(const LargeDataSet::Key& key) const
{
    Update();
    key.Update();

    if(memcmp(mKey.Ptr(), key.mKey.Ptr(), mKey.Length()) <= 0)
    {
        return true;
    }
    return false;
}


LargeDataSet::Key& LargeDataSet::Key::operator=(const LargeDataSet::Key& key)
{
    if(this != &key)
    {
        mSourceID = key.mSourceID;
        mMessageCode = key.mMessageCode;
        mPresenceVector = key.mPresenceVector;
        mIdentifier = key.mIdentifier;
        mKey = key.mKey;
    }
    return *this;
}


LargeDataSet::LargeDataSet()
{
    mCompleteFlag = false;
    mUpdateTimeMs = 0;
    mMessageCode = 0;
    mHaveLastFlag = false;
}


LargeDataSet::~LargeDataSet()
{
}


void LargeDataSet::Clear()
{
    mHeader = Header();
    mMessageCode = 0;
    mHaveLastFlag = false;
    mCompleteFlag = false;
    mStream.clear();
    mMissing.clear();
    mUpdateTimeMs = 0;
}


bool LargeDataSet::AddPacket(const Packet& message)
{
    Header header;
    UShort messageCode;
    message.SetReadPos(0);
    if(header.Read(message) &&
       message.Read(messageCode))
    {
        if(header.mControlFlag != Header::DataControl::Single)
        {
            return AddPacket(header, messageCode, message);
        }
    }
    return false;
}


bool LargeDataSet::AddPacket(const Header& header, 
                             const UShort messageCode, 
                             const Packet& packet)
{
    if(header.mControlFlag == Header::DataControl::Single || mCompleteFlag)
    {
        return false; // Not multi-packet stream.
    }
    if(mStream.size() == 0)
    {
        mMessageCode = messageCode;
        mHeader.mSourceID = header.mSourceID;
        mHeader.mDestinationID = header.mDestinationID;
        mHeader.mSequenceNumber = header.mSequenceNumber;
        /*
        for(UShort m = 0; m < header.mSequenceNumber; m++)
        {
            mMissing.insert(m);
        }
        */
        mStream[header.mSequenceNumber] = packet;
        mUpdateTimeMs = Time::GetUtcTimeMs();
        return true;
    }
    
    if(messageCode != mMessageCode || 
       header.mSourceID != mHeader.mSourceID)
    {
        return false; // Wrong data set.
    }
    /*
    std::set<UShort>::iterator m;
    // Update count of packets missing.
    if(header.mSequenceNumber > mHeader.mSequenceNumber)
    {
        for(UShort n = mHeader.mSequenceNumber + 1; n < header.mSequenceNumber; n++)
        {
            mMissing.insert(n);
        }
        mHeader.mSequenceNumber = header.mSequenceNumber;
    }
    else
    {
        m = mMissing.find(header.mSequenceNumber);
        if(m == mMissing.end())
        {
            return false; // This packet is not missing from the data set.
        }
        else
        {
            // Remove found data.
            mMissing.erase(m);
        }
    }
    */

    if(header.mControlFlag == Header::DataControl::Last)
    {
        mHaveLastFlag = true;
    }

    mStream[header.mSequenceNumber] = packet;
    if(/*mMissing.size() == 0 && */mHaveLastFlag)
    {
        // See if all the data is present.
        Stream::iterator p;
        UShort prevSequenceNumber = 0;
        bool first = true;
        mCompleteFlag = true;
        for(p = mStream.begin(); p != mStream.end(); p++)
        {
            if(first)
            {
                first = false;
            }
            else if(prevSequenceNumber + 1 != p->first)
            {
                mCompleteFlag = false;
                break;
            }
            prevSequenceNumber = p->first;
        }
    }
    mUpdateTimeMs = Time::GetUtcTimeMs();
    return true;
}


void LargeDataSet::CreateLargeDataSet(const Header& header,
                                      const UShort messageCode,
                                      const Packet& payload,
                                      Packet::List& stream,
                                      Header::List& streamHeaders,
                                      const Packet* transportHeader,
                                      const UShort maxPayloadSize,
                                      const UShort startingSequenceNumber)
{
    stream.clear();
    streamHeaders.clear();

    Header sHeader(header);
    Packet sPacket;
    
    unsigned int total = 0;                     // Total number of bytes in payload converted.
    unsigned int toWrite = 0;                   // How much data to write for a given packet.
    const unsigned char* ptr = payload.Ptr();   // Pointer to payload data to write.
    unsigned int transportHeaderSize = transportHeader ? transportHeader->Length() : 0;

    // Reserve memory for data
    sPacket.Reserve(maxPayloadSize + Header::MinSize + transportHeaderSize + USHORT_SIZE);
    sHeader.mSequenceNumber = startingSequenceNumber;

    while(total < payload.Length())
    {
        // Set the data control flags.
        if(sHeader.mSequenceNumber == startingSequenceNumber)
        {
            sHeader.mControlFlag = Header::DataControl::First;
        }
        else
        {
            sHeader.mControlFlag = Header::DataControl::Normal;
        }
        // Calculate the amount of data being written.
        toWrite = maxPayloadSize;
        if(total + toWrite > payload.Length())
        {
            toWrite = payload.Length() - total;
            // Check for last packet in sequence or if
            // this is not really a large data set.
            if(sHeader.mSequenceNumber != 0)
            {
                sHeader.mControlFlag = Header::DataControl::Last;
            }
            else
            {
                sHeader.mControlFlag = Header::DataControl::Single;
            }
        }
        sHeader.mSize = (UShort)(Header::MinSize + USHORT_SIZE + toWrite);
        // Clear the stream packet
        sPacket.Clear();
        if(transportHeaderSize > 0)
        {
            sPacket.Write(*transportHeader);  //  Transport Header.
        }
        sHeader.Write(sPacket);              //  General Transport Header.
        sPacket.Write(messageCode);          //  Message type.
        total += (unsigned int)sPacket.Write((unsigned char *)(ptr), toWrite); // Write payload data
        stream.push_back(sPacket);           //  Add to multi-packet sequence.
        streamHeaders.push_back(sHeader);    //  Add header to sequence.
        ptr += toWrite;                      //  Advance the pointer.
        sHeader.mSequenceNumber++;           //  Increase the sequence number.        
    }
}


bool LargeDataSet::MergeLargeDataSet(Header& header,
                                     UShort& messageCode,
                                     Packet& payload,
                                     const Packet::List& stream,
                                     const Packet* transportHeader)
{
    Header sHeader;
    Packet::List::const_iterator sPacket;
    UShort sequenceNumber = 0;
    std::map<UShort, const Packet*> orderedPackets;
    std::map<UShort, Header> orderedHeaders;

    // Clear data.
    header.mSize = Header::MinSize;
    header.mAckNackFlag = Header::AckNack::None;
    header.mSequenceNumber = 0;
    header.mControlFlag = Header::DataControl::Single;
    payload.Clear();
    
    unsigned int transportHeaderSize = transportHeader ? transportHeader->Length() : 0;
    for(sPacket = stream.begin(); sPacket != stream.end(); sPacket++)
    {
        // Read after the transport header (don't do validation).
        if(transportHeaderSize > 0)
        {
            sPacket->SetReadPos(transportHeaderSize);
        }
        else
        {
            sPacket->SetReadPos(0);
        }
        if(sHeader.Read(*sPacket) && sPacket->Read(messageCode))
        {
            orderedPackets[sHeader.mSequenceNumber] = &(*sPacket);
            orderedHeaders[sHeader.mSequenceNumber] = sHeader;
            /*
            if(sequenceNumber == 0)
            {
                header.mSourceID = sHeader.mSourceID;
                header.mDestinationID = sHeader.mDestinationID;
                header.mPriorityFlag = sHeader.mPriorityFlag;
            }
            // Check for out of order data.
            if(sHeader.mSequenceNumber != sequenceNumber)
            {
                payload.Clear();
                break;
            }
            // Write the data to the payload.
            payload.Write((unsigned char *)(sPacket->Ptr() + sPacket->GetReadPos()), 
                                            sHeader.mSize - (Header::MinSize + USHORT_SIZE));
            sequenceNumber++;
            */
        }
        else
        {
            payload.Clear();
            break;
        }
    }
    if(stream.size() == orderedPackets.size())
    {
        std::map<UShort, const Packet*>::iterator data;
        std::map<UShort, Header>::iterator dataHeader;

        bool first = true;
        UShort prevSequenceNumber = 0;
        for(data = orderedPackets.begin(), dataHeader = orderedHeaders.begin(); 
            data != orderedPackets.end() && dataHeader != orderedHeaders.end(); 
            data++, dataHeader++)
        {
            if(first)
            {
                if(dataHeader->second.mControlFlag != Header::DataControl::First)
                {
                    break;
                }
                header.mSourceID = dataHeader->second.mSourceID;
                header.mDestinationID = dataHeader->second.mDestinationID;
                header.mPriorityFlag = dataHeader->second.mPriorityFlag;
                first = false;
            }
            else if(prevSequenceNumber + 1 != data->first)
            {
                break;
            }
            prevSequenceNumber = header.mSequenceNumber = data->first;
            // Write the data to the payload.
            payload.Write((unsigned char *)(data->second->Ptr() + data->second->GetReadPos()), 
                                            dataHeader->second.mSize - (Header::MinSize + USHORT_SIZE));
        }
    }
    
    return payload.Length() > 0;
}


bool LargeDataSet::MergeLargeDataSet(Header& header,
                                     UShort& messageCode,
                                     Packet& payload,
                                     const Stream& stream,
                                     const Packet* transportHeader)
{
    Header sHeader;
    Stream::const_iterator sPacket;
    UShort prevSequenceNumber = 0;
    
    // Clear data.
    header.mSize = Header::MinSize;
    header.mAckNackFlag = Header::AckNack::None;
    header.mSequenceNumber = 0;
    header.mControlFlag = Header::DataControl::Single;
    payload.Clear();
    
    bool first = true;
    unsigned int transportHeaderSize = transportHeader ? transportHeader->Length() : 0;
    for(sPacket = stream.begin(); sPacket != stream.end(); sPacket++)
    {
        // Read after the transport header (don't do validation).
        if(transportHeaderSize > 0)
        {
            sPacket->second.SetReadPos(transportHeaderSize);
        }
        else
        {
            sPacket->second.SetReadPos(0);
        }
        if(sHeader.Read(sPacket->second) && sPacket->second.Read(messageCode))
        {
            if(first)
            {
                header.mSourceID = sHeader.mSourceID;
                header.mDestinationID = sHeader.mDestinationID;
                header.mPriorityFlag = sHeader.mPriorityFlag;
                first = false;
            }
            // Check for out of order data.
            else if(prevSequenceNumber + 1 != sHeader.mSequenceNumber)
            {
                payload.Clear();
                break;
            }
            // Write the data to the payload.
            payload.Write((unsigned char *)(sPacket->second.Ptr() + sPacket->second.GetReadPos()), 
                                            sHeader.mSize - (Header::MinSize + USHORT_SIZE));
            header.mSequenceNumber = prevSequenceNumber = sHeader.mSequenceNumber;
        }
        else
        {
            payload.Clear();
            break;
        }
    }
    
    return payload.Length() > 0;
}

/*  End of File */