FIX8::FIXReader Class Reference

Fix message reader. More...

#include <connection.hpp>

Inheritance diagram for FIX8::FIXReader:

Public Member Functions
	FIXReader (Poco::Net::StreamSocket *sock, Session &session, const ProcessModel pmodel=pm_pipeline)
virtual	~FIXReader ()
	Dtor. More...
virtual void	start ()
	Start the processing threads. More...
virtual void	quit ()
	Stop the processing threads and quit. More...
virtual void	stop ()
	Send a message to the processing method instructing it to quit. More...
virtual F8API int	execute (f8_thread_cancellation_token &cancellation_token)
int	join ()
F8API void	set_preamble_sz ()
	Calculate the length of the Fix message preamble, e.g. "8=FIX.4.4^A9=". More...
bool	is_socket_error () const
bool	poll (const Poco::Timespan &ts=Poco::Timespan()) const
f8_thread_cancellation_token &	callback_cancellation_token ()
Public Member Functions inherited from FIX8::AsyncSocket< f8String >
	AsyncSocket (Poco::Net::StreamSocket *sock, Session &session, const ProcessModel pmodel=pm_pipeline)
virtual	~AsyncSocket ()=default
	Dtor. More...
size_t	queued () const
bool	started () const
int	operator() ()
virtual void	request_stop ()
	Start the processing thread. More...
Poco::Net::StreamSocket *	socket ()
int	join ()
f8_thread_cancellation_token &	cancellation_token ()

Private Types
enum	{ _max_msg_len = FIX8_MAX_MSG_LENGTH, _chksum_sz = 7 }

Private Member Functions
F8API int	callback_processor ()
bool	read (f8String &to)
int	sockRead (char *where, const size_t sz)

Private Attributes
f8_atomic< bool >	_socket_error
f8_thread< FIXReader >	_callback_thread
f8_thread_cancellation_token	_callback_cancellation_token
size_t	_bg_sz

Additional Inherited Members
Protected Attributes inherited from FIX8::AsyncSocket< f8String >
coroutine	_coro
Poco::Net::StreamSocket *	_sock
f8_concurrent_queue< f8String >	_msg_queue
Session &	_session
ProcessModel	_pmodel
f8_thread_cancellation_token	_cancellation_token
volatile bool	_started
f8_mutex	_start_mutex

Detailed Description

Fix message reader.

Definition at line 134 of file connection.hpp.

Member Enumeration Documentation

anonymous enum

private

Enumerator
_max_msg_len
_chksum_sz

Definition at line 136 of file connection.hpp.

{ _max_msg_len = FIX8_MAX_MSG_LENGTH, _chksum_sz = 7 };

Constructor & Destructor Documentation

FIX8::FIXReader::FIXReader ( Poco::Net::StreamSocket *  sock, Session &  session, const ProcessModel  pmodel = pm_pipeline  )

inline

Ctor.

Parameters

sock	connected socket
session	session
pmodel	process model

Definition at line 245 of file connection.hpp.

References set_preamble_sz().

      : AsyncSocket<f8String>(sock, session, pmodel), _callback_thread(std::ref(*this), &FIXReader::callback_processor)
#if FIX8_EXPERIMENTAL_BUFFERED_SOCKET_READ
      , _read_buffer(), _read_buffer_rptr(_read_buffer), _read_buffer_wptr(_read_buffer)
#endif
      , _bg_sz()
   {
      set_preamble_sz();
   }

virtual FIX8::FIXReader::~FIXReader ( )

inlinevirtual

Dtor.

Definition at line 256 of file connection.hpp.

References join(), and stop().

   {
      stop();
      join();
   }

Member Function Documentation

f8_thread_cancellation_token& FIX8::FIXReader::callback_cancellation_token ( )

inline

Definition at line 335 of file connection.hpp.

References _callback_cancellation_token.

{ return _callback_cancellation_token; }

int FIXReader::callback_processor ( )

private

Process messages from inbound queue, calls session process method.

Returns

number of messages processed

Definition at line 164 of file connection.cpp.

References scout_info, and scout_warn.

{
   int processed(0), ignored(0);

   for (; !_cancellation_token && !_session.is_shutdown();)
   {
      f8String *msg_ptr(0);
#if (FIX8_MPMC_SYSTEM == FIX8_MPMC_TBB)
      f8String msg;
      _msg_queue.pop (msg); // will block
      if (msg.empty())  // means exit
         break;
      msg_ptr = &msg;
#else
      _msg_queue.pop (msg_ptr); // will block
      if (msg_ptr->empty())  // means exit
         break;
#endif

      if (!_session.process(*msg_ptr))
      {
         scout_warn << "Unhandled message: " << *msg_ptr;
         ++ignored;
      }
      else
         ++processed;
#if (FIX8_MPMC_SYSTEM == FIX8_MPMC_FF)
      _msg_queue.release(msg_ptr);
#endif
   }

   scout_info << "FIXReaderCallback: " << processed << " messages processed, " << ignored << " ignored";

   return 0;
}

int FIXReader::execute ( f8_thread_cancellation_token &  cancellation_token )

virtual

Reader thread method. Reads messages and places them on the queue for processing. Supports pipelined, threaded and coroutine process models.

Returns

0 on success

Reimplemented from FIX8::AsyncSocket< f8String >.

Definition at line 45 of file connection.cpp.

References coro_yield, FIX8::pm_coro, FIX8::pm_pipeline, reenter, scout_error, scout_info, scout_warn, FIX8::States::st_session_terminated, and FIX8::f8Exception::what().

Referenced by FIX8::Connection::reader_execute().

{
   unsigned processed(0), dropped(0), invalid(0);
   int retval(0);

   if (_pmodel == pm_coro)
   {
      f8String msg;

      try
      {
         reenter(_coro)
         {
            while (!_session.is_shutdown())
            {
               if (poll()) // it is a fair assumption that there is a complete message waiting to be read
               {
                  if (read(msg)) // if our assumption is wrong we could block here
                  {
                     if (!_session.process(msg))
                     {
                        scout_warn << "Unhandled message: " << msg;
                        ++invalid;
                     }
                     else
                        ++processed;
                  }
                  else
                     ++invalid;
               }
               coro_yield;
            }
         }
      }
      catch (Poco::Net::NetException& e)
      {
         scout_error << e.what();
         _session.do_state_change(States::st_session_terminated);
         retval = -1;
      }
      catch (PeerResetConnection& e)
      {
         scout_error << e.what();
         _session.do_state_change(States::st_session_terminated);
         //_session.stop();
         retval = -1;
      }
      catch (exception& e)
      {
         scout_error << e.what();
         retval = -2;
         ++invalid;
      }
      catch (...)
      {
         scout_error << "reader(coro): Unknown exception";
         retval = -2;
      }
   }
   else
   {
      while (!cancellation_token && !_session.is_shutdown())
      {
         f8String msg;

         try
         {
            if (read(msg)) // will block
            {
               if (_pmodel == pm_pipeline)
               {
                  if (!_msg_queue.try_push (msg))
                  {
                     scout_warn << "FIXReader: message queue is full";
                     ++dropped;
                  }
                  else
                     ++processed;
               }
               else // _pmodel == pm_thread
               {
                  if (!_session.process(msg))
                  {
                     scout_warn << "Unhandled message: " << msg;
                     ++invalid;
                  }
                  else
                     ++processed;
               }
            }
            else
               ++invalid;
         }
         catch (exception& e)
         {
            scout_error << e.what();
            if (!_session.is_shutdown())
               _session.do_state_change(States::st_session_terminated);
            retval = -1;
            ++invalid;
            break;
         }
      }
      scout_info << "FIXReader: " << processed << " messages processed, " << dropped << " dropped, "
         << invalid << " invalid";
      if (retval && errno)
      {
         scout_warn << "socket error=" << strerror(errno) << '(' << errno << ')';
      }

      {
         f8_scoped_lock guard(_start_mutex);
         _started = false;
      }
   }
   return retval;
}

bool FIX8::FIXReader::is_socket_error ( ) const

inline

Check to see if the socket is in error

Returns

true if there was a socket error

Definition at line 325 of file connection.hpp.

References _socket_error.

Referenced by FIX8::Connection::is_socket_error().

{ return _socket_error; }

int FIX8::FIXReader::join ( )

inline

Wait till writer thread has finished.

Returns

0 on success

Definition at line 318 of file connection.hpp.

References FIX8::AsyncSocket< f8String >::_pmodel, FIX8::AsyncSocket< T >::join(), and FIX8::pm_coro.

Referenced by FIX8::Connection::join(), and ~FIXReader().

{ return _pmodel != pm_coro ? AsyncSocket<f8String>::join() : -1; }

bool FIX8::FIXReader::poll ( const Poco::Timespan &  ts = Poco::Timespan() ) const

inline

Check to see if there is any data waiting to be read

Parameters

ts

timeout

Returns

true of data ready

Definition at line 330 of file connection.hpp.

References FIX8::AsyncSocket< f8String >::_sock.

Referenced by FIX8::Connection::reader_poll().

   {
      return _sock->poll(ts, Poco::Net::Socket::SELECT_READ);
   }

virtual void FIX8::FIXReader::quit ( )

inlinevirtual

Stop the processing threads and quit.

Reimplemented from FIX8::AsyncSocket< f8String >.

Definition at line 276 of file connection.hpp.

References FIX8::AsyncSocket< f8String >::_pmodel, FIX8::_f8_threadcore::join(), FIX8::pm_coro, FIX8::pm_pipeline, FIX8::AsyncSocket< T >::quit(), and FIX8::f8_thread< T >::request_stop().

   {
      if (_pmodel == pm_pipeline)
      {
         _callback_thread.request_stop();
         _callback_thread.join();
      }
      if (_pmodel != pm_coro)
         AsyncSocket<f8String>::quit();
   }

bool FIXReader::read ( f8String &  to )

private

Read a Fix message. Throws InvalidBodyLength, IllegalMessage.

Parameters

to	string to place message in

Returns

true on success

Definition at line 207 of file connection.cpp.

References FIX8::default_field_separator(), FIX8::MessageBase::extract_element(), FILE_LINE, FIX8_MAX_FLD_LENGTH, and FIX8::MAX_MSGTYPE_FIELD_LEN().

{
   char msg_buf[_max_msg_len] {};
   int result(sockRead(msg_buf, _bg_sz));

   if (result == static_cast<int>(_bg_sz))
   {
      char bt;
      size_t offs(_bg_sz);
      do // get the last chrs of bodylength and ^A
      {
         if (sockRead(&bt, 1) != 1)
            return false;
         if (!isdigit(bt) && bt != default_field_separator)
            throw IllegalMessage(msg_buf, FILE_LINE);
         msg_buf[offs++] = bt;
      }
      while (bt != default_field_separator && offs < _max_msg_len);
      to.assign(msg_buf, offs);

      char tag[MAX_MSGTYPE_FIELD_LEN], val[FIX8_MAX_FLD_LENGTH];
      unsigned result;
      if ((result = MessageBase::extract_element(to.data(), static_cast<unsigned>(to.size()), tag, val)))
      {
         if (*tag != '8')
            throw IllegalMessage(to, FILE_LINE);

         if (_session.get_ctx()._beginStr.compare(val))  // invalid FIX version
            throw InvalidVersion(string(val));

         if ((result = MessageBase::extract_element(to.data() + result, static_cast<unsigned>(to.size()) - result, tag, val)))
         {
            if (*tag != '9')
               throw IllegalMessage(to, FILE_LINE);

            const unsigned mlen(fast_atoi<unsigned>(val));
            if (mlen == 0 || mlen > _max_msg_len - _bg_sz - _chksum_sz) // invalid msglen
               throw InvalidBodyLength(mlen);

            // read the body
            if ((result = sockRead(msg_buf, mlen) != static_cast<int>(mlen)))
               return false;

            // read the checksum
            if ((result = sockRead(msg_buf + mlen, _chksum_sz) != static_cast<int>(_chksum_sz)))
               return false;

            to.append(msg_buf, mlen + _chksum_sz);
            _session.update_received();
            //string ts;
            //cerr << GetTimeAsStringMS(ts, &_session.get_last_received(), 9) << endl;
            return true;
         }
      }

      throw IllegalMessage(to, FILE_LINE);
   }

   return false;
}

void FIXReader::set_preamble_sz

(

)

Calculate the length of the Fix message preamble, e.g. "8=FIX.4.4^A9=".

Definition at line 201 of file connection.cpp.

Referenced by FIXReader().

{
   _bg_sz = 2 + _session.get_ctx()._beginStr.size() + 1 + 3;
}

int FIX8::FIXReader::sockRead ( char *  where, const size_t  sz  )

inlineprivate

Read bytes from the socket layer, throws PeerResetConnection.

Parameters

where	buffer to place bytes in
sz	number of bytes to read

Returns

number of bytes read

Definition at line 217 of file connection.hpp.

References FIX8::AsyncSocket< f8String >::_sock.

   {
      unsigned remaining(static_cast<unsigned>(sz)), rddone(0);
      while (remaining > 0)
      {
         const int rdSz(_sock->receiveBytes(where + rddone, remaining));
         if (rdSz <= 0)
         {
            if (errno == EAGAIN
#if defined EWOULDBLOCK && EAGAIN != EWOULDBLOCK
               || errno == EWOULDBLOCK
#endif
            )
               continue;
            throw PeerResetConnection("sockRead: connection gone");
         }
         rddone += rdSz;
         remaining -= rdSz;
      }
      return rddone;
   }

virtual void FIX8::FIXReader::start ( )

inlinevirtual

Start the processing threads.

Reimplemented from FIX8::AsyncSocket< f8String >.

Definition at line 263 of file connection.hpp.

References FIX8::AsyncSocket< f8String >::_pmodel, FIX8::pm_coro, FIX8::pm_pipeline, FIX8::AsyncSocket< T >::start(), and FIX8::f8_thread< T >::start().

   {
      _socket_error = false;
      if (_pmodel != pm_coro)
         AsyncSocket<f8String>::start();
      if (_pmodel == pm_pipeline)
      {
         if (_callback_thread.start())
            _socket_error = true;
      }
   }

virtual void FIX8::FIXReader::stop ( )

inlinevirtual

Send a message to the processing method instructing it to quit.

Definition at line 288 of file connection.hpp.

References FIX8::AsyncSocket< f8String >::_msg_queue, FIX8::AsyncSocket< f8String >::_pmodel, FIX8::AsyncSocket< f8String >::_start_mutex, FIX8::AsyncSocket< f8String >::_started, glout_warn, FIX8::pm_coro, FIX8::pm_pipeline, FIX8::AsyncSocket< T >::request_stop(), and FIX8::f8_thread< T >::request_stop().

Referenced by ~FIXReader().

   {
      if (_started)
      {
         f8_scoped_lock guard(_start_mutex);
         if (_started)
         {
            if (_pmodel == pm_pipeline)
            {
               const f8String from;
               _msg_queue.try_push(from);
               _callback_thread.request_stop();
            }
            if (_pmodel != pm_coro)
               AsyncSocket<f8String>::request_stop();
         }
      }
      else
      {
         glout_warn << "FIXReader: AsyncSocket already stopped.";
      }
   }

Member Data Documentation

size_t FIX8::FIXReader::_bg_sz

private

Definition at line 151 of file connection.hpp.

f8_thread_cancellation_token FIX8::FIXReader::_callback_cancellation_token

private

Definition at line 140 of file connection.hpp.

Referenced by callback_cancellation_token().

f8_thread<FIXReader> FIX8::FIXReader::_callback_thread

private

Definition at line 139 of file connection.hpp.

f8_atomic<bool> FIX8::FIXReader::_socket_error

private

Definition at line 137 of file connection.hpp.

Referenced by is_socket_error().

---

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

• connection.hpp
• connection.cpp