FIX8::FilePersister Class Reference

#include <persist.hpp>

Inheritance diagram for FIX8::FilePersister:

Public Member Functions
	FilePersister (unsigned rotnum=0)
	Ctor. More...
virtual F8API	~FilePersister ()
	Dtor. More...
virtual F8API bool	initialise (const f8String &dbDir, const f8String &dbFname, bool purge=false)
virtual F8API bool	put (const unsigned seqnum, const f8String &what)
virtual F8API bool	put (const unsigned sender_seqnum, const unsigned target_seqnum)
virtual F8API bool	get (const unsigned seqnum, f8String &to) const
virtual F8API unsigned	get (const unsigned from, const unsigned to, Session &session, bool(Session::*)(const Session::SequencePair &with, Session::RetransmissionContext &rctx)) const
virtual F8API unsigned	get_last_seqnum (unsigned &to) const
virtual F8API bool	get (unsigned &sender_seqnum, unsigned &target_seqnum) const
virtual F8API unsigned	find_nearest_highest_seqnum (const unsigned requested, const unsigned last) const
Public Member Functions inherited from FIX8::Persister
	Persister ()=default
	Ctor. More...
virtual	~Persister ()
	Dtor. More...
	Persister (const Persister &)=delete
Persister &	operator= (const Persister &)=delete
virtual bool	put (const f8String &key, const f8String &what)
virtual bool	get (const f8String &key, f8String &to) const
virtual bool	del (const f8String &key)
virtual bool	purge ()
virtual void	stop ()
	Stop the persister thread. More...

Private Types
using	Index = std::map< uint32_t, Prec >

Private Attributes
f8String	_dbFname
f8String	_dbIname
int	_fod
int	_iod
unsigned	_rotnum
bool	_wasCreated
Index	_index

Additional Inherited Members
Public Types inherited from FIX8::Persister
enum	{ MaxMsgLen = FIX8_MAX_MSG_LENGTH }
	Maximum length of persisted FIX message. More...
Protected Attributes inherited from FIX8::Persister
bool	_opened = false

Detailed Description

Definition at line 391 of file persist.hpp.

Member Typedef Documentation

using FIX8::FilePersister::Index = std::map<uint32_t, Prec>

private

Definition at line 398 of file persist.hpp.

Constructor & Destructor Documentation

FIX8::FilePersister::FilePersister ( unsigned  rotnum = 0 )

inline

Ctor.

Definition at line 403 of file persist.hpp.

: _fod(-1), _iod(-1), _rotnum(rotnum), _wasCreated() {}

FilePersister::~FilePersister ( )

virtual

Dtor.

Definition at line 156 of file filepersist.cpp.

{
   close(_fod);
   close(_iod);
}

Member Function Documentation

unsigned FilePersister::find_nearest_highest_seqnum ( const unsigned  requested, const unsigned  last  ) const

virtual

Find the nearest highest sequence number from the sequence to last provided.

Parameters

requested	sequence number to start
last	highest sequence

Returns

the nearest sequence number or 0 if not found

Implements FIX8::Persister.

Definition at line 339 of file filepersist.cpp.

{
   if (last)
   {
      for (unsigned startseqnum(requested); startseqnum <= last; ++startseqnum)
      {
         Index::const_iterator itr(_index.find(startseqnum));
         if (itr != _index.end())
            return itr->first;
      }
   }

   return 0;
}

bool FilePersister::get ( const unsigned  seqnum, f8String &  to  ) const

virtual

Retrieve a persisted message.

Parameters

seqnum	sequence number of message
to	target message string

Returns

true on success

Implements FIX8::Persister.

Definition at line 310 of file filepersist.cpp.

References FIX8_MAX_MSG_LENGTH, glout_error, and glout_warn.

{
   if (!_opened || !seqnum || _index.empty())
      return false;
   Index::const_iterator itr(_index.find(seqnum));
   if (itr == _index.end())
   {
      glout_warn << "Warning: index does not contain seqnum: " << seqnum << " in: " << _dbIname;
      return false;
   }

   if (lseek(_fod, itr->second._offset, SEEK_SET) < 0)
   {
      glout_error << "Error: could not seek to correct index location for get: " << _dbFname;
      return false;
   }

   char buff[FIX8_MAX_MSG_LENGTH];
   if (read (_fod, buff, itr->second._size) != itr->second._size)
   {
      glout_error << "Error: could not read message record for seqnum " << seqnum << " from: " << _dbFname;
      return false;
   }

   to.assign(buff, itr->second._size);
   return true;
}

unsigned FilePersister::get ( const unsigned  from, const unsigned  to, Session &  session, bool(Session::*)(const Session::SequencePair &with, Session::RetransmissionContext &rctx)  callback  ) const

virtual

Retrieve a range of persisted messages.

Parameters

from	start at sequence number
to	end sequence number
session	session containing callback method
callback	method to call with each retrieved message

Returns

number of messages retrieved

Implements FIX8::Persister.

Definition at line 169 of file filepersist.cpp.

References FIX8::Session::RetransmissionContext::_no_more_records, FIX8_MAX_MSG_LENGTH, FIX8::Session::get_next_send_seq(), glout_debug, glout_error, and glout_info.

{
   unsigned last_seq(0);
   get_last_seqnum(last_seq);
   unsigned recs_sent(0), startSeqNum(find_nearest_highest_seqnum (from, last_seq));
   const unsigned finish(to == 0 ? last_seq : to);
   Session::RetransmissionContext rctx(from, to, session.get_next_send_seq());

   if (!startSeqNum || from > finish)
   {
      glout_info << "No records found";
      rctx._no_more_records = true;
      (session.*callback)(Session::SequencePair(0, ""), rctx);
      return 0;
   }

   Index::const_iterator itr(_index.find(startSeqNum));
   if (itr != _index.end())
   {
      char buff[FIX8_MAX_MSG_LENGTH];

      do
      {
         if (!itr->first || itr->first > finish)
            break;
         if (lseek(_fod, itr->second._offset, SEEK_SET) < 0)
         {
            glout_error << "Error: could not seek to correct index location for get: " << _dbFname;
            break;
         }

         if (read (_fod, buff, itr->second._size) != itr->second._size)
         {
            glout_error << "Error: could not read message record for seqnum " << itr->first << " from: " << _dbFname;
            break;
         }

         Session::SequencePair txresult(itr->first, f8String(buff, itr->second._size));
         ++recs_sent;
         if (!(session.*callback)(txresult, rctx))
         {
            glout_debug << "Retransmission callback signalled an error, not sending any more records from: " << _dbFname;
            break;
         }
      }
      while(++itr != _index.end());

      rctx._no_more_records = true;
      (session.*callback)(Session::SequencePair(0, ""), rctx);
   }
   else
   {
      glout_error << "record not found (" << startSeqNum << ')';
   }

   return recs_sent;
}

bool FilePersister::get ( unsigned &  sender_seqnum, unsigned &  target_seqnum  ) const

virtual

Retrieve a sequence control record.

Parameters

sender_seqnum	sequence number of last sent message
target_seqnum	sequence number of last received message

Returns

true on success

Implements FIX8::Persister.

Definition at line 286 of file filepersist.cpp.

References glout_error, and glout_warn.

{
   if (!_opened)
      return false;

   if (_index.empty())
   {
      glout_warn << "Warning: index is empty: " << _dbIname;
      return false;
   }

   Index::const_iterator itr(_index.find(0));
   if (itr == _index.end())
   {
      glout_error << "Error: index does not contain control record: " << _dbIname;
      return false;
   }

   sender_seqnum = itr->second._offset;
   target_seqnum = itr->second._size;
   return true;
}

unsigned FilePersister::get_last_seqnum ( unsigned &  to ) const

virtual

Retrieve sequence number of last peristed message.

Parameters

to	target sequence number

Returns

sequence number of last peristed message on success

Implements FIX8::Persister.

Definition at line 163 of file filepersist.cpp.

{
   return sequence = _index.empty() ? 0 : _index.rbegin()->first;
}

bool FilePersister::initialise ( const f8String &  dbDir, const f8String &  dbFname, bool  purge = false  )

virtual

Open existing database or create new database.

Parameters

dbDir	database directory
dbFname	database name
purge	if true, empty database if found

Returns

true on success

Definition at line 45 of file filepersist.cpp.

References FIX8::IPrec::_seq, FIX8::CheckAddTrailingSlash(), FIX8::exist(), glout_error, glout_info, glout_warn, FIX8::Logger::max_rotation, and O_BINARY.

{
   if (_opened)
      return true;

   f8String odbdir(dbDir);
   ostringstream ostr;
   ostr << CheckAddTrailingSlash(odbdir) << dbFname;
   _dbFname = ostr.str();
   ostr << ".idx";
   _dbIname = ostr.str();

   bool nof;

   if ((nof = !exist(_dbFname)) || purge)
   {
      if (purge)
      {
         if (_rotnum > 0)
         {
            vector<string> dblst, idxlst;
            dblst.push_back(_dbFname);
            idxlst.push_back(_dbIname);

            for (unsigned ii(0); ii < _rotnum && ii < Logger::max_rotation; ++ii)
            {
               ostringstream ostr;
               ostr << _dbFname << '.' << (ii + 1);
               dblst.push_back(ostr.str());
               ostr << ".idx";
               idxlst.push_back(ostr.str());
            }

            for (unsigned ii(_rotnum); ii; --ii)
            {
               rename (dblst[ii - 1].c_str(), dblst[ii].c_str());   // ignore errors
               rename (idxlst[ii - 1].c_str(), idxlst[ii].c_str()); // ignore errors
            }
         }
      }

      if ((_fod = open(_dbFname.c_str(), O_RDWR | O_CREAT | O_TRUNC | O_BINARY, 0600)) < 0)
      {
         glout_error << "Error: creating database: " << _dbFname << " (" << strerror(errno) << ')';
         return false;
      }
      if ((_iod = open(_dbIname.c_str(), O_RDWR | O_CREAT | O_TRUNC | O_BINARY, 0600)) < 0)
      {
         glout_error << "Error: creating database index: " << _dbIname << " (" << strerror(errno) << ')';
         return false;
      }

      _wasCreated = true;

      if (purge && !nof)
      {
         glout_info << (_rotnum ? "Rotated and purged perist db" : "Purged perist db");
      }
   }
   else
   {
      if ((_fod = open(_dbFname.c_str(), O_RDWR | O_BINARY)) < 0)
      {
         glout_error << "Error: opening existing database: " << _dbFname << " (" << strerror(errno) << ')';
         return false;
      }
      if ((_iod = open(_dbIname.c_str(), O_RDWR | O_BINARY)) < 0)
      {
         glout_error << "Error: opening existing database index: " << _dbIname << " (" << strerror(errno) << ')';
         return false;
      }

      IPrec iprec;
      while (true)
      {
         const ssize_t blrd(read(_iod, static_cast<void *>(&iprec), sizeof(IPrec)));
         if (blrd < 0)
         {
            glout_error << "Error: reading existing database index: " << _dbIname << " (" << strerror(errno) << ')';
            return false;
         }
         else if (blrd == 0)
            break; // eof

         if (iprec._seq == 0)
         {
            glout_info << iprec;
         }

         if (!_index.insert({iprec._seq, iprec._prec}).second)
         {
            glout_warn << "Warning: inserting index record into database index: " << _dbIname << " (" << iprec << "). Ignoring.";
         }
      }

      if (_index.size())
      {
         glout_info << "Database " << _dbFname << " indexed " << _index.size() << " records.";
      }

      unsigned last;
      if (get_last_seqnum(last))
      {
         glout_info << _dbFname << ": Last sequence is " << last;
      }
   }

   return _opened = true;
}

bool FilePersister::put ( const unsigned  seqnum, const f8String &  what  )

virtual

Persist a message.

Parameters

seqnum	sequence number of message
what	message string

Returns

true on success

Implements FIX8::Persister.

Definition at line 249 of file filepersist.cpp.

References FIX8::IPrec::_prec, and glout_error.

{
   if (!_opened || !seqnum)
      return false;

   if (_index.find(seqnum) != _index.end())
   {
      glout_error << "Error: seqnum " << seqnum << " already persisted in: " << _dbIname;
      return false;
   }
   if (lseek(_iod, 0, SEEK_END) < 0)
   {
      glout_error << "Error: could not seek to index end for seqnum persitence: " << _dbIname;
      return false;
   }
   off_t offset;
   if ((offset = lseek(_fod, 0, SEEK_END)) < 0)
   {
      glout_error << "Error: could not seek to end for seqnum persitence: " << _dbFname;
      return false;
   }
   IPrec iprec(seqnum, offset, static_cast<unsigned>(what.size()));
   if (write (_iod, static_cast<void *>(&iprec), sizeof(IPrec)) != sizeof(IPrec))
   {
      glout_error << "Error: could not write index record for seqnum " << seqnum << " to: " << _dbIname;
      return false;
   }
   if (write (_fod, what.data(), static_cast<unsigned>(what.size())) != static_cast<ssize_t>(what.size()))
   {
      glout_error << "Error: could not write record for seqnum " << seqnum << " to: " << _dbFname;
      return false;
   }

   return _index.insert({seqnum, iprec._prec}).second;
}

bool FilePersister::put ( const unsigned  sender_seqnum, const unsigned  target_seqnum  )

virtual

Persist a sequence control record.

Parameters

sender_seqnum	sequence number of last sent message
target_seqnum	sequence number of last received message

Returns

true on success

Implements FIX8::Persister.

Definition at line 229 of file filepersist.cpp.

References FIX8::IPrec::_prec, and glout_error.

{
   if (!_opened)
      return false;
   IPrec iprec(0, sender_seqnum, target_seqnum);
   Index::iterator itr(_index.find(0));
   if (itr == _index.end())
      _index.insert({0, iprec._prec});
   else
      itr->second = iprec._prec;

   if (lseek(_iod, 0, SEEK_SET) < 0)
   {
      glout_error << "Error: could not seek to 0 for seqnum persitence: " << _dbIname;
      return false;
   }
   return write (_iod, static_cast<void *>(&iprec), sizeof(IPrec)) == sizeof(IPrec);
}

Member Data Documentation

f8String FIX8::FilePersister::_dbFname

private

Definition at line 393 of file persist.hpp.

f8String FIX8::FilePersister::_dbIname

private

Definition at line 393 of file persist.hpp.

int FIX8::FilePersister::_fod

private

Definition at line 394 of file persist.hpp.

Index FIX8::FilePersister::_index

private

Definition at line 399 of file persist.hpp.

int FIX8::FilePersister::_iod

private

Definition at line 394 of file persist.hpp.

unsigned FIX8::FilePersister::_rotnum

private

Definition at line 395 of file persist.hpp.

bool FIX8::FilePersister::_wasCreated

private

Definition at line 396 of file persist.hpp.

---

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

• persist.hpp
• filepersist.cpp