FIX::ThreadedSocketConnection Class Reference

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

`#include <ThreadedSocketConnection.h>`

Inheritance diagram for FIX::ThreadedSocketConnection:
[legend]
Collaboration diagram for FIX::ThreadedSocketConnection:
[legend]

## 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)
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
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 35 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 );
}
FIX::ThreadedSocketConnection::m_pSession
Session * m_pSession
Definition: ThreadedSocketConnection.h:76
FIX::ThreadedSocketConnection::m_socket
int m_socket
Definition: ThreadedSocketConnection.h:67
FIX::ThreadedSocketConnection::m_fds
fd_set m_fds
Definition: ThreadedSocketConnection.h:78
FIX::ThreadedSocketConnection::m_pLog
Log * m_pLog
Definition: ThreadedSocketConnection.h:73
FIX::ThreadedSocketConnection::m_sessions
Sessions m_sessions
Definition: ThreadedSocketConnection.h:75
FIX::ThreadedSocketConnection::m_disconnect
bool m_disconnect
Definition: ThreadedSocketConnection.h:77

## ThreadedSocketConnection() [2/2]

FIX::ThreadedSocketConnection::ThreadedSocketConnection | ( | const SessionID & | sessionID,
|  | int | s,
|  | const std::string & | address,
|  | short | port,
|  | Log * | pLog
| ) |  |

Definition at line 45 of file ThreadedSocketConnection.cpp.

: m_socket( s ), m_address( address ), m_port( port ),
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 );
}
FIX::Session::lookupSession
static Session * lookupSession(const SessionID &)
Definition: Session.cpp:1403
FIX::ThreadedSocketConnection::m_port
int m_port
Definition: ThreadedSocketConnection.h:71
FIX::ThreadedSocketConnection::m_pSession
Session * m_pSession
Definition: ThreadedSocketConnection.h:76
FIX::ThreadedSocketConnection::m_socket
int m_socket
Definition: ThreadedSocketConnection.h:67
FIX::ThreadedSocketConnection::m_fds
fd_set m_fds
Definition: ThreadedSocketConnection.h:78
FIX::ThreadedSocketConnection::m_pLog
Log * m_pLog
Definition: ThreadedSocketConnection.h:73
FIX::Session::setResponder
void setResponder(Responder *pR)
Definition: Session.h:197
FIX::ThreadedSocketConnection::m_disconnect
bool m_disconnect
Definition: ThreadedSocketConnection.h:77
FIX::ThreadedSocketConnection::m_address
std::string m_address
Definition: ThreadedSocketConnection.h:70

## ~ThreadedSocketConnection()

FIX::ThreadedSocketConnection::~ThreadedSocketConnection | ( |  | ) |

virtual

Definition at line 58 of file ThreadedSocketConnection.cpp.

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

{
if ( m_pSession )
{
m_pSession->setResponder( 0 );
Session::unregisterSession( m_pSession->getSessionID() );
}
}
FIX::Session::unregisterSession
static void unregisterSession(const SessionID &)
Definition: Session.cpp:1454
FIX::ThreadedSocketConnection::m_pSession
Session * m_pSession
Definition: ThreadedSocketConnection.h:76
FIX::Session::getSessionID
const SessionID & getSessionID() const
Definition: Session.h:75
FIX::Session::setResponder
void setResponder(Responder *pR)
Definition: Session.h:197

## Member Function Documentation

## connect()

bool FIX::ThreadedSocketConnection::connect | ( |  | ) |

Definition at line 80 of file ThreadedSocketConnection.cpp.

References getSocket(), m_address, m_port, and FIX::socket_connect().

Referenced by getSocket(), and FIX::ThreadedSocketInitiator::socketThread().

{
return socket_connect(getSocket(), m_address.c_str(), m_port) >= 0;
}
FIX::ThreadedSocketConnection::getSocket
int getSocket() const
Definition: ThreadedSocketConnection.h:56
FIX::socket_connect
int socket_connect(int socket, const char *address, int port)
Definition: Utility.cpp:131
FIX::ThreadedSocketConnection::m_port
int m_port
Definition: ThreadedSocketConnection.h:71
FIX::ThreadedSocketConnection::m_address
std::string m_address
Definition: ThreadedSocketConnection.h:70

## disconnect()

void FIX::ThreadedSocketConnection::disconnect | ( |  | ) |

virtual

Implements FIX::Responder.

Definition at line 85 of file ThreadedSocketConnection.cpp.

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

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

{
m_disconnect = true;
socket_close( m_socket );
}
FIX::ThreadedSocketConnection::m_socket
int m_socket
Definition: ThreadedSocketConnection.h:67
FIX::ThreadedSocketConnection::m_disconnect
bool m_disconnect
Definition: ThreadedSocketConnection.h:77
FIX::socket_close
void socket_close(int s)
Definition: Utility.cpp:158

## getSession()

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

inline

Definition at line 55 of file ThreadedSocketConnection.h.

References m_pSession.

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

{ return m_pSession; }
FIX::ThreadedSocketConnection::m_pSession
Session * m_pSession
Definition: ThreadedSocketConnection.h:76

## getSocket()

int FIX::ThreadedSocketConnection::getSocket | ( |  | ) | const

inline

Definition at line 56 of file ThreadedSocketConnection.h.

References connect(), disconnect(), m_socket, processStream(), read(), readMessage(), send(), and setSession().

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

{ return m_socket; }
FIX::ThreadedSocketConnection::m_socket
int m_socket
Definition: ThreadedSocketConnection.h:67

## processStream()

void FIX::ThreadedSocketConnection::processStream | ( |  | ) |

private

Definition at line 150 of file ThreadedSocketConnection.cpp.

References disconnect(), FIX::Session::isLoggedOn(), m_pSession, FIX::Session::next(), readMessage(), setSession(), and FIX::TYPE::UtcTimeStamp.

Referenced by getSocket(), and read().

{
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;
}
}
}
}
FIX::ThreadedSocketConnection::readMessage
bool readMessage(std::string &msg)
Definition: ThreadedSocketConnection.cpp:139
FIX::Session::next
void next()
Definition: Session.cpp:125
FIX::ThreadedSocketConnection::m_pSession
Session * m_pSession
Definition: ThreadedSocketConnection.h:76
FIX::ThreadedSocketConnection::setSession
bool setSession(const std::string &msg)
Definition: ThreadedSocketConnection.cpp:175
FIX::ThreadedSocketConnection::disconnect
void disconnect()
Definition: ThreadedSocketConnection.cpp:85
FIX::TYPE::UtcTimeStamp
Definition: FieldTypes.h:674
FIX::Session::isLoggedOn
bool isLoggedOn()
Definition: Session.h:65

## read()

bool FIX::ThreadedSocketConnection::read | ( |  | ) |

Definition at line 91 of file ThreadedSocketConnection.cpp.

References FIX::Parser::addToStream(), disconnect(), FIX::Session::disconnect(), FIX::Session::getLog(), m_buffer, m_disconnect, m_fds, m_parser, m_pSession, m_socket, FIX::Session::next(), FIX::Log::onEvent(), and processStream().

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

{
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 = recv( m_socket, m_buffer, sizeof(m_buffer), 0 );
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;
}
}
FIX::Session::next
void next()
Definition: Session.cpp:125
FIX::ThreadedSocketConnection::m_pSession
Session * m_pSession
Definition: ThreadedSocketConnection.h:76
FIX::ThreadedSocketConnection::m_socket
int m_socket
Definition: ThreadedSocketConnection.h:67
FIX::ThreadedSocketConnection::processStream
void processStream()
Definition: ThreadedSocketConnection.cpp:150
FIX::Log::onEvent
virtual void onEvent(const std::string &)=0
FIX::Session::getLog
Log * getLog()
Definition: Session.h:214
FIX::ThreadedSocketConnection::m_fds
fd_set m_fds
Definition: ThreadedSocketConnection.h:78
FIX::Session::disconnect
void disconnect()
Definition: Session.cpp:542
FIX::ThreadedSocketConnection::m_buffer
char m_buffer[BUFSIZ]
Definition: ThreadedSocketConnection.h:68
FIX::ThreadedSocketConnection::m_parser
Parser m_parser
Definition: ThreadedSocketConnection.h:74
FIX::Parser::addToStream
void addToStream(const char *str, size_t len)
Definition: Parser.h:48
FIX::ThreadedSocketConnection::disconnect
void disconnect()
Definition: ThreadedSocketConnection.cpp:85
FIX::ThreadedSocketConnection::m_disconnect
bool m_disconnect
Definition: ThreadedSocketConnection.h:77

## readMessage()

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

private

Definition at line 139 of file ThreadedSocketConnection.cpp.

References m_parser, and FIX::Parser::readFixMessage().

Referenced by getSocket(), and processStream().

{
try
{
return m_parser.readFixMessage( msg );
}
catch ( MessageParseError& ) {}
return true;
}
FIX::ThreadedSocketConnection::m_parser
Parser m_parser
Definition: ThreadedSocketConnection.h:74
FIX::Parser::readFixMessage
bool readFixMessage(std::string &str)
Definition: Parser.cpp:59

## send()

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

private virtual

Implements FIX::Responder.

Definition at line 67 of file ThreadedSocketConnection.cpp.

References m_socket, and FIX::socket_send().

Referenced by getSocket().

{
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;
}
FIX::ThreadedSocketConnection::m_socket
int m_socket
Definition: ThreadedSocketConnection.h:67
FIX::socket_send
ssize_t socket_send(int s, const char *msg, size_t length)
Definition: Utility.cpp:153

## setSession()

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

private

Definition at line 175 of file ThreadedSocketConnection.cpp.

References FIX::Session::getSessionID(), FIX::Session::isSessionRegistered(), FIX::Session::lookupSession(), m_pLog, m_pSession, m_sessions, FIX::Log::onEvent(), FIX::Log::onIncoming(), FIX::process_sleep(), FIX::Session::registerSession(), and FIX::Session::setResponder().

Referenced by getSocket(), and processStream().

{
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;
}
FIX::Session::lookupSession
static Session * lookupSession(const SessionID &)
Definition: Session.cpp:1403
FIX::ThreadedSocketConnection::m_pSession
Session * m_pSession
Definition: ThreadedSocketConnection.h:76
FIX::process_sleep
void process_sleep(double s)
Definition: Utility.cpp:444
FIX::Session::isSessionRegistered
static bool isSessionRegistered(const SessionID &)
Definition: Session.cpp:1438
FIX::Session::registerSession
static Session * registerSession(const SessionID &)
Definition: Session.cpp:1444
FIX::Log::onEvent
virtual void onEvent(const std::string &)=0
FIX::ThreadedSocketConnection::m_pLog
Log * m_pLog
Definition: ThreadedSocketConnection.h:73
FIX::Session::getSessionID
const SessionID & getSessionID() const
Definition: Session.h:75
FIX::ThreadedSocketConnection::m_sessions
Sessions m_sessions
Definition: ThreadedSocketConnection.h:75
FIX::Log::onIncoming
virtual void onIncoming(const std::string &)=0
FIX::Session::setResponder
void setResponder(Responder *pR)
Definition: Session.h:197

## Member Data Documentation

## m_address

std::string FIX::ThreadedSocketConnection::m_address

private

Definition at line 70 of file ThreadedSocketConnection.h.

Referenced by connect().

## m_buffer

char FIX::ThreadedSocketConnection::m_buffer[BUFSIZ]

private

Definition at line 68 of file ThreadedSocketConnection.h.

Referenced by read().

## m_disconnect

bool FIX::ThreadedSocketConnection::m_disconnect

private

Definition at line 77 of file ThreadedSocketConnection.h.

Referenced by disconnect(), and read().

## m_fds

fd_set FIX::ThreadedSocketConnection::m_fds

private

Definition at line 78 of file ThreadedSocketConnection.h.

Referenced by read().

## m_parser

Parser FIX::ThreadedSocketConnection::m_parser

private

Definition at line 74 of file ThreadedSocketConnection.h.

Referenced by read(), and readMessage().

## m_pLog

Log* FIX::ThreadedSocketConnection::m_pLog

private

Definition at line 73 of file ThreadedSocketConnection.h.

Referenced by setSession().

## m_port

int FIX::ThreadedSocketConnection::m_port

private

Definition at line 71 of file ThreadedSocketConnection.h.

Referenced by connect().

## m_pSession

Session* FIX::ThreadedSocketConnection::m_pSession

private

Definition at line 76 of file ThreadedSocketConnection.h.

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

## m_sessions

Sessions FIX::ThreadedSocketConnection::m_sessions

private

Definition at line 75 of file ThreadedSocketConnection.h.

Referenced by setSession().

## m_socket

int FIX::ThreadedSocketConnection::m_socket

private

Definition at line 67 of file ThreadedSocketConnection.h.

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

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

- ThreadedSocketConnection.h
- ThreadedSocketConnection.cpp