FIX8::FIXWriter Class Reference

Fix message writer. More...

#include <connection.hpp>

Inheritance diagram for FIX8::FIXWriter:

Public Member Functions
	FIXWriter (Poco::Net::StreamSocket *sock, Session &session, const ProcessModel pmodel=pm_pipeline)
virtual	~FIXWriter ()
	Dtor. More...
bool	write (Message *from, bool destroy)
size_t	write_batch (const std::vector< Message * > &msgs, bool destroy)
int	join ()
bool	write (Message &from)
bool	poll (const Poco::Timespan &ts=Poco::Timespan()) const
int	send (const char *data, size_t remaining, bool nb=false)
virtual void	start ()
	Start the processing threads. 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)
virtual void	quit ()
	Stop the processing threads and quit. More...
Public Member Functions inherited from FIX8::AsyncSocket< Message * >
	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 Attributes
f8_spin_lock	_con_spl

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

Detailed Description

Fix message writer.

Definition at line 340 of file connection.hpp.

Constructor & Destructor Documentation

FIX8::FIXWriter::FIXWriter ( 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 349 of file connection.hpp.

      : AsyncSocket<Message *>(sock, session, pmodel) {}

virtual FIX8::FIXWriter::~FIXWriter ( )

inlinevirtual

Dtor.

Definition at line 353 of file connection.hpp.

References quit().

   {
      quit();
   }

Member Function Documentation

int FIXWriter::execute ( f8_thread_cancellation_token &  cancellation_token )

virtual

Writer thread method. Reads messages from the queue and sends them over the socket.

Returns

0 on success

Reimplemented from FIX8::AsyncSocket< Message * >.

Definition at line 269 of file connection.cpp.

References scout_error, scout_info, and FIX8::f8Exception::what().

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

{
   int result(0), processed(0), invalid(0);

   while (!cancellation_token && !_session.is_shutdown())
   {
      try
      {
         Message *inmsg(0);
         _msg_queue.pop (inmsg); // will block
         if (!inmsg)
            break;
         unique_ptr<Message> msg(inmsg);
         _session.send_process(msg.get());
         ++processed;
      }
      catch (PeerResetConnection& e)
      {
         scout_error << e.what();
         result = -1;
         break;
      }
      catch (Poco::Net::NetException& e)
      {
         scout_error << e.what();
         ++invalid;
         break;
      }
      catch (exception& e)
      {
         scout_error << e.what();
         ++invalid;
         break; //?
      }
   }

   scout_info << "FIXWriter: " << processed << " messages processed, " << invalid << " invalid";

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

int FIX8::FIXWriter::join ( )

inline

Wait till writer thead has finished.

Returns

0 on success

Definition at line 412 of file connection.hpp.

References FIX8::AsyncSocket< Message * >::_pmodel, FIX8::AsyncSocket< T >::join(), and FIX8::pm_pipeline.

{ return _pmodel == pm_pipeline ? AsyncSocket<Message *>::join() : -1; }

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

inline

Check to see if a write would block

Parameters

ts

timeout

Returns

true if a write would block

Definition at line 428 of file connection.hpp.

References FIX8::AsyncSocket< Message * >::_sock.

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

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

virtual void FIX8::FIXWriter::quit ( )

inlinevirtual

Stop the processing threads and quit.

Reimplemented from FIX8::AsyncSocket< Message * >.

Definition at line 498 of file connection.hpp.

References FIX8::AsyncSocket< Message * >::_pmodel, FIX8::pm_pipeline, and FIX8::AsyncSocket< T >::quit().

Referenced by ~FIXWriter().

   {
      if (_pmodel == pm_pipeline)
         AsyncSocket<Message *>::quit();
   }

int FIX8::FIXWriter::send ( const char *  data, size_t  remaining, bool  nb = false  )

inline

Send message over socket.

Parameters

data	char * buffer to send
remaining	number of bytes
nb	- if true, don't block

Returns

number of bytes sent

Definition at line 438 of file connection.hpp.

References FIX8::AsyncSocket< Message * >::_sock.

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

   {
      unsigned wrdone(0);

      while (remaining > 0)
      {
         const int wrtSz(_sock->sendBytes(data + wrdone, static_cast<int>(remaining)));
         if (wrtSz < 0)
         {
            if (errno == EAGAIN
#if defined EWOULDBLOCK && EAGAIN != EWOULDBLOCK
               || errno == EWOULDBLOCK
#endif
            )
            {
               if (nb)
                  return wrdone;
               continue;
            }
            throw PeerResetConnection("send: connection gone");
         }

         wrdone += wrtSz;
         remaining -= wrtSz;
      }

      return wrdone;
   }

virtual void FIX8::FIXWriter::start ( )

inlinevirtual

Start the processing threads.

Reimplemented from FIX8::AsyncSocket< Message * >.

Definition at line 468 of file connection.hpp.

References FIX8::AsyncSocket< Message * >::_pmodel, FIX8::pm_pipeline, and FIX8::AsyncSocket< T >::start().

   {
      if (_pmodel == pm_pipeline)
         AsyncSocket<Message *>::start();
   }

virtual void FIX8::FIXWriter::stop ( )

inlinevirtual

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

Definition at line 475 of file connection.hpp.

References FIX8::AsyncSocket< Message * >::_msg_queue, FIX8::AsyncSocket< Message * >::_pmodel, FIX8::AsyncSocket< Message * >::_start_mutex, FIX8::AsyncSocket< Message * >::_started, glout_warn, FIX8::pm_pipeline, and FIX8::AsyncSocket< T >::request_stop().

   {
      if (_started)
      {
         f8_scoped_lock guard(_start_mutex);
         if (_started)
         {
            _msg_queue.try_push(0);
            if (_pmodel == pm_pipeline)
               AsyncSocket::request_stop();
         }
      }
      else
      {
         glout_warn << "FIXWriter: AsyncSocket already stopped.";
      }
   }

bool FIX8::FIXWriter::write ( Message *  from, bool  destroy  )

inline

Place Fix message on outbound message queue.

Parameters

from	message to send
destroy	if true delete after send

Returns

true in success

Definition at line 362 of file connection.hpp.

References FIX8::AsyncSocket< Message * >::_msg_queue, FIX8::AsyncSocket< Message * >::_pmodel, FIX8::AsyncSocket< Message * >::_session, FIX8::pm_pipeline, and FIX8::Session::send_process().

Referenced by FIX8::Connection::write(), and write_batch().

   {
      if (_pmodel == pm_pipeline) // pipeline mode ignores destroy flag
         return _msg_queue.try_push(from);
      f8_scoped_spin_lock guard(_con_spl);
      if (destroy)
      {
         std::unique_ptr<Message> msg(from);
         return _session.send_process(msg.get());
      }
      return _session.send_process(from);
   }

bool FIX8::FIXWriter::write ( Message &  from )

inline

Send Fix message directly

Parameters

from	message to send

Returns

true in success

Definition at line 417 of file connection.hpp.

References FIX8::AsyncSocket< Message * >::_pmodel, FIX8::AsyncSocket< Message * >::_session, FIX8::pm_pipeline, and FIX8::Session::send_process().

   {
      if (_pmodel == pm_pipeline) // not permitted if pipeling
         throw f8Exception("cannot send message directly if pipelining");
      f8_scoped_spin_lock guard(_con_spl);
      return _session.send_process(&from);
   }

size_t FIX8::FIXWriter::write_batch ( const std::vector< Message * > &  msgs, bool  destroy  )

inline

Place Fix messages on outbound message queue as a single batch.

Parameters

msgs	messages to send
destroy	if true delete after send

Returns

count of messages written

Todo:

: need assert on result==msgs.size()

Definition at line 379 of file connection.hpp.

References FIX8::AsyncSocket< Message * >::_msg_queue, FIX8::AsyncSocket< Message * >::_pmodel, FIX8::AsyncSocket< Message * >::_session, FIX8::pm_pipeline, FIX8::Session::send_process(), FIX8::Message::set_end_of_batch(), and write().

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

   {
      if (msgs.empty())
         return 0;
      if (msgs.size() == 1)
         return write(msgs.front(), destroy) ? 1 : 0;
      size_t result(0);
      f8_scoped_spin_lock guard(_con_spl);
      for (std::vector<Message *>::const_iterator itr(msgs.begin()), eitr(msgs.end()), litr(eitr-1); itr != eitr; ++itr)
      {
         Message* msg = *itr;
         msg->set_end_of_batch(itr == litr);
         if (_pmodel == pm_pipeline) // pipeline mode ignores destroy flag
         {
            _msg_queue.try_push(msg);
            ++result;
            continue;
         }
         if (_session.send_process(msg))
            ++result;
      }
      if (destroy && _pmodel != pm_pipeline)
      {
         for (std::vector<Message *>::const_iterator itr(msgs.begin()), eitr(msgs.end()); itr != eitr; ++itr)
         {
            std::unique_ptr<Message> smsg(*itr);
         }
      }
      return result;
   }

Member Data Documentation

f8_spin_lock FIX8::FIXWriter::_con_spl

private

Definition at line 342 of file connection.hpp.

---

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

• connection.hpp
• connection.cpp