Public Member Functions | Private Member Functions | Private Attributes

fsc_nsAttrStream Class Reference
[G_new_group]

is a nsStream who take care to store/read attribute to/from FSCDB. More...

#include <fsc_nsAttrStream.h>

Inheritance diagram for fsc_nsAttrStream:
Inheritance graph
[legend]
Collaboration diagram for fsc_nsAttrStream:
Collaboration graph
[legend]

List of all members.

Public Member Functions

 fsc_nsAttrStream (fio_RelFileTrans &a_relFile, UInt8_t a_size, ivd_RecordIDX_t a_idx)
virtual ~fsc_nsAttrStream (void)
fsc_MergeStatus_e MergeAttributes (df_SplitInfo &a_splitInfo)
 return the status of merging: already in completed inserted
void ReadSplitsFromDB ()
void Dump (ostream &os)
df_StandardAttr_tGetStandAttrRef ()
void Pack (df_Packer &a_packer)

Private Member Functions

df_AttrType_e ReadSplitStreamFromDfStream (df_Stream &a_dfStream, fsc_SplitStream_t &a_splitStream)
 NOTE! skip standard attribute in dfStream before use return the type of last read data.
bool CheckSplitSeq (fsc_SplitStream_l_i a_it, ivd_FilePosition_t a_offset)
void WriteAllSplitStreamsToDfStream (df_Stream &a_dfStream, bool a_completted)
 If attributes are complete then do not store splitInfo headers and realy merge splitted data together.

Private Attributes

 log_CLASSID_m
df_StandardAttr_t m_standardAttr
fsc_SplitStream_l m_lstOfSplitsOfAttrStream

Detailed Description

is a nsStream who take care to store/read attribute to/from FSCDB.

This class take care to collect all parts of attribute using list of fsc_splitStream_t Each splitInfo attributes are added whethet as separate fsc_splitStream_t in list or is concatenate to existing one. When all parts are collected then only one stream remain in list.

Before stored to FSC db a nsStream is filled.

df_AttrHeader_t is used whether for particular file meta data or for internal split handling.

Definition at line 90 of file fsc_nsAttrStream.h.


Constructor & Destructor Documentation

fsc_nsAttrStream::fsc_nsAttrStream ( fio_RelFileTrans a_relFile,
UInt8_t  a_size,
ivd_RecordIDX_t  a_idx 
) [inline]

Definition at line 92 of file fsc_nsAttrStream.h.

            :
            fsc_nsStream(a_relFile, a_size, a_idx)
    {
        // empty
    };

virtual fsc_nsAttrStream::~fsc_nsAttrStream ( void   )  [inline, virtual]

Definition at line 101 of file fsc_nsAttrStream.h.

{};


Member Function Documentation

bool fsc_nsAttrStream::CheckSplitSeq ( fsc_SplitStream_l_i  a_it,
ivd_FilePosition_t  a_offset 
) [private]

Definition at line 383 of file fsc_nsAttrStream.cpp.

References at_LASTSPLIT, df_AttrHeader_t::attrOfs, df_AttrHeader_t::attrSize, df_AttrHeader_t::attrType, dbg_DETAIL, log_DBG_m, log_FUNC_m, and m_lstOfSplitsOfAttrStream.

Referenced by MergeAttributes().

                                                                  {
    log_FUNC_m(CheckSplitSeq);
    fsc_SplitStream_l_i start(++a_it);
    log_DBG_m(dbg_DETAIL, "Attributes:  start ofs " << start->splitInfo.attrOfs
                        << "  size " << start->splitInfo.attrSize);
    while (a_it != m_lstOfSplitsOfAttrStream.end()) {
        df_AttrHeader_t &ah = a_it->splitInfo;
        log_DBG_m(dbg_DETAIL, "Attributes:  test (ah.attrOfs < a_offset) " 
                        << ah.attrOfs << " < " << a_offset);
        if (ah.attrOfs < a_offset) {
            log_DBG_m(dbg_DETAIL, "Attributes:  when complete will be removed " 
                        << " ofs " << ah.attrOfs << " size " << ah.attrSize);
            ++a_it;
            // this stream will be removed if complete sequence is found.
            continue;
        }
        else if (ah.attrOfs == a_offset) {
            if (ah.attrType == at_LASTSPLIT) { // found last one
                log_DBG_m(dbg_DETAIL, "Attributes:  lastSplit was found.")
                // remove unneccesary streams of other copies
                log_DBG_m(dbg_DETAIL, "Attributes:  remove from start up to " 
                        << " ofs " << a_it->splitInfo.attrOfs 
                        << " size " << a_it->splitInfo.attrSize);
                m_lstOfSplitsOfAttrStream.erase(start, a_it);
                m_lstOfSplitsOfAttrStream.erase(++a_it, m_lstOfSplitsOfAttrStream.end());
                return true;
            } 
            // check next in sequence
            else if (CheckSplitSeq(a_it, ah.attrOfs + ah.attrSize)) {
                log_DBG_m(dbg_DETAIL, "Attributes:  lastSplit was found, remove other copies."
                                << " num of splits " << m_lstOfSplitsOfAttrStream.size());
                // remove unneccesary streams of other copies
                // from start to a_it (a_it exclusive)
                log_DBG_m(dbg_DETAIL, "Attributes:  remove from start up to " 
                        << " ofs " << a_it->splitInfo.attrOfs 
                        << " size " << a_it->splitInfo.attrSize);
                m_lstOfSplitsOfAttrStream.erase(start, a_it);
                log_DBG_m(dbg_DETAIL, "Attributes:  after remove "
                                << " num of splits " << m_lstOfSplitsOfAttrStream.size());
                return true;
            }
            log_DBG_m(dbg_DETAIL, "Attributes are not complete 1.")
            return false;
        }
            log_DBG_m(dbg_DETAIL, "Attributes are not complete 2.")
        return false;
    }
            log_DBG_m(dbg_DETAIL, "Attributes are not complete 3.")
    return false; 
}

Here is the caller graph for this function:

void fsc_nsAttrStream::Dump ( ostream &  os  ) 

Reimplemented from fio_Vector.

Definition at line 487 of file fsc_nsAttrStream.cpp.

References df_StandardAttr_t::Dump(), df_LimitedStream::GetLimStreamBufRef(), fio_Vector::GetVectorIndex(), fio_Vector::GetVectorSize(), log_FUNC_m, fsc_nsStream::m_firstStream_p, m_lstOfSplitsOfAttrStream, m_standardAttr, df_StandardAttr_t::NetToHost(), NULL, and ReadSplitsFromDB().

Referenced by fsc_nsAttrMgr::Dump().

                                       {    
    log_FUNC_m(Dump)    ;
    if (m_firstStream_p != NULL) {
        os << " m_firstStream_p is not NULL " << m_firstStream_p << endl;
        string &limStream = m_firstStream_p->GetLimStreamBufRef();
        if (limStream.size() >= sizeof(df_StandardAttr_t)) {
            os << " got from first stream. " << endl;
            df_StandardAttr_t sa; 
            memcpy(&sa, limStream.data(), sizeof(df_StandardAttr_t));
            sa.NetToHost();
            sa.Dump(os);
        }
        else {
            os << " first stream is not long enought. size = " << limStream.size() << endl;
        }
    }    
    else {        
        if (GetVectorIndex() != 0) {            
            os << " got from DB. size " << GetVectorSize() 
               << " idx " << GetVectorIndex() << endl;
            ReadSplitsFromDB();
            df_StandardAttr_t sa(m_standardAttr);
            sa.NetToHost();
            sa.Dump(os);

            fsc_SplitStream_l_i iter = m_lstOfSplitsOfAttrStream.begin();
            for (; iter != m_lstOfSplitsOfAttrStream.end(); ++iter) {
                iter->Dump(os);
                //df_AttrHeader_t &ah = iter->splitInfo;
                //if (ah.attrType == at_ENDOFDATA) {
                //    os << "  Complete attributes." 
                //}
                //else {
                //    os << "  ofs " << ah.attrOfs  
                //    << "  size " << ah.attrSize 
                //    << "  lastSplit " <<  boolalpha << (ah.attrType == at_LASTSPLIT) << endl;
                //}
            }
        }
    }
}

Here is the call graph for this function:

Here is the caller graph for this function:

df_StandardAttr_t& fsc_nsAttrStream::GetStandAttrRef (  )  [inline]

Definition at line 115 of file fsc_nsAttrStream.h.

References m_lstOfSplitsOfAttrStream, m_standardAttr, and ReadSplitsFromDB().

Referenced by fsc_nsAttrMgr::GetStandAttrRef().

                                         { 
        if (m_lstOfSplitsOfAttrStream.size() == 0) {
            ReadSplitsFromDB();
        }
        return m_standardAttr;
    };

Here is the call graph for this function:

Here is the caller graph for this function:

fsc_MergeStatus_e fsc_nsAttrStream::MergeAttributes ( df_SplitInfo a_splitInfo  ) 

return the status of merging: already in completed inserted

Definition at line 199 of file fsc_nsAttrStream.cpp.

References at_FIRSTSPLIT, at_LASTSPLIT, df_AttrHeader_t::attrOfs, df_AttrHeader_t::attrSize, df_SplitInfo::attrStream, df_AttrHeader_t::attrType, CheckSplitSeq(), df_Stream::Clear(), cmn_Num2Str(), dbg_DETAIL, ie_DF_INVSEQ, df_SplitInfo::lastSplit, log_DBG_m, log_FUNC_m, m_lstOfSplitsOfAttrStream, m_standardAttr, mrgCOMPLETED, df_Stream::Read(), df_Stream::ReadReset(), ReadSplitsFromDB(), ReadSplitStreamFromDfStream(), df_SplitInfo::splitOffset, df_SplitInfo::splitSize, and WriteAllSplitStreamsToDfStream().

Referenced by fsc_nsAttrMgr::MergeAttributes().

                                                                            {
    log_FUNC_m(MergeAttributes);

    fsc_MergeStatus_e completed = mrgINSERTED;
    ReadSplitsFromDB();
    // check where to add splited stream
    fsc_SplitStream_l_i iter = m_lstOfSplitsOfAttrStream.begin();
    for (; iter != m_lstOfSplitsOfAttrStream.end(); ++iter) {
        df_AttrHeader_t &ah = iter->splitInfo;
        if (ah.attrOfs < a_splitInfo.splitOffset) {
            continue;
            
        }   
        // in case that first split size is 0
        else if (   ah.attrOfs  == a_splitInfo.splitOffset 
                 && ah.attrSize < a_splitInfo.splitSize) {
            continue;
        } 
        else if (    ah.attrOfs  == a_splitInfo.splitOffset
                 &&  ah.attrSize == a_splitInfo.splitSize
                 && (ah.attrType == at_LASTSPLIT) == a_splitInfo.lastSplit) {
            return mrgALREADY_IN; // nothing to do split already in 
        }
        break; // insert it at this point
    }
    
    // insert a_dfStream
    iter = m_lstOfSplitsOfAttrStream.insert(iter, fsc_SplitStream_t());

    a_splitInfo.attrStream.ReadReset();
    // skip standard attributes first
    int read = a_splitInfo.attrStream.Read(reinterpret_cast<char*>(&m_standardAttr), 
                                           sizeof(df_StandardAttr_t));
    if (read != sizeof(df_StandardAttr_t)) {
        throw ivd_InternalError(ie_DF_INVSEQ, 
                string("Expect standard attributes. bytes read = ") + cmn_Num2Str(read));
    }
    // fill splitAttrStream from a_splitInfo.attrStream
    log_DBG_m(dbg_DETAIL, "fill splitAttrStream from a_splitInfo.attrStream");
    ReadSplitStreamFromDfStream(a_splitInfo.attrStream, *iter);

    // check if all attributes are in.
    fsc_SplitStream_l_i it  = m_lstOfSplitsOfAttrStream.begin();
    for (; it != m_lstOfSplitsOfAttrStream.end(); ++it) {
        // check all seq that starts with first split
        if (it->splitInfo.attrType == at_FIRSTSPLIT) {
            // check split sequence endded by lastSplit
            if (CheckSplitSeq(it, it->splitInfo.attrSize)) {
                completed = mrgCOMPLETED;
                log_DBG_m(dbg_DETAIL, "Attributes completed.")
                break; 
            }
        }
    }

    // repack merged attribute stream to a_splitInfo.attrStream
    a_splitInfo.attrStream.Clear();
    WriteAllSplitStreamsToDfStream(a_splitInfo.attrStream, completed == mrgCOMPLETED);

    return completed;
}

Here is the call graph for this function:

Here is the caller graph for this function:

void fsc_nsAttrStream::Pack ( df_Packer a_packer  ) 

Definition at line 532 of file fsc_nsAttrStream.cpp.

References dbg_DETAIL, log_DBG_m, log_FUNC_m, m_lstOfSplitsOfAttrStream, and ReadSplitsFromDB().

Referenced by fsc_nsAttrMgr::Pack().

                                               {
    log_FUNC_m(Pack);
    if (m_lstOfSplitsOfAttrStream.size() == 0) {
        ReadSplitsFromDB();
    }
    if (m_lstOfSplitsOfAttrStream.size() != 1) {
        log_DBG_m(dbg_DETAIL, " File streams are not completted yet. Skip recovery of streams.");
        return;
    }
    m_lstOfSplitsOfAttrStream.front().Pack(a_packer);
}

Here is the call graph for this function:

Here is the caller graph for this function:

void fsc_nsAttrStream::ReadSplitsFromDB (  ) 

Definition at line 263 of file fsc_nsAttrStream.cpp.

References at_ENDOFDATA, at_FIRSTSPLIT, at_LASTSPLIT, at_SPLIT, cmn_Num2Str(), dbg_DETAIL, ie_DF_INVSEQ, log_DBG_m, log_FUNC_m, m_lstOfSplitsOfAttrStream, fsc_nsStream::m_recSize, m_standardAttr, df_Stream::Read(), fsc_nsStream::ReadFromDB(), df_Stream::ReadReset(), and ReadSplitStreamFromDfStream().

Referenced by Dump(), GetStandAttrRef(), MergeAttributes(), and Pack().

                                       {
    log_FUNC_m(ReadSplitsFromDB);

    if (m_lstOfSplitsOfAttrStream.size() > 0) {
        log_DBG_m(dbg_DETAIL, " Splits were already read. split no = " 
                            << m_lstOfSplitsOfAttrStream.size());
        return; // already read
    }
    df_Stream attrStream(m_recSize);

    fsc_nsStream::ReadFromDB(attrStream);

    // get standard attributes first
    attrStream.ReadReset();
    int read = attrStream.Read(reinterpret_cast<char*>(&m_standardAttr), sizeof(df_StandardAttr_t));
    if (read != sizeof(df_StandardAttr_t)) {
        throw ivd_InternalError(ie_DF_INVSEQ, 
                string("Expect standard attributes. bytes read = ") + cmn_Num2Str(read));
    }

    m_lstOfSplitsOfAttrStream.push_back(fsc_SplitStream_t());

    while (true) {
        df_AttrType_e ret;
        switch (ret = ReadSplitStreamFromDfStream(attrStream, m_lstOfSplitsOfAttrStream.back())) {
            case at_SPLIT :    
            case at_LASTSPLIT :
            case at_FIRSTSPLIT :    
                m_lstOfSplitsOfAttrStream.push_back(fsc_SplitStream_t());
                break;
            case at_ENDOFDATA :
                goto endwhile;
                break;
            default :
                throw ivd_InternalError(ie_DF_INVSEQ, 
                        string("Unknown record type " + cmn_Num2Str((int)ret)));
        }
    }
endwhile:
    if (   m_lstOfSplitsOfAttrStream.back().splitInfo.attrType == at_ENDOFDATA
        && m_lstOfSplitsOfAttrStream.back().splitData.size() == 0) {
        m_lstOfSplitsOfAttrStream.pop_back();
    }
}

Here is the call graph for this function:

Here is the caller graph for this function:

df_AttrType_e fsc_nsAttrStream::ReadSplitStreamFromDfStream ( df_Stream a_dfStream,
fsc_SplitStream_t a_splitStream 
) [private]

NOTE! skip standard attribute in dfStream before use return the type of last read data.

possible values are at_ENDOFDATA, at_SPLIT, and at_LASTSPLIT

Definition at line 310 of file fsc_nsAttrStream.cpp.

References at_ADS, at_ENDOFDATA, at_FIRSTSPLIT, at_LASTSPLIT, at_META_MASK, at_SPLIT, df_AttrHeader_t::attrOfs, df_AttrHeader_t::attrSize, df_AttrHeader_t::attrType, cmn_HexDump(), cmn_Num2Str(), dbg_DETAIL, ie_DF_INVSEQ, log_DBG_m, log_FUNC_m, df_Stream::Read(), fsc_SplitStream_t::splitData, and fsc_SplitStream_t::splitInfo.

Referenced by MergeAttributes(), and ReadSplitsFromDB().

                                                                                              {
    log_FUNC_m(ReadSplitStreamFromDfStream);

    string &splitStream = a_splitStream.splitData;
    splitStream.erase();
    log_DBG_m(dbg_DETAIL, "fill new splitStream. size after insert  " << splitStream.size()
                        << " split info must be zerro "
                   << "  ofs " << a_splitStream.splitInfo.attrOfs  
                   << "  size " << a_splitStream.splitInfo.attrSize 
                   << "  lastSplit " <<  boolalpha 
                   << (a_splitStream.splitInfo.attrType == at_LASTSPLIT));

    df_AttrHeader_t ah;
    while (true) {
        int read = a_dfStream.Read(reinterpret_cast<char*>(&ah), sizeof(df_AttrHeader_t));
        log_DBG_m(dbg_DETAIL, "R AttrHeader " << endl <<
            cmn_HexDump(&ah, sizeof(df_AttrHeader_t), sizeof(df_AttrHeader_t), false) );
        if (read != sizeof(df_AttrHeader_t)) { // no more data
            log_DBG_m(dbg_DETAIL, "End of filling split stream. stream size = " 
                                << splitStream.size());
            return at_ENDOFDATA;
        }

        df_AttrType_e at = (df_AttrType_e)ah.attrType & at_META_MASK
                         ? at_META_MASK
                         : (df_AttrType_e)ah.attrType;

        switch (at) {
            case at_META_MASK :
            case at_ADS : {
                    splitStream.append(reinterpret_cast<char*>(&ah), sizeof(df_AttrHeader_t));
                    UInt32_t currPos = splitStream.size();
                    log_DBG_m(dbg_DETAIL, " Got some data. size = " << ah.attrSize
                                        << " stream size after header append = " 
                                        << splitStream.size());
                    // NOTE! Dirty trick, expand string for new data. then fill it
                    // string doesn't know about that.
                    splitStream.resize(currPos + ah.attrSize); // reserve space for attr data
                    log_DBG_m(dbg_DETAIL, " Got some data. size = " << ah.attrSize
                                        << " written on stream pos = " <<  currPos
                                        << " stream size after resize = " << splitStream.size());
                    read = a_dfStream.Read(reinterpret_cast<char*>(&splitStream[currPos]), 
                                           ah.attrSize);
                    if (read != ah.attrSize) {
                        throw ivd_InternalError(ie_DF_INVSEQ, 
                                string("Expect part of attributeStream. Size ")
                                + cmn_Num2Str(ah.attrSize)
                                + " type " + cmn_Num2Str((int)ah.attrType));
                    }
                }
                break;
            case at_SPLIT :    
            case at_LASTSPLIT :
            case at_FIRSTSPLIT :    
                a_splitStream.splitInfo = ah;
                return static_cast<df_AttrType_e>(ah.attrType);
                break;
            case at_ENDOFDATA : 
                log_DBG_m(dbg_DETAIL, "End of filling split stream. stream size = " 
                                    << splitStream.size());
                return at_ENDOFDATA;
                break;
            default : 
                throw ivd_InternalError(ie_DF_INVSEQ, 
                        string("Expect part of attributeStream. Got unknown type ")
                        + cmn_Num2Str((int)ah.attrType));
        }
    }
}

Here is the call graph for this function:

Here is the caller graph for this function:

void fsc_nsAttrStream::WriteAllSplitStreamsToDfStream ( df_Stream a_dfStream,
bool  a_completted 
) [private]

If attributes are complete then do not store splitInfo headers and realy merge splitted data together.

Definition at line 437 of file fsc_nsAttrStream.cpp.

References at_ENDOFDATA, at_LASTSPLIT, df_AttrHeader_t::attrOfs, df_AttrHeader_t::attrSize, df_AttrHeader_t::attrType, dbg_DETAIL, log_DBG_m, log_FUNC_m, m_lstOfSplitsOfAttrStream, m_standardAttr, size, fsc_SplitStream_t::splitData, fsc_SplitStream_t::splitInfo, and df_Stream::Write().

Referenced by MergeAttributes().

                                                                               {
    log_FUNC_m(WriteAllSplitStreamsToDfStream);
    a_dfStream.Write(reinterpret_cast<char*>(&m_standardAttr), sizeof(df_StandardAttr_t));
    fsc_SplitStream_l_i iter = m_lstOfSplitsOfAttrStream.begin();
    fsc_SplitStream_l_i next = m_lstOfSplitsOfAttrStream.begin();

    bool dataMerged = false;
    for (; iter != m_lstOfSplitsOfAttrStream.end(); ++iter) {
        fsc_SplitStream_t &ss = *iter;
        UInt32_t size = ss.splitData.size();
        df_AttrHeader_t &ah = ss.splitInfo;
        log_DBG_m(dbg_DETAIL, "W to df_Stream splitInfo :" << endl 
                   << "  ofs " << ah.attrOfs  
                   << "  size " << ah.attrSize 
                   << "  lastSplit " <<  boolalpha << (ah.attrType == at_LASTSPLIT)
                   << "  attribute split size = " << size);
        // NOTE! If attribute are completted and exist next split,
        // check if next has enought data
        // then offset of next's first header, represents amount of space filled by zeroes
        // so these zeroes have to be skipped. Size is reduced by offset of data in next split.
        // header in next split must be skipped then
        if (   a_completted
            && ++next != m_lstOfSplitsOfAttrStream.end()
            && next->splitData.size() >= sizeof(df_AttrHeader_t)) {
            const df_AttrHeader_t &ah = *reinterpret_cast<const df_AttrHeader_t*>(next->splitData.data());
            size -=ah.attrOfs;
            log_DBG_m(dbg_DETAIL, " Splits will be merged offset = " << ah.attrOfs
                                << " next split ofs = " << next->splitInfo.attrOfs
                                <<  " new data size = " << size);
        }
        if (dataMerged) { // skip first header
            log_DBG_m(dbg_DETAIL, " previous and this splits are merging.");
            a_dfStream.Write(ss.splitData.data() + sizeof(df_AttrHeader_t), 
                             size - sizeof(df_AttrHeader_t));
        }
        else {
            a_dfStream.Write(ss.splitData.data(), size);
        }
        dataMerged = size != ss.splitData.size(); // to skip header in next split if true

        if (  !a_completted 
            && ss.splitInfo.attrType != at_ENDOFDATA) {
            a_dfStream.Write(reinterpret_cast<char*>(&ss.splitInfo), sizeof(df_AttrHeader_t));
        }
    }
}

Here is the call graph for this function:

Here is the caller graph for this function:


Member Data Documentation

Reimplemented from fsc_nsStream.

Definition at line 125 of file fsc_nsAttrStream.h.


The documentation for this class was generated from the following files: