FIX::ThreadedSocketAcceptor Class Reference

Threaded Socket implementation of Acceptor. More...

`#include <ThreadedSocketAcceptor.h>`

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

## Classes

struct | AcceptorThreadInfo
struct | ConnectionThreadInfo

## Public Member Functions

| ThreadedSocketAcceptor (Application &, MessageStoreFactory &, const SessionSettings &) throw ( ConfigError )
| ThreadedSocketAcceptor (Application &, MessageStoreFactory &, const SessionSettings &, LogFactory &) throw ( ConfigError )
virtual | ~ThreadedSocketAcceptor ()
Public Member Functions inherited from FIX::Acceptor
| Acceptor (Application &, MessageStoreFactory &, const SessionSettings &) throw ( ConfigError )
| Acceptor (Application &, MessageStoreFactory &, const SessionSettings &, LogFactory &) throw ( ConfigError )
virtual | ~Acceptor ()
Log * | getLog ()
void | start () throw ( ConfigError, RuntimeError )
| Start acceptor. More...
void | block () throw ( ConfigError, RuntimeError )
| Block on the acceptor. More...
bool | poll (double timeout=0.0) throw ( ConfigError, RuntimeError )
| Poll the acceptor. More...
void | stop (bool force=false)
| Stop acceptor. More...
bool | isLoggedOn ()
| Check to see if any sessions are currently logged on. More...
Session * | getSession (const std::string &msg, Responder &)
const std::set< SessionID > & | getSessions () const
Session * | getSession (const SessionID &sessionID) const
const Dictionary *const | getSessionSettings (const SessionID &sessionID) const
bool | has (const SessionID &id)
bool | isStopped ()
Application & | getApplication ()
MessageStoreFactory & | getMessageStoreFactory ()

## Private Types

typedef std::set< int > | Sockets
typedef std::set< SessionID > | Sessions
typedef std::map< int, Sessions > | PortToSessions
typedef std::map< int, int > | SocketToPort
typedef std::map< int, thread_id > | SocketToThread

## Private Member Functions

bool | readSettings (const SessionSettings &)
void | onConfigure (const SessionSettings &) throw ( ConfigError )
| Implemented to configure acceptor. More...
void | onInitialize (const SessionSettings &) throw ( RuntimeError )
| Implemented to initialize acceptor. More...
void | onStart ()
| Implemented to start listening for connections. More...
bool | onPoll (double timeout)
| Implemented to connect and poll for events. More...
void | onStop ()
| Implemented to stop a running acceptor. More...
void | addThread (int s, thread_id t)
void | removeThread (int s)

## Static Private Member Functions

static THREAD_PROC | socketAcceptorThread (void *p)
static THREAD_PROC | socketConnectionThread (void *p)

## Private Attributes

Sockets | m_sockets
PortToSessions | m_portToSessions
SocketToPort | m_socketToPort
SocketToThread | m_threads
Mutex | m_mutex

## Friends

class | SocketConnection

## Detailed Description

Threaded Socket implementation of Acceptor.

Definition at line 36 of file ThreadedSocketAcceptor.h.

## Member Typedef Documentation

## PortToSessions

typedef std::map< int, Sessions > FIX::ThreadedSocketAcceptor::PortToSessions

private

Definition at line 73 of file ThreadedSocketAcceptor.h.

## Sessions

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

private

Definition at line 72 of file ThreadedSocketAcceptor.h.

## Sockets

typedef std::set< int > FIX::ThreadedSocketAcceptor::Sockets

private

Definition at line 71 of file ThreadedSocketAcceptor.h.

## SocketToPort

typedef std::map< int, int > FIX::ThreadedSocketAcceptor::SocketToPort

private

Definition at line 74 of file ThreadedSocketAcceptor.h.

## SocketToThread

typedef std::map< int, thread_id > FIX::ThreadedSocketAcceptor::SocketToThread

private

Definition at line 75 of file ThreadedSocketAcceptor.h.

## Constructor & Destructor Documentation

## ThreadedSocketAcceptor() [1/2]

FIX::ThreadedSocketAcceptor::ThreadedSocketAcceptor | ( | Application & | application,
|  | MessageStoreFactory & | factory,
|  | const SessionSettings & | settings
| ) |  |
throw | ( | ConfigError
| ) |  |

Definition at line 32 of file ThreadedSocketAcceptor.cpp.

References FIX::socket_init().

: Acceptor( application, factory, settings )
{ socket_init(); }
FIX::socket_init
void socket_init()
Definition: Utility.cpp:81
FIX::Acceptor::Acceptor
Acceptor(Application &, MessageStoreFactory &, const SessionSettings &)
Definition: Acceptor.cpp:36

## ThreadedSocketAcceptor() [2/2]

FIX::ThreadedSocketAcceptor::ThreadedSocketAcceptor | ( | Application & | application,
|  | MessageStoreFactory & | factory,
|  | const SessionSettings & | settings,
|  | LogFactory & | logFactory
| ) |  |
throw | ( | ConfigError
| ) |  |

Definition at line 39 of file ThreadedSocketAcceptor.cpp.

References FIX::socket_init().

: Acceptor( application, factory, settings, logFactory )
{
socket_init();
}
FIX::socket_init
void socket_init()
Definition: Utility.cpp:81
FIX::Acceptor::Acceptor
Acceptor(Application &, MessageStoreFactory &, const SessionSettings &)
Definition: Acceptor.cpp:36

## ~ThreadedSocketAcceptor()

FIX::ThreadedSocketAcceptor::~ThreadedSocketAcceptor | ( |  | ) |

virtual

Definition at line 49 of file ThreadedSocketAcceptor.cpp.

References FIX::socket_term().

{
socket_term();
}
FIX::socket_term
void socket_term()
Definition: Utility.cpp:96

## Member Function Documentation

## addThread()

void FIX::ThreadedSocketAcceptor::addThread | ( | int | s,
|  | thread_id | t
| ) |  |

private

Definition at line 168 of file ThreadedSocketAcceptor.cpp.

References m_mutex, and m_threads.

Referenced by onStart().

{
Locker l(m_mutex);
m_threads[ s ] = t;
}
FIX::ThreadedSocketAcceptor::m_threads
SocketToThread m_threads
Definition: ThreadedSocketAcceptor.h:92
FIX::ThreadedSocketAcceptor::m_mutex
Mutex m_mutex
Definition: ThreadedSocketAcceptor.h:93

## onConfigure()

void FIX::ThreadedSocketAcceptor::onConfigure | ( | const SessionSettings & |  | ) |
throw | ( | ConfigError
| ) |  |

private virtual

Implemented to configure acceptor.

Reimplemented from FIX::Acceptor.

Definition at line 54 of file ThreadedSocketAcceptor.cpp.

References FIX::Dictionary::getBool(), FIX::Dictionary::getInt(), FIX::Dictionary::has(), FIX::SOCKET_ACCEPT_PORT, FIX::SOCKET_NODELAY, and FIX::SOCKET_REUSE_ADDRESS.

{
std::set<SessionID> sessions = s.getSessions();
std::set<SessionID>::iterator i;
for( i = sessions.begin(); i != sessions.end(); ++i )
{
const Dictionary& settings = s.get( *i );
settings.getInt( SOCKET_ACCEPT_PORT );
if( settings.has(SOCKET_REUSE_ADDRESS) )
settings.getBool( SOCKET_REUSE_ADDRESS );
if( settings.has(SOCKET_NODELAY) )
settings.getBool( SOCKET_NODELAY );
}
}
FIX::SOCKET_NODELAY
const char SOCKET_NODELAY[]
Definition: SessionSettings.h:66
FIX::SOCKET_ACCEPT_PORT
const char SOCKET_ACCEPT_PORT[]
Definition: SessionSettings.h:62
FIX::SOCKET_REUSE_ADDRESS
const char SOCKET_REUSE_ADDRESS[]
Definition: SessionSettings.h:63

## onInitialize()

void FIX::ThreadedSocketAcceptor::onInitialize | ( | const SessionSettings & |  | ) |
throw | ( | RuntimeError
| ) |  |

private virtual

Implemented to initialize acceptor.

Reimplemented from FIX::Acceptor.

Definition at line 70 of file ThreadedSocketAcceptor.cpp.

References FIX::IntConvertor::convert(), FIX::Dictionary::getBool(), FIX::Dictionary::getInt(), FIX::Dictionary::has(), m_portToSessions, m_sockets, m_socketToPort, FIX::SOCKET_ACCEPT_PORT, FIX::socket_close(), FIX::socket_createAcceptor(), FIX::SOCKET_NODELAY, FIX::SOCKET_RECEIVE_BUFFER_SIZE, FIX::SOCKET_REUSE_ADDRESS, FIX::SOCKET_SEND_BUFFER_SIZE, and FIX::socket_setsockopt().

{
short port = 0;
std::set<int> ports;
std::set<SessionID> sessions = s.getSessions();
std::set<SessionID>::iterator i = sessions.begin();
for( ; i != sessions.end(); ++i )
{
const Dictionary& settings = s.get( *i );
port = (short)settings.getInt( SOCKET_ACCEPT_PORT );
m_portToSessions[port].insert( *i );
if( ports.find(port) != ports.end() )
continue;
ports.insert( port );
const bool reuseAddress = settings.has( SOCKET_REUSE_ADDRESS ) ?
settings.getBool( SOCKET_REUSE_ADDRESS ) : true;
const bool noDelay = settings.has( SOCKET_NODELAY ) ?
settings.getBool( SOCKET_NODELAY ) : false;
const int sendBufSize = settings.has( SOCKET_SEND_BUFFER_SIZE ) ?
settings.getInt( SOCKET_SEND_BUFFER_SIZE ) : 0;
const int rcvBufSize = settings.has( SOCKET_RECEIVE_BUFFER_SIZE ) ?
settings.getInt( SOCKET_RECEIVE_BUFFER_SIZE ) : 0;
int socket = socket_createAcceptor( port, reuseAddress );
if( socket < 0 )
{
SocketException e;
socket_close( socket );
throw RuntimeError( "Unable to create, bind, or listen to port "
+ IntConvertor::convert( (unsigned short)port ) + " (" + e.what() + ")" );
}
if( noDelay )
socket_setsockopt( socket, TCP_NODELAY );
if( sendBufSize )
socket_setsockopt( socket, SO_SNDBUF, sendBufSize );
if( rcvBufSize )
socket_setsockopt( socket, SO_RCVBUF, rcvBufSize );
m_socketToPort[socket] = port;
m_sockets.insert( socket );
}
}
FIX::ThreadedSocketAcceptor::m_socketToPort
SocketToPort m_socketToPort
Definition: ThreadedSocketAcceptor.h:91
FIX::SOCKET_NODELAY
const char SOCKET_NODELAY[]
Definition: SessionSettings.h:66
FIX::SOCKET_ACCEPT_PORT
const char SOCKET_ACCEPT_PORT[]
Definition: SessionSettings.h:62
FIX::SOCKET_SEND_BUFFER_SIZE
const char SOCKET_SEND_BUFFER_SIZE[]
Definition: SessionSettings.h:67
FIX::IntConvertor::convert
static std::string convert(signed_int value)
Definition: FieldConvertors.h:154
FIX::SOCKET_RECEIVE_BUFFER_SIZE
const char SOCKET_RECEIVE_BUFFER_SIZE[]
Definition: SessionSettings.h:68
FIX::socket_setsockopt
int socket_setsockopt(int s, int opt)
Definition: Utility.cpp:186
FIX::ThreadedSocketAcceptor::m_portToSessions
PortToSessions m_portToSessions
Definition: ThreadedSocketAcceptor.h:90
FIX::socket_createAcceptor
int socket_createAcceptor(int port, bool reuse)
Definition: Utility.cpp:103
FIX::SOCKET_REUSE_ADDRESS
const char SOCKET_REUSE_ADDRESS[]
Definition: SessionSettings.h:63
FIX::socket_close
void socket_close(int s)
Definition: Utility.cpp:158
FIX::ThreadedSocketAcceptor::m_sockets
Sockets m_sockets
Definition: ThreadedSocketAcceptor.h:89

## onPoll()

bool FIX::ThreadedSocketAcceptor::onPoll | ( | double | second | ) |

private virtual

Implemented to connect and poll for events.

Implements FIX::Acceptor.

Definition at line 135 of file ThreadedSocketAcceptor.cpp.

{
return false;
}

## onStart()

void FIX::ThreadedSocketAcceptor::onStart | ( |  | ) |

private virtual

Implemented to start listening for connections.

Implements FIX::Acceptor.

Definition at line 121 of file ThreadedSocketAcceptor.cpp.

References addThread(), m_mutex, m_sockets, m_socketToPort, socketAcceptorThread(), and FIX::thread_spawn().

{
Sockets::iterator i;
for( i = m_sockets.begin(); i != m_sockets.end(); ++i )
{
Locker l( m_mutex );
int port = m_socketToPort[*i];
AcceptorThreadInfo* info = new AcceptorThreadInfo( this, *i, port );
thread_id thread;
thread_spawn( &socketAcceptorThread, info, thread );
addThread( *i, thread );
}
}
FIX::ThreadedSocketAcceptor::m_socketToPort
SocketToPort m_socketToPort
Definition: ThreadedSocketAcceptor.h:91
FIX::thread_spawn
bool thread_spawn(THREAD_START_ROUTINE func, void *var, thread_id &thread)
Definition: Utility.cpp:394
FIX::ThreadedSocketAcceptor::socketAcceptorThread
static THREAD_PROC socketAcceptorThread(void *p)
Definition: ThreadedSocketAcceptor.cpp:186
FIX::ThreadedSocketAcceptor::addThread
void addThread(int s, thread_id t)
Definition: ThreadedSocketAcceptor.cpp:168
FIX::thread_id
pthread_t thread_id
Definition: Utility.h:164
FIX::ThreadedSocketAcceptor::m_mutex
Mutex m_mutex
Definition: ThreadedSocketAcceptor.h:93
FIX::ThreadedSocketAcceptor::m_sockets
Sockets m_sockets
Definition: ThreadedSocketAcceptor.h:89

## onStop()

void FIX::ThreadedSocketAcceptor::onStop | ( |  | ) |

private virtual

Implemented to stop a running acceptor.

Implements FIX::Acceptor.

Definition at line 140 of file ThreadedSocketAcceptor.cpp.

References FIX::Acceptor::isLoggedOn(), m_mutex, m_threads, FIX::socket_close(), FIX::Acceptor::start(), and FIX::thread_join().

{
SocketToThread threads;
SocketToThread::iterator i;
{
Locker l(m_mutex);
time_t start = 0;
time_t now = 0;
::time( &start );
while ( isLoggedOn() )
{
if( ::time(&now) -5 >= start )
break;
}
threads = m_threads;
m_threads.clear();
}
for ( i = threads.begin(); i != threads.end(); ++i )
socket_close( i->first );
for ( i = threads.begin(); i != threads.end(); ++i )
thread_join( i->second );
}
FIX::Acceptor::isLoggedOn
bool isLoggedOn()
Check to see if any sessions are currently logged on.
Definition: Acceptor.cpp:230
FIX::thread_join
void thread_join(thread_id thread)
Definition: Utility.cpp:415
FIX::ThreadedSocketAcceptor::m_threads
SocketToThread m_threads
Definition: ThreadedSocketAcceptor.h:92
FIX::ThreadedSocketAcceptor::SocketToThread
std::map< int, thread_id > SocketToThread
Definition: ThreadedSocketAcceptor.h:75
FIX::Acceptor::start
void start()
Start acceptor.
Definition: Acceptor.cpp:158
FIX::ThreadedSocketAcceptor::m_mutex
Mutex m_mutex
Definition: ThreadedSocketAcceptor.h:93
FIX::socket_close
void socket_close(int s)
Definition: Utility.cpp:158

## readSettings()

bool FIX::ThreadedSocketAcceptor::readSettings | ( | const SessionSettings & |  | ) |

private

## removeThread()

void FIX::ThreadedSocketAcceptor::removeThread | ( | int | s | ) |

private

Definition at line 175 of file ThreadedSocketAcceptor.cpp.

References m_mutex, m_threads, and FIX::thread_detach().

{
Locker l(m_mutex);
SocketToThread::iterator i = m_threads.find( s );
if ( i != m_threads.end() )
{
thread_detach( i->second );
m_threads.erase( i );
}
}
FIX::thread_detach
void thread_detach(thread_id thread)
Definition: Utility.cpp:425
FIX::ThreadedSocketAcceptor::m_threads
SocketToThread m_threads
Definition: ThreadedSocketAcceptor.h:92
FIX::ThreadedSocketAcceptor::m_mutex
Mutex m_mutex
Definition: ThreadedSocketAcceptor.h:93

## socketAcceptorThread()

THREAD_PROC FIX::ThreadedSocketAcceptor::socketAcceptorThread | ( | void * | p | ) |

static private

Definition at line 186 of file ThreadedSocketAcceptor.cpp.

References FIX::ThreadedSocketAcceptor::AcceptorThreadInfo::m_pAcceptor, FIX::ThreadedSocketAcceptor::AcceptorThreadInfo::m_port, FIX::ThreadedSocketAcceptor::AcceptorThreadInfo::m_socket, FIX::socket_accept(), FIX::socket_getsockopt(), FIX::socket_peername(), FIX::socket_setsockopt(), socketConnectionThread(), and FIX::thread_spawn().

Referenced by onStart().

{
AcceptorThreadInfo * info = reinterpret_cast < AcceptorThreadInfo* > ( p );
ThreadedSocketAcceptor* pAcceptor = info->m_pAcceptor;
int s = info->m_socket;
int port = info->m_port;
delete info;
int noDelay = 0;
int sendBufSize = 0;
int rcvBufSize = 0;
socket_getsockopt( s, TCP_NODELAY, noDelay );
socket_getsockopt( s, SO_SNDBUF, sendBufSize );
socket_getsockopt( s, SO_RCVBUF, rcvBufSize );
int socket = 0;
while ( ( !pAcceptor->isStopped() && ( socket = socket_accept( s ) ) >= 0 ) )
{
if( noDelay )
socket_setsockopt( socket, TCP_NODELAY );
if( sendBufSize )
socket_setsockopt( socket, SO_SNDBUF, sendBufSize );
if( rcvBufSize )
socket_setsockopt( socket, SO_RCVBUF, rcvBufSize );
Sessions sessions = pAcceptor->m_portToSessions[port];
ThreadedSocketConnection * pConnection =
new ThreadedSocketConnection
( socket, sessions, pAcceptor->getLog() );
ConnectionThreadInfo* info = new ConnectionThreadInfo( pAcceptor, pConnection );
{
Locker l( pAcceptor->m_mutex );
std::stringstream stream;
stream << "Accepted connection from " << socket_peername( socket ) << " on port " << port;
if( pAcceptor->getLog() )
pAcceptor->getLog()->onEvent( stream.str() );
thread_id thread;
if ( !thread_spawn( &socketConnectionThread, info, thread ) )
delete info;
pAcceptor->addThread( socket, thread );
}
}
if( !pAcceptor->isStopped() )
pAcceptor->removeThread( s );
return 0;
}
FIX::socket_getsockopt
int socket_getsockopt(int s, int opt, int &optval)
Definition: Utility.cpp:211
FIX::socket_accept
int socket_accept(int s)
Definition: Utility.cpp:147
FIX::thread_spawn
bool thread_spawn(THREAD_START_ROUTINE func, void *var, thread_id &thread)
Definition: Utility.cpp:394
FIX::socket_setsockopt
int socket_setsockopt(int s, int opt)
Definition: Utility.cpp:186
FIX::ThreadedSocketAcceptor::Sessions
std::set< SessionID > Sessions
Definition: ThreadedSocketAcceptor.h:72
FIX::ThreadedSocketAcceptor::socketConnectionThread
static THREAD_PROC socketConnectionThread(void *p)
Definition: ThreadedSocketAcceptor.cpp:242
FIX::ThreadedSocketAcceptor::ThreadedSocketAcceptor
ThreadedSocketAcceptor(Application &, MessageStoreFactory &, const SessionSettings &)
Definition: ThreadedSocketAcceptor.cpp:32
FIX::thread_id
pthread_t thread_id
Definition: Utility.h:164
FIX::socket_peername
const char * socket_peername(int socket)
Definition: Utility.cpp:331

## socketConnectionThread()

THREAD_PROC FIX::ThreadedSocketAcceptor::socketConnectionThread | ( | void * | p | ) |

static private

Definition at line 242 of file ThreadedSocketAcceptor.cpp.

References FIX::ThreadedSocketConnection::getSocket(), FIX::ThreadedSocketAcceptor::ConnectionThreadInfo::m_pAcceptor, FIX::ThreadedSocketAcceptor::ConnectionThreadInfo::m_pConnection, and FIX::ThreadedSocketConnection::read().

Referenced by socketAcceptorThread().

{
ConnectionThreadInfo * info = reinterpret_cast < ConnectionThreadInfo* > ( p );
ThreadedSocketAcceptor* pAcceptor = info->m_pAcceptor;
ThreadedSocketConnection* pConnection = info->m_pConnection;
delete info;
int socket = pConnection->getSocket();
while ( pConnection->read() ) {}
delete pConnection;
if( !pAcceptor->isStopped() )
pAcceptor->removeThread( socket );
return 0;
}
FIX::ThreadedSocketAcceptor::ThreadedSocketAcceptor
ThreadedSocketAcceptor(Application &, MessageStoreFactory &, const SessionSettings &)
Definition: ThreadedSocketAcceptor.cpp:32

## Friends And Related Function Documentation

## SocketConnection

friend class SocketConnection

friend

Definition at line 38 of file ThreadedSocketAcceptor.h.

## Member Data Documentation

## m_mutex

Mutex FIX::ThreadedSocketAcceptor::m_mutex

private

Definition at line 93 of file ThreadedSocketAcceptor.h.

Referenced by addThread(), onStart(), onStop(), and removeThread().

## m_portToSessions

PortToSessions FIX::ThreadedSocketAcceptor::m_portToSessions

private

Definition at line 90 of file ThreadedSocketAcceptor.h.

Referenced by onInitialize().

## m_sockets

Sockets FIX::ThreadedSocketAcceptor::m_sockets

private

Definition at line 89 of file ThreadedSocketAcceptor.h.

Referenced by onInitialize(), and onStart().

## m_socketToPort

SocketToPort FIX::ThreadedSocketAcceptor::m_socketToPort

private

Definition at line 91 of file ThreadedSocketAcceptor.h.

Referenced by onInitialize(), and onStart().

## m_threads

SocketToThread FIX::ThreadedSocketAcceptor::m_threads

private

Definition at line 92 of file ThreadedSocketAcceptor.h.

Referenced by addThread(), onStop(), and removeThread().

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

- ThreadedSocketAcceptor.h
- ThreadedSocketAcceptor.cpp