ThreadedSocketInitiator.cpp
Go to the documentation of this file.
/****************************************************************************
** Copyright (c) 2001-2014
**
** This file is part of the QuickFIX FIX Engine
**
** This file may be distributed under the terms of the quickfixengine.org
** license as defined by quickfixengine.org and appearing in the file
** LICENSE included in the packaging of this file.
**
** This file is provided AS IS with NO WARRANTY OF ANY KIND, INCLUDING THE
** WARRANTY OF DESIGN, MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE.
**
** See http://www.quickfixengine.org/LICENSE for licensing information.
**
** Contact ask@quickfixengine.org if any conditions of this licensing are
** not clear to you.
**
****************************************************************************/
#ifdef _MSC_VER
#include "stdafx.h"
#else
#include "config.h"
#endif
#include "ThreadedSocketInitiator.h"
#include "Session.h"
#include "Settings.h"
namespace FIX
{
ThreadedSocketInitiator::ThreadedSocketInitiator(
Application& application,
MessageStoreFactory& factory,
const SessionSettings& settings ) throw( ConfigError )
: Initiator( application, factory, settings ),
m_lastConnect( 0 ), m_reconnectInterval( 30 ), m_noDelay( false ),
m_sendBufSize( 0 ), m_rcvBufSize( 0 )
{
socket_init();
}
ThreadedSocketInitiator::ThreadedSocketInitiator(
Application& application,
MessageStoreFactory& factory,
const SessionSettings& settings,
LogFactory& logFactory ) throw( ConfigError )
: Initiator( application, factory, settings, logFactory ),
m_lastConnect( 0 ), m_reconnectInterval( 30 ), m_noDelay( false ),
m_sendBufSize( 0 ), m_rcvBufSize( 0 )
{
socket_init();
}
ThreadedSocketInitiator::~ThreadedSocketInitiator()
{
socket_term();
}
void ThreadedSocketInitiator::onConfigure( const SessionSettings& s )
throw ( ConfigError )
{
const Dictionary& dict = s.get();
if( dict.has( RECONNECT_INTERVAL ) )
m_reconnectInterval = dict.getInt( RECONNECT_INTERVAL );
if( dict.has( SOCKET_NODELAY ) )
m_noDelay = dict.getBool( SOCKET_NODELAY );
if( dict.has( SOCKET_SEND_BUFFER_SIZE ) )
m_sendBufSize = dict.getInt( SOCKET_SEND_BUFFER_SIZE );
if( dict.has( SOCKET_RECEIVE_BUFFER_SIZE ) )
m_rcvBufSize = dict.getInt( SOCKET_RECEIVE_BUFFER_SIZE );
}
void ThreadedSocketInitiator::onInitialize( const SessionSettings& s )
throw ( RuntimeError )
{
}
void ThreadedSocketInitiator::onStart()
{
while ( !isStopped() )
{
time_t now;
::time( &now );
if ( (now - m_lastConnect) >= m_reconnectInterval )
{
Locker l( m_mutex );
connect();
m_lastConnect = now;
}
process_sleep( 1 );
}
}
bool ThreadedSocketInitiator::onPoll( double timeout )
{
return false;
}
void ThreadedSocketInitiator::onStop()
{
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 );
threads.clear();
}
void ThreadedSocketInitiator::doConnect( const SessionID& s, const Dictionary& d )
{
try
{
Session* session = Session::lookupSession( s );
if( !session->isSessionTime(UtcTimeStamp()) ) return;
Log* log = session->getLog();
std::string address;
short port = 0;
getHost( s, d, address, port );
int socket = socket_createConnector();
if( m_noDelay )
socket_setsockopt( socket, TCP_NODELAY );
if( m_sendBufSize )
socket_setsockopt( socket, SO_SNDBUF, m_sendBufSize );
if( m_rcvBufSize )
socket_setsockopt( socket, SO_RCVBUF, m_rcvBufSize );
setPending( s );
log->onEvent( "Connecting to " + address + " on port " + IntConvertor::convert((unsigned short)port) );
ThreadedSocketConnection* pConnection =
new ThreadedSocketConnection( s, socket, address, port, getLog() );
ThreadPair* pair = new ThreadPair( this, pConnection );
{
Locker l( m_mutex );
thread_id thread;
if ( thread_spawn( &socketThread, pair, thread ) )
{
addThread( socket, thread );
}
else
{
delete pair;
pConnection->disconnect();
delete pConnection;
setDisconnected( s );
}
}
}
catch ( std::exception& ) {}
}
void ThreadedSocketInitiator::addThread( int s, thread_id t )
{
Locker l(m_mutex);
m_threads[ s ] = t;
}
void ThreadedSocketInitiator::removeThread( int s )
{
Locker l(m_mutex);
SocketToThread::iterator i = m_threads.find( s );
if ( i != m_threads.end() )
{
thread_detach( i->second );
m_threads.erase( i );
}
}
THREAD_PROC ThreadedSocketInitiator::socketThread( void* p )
{
ThreadPair * pair = reinterpret_cast < ThreadPair* > ( p );
ThreadedSocketInitiator* pInitiator = pair->first;
ThreadedSocketConnection* pConnection = pair->second;
FIX::SessionID sessionID = pConnection->getSession()->getSessionID();
FIX::Session* pSession = FIX::Session::lookupSession( sessionID );
int socket = pConnection->getSocket();
delete pair;
pInitiator->lock();
if( !pConnection->connect() )
{
pInitiator->getLog()->onEvent( "Connection failed" );
pConnection->disconnect();
delete pConnection;
pInitiator->removeThread( socket );
pInitiator->setDisconnected( sessionID );
return 0;
}
pInitiator->setConnected( sessionID );
pInitiator->getLog()->onEvent( "Connection succeeded" );
pSession->next();
while ( pConnection->read() ) {}
delete pConnection;
if( !pInitiator->isStopped() )
pInitiator->removeThread( socket );
pInitiator->setDisconnected( sessionID );
return 0;
}
void ThreadedSocketInitiator::getHost( const SessionID& s, const Dictionary& d,
std::string& address, short& port )
{
int num = 0;
SessionToHostNum::iterator i = m_sessionToHostNum.find( s );
if ( i != m_sessionToHostNum.end() ) num = i->second;
std::stringstream hostStream;
hostStream << SOCKET_CONNECT_HOST << num;
std::string hostString = hostStream.str();
std::stringstream portStream;
portStream << SOCKET_CONNECT_PORT << num;
std::string portString = portStream.str();
if( d.has(hostString) && d.has(portString) )
{
address = d.getString( hostString );
port = ( short ) d.getInt( portString );
}
else
{
num = 0;
address = d.getString( SOCKET_CONNECT_HOST );
port = ( short ) d.getInt( SOCKET_CONNECT_PORT );
}
m_sessionToHostNum[ s ] = ++num;
}
}
FIX::ThreadedSocketInitiator::getHost
void getHost(const SessionID &, const Dictionary &, std::string &, short &)
Definition: ThreadedSocketInitiator.cpp:238
Settings.h
FIX::thread_join
void thread_join(thread_id thread)
Definition: Utility.cpp:415
FIX::Session::lookupSession
static Session * lookupSession(const SessionID &)
Definition: Session.cpp:1403
FIX::ThreadedSocketConnection::getSocket
int getSocket() const
Definition: ThreadedSocketConnection.h:56
FIX::socket_init
void socket_init()
Definition: Utility.cpp:81
FIX::ThreadedSocketInitiator::m_lastConnect
time_t m_lastConnect
Definition: ThreadedSocketInitiator.h:73
FIX::Session
Maintains the state and implements the logic of a FIX session.
Definition: Session.h:45
FIX::ThreadedSocketConnection
Encapsulates a socket file descriptor (multi-threaded).
Definition: ThreadedSocketConnection.h:44
FIX::ThreadedSocketInitiator::ThreadPair
std::pair< ThreadedSocketInitiator *, ThreadedSocketConnection *> ThreadPair
Definition: ThreadedSocketInitiator.h:53
FIX::socket_term
void socket_term()
Definition: Utility.cpp:96
FIX::RECONNECT_INTERVAL
const char RECONNECT_INTERVAL[]
Definition: SessionSettings.h:69
FIX::Session::next
void next()
Definition: Session.cpp:125
FIX::thread_detach
void thread_detach(thread_id thread)
Definition: Utility.cpp:425
FIX::ThreadedSocketInitiator::m_noDelay
bool m_noDelay
Definition: ThreadedSocketInitiator.h:75
FIX::ThreadedSocketInitiator::removeThread
void removeThread(int s)
Definition: ThreadedSocketInitiator.cpp:188
FIX::socket_createConnector
int socket_createConnector()
Definition: Utility.cpp:126
FIX::SOCKET_NODELAY
const char SOCKET_NODELAY[]
Definition: SessionSettings.h:66
FIX::Initiator::connect
void connect()
Definition: Initiator.cpp:131
FIX::process_sleep
void process_sleep(double s)
Definition: Utility.cpp:444
FIX::ThreadedSocketInitiator::SocketToThread
std::map< int, thread_id > SocketToThread
Definition: ThreadedSocketInitiator.h:51
FIX::SOCKET_CONNECT_HOST
const char SOCKET_CONNECT_HOST[]
Definition: SessionSettings.h:64
FIX::SOCKET_SEND_BUFFER_SIZE
const char SOCKET_SEND_BUFFER_SIZE[]
Definition: SessionSettings.h:67
FIX::thread_spawn
bool thread_spawn(THREAD_START_ROUTINE func, void *var, thread_id &thread)
Definition: Utility.cpp:394
FIX::Dictionary::has
bool has(const std::string &) const
Check if the dictionary contains a value for key.
Definition: Dictionary.cpp:150
FIX::ThreadedSocketInitiator::onStop
void onStop()
Implemented to stop a running initiator.
Definition: ThreadedSocketInitiator.cpp:103
ThreadedSocketInitiator.h
FIX::ThreadedSocketInitiator::m_rcvBufSize
int m_rcvBufSize
Definition: ThreadedSocketInitiator.h:77
FIX::SOCKET_CONNECT_PORT
const char SOCKET_CONNECT_PORT[]
Definition: SessionSettings.h:65
FIX::IntConvertor::convert
static std::string convert(signed_int value)
Definition: FieldConvertors.h:154
FIX::ThreadedSocketInitiator::onPoll
bool onPoll(double timeout)
Implemented to connect and poll for events.
Definition: ThreadedSocketInitiator.cpp:98
FIX::ThreadedSocketInitiator::ThreadedSocketInitiator
ThreadedSocketInitiator(Application &, MessageStoreFactory &, const SessionSettings &)
Definition: ThreadedSocketInitiator.cpp:32
FIX::ThreadedSocketInitiator::socketThread
static THREAD_PROC socketThread(void *p)
Definition: ThreadedSocketInitiator.cpp:200
FIX::ThreadedSocketConnection::read
bool read()
Definition: ThreadedSocketConnection.cpp:91
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::Log
This interface must be implemented to log messages and events.
Definition: Log.h:81
FIX
Definition: Acceptor.cpp:34
FIX::Log::onEvent
virtual void onEvent(const std::string &)=0
FIX::Session::getLog
Log * getLog()
Definition: Session.h:214
FIX::Application
This interface must be implemented to define what your FIX application does.
Definition: Application.h:43
FIX::ThreadedSocketInitiator::~ThreadedSocketInitiator
virtual ~ThreadedSocketInitiator()
Definition: ThreadedSocketInitiator.cpp:55
FIX::RuntimeError
Application encountered serious error during runtime
Definition: Exceptions.h:94
FIX::ConfigError
Application is not configured correctly
Definition: Exceptions.h:87
FIX::Locker
Locks/Unlocks a mutex using RAII.
Definition: Mutex.h:95
FIX::ThreadedSocketInitiator::m_reconnectInterval
int m_reconnectInterval
Definition: ThreadedSocketInitiator.h:74
FIX::ThreadedSocketInitiator::onConfigure
void onConfigure(const SessionSettings &)
Implemented to configure acceptor.
Definition: ThreadedSocketInitiator.cpp:60
FIX::SessionSettings
Container for setting dictionaries mapped to sessions.
Definition: SessionSettings.h:130
FIX::Initiator::start
void start()
Start initiator.
Definition: Initiator.cpp:190
FIX::ThreadedSocketInitiator::onInitialize
void onInitialize(const SessionSettings &)
Implemented to initialize initiator.
Definition: ThreadedSocketInitiator.cpp:75
FIX::LogFactory
This interface must be implemented to create a Log.
Definition: Log.h:42
FIX::Session::getSessionID
const SessionID & getSessionID() const
Definition: Session.h:75
FIX::Initiator::getLog
Log * getLog()
Definition: Initiator.h:90
FIX::ThreadedSocketInitiator::addThread
void addThread(int s, thread_id t)
Definition: ThreadedSocketInitiator.cpp:181
FIX::MessageStoreFactory
This interface must be implemented to create a MessageStore.
Definition: MessageStore.h:41
FIX::ThreadedSocketInitiator::m_sessionToHostNum
SessionToHostNum m_sessionToHostNum
Definition: ThreadedSocketInitiator.h:72
FIX::ThreadedSocketInitiator::m_threads
SocketToThread m_threads
Definition: ThreadedSocketInitiator.h:78
FIX::ThreadedSocketInitiator::onStart
void onStart()
Implemented to start connecting to targets.
Definition: ThreadedSocketInitiator.cpp:80
FIX::ThreadedSocketConnection::disconnect
void disconnect()
Definition: ThreadedSocketConnection.cpp:85
FIX::Initiator::isLoggedOn
bool isLoggedOn()
Check to see if any sessions are currently logged on.
Definition: Initiator.cpp:269
FIX::thread_id
pthread_t thread_id
Definition: Utility.h:164
FIX::UtcTimeStamp
Date and Time represented in UTC.
Definition: FieldTypes.h:399
FIX::Dictionary
For storage and retrieval of key/value pairs.
Definition: Dictionary.h:36
FIX::ThreadedSocketConnection::getSession
Session * getSession() const
Definition: ThreadedSocketConnection.h:55
FIX::Dictionary::getString
std::string getString(const std::string &, bool capitalize=false) const
Get a value as a string.
Definition: Dictionary.cpp:32
FIX::ThreadedSocketConnection::connect
bool connect()
Definition: ThreadedSocketConnection.cpp:80
FIX::ThreadedSocketInitiator::m_sendBufSize
int m_sendBufSize
Definition: ThreadedSocketInitiator.h:76
FIX::ThreadedSocketInitiator::doConnect
void doConnect(const SessionID &s, const Dictionary &d)
Implemented to connect a session to its target.
Definition: ThreadedSocketInitiator.cpp:133
FIX::SessionID
Unique session id consists of BeginString, SenderCompID and TargetCompID.
Definition: SessionID.h:30
FIX::socket_close
void socket_close(int s)
Definition: Utility.cpp:158
FIX::Session::isSessionTime
bool isSessionTime(const UtcTimeStamp &time)
Definition: Session.h:108
FIX::Dictionary::getBool
bool getBool(const std::string &) const
Get a value as a bool.
Definition: Dictionary.cpp:71
THREAD_PROC
#define THREAD_PROC
Definition: Utility.h:158
FIX::Dictionary::getInt
int getInt(const std::string &) const
Get a value as a int.
Definition: Dictionary.cpp:45
FIX::ThreadedSocketInitiator
Threaded Socket implementation of Initiator.
Definition: ThreadedSocketInitiator.h:39
FIX::Initiator::isStopped
bool isStopped()
Definition: Initiator.h:83
Session.h
FIX::Initiator
Base for classes which act as an initiator for establishing connections.
Definition: Initiator.h:51
FIX::Initiator::setPending
void setPending(const SessionID &)
Definition: Initiator.cpp:145
FIX::ThreadedSocketInitiator::m_mutex
Mutex m_mutex
Definition: ThreadedSocketInitiator.h:79
FIX::Initiator::setDisconnected
void setDisconnected(const SessionID &)
Definition: Initiator.cpp:163