#include <fsc_nsAttrStream.h>


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 76 of file fsc_nsAttrStream.h.
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_t & | GetStandAttrRef () |
| 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 |
| fsc_nsAttrStream::fsc_nsAttrStream | ( | fio_RelFileTrans & | a_relFile, | |
| UInt8_t | a_size, | |||
| ivd_RecordIDX_t | a_idx | |||
| ) | [inline] |
Definition at line 78 of file fsc_nsAttrStream.h.
00081 : 00082 fsc_nsStream(a_relFile, a_size, a_idx) 00083 { 00084 // empty 00085 };
| virtual fsc_nsAttrStream::~fsc_nsAttrStream | ( | void | ) | [inline, virtual] |
| fsc_MergeStatus_e fsc_nsAttrStream::MergeAttributes | ( | df_SplitInfo & | a_splitInfo | ) |
return the status of merging: already in completed inserted
Definition at line 185 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, mrgALREADY_IN, mrgCOMPLETED, mrgINSERTED, df_Stream::Read(), df_Stream::ReadReset(), ReadSplitsFromDB(), ReadSplitStreamFromDfStream(), df_SplitInfo::splitOffset, df_SplitInfo::splitSize, and WriteAllSplitStreamsToDfStream().
Referenced by fsc_nsAttrMgr::MergeAttributes().
00185 { 00186 log_FUNC_m(MergeAttributes); 00187 00188 fsc_MergeStatus_e completed = mrgINSERTED; 00189 ReadSplitsFromDB(); 00190 // check where to add splited stream 00191 fsc_SplitStream_l_i iter = m_lstOfSplitsOfAttrStream.begin(); 00192 for (; iter != m_lstOfSplitsOfAttrStream.end(); ++iter) { 00193 df_AttrHeader_t &ah = iter->splitInfo; 00194 if (ah.attrOfs < a_splitInfo.splitOffset) { 00195 continue; 00196 00197 } 00198 // in case that first split size is 0 00199 else if ( ah.attrOfs == a_splitInfo.splitOffset 00200 && ah.attrSize < a_splitInfo.splitSize) { 00201 continue; 00202 } 00203 else if ( ah.attrOfs == a_splitInfo.splitOffset 00204 && ah.attrSize == a_splitInfo.splitSize 00205 && (ah.attrType == at_LASTSPLIT) == a_splitInfo.lastSplit) { 00206 return mrgALREADY_IN; // nothing to do split already in 00207 } 00208 break; // insert it at this point 00209 } 00210 00211 // insert a_dfStream 00212 iter = m_lstOfSplitsOfAttrStream.insert(iter, fsc_SplitStream_t()); 00213 00214 a_splitInfo.attrStream.ReadReset(); 00215 // skip standard attributes first 00216 int read = a_splitInfo.attrStream.Read(reinterpret_cast<char*>(&m_standardAttr), 00217 sizeof(df_StandardAttr_t)); 00218 if (read != sizeof(df_StandardAttr_t)) { 00219 throw ivd_InternalError(ie_DF_INVSEQ, 00220 string("Expect standard attributes. bytes read = ") + cmn_Num2Str(read)); 00221 } 00222 // fill splitAttrStream from a_splitInfo.attrStream 00223 log_DBG_m(dbg_DETAIL, "fill splitAttrStream from a_splitInfo.attrStream"); 00224 ReadSplitStreamFromDfStream(a_splitInfo.attrStream, *iter); 00225 00226 // check if all attributes are in. 00227 fsc_SplitStream_l_i it = m_lstOfSplitsOfAttrStream.begin(); 00228 for (; it != m_lstOfSplitsOfAttrStream.end(); ++it) { 00229 // check all seq that starts with first split 00230 if (it->splitInfo.attrType == at_FIRSTSPLIT) { 00231 // check split sequence endded by lastSplit 00232 if (CheckSplitSeq(it, it->splitInfo.attrSize)) { 00233 completed = mrgCOMPLETED; 00234 log_DBG_m(dbg_DETAIL, "Attributes completed.") 00235 break; 00236 } 00237 } 00238 } 00239 00240 // repack merged attribute stream to a_splitInfo.attrStream 00241 a_splitInfo.attrStream.Clear(); 00242 WriteAllSplitStreamsToDfStream(a_splitInfo.attrStream, completed == mrgCOMPLETED); 00243 00244 return completed; 00245 }


| void fsc_nsAttrStream::ReadSplitsFromDB | ( | ) |
Definition at line 249 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().
00249 { 00250 log_FUNC_m(ReadSplitsFromDB); 00251 00252 if (m_lstOfSplitsOfAttrStream.size() > 0) { 00253 log_DBG_m(dbg_DETAIL, " Splits were already read. split no = " 00254 << m_lstOfSplitsOfAttrStream.size()); 00255 return; // already read 00256 } 00257 df_Stream attrStream(m_recSize); 00258 00259 fsc_nsStream::ReadFromDB(attrStream); 00260 00261 // get standard attributes first 00262 attrStream.ReadReset(); 00263 int read = attrStream.Read(reinterpret_cast<char*>(&m_standardAttr), sizeof(df_StandardAttr_t)); 00264 if (read != sizeof(df_StandardAttr_t)) { 00265 throw ivd_InternalError(ie_DF_INVSEQ, 00266 string("Expect standard attributes. bytes read = ") + cmn_Num2Str(read)); 00267 } 00268 00269 m_lstOfSplitsOfAttrStream.push_back(fsc_SplitStream_t()); 00270 00271 while (true) { 00272 df_AttrType_e ret; 00273 switch (ret = ReadSplitStreamFromDfStream(attrStream, m_lstOfSplitsOfAttrStream.back())) { 00274 case at_SPLIT : 00275 case at_LASTSPLIT : 00276 case at_FIRSTSPLIT : 00277 m_lstOfSplitsOfAttrStream.push_back(fsc_SplitStream_t()); 00278 break; 00279 case at_ENDOFDATA : 00280 goto endwhile; 00281 break; 00282 default : 00283 throw ivd_InternalError(ie_DF_INVSEQ, 00284 string("Unknown record type " + cmn_Num2Str((int)ret))); 00285 } 00286 } 00287 endwhile: 00288 if ( m_lstOfSplitsOfAttrStream.back().splitInfo.attrType == at_ENDOFDATA 00289 && m_lstOfSplitsOfAttrStream.back().splitData.size() == 0) { 00290 m_lstOfSplitsOfAttrStream.pop_back(); 00291 } 00292 }


| void fsc_nsAttrStream::Dump | ( | ostream & | os | ) |
Reimplemented from fio_Vector.
Definition at line 473 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().
00473 { 00474 log_FUNC_m(Dump) ; 00475 if (m_firstStream_p != NULL) { 00476 os << " m_firstStream_p is not NULL " << m_firstStream_p << endl; 00477 string &limStream = m_firstStream_p->GetLimStreamBufRef(); 00478 if (limStream.size() >= sizeof(df_StandardAttr_t)) { 00479 os << " got from first stream. " << endl; 00480 df_StandardAttr_t sa; 00481 memcpy(&sa, limStream.data(), sizeof(df_StandardAttr_t)); 00482 sa.NetToHost(); 00483 sa.Dump(os); 00484 } 00485 else { 00486 os << " first stream is not long enought. size = " << limStream.size() << endl; 00487 } 00488 } 00489 else { 00490 if (GetVectorIndex() != 0) { 00491 os << " got from DB. size " << GetVectorSize() 00492 << " idx " << GetVectorIndex() << endl; 00493 ReadSplitsFromDB(); 00494 df_StandardAttr_t sa(m_standardAttr); 00495 sa.NetToHost(); 00496 sa.Dump(os); 00497 00498 fsc_SplitStream_l_i iter = m_lstOfSplitsOfAttrStream.begin(); 00499 for (; iter != m_lstOfSplitsOfAttrStream.end(); ++iter) { 00500 iter->Dump(os); 00501 //df_AttrHeader_t &ah = iter->splitInfo; 00502 //if (ah.attrType == at_ENDOFDATA) { 00503 // os << " Complete attributes." 00504 //} 00505 //else { 00506 // os << " ofs " << ah.attrOfs 00507 // << " size " << ah.attrSize 00508 // << " lastSplit " << boolalpha << (ah.attrType == at_LASTSPLIT) << endl; 00509 //} 00510 } 00511 } 00512 } 00513 }


| df_StandardAttr_t& fsc_nsAttrStream::GetStandAttrRef | ( | ) | [inline] |
Definition at line 101 of file fsc_nsAttrStream.h.
References m_lstOfSplitsOfAttrStream, m_standardAttr, and ReadSplitsFromDB().
Referenced by fsc_nsAttrMgr::GetStandAttrRef().
00101 { 00102 if (m_lstOfSplitsOfAttrStream.size() == 0) { 00103 ReadSplitsFromDB(); 00104 } 00105 return m_standardAttr; 00106 };


| void fsc_nsAttrStream::Pack | ( | df_Packer & | a_packer | ) |
Definition at line 518 of file fsc_nsAttrStream.cpp.
References dbg_DETAIL, log_DBG_m, log_FUNC_m, m_lstOfSplitsOfAttrStream, and ReadSplitsFromDB().
Referenced by fsc_nsAttrMgr::Pack().
00518 { 00519 log_FUNC_m(Pack); 00520 if (m_lstOfSplitsOfAttrStream.size() == 0) { 00521 ReadSplitsFromDB(); 00522 } 00523 if (m_lstOfSplitsOfAttrStream.size() != 1) { 00524 log_DBG_m(dbg_DETAIL, " File streams are not completted yet. Skip recovery of streams."); 00525 return; 00526 } 00527 m_lstOfSplitsOfAttrStream.front().Pack(a_packer); 00528 }


| 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 296 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().
00297 { 00298 log_FUNC_m(ReadSplitStreamFromDfStream); 00299 00300 string &splitStream = a_splitStream.splitData; 00301 splitStream.erase(); 00302 log_DBG_m(dbg_DETAIL, "fill new splitStream. size after insert " << splitStream.size() 00303 << " split info must be zerro " 00304 << " ofs " << a_splitStream.splitInfo.attrOfs 00305 << " size " << a_splitStream.splitInfo.attrSize 00306 << " lastSplit " << boolalpha 00307 << (a_splitStream.splitInfo.attrType == at_LASTSPLIT)); 00308 00309 df_AttrHeader_t ah; 00310 while (true) { 00311 int read = a_dfStream.Read(reinterpret_cast<char*>(&ah), sizeof(df_AttrHeader_t)); 00312 log_DBG_m(dbg_DETAIL, "R AttrHeader " << endl << 00313 cmn_HexDump(&ah, sizeof(df_AttrHeader_t), sizeof(df_AttrHeader_t), false) ); 00314 if (read != sizeof(df_AttrHeader_t)) { // no more data 00315 log_DBG_m(dbg_DETAIL, "End of filling split stream. stream size = " 00316 << splitStream.size()); 00317 return at_ENDOFDATA; 00318 } 00319 00320 df_AttrType_e at = (df_AttrType_e)ah.attrType & at_META_MASK 00321 ? at_META_MASK 00322 : (df_AttrType_e)ah.attrType; 00323 00324 switch (at) { 00325 case at_META_MASK : 00326 case at_ADS : { 00327 splitStream.append(reinterpret_cast<char*>(&ah), sizeof(df_AttrHeader_t)); 00328 UInt32_t currPos = splitStream.size(); 00329 log_DBG_m(dbg_DETAIL, " Got some data. size = " << ah.attrSize 00330 << " stream size after header append = " 00331 << splitStream.size()); 00332 // NOTE! Dirty trick, expand string for new data. then fill it 00333 // string doesn't know about that. 00334 splitStream.resize(currPos + ah.attrSize); // reserve space for attr data 00335 log_DBG_m(dbg_DETAIL, " Got some data. size = " << ah.attrSize 00336 << " written on stream pos = " << currPos 00337 << " stream size after resize = " << splitStream.size()); 00338 read = a_dfStream.Read(reinterpret_cast<char*>(&splitStream[currPos]), 00339 ah.attrSize); 00340 if (read != ah.attrSize) { 00341 throw ivd_InternalError(ie_DF_INVSEQ, 00342 string("Expect part of attributeStream. Size ") 00343 + cmn_Num2Str(ah.attrSize) 00344 + " type " + cmn_Num2Str((int)ah.attrType)); 00345 } 00346 } 00347 break; 00348 case at_SPLIT : 00349 case at_LASTSPLIT : 00350 case at_FIRSTSPLIT : 00351 a_splitStream.splitInfo = ah; 00352 return static_cast<df_AttrType_e>(ah.attrType); 00353 break; 00354 case at_ENDOFDATA : 00355 log_DBG_m(dbg_DETAIL, "End of filling split stream. stream size = " 00356 << splitStream.size()); 00357 return at_ENDOFDATA; 00358 break; 00359 default : 00360 throw ivd_InternalError(ie_DF_INVSEQ, 00361 string("Expect part of attributeStream. Got unknown type ") 00362 + cmn_Num2Str((int)ah.attrType)); 00363 } 00364 } 00365 }


| bool fsc_nsAttrStream::CheckSplitSeq | ( | fsc_SplitStream_l_i | a_it, | |
| ivd_FilePosition_t | a_offset | |||
| ) | [private] |
Definition at line 369 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().
00370 { 00371 log_FUNC_m(CheckSplitSeq); 00372 fsc_SplitStream_l_i start(++a_it); 00373 log_DBG_m(dbg_DETAIL, "Attributes: start ofs " << start->splitInfo.attrOfs 00374 << " size " << start->splitInfo.attrSize); 00375 while (a_it != m_lstOfSplitsOfAttrStream.end()) { 00376 df_AttrHeader_t &ah = a_it->splitInfo; 00377 log_DBG_m(dbg_DETAIL, "Attributes: test (ah.attrOfs < a_offset) " 00378 << ah.attrOfs << " < " << a_offset); 00379 if (ah.attrOfs < a_offset) { 00380 log_DBG_m(dbg_DETAIL, "Attributes: when complete will be removed " 00381 << " ofs " << ah.attrOfs << " size " << ah.attrSize); 00382 ++a_it; 00383 // this stream will be removed if complete sequence is found. 00384 continue; 00385 } 00386 else if (ah.attrOfs == a_offset) { 00387 if (ah.attrType == at_LASTSPLIT) { // found last one 00388 log_DBG_m(dbg_DETAIL, "Attributes: lastSplit was found.") 00389 // remove unneccesary streams of other copies 00390 log_DBG_m(dbg_DETAIL, "Attributes: remove from start up to " 00391 << " ofs " << a_it->splitInfo.attrOfs 00392 << " size " << a_it->splitInfo.attrSize); 00393 m_lstOfSplitsOfAttrStream.erase(start, a_it); 00394 m_lstOfSplitsOfAttrStream.erase(++a_it, m_lstOfSplitsOfAttrStream.end()); 00395 return true; 00396 } 00397 // check next in sequence 00398 else if (CheckSplitSeq(a_it, ah.attrOfs + ah.attrSize)) { 00399 log_DBG_m(dbg_DETAIL, "Attributes: lastSplit was found, remove other copies." 00400 << " num of splits " << m_lstOfSplitsOfAttrStream.size()); 00401 // remove unneccesary streams of other copies 00402 // from start to a_it (a_it exclusive) 00403 log_DBG_m(dbg_DETAIL, "Attributes: remove from start up to " 00404 << " ofs " << a_it->splitInfo.attrOfs 00405 << " size " << a_it->splitInfo.attrSize); 00406 m_lstOfSplitsOfAttrStream.erase(start, a_it); 00407 log_DBG_m(dbg_DETAIL, "Attributes: after remove " 00408 << " num of splits " << m_lstOfSplitsOfAttrStream.size()); 00409 return true; 00410 } 00411 log_DBG_m(dbg_DETAIL, "Attributes are not complete 1.") 00412 return false; 00413 } 00414 log_DBG_m(dbg_DETAIL, "Attributes are not complete 2.") 00415 return false; 00416 } 00417 log_DBG_m(dbg_DETAIL, "Attributes are not complete 3.") 00418 return false; 00419 }

| 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 423 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().
00424 { 00425 log_FUNC_m(WriteAllSplitStreamsToDfStream); 00426 a_dfStream.Write(reinterpret_cast<char*>(&m_standardAttr), sizeof(df_StandardAttr_t)); 00427 fsc_SplitStream_l_i iter = m_lstOfSplitsOfAttrStream.begin(); 00428 fsc_SplitStream_l_i next = m_lstOfSplitsOfAttrStream.begin(); 00429 00430 bool dataMerged = false; 00431 for (; iter != m_lstOfSplitsOfAttrStream.end(); ++iter) { 00432 fsc_SplitStream_t &ss = *iter; 00433 UInt32_t size = ss.splitData.size(); 00434 df_AttrHeader_t &ah = ss.splitInfo; 00435 log_DBG_m(dbg_DETAIL, "W to df_Stream splitInfo :" << endl 00436 << " ofs " << ah.attrOfs 00437 << " size " << ah.attrSize 00438 << " lastSplit " << boolalpha << (ah.attrType == at_LASTSPLIT) 00439 << " attribute split size = " << size); 00440 // NOTE! If attribute are completted and exist next split, 00441 // check if next has enought data 00442 // then offset of next's first header, represents amount of space filled by zeroes 00443 // so these zeroes have to be skipped. Size is reduced by offset of data in next split. 00444 // header in next split must be skipped then 00445 if ( a_completted 00446 && ++next != m_lstOfSplitsOfAttrStream.end() 00447 && next->splitData.size() >= sizeof(df_AttrHeader_t)) { 00448 const df_AttrHeader_t &ah = *reinterpret_cast<const df_AttrHeader_t*>(next->splitData.data()); 00449 size -=ah.attrOfs; 00450 log_DBG_m(dbg_DETAIL, " Splits will be merged offset = " << ah.attrOfs 00451 << " next split ofs = " << next->splitInfo.attrOfs 00452 << " new data size = " << size); 00453 } 00454 if (dataMerged) { // skip first header 00455 log_DBG_m(dbg_DETAIL, " previous and this splits are merging."); 00456 a_dfStream.Write(ss.splitData.data() + sizeof(df_AttrHeader_t), 00457 size - sizeof(df_AttrHeader_t)); 00458 } 00459 else { 00460 a_dfStream.Write(ss.splitData.data(), size); 00461 } 00462 dataMerged = size != ss.splitData.size(); // to skip header in next split if true 00463 00464 if ( !a_completted 00465 && ss.splitInfo.attrType != at_ENDOFDATA) { 00466 a_dfStream.Write(reinterpret_cast<char*>(&ss.splitInfo), sizeof(df_AttrHeader_t)); 00467 } 00468 } 00469 }


fsc_nsAttrStream::log_CLASSID_m [private] |
Definition at line 113 of file fsc_nsAttrStream.h.
Referenced by Dump(), GetStandAttrRef(), MergeAttributes(), ReadSplitsFromDB(), and WriteAllSplitStreamsToDfStream().
Definition at line 115 of file fsc_nsAttrStream.h.
Referenced by CheckSplitSeq(), Dump(), GetStandAttrRef(), MergeAttributes(), Pack(), ReadSplitsFromDB(), and WriteAllSplitStreamsToDfStream().
1.5.6