FIX::ThreadedSocketConnection Class Reference

Encapsulates a socket file descriptor (multi-threaded). More...

#include <ThreadedSocketConnection.h>

Inheritance diagram for FIX::ThreadedSocketConnection:

Collaboration diagram for FIX::ThreadedSocketConnection:

Public Types
typedef std::set< SessionID >	Sessions

Public Member Functions
	ThreadedSocketConnection (int s, Sessions sessions, Log *pLog)
	ThreadedSocketConnection (const SessionID &, int s, const std::string &address, short port, Log *pLog, const std::string &sourceAddress="", short sourcePort=0)
virtual	~ThreadedSocketConnection ()
Session *	getSession () const
int	getSocket () const
bool	connect ()
void	disconnect ()
bool	read ()

Private Member Functions
bool	readMessage (std::string &msg) throw ( SocketRecvFailed )
void	processStream ()
bool	send (const std::string &)
bool	setSession (const std::string &msg)
Private Member Functions inherited from FIX::Responder
virtual	~Responder ()

Private Attributes
int	m_socket
char	m_buffer [BUFSIZ]
std::string	m_address
int	m_port
std::string	m_sourceAddress
int	m_sourcePort
Log *	m_pLog
Parser	m_parser
Sessions	m_sessions
Session *	m_pSession
bool	m_disconnect
fd_set	m_fds

Detailed Description

Encapsulates a socket file descriptor (multi-threaded).

Definition at line 44 of file ThreadedSocketConnection.h.

Member Typedef Documentation

Sessions

typedef std::set<SessionID> FIX::ThreadedSocketConnection::Sessions

Definition at line 47 of file ThreadedSocketConnection.h.

Constructor & Destructor Documentation

ThreadedSocketConnection() [1/2]

FIX::ThreadedSocketConnection::ThreadedSocketConnection	(	int	s,
		Sessions	sessions,
		Log *	pLog )

Definition at line 34 of file ThreadedSocketConnection.cpp.

: m_socket( s ), m_pLog( pLog ),
  m_sessions( sessions ), m_pSession( 0 ),
  m_disconnect( false )
{
  FD_ZERO( &m_fds );
  FD_SET( m_socket, &m_fds );
}

References m_disconnect, m_fds, m_pLog, m_pSession, m_sessions, and m_socket.

ThreadedSocketConnection() [2/2]

FIX::ThreadedSocketConnection::ThreadedSocketConnection	(	const SessionID &	sessionID,
		int	s,
		const std::string &	address,
		short	port,
		Log *	pLog,
		const std::string &	sourceAddress = "",
		short	sourcePort = 0 )

Definition at line 44 of file ThreadedSocketConnection.cpp.

  : m_socket( s ), m_address( address ), m_port( port ),
    m_sourceAddress( sourceAddress ), m_sourcePort( sourcePort ),
    m_pLog( pLog ),
    m_pSession( Session::lookupSession( sessionID ) ),
    m_disconnect( false )
{
  FD_ZERO( &m_fds );
  FD_SET( m_socket, &m_fds );
  if ( m_pSession ) m_pSession->setResponder( this );
}

References m_address, m_disconnect, m_fds, m_pLog, m_port, m_pSession, m_socket, m_sourceAddress, and m_sourcePort.

~ThreadedSocketConnection()

FIX::ThreadedSocketConnection::~ThreadedSocketConnection ( )

virtual

Definition at line 60 of file ThreadedSocketConnection.cpp.

{
  if ( m_pSession )
  {
    m_pSession->setResponder( 0 );
    Session::unregisterSession( m_pSession->getSessionID() );
  }
}

References m_pSession, and FIX::Session::unregisterSession().

Member Function Documentation

connect()

bool FIX::ThreadedSocketConnection::connect

(

)

Definition at line 82 of file ThreadedSocketConnection.cpp.

{
  // do the bind in the thread as name resolution may block
  if ( !m_sourceAddress.empty() || m_sourcePort )
    socket_bind( m_socket, m_sourceAddress.c_str(), m_sourcePort );
 
  return socket_connect(getSocket(), m_address.c_str(), m_port) >= 0;
}

References getSocket(), m_address, m_port, m_socket, m_sourceAddress, m_sourcePort, FIX::socket_bind(), and FIX::socket_connect().

Referenced by FIX::ThreadedSocketInitiator::socketThread().

disconnect()

void FIX::ThreadedSocketConnection::disconnect ( )

virtual

Implements FIX::Responder.

Definition at line 91 of file ThreadedSocketConnection.cpp.

{  
  m_disconnect = true;
  socket_close( m_socket );
}

References m_disconnect, m_socket, and FIX::socket_close().

Referenced by FIX::ThreadedSocketInitiator::doConnect(), processStream(), read(), and FIX::ThreadedSocketInitiator::socketThread().

getSession()

Session * FIX::ThreadedSocketConnection::getSession ( ) const

inline

Definition at line 56 of file ThreadedSocketConnection.h.

{ return m_pSession; }

References m_pSession.

Referenced by FIX::ThreadedSocketInitiator::socketThread().

getSocket()

int FIX::ThreadedSocketConnection::getSocket ( ) const

inline

Definition at line 57 of file ThreadedSocketConnection.h.

{ return m_socket; }

References m_socket.

Referenced by connect(), FIX::ThreadedSocketAcceptor::socketConnectionThread(), and FIX::ThreadedSocketInitiator::socketThread().

processStream()

void FIX::ThreadedSocketConnection::processStream ( )

private

Definition at line 156 of file ThreadedSocketConnection.cpp.

{
  std::string msg;
  while( readMessage(msg) )
  {
    if ( !m_pSession )
    {
      if ( !setSession( msg ) )
      { disconnect(); continue; }
    }
    try
    {
      m_pSession->next( msg, UtcTimeStamp() );
    }
    catch( InvalidMessage& )
    {
      if( !m_pSession->isLoggedOn() )
      {
        disconnect();
        return;
      }
    }
  }
}

References disconnect(), m_pSession, readMessage(), and setSession().

Referenced by read().

read()

bool FIX::ThreadedSocketConnection::read

(

)

Definition at line 97 of file ThreadedSocketConnection.cpp.

{
  struct timeval timeout = { 1, 0 };
  fd_set readset = m_fds;
 
  try
  {
    // Wait for input (1 second timeout)
    int result = select( 1 + m_socket, &readset, 0, 0, &timeout );
 
    if( result > 0 ) // Something to read
    {
      // We can read without blocking
      ssize_t size = socket_recv( m_socket, m_buffer, sizeof(m_buffer) );
      if ( size <= 0 ) { throw SocketRecvFailed( size ); }
      m_parser.addToStream( m_buffer, size );
    }
    else if( result == 0 && m_pSession ) // Timeout
    {
      m_pSession->next();
    }
    else if( result < 0 ) // Error
    {
      throw SocketRecvFailed( result );
    }
 
    processStream();
    return true;
  }
  catch ( SocketRecvFailed& e )
  {
    if( m_disconnect )
      return false;
 
    if( m_pSession )
    {
      m_pSession->getLog()->onEvent( e.what() );
      m_pSession->disconnect();
    }
    else
    {
      disconnect();
    }
 
    return false;
  }
}

References disconnect(), m_buffer, m_disconnect, m_fds, m_parser, m_pSession, m_socket, processStream(), and FIX::socket_recv().

Referenced by FIX::ThreadedSocketAcceptor::socketConnectionThread(), and FIX::ThreadedSocketInitiator::socketThread().

readMessage()

bool FIX::ThreadedSocketConnection::readMessage ( std::string & msg ) throw ( SocketRecvFailed )

private

Definition at line 145 of file ThreadedSocketConnection.cpp.

{
  try
  {
    return m_parser.readFixMessage( msg );
  }
  catch ( MessageParseError& ) {}
  return true;
}

References m_parser.

Referenced by processStream().

send()

bool FIX::ThreadedSocketConnection::send ( const std::string & msg )

privatevirtual

Implements FIX::Responder.

Definition at line 69 of file ThreadedSocketConnection.cpp.

{
  int totalSent = 0;
  while(totalSent < (int)msg.length())
  {
    ssize_t sent = socket_send( m_socket, msg.c_str() + totalSent, msg.length() );
    if(sent < 0) return false;
    totalSent += sent;
  }
 
  return true;
}

References m_socket, and FIX::socket_send().

setSession()

bool FIX::ThreadedSocketConnection::setSession ( const std::string & msg )

private

Definition at line 181 of file ThreadedSocketConnection.cpp.

{
  m_pSession = Session::lookupSession( msg, true );
  if ( !m_pSession ) 
  {
    if( m_pLog )
    {
      m_pLog->onEvent( "Session not found for incoming message: " + msg );
      m_pLog->onIncoming( msg );
    }
    return false;
  }
 
  SessionID sessionID = m_pSession->getSessionID();
  m_pSession = 0;
 
  // see if the session frees up within 5 seconds
  for( int i = 1; i <= 5; i++ )
  {
    if( !Session::isSessionRegistered( sessionID ) )
      m_pSession = Session::registerSession( sessionID );
    if( m_pSession ) break;
    process_sleep( 1 );
  }
 
  if ( !m_pSession ) 
    return false;
  if ( m_sessions.find(m_pSession->getSessionID()) == m_sessions.end() )
    return false;
 
  m_pSession->setResponder( this );
  return true;
}

References FIX::Session::isSessionRegistered(), FIX::Session::lookupSession(), m_pLog, m_pSession, m_sessions, FIX::process_sleep(), and FIX::Session::registerSession().

Referenced by processStream().

Member Data Documentation

m_address

std::string FIX::ThreadedSocketConnection::m_address

private

Definition at line 71 of file ThreadedSocketConnection.h.

Referenced by connect(), and ThreadedSocketConnection().

m_buffer

char FIX::ThreadedSocketConnection::m_buffer[BUFSIZ]

private

Definition at line 69 of file ThreadedSocketConnection.h.

Referenced by read().

m_disconnect

bool FIX::ThreadedSocketConnection::m_disconnect

private

Definition at line 80 of file ThreadedSocketConnection.h.

Referenced by disconnect(), read(), ThreadedSocketConnection(), and ThreadedSocketConnection().

m_fds

fd_set FIX::ThreadedSocketConnection::m_fds

private

Definition at line 81 of file ThreadedSocketConnection.h.

Referenced by read(), ThreadedSocketConnection(), and ThreadedSocketConnection().

m_parser

Parser FIX::ThreadedSocketConnection::m_parser

private

Definition at line 77 of file ThreadedSocketConnection.h.

Referenced by read(), and readMessage().

m_pLog

Log* FIX::ThreadedSocketConnection::m_pLog

private

Definition at line 76 of file ThreadedSocketConnection.h.

Referenced by setSession(), ThreadedSocketConnection(), and ThreadedSocketConnection().

m_port

int FIX::ThreadedSocketConnection::m_port

private

Definition at line 72 of file ThreadedSocketConnection.h.

Referenced by connect(), and ThreadedSocketConnection().

m_pSession

Session* FIX::ThreadedSocketConnection::m_pSession

private

Definition at line 79 of file ThreadedSocketConnection.h.

Referenced by getSession(), processStream(), read(), setSession(), ThreadedSocketConnection(), ThreadedSocketConnection(), and ~ThreadedSocketConnection().

m_sessions

Sessions FIX::ThreadedSocketConnection::m_sessions

private

Definition at line 78 of file ThreadedSocketConnection.h.

Referenced by setSession(), and ThreadedSocketConnection().

m_socket

int FIX::ThreadedSocketConnection::m_socket

private

Definition at line 68 of file ThreadedSocketConnection.h.

Referenced by connect(), disconnect(), getSocket(), read(), send(), ThreadedSocketConnection(), and ThreadedSocketConnection().

m_sourceAddress

std::string FIX::ThreadedSocketConnection::m_sourceAddress

private

Definition at line 73 of file ThreadedSocketConnection.h.

Referenced by connect(), and ThreadedSocketConnection().

m_sourcePort

int FIX::ThreadedSocketConnection::m_sourcePort

private

Definition at line 74 of file ThreadedSocketConnection.h.

Referenced by connect(), and ThreadedSocketConnection().

---

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

• ThreadedSocketConnection.h
• ThreadedSocketConnection.cpp