SocketInitiator.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 "SocketInitiator.h"
#include "Session.h"
#include "Settings.h"
namespace FIX
{
SocketInitiator::SocketInitiator( Application& application,
MessageStoreFactory& factory,
const SessionSettings& settings )
throw( ConfigError )
: Initiator( application, factory, settings ),
m_connector( 1 ), m_lastConnect( 0 ),
m_reconnectInterval( 30 ), m_noDelay( false ), m_sendBufSize( 0 ),
m_rcvBufSize( 0 )
{
}
SocketInitiator::SocketInitiator( Application& application,
MessageStoreFactory& factory,
const SessionSettings& settings,
LogFactory& logFactory )
throw( ConfigError )
: Initiator( application, factory, settings, logFactory ),
m_connector( 1 ), m_lastConnect( 0 ),
m_reconnectInterval( 30 ), m_noDelay( false ), m_sendBufSize( 0 ),
m_rcvBufSize( 0 )
{
}
SocketInitiator::~SocketInitiator()
{
SocketConnections::iterator i;
for (i = m_connections.begin();
i != m_connections.end(); ++i)
delete i->second;
for (i = m_pendingConnections.begin();
i != m_pendingConnections.end(); ++i)
delete i->second;
}
void SocketInitiator::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 SocketInitiator::onInitialize( const SessionSettings& s )
throw ( RuntimeError )
{
}
void SocketInitiator::onStart()
{
connect();
while ( !isStopped() ) {
m_connector.block( *this, false, 1.0 );
onTimeout( m_connector );
}
time_t start = 0;
time_t now = 0;
::time( &start );
while ( isLoggedOn() )
{
m_connector.block( *this );
if( ::time(&now) -5 >= start )
break;
}
}
bool SocketInitiator::onPoll( double timeout )
{
time_t start = 0;
time_t now = 0;
if( isStopped() )
{
if( start == 0 )
::time( &start );
if( !isLoggedOn() )
return false;
if( ::time(&now) - 5 >= start )
return false;
}
m_connector.block( *this, true, timeout );
return true;
}
void SocketInitiator::onStop()
{
}
void SocketInitiator::doConnect( const SessionID& s, const Dictionary& d )
{
try
{
std::string address;
short port = 0;
Session* session = Session::lookupSession( s );
if( !session->isSessionTime(UtcTimeStamp()) ) return;
Log* log = session->getLog();
getHost( s, d, address, port );
log->onEvent( "Connecting to " + address + " on port " + IntConvertor::convert((unsigned short)port) );
int result = m_connector.connect( address, port, m_noDelay, m_sendBufSize, m_rcvBufSize );
setPending( s );
m_pendingConnections[ result ]
= new SocketConnection( *this, s, result, &m_connector.getMonitor() );
}
catch ( std::exception& ) {}
}
void SocketInitiator::onConnect( SocketConnector&, int s )
{
SocketConnections::iterator i = m_pendingConnections.find( s );
if( i == m_pendingConnections.end() ) return;
SocketConnection* pSocketConnection = i->second;
m_connections[s] = pSocketConnection;
m_pendingConnections.erase( i );
setConnected( pSocketConnection->getSession()->getSessionID() );
pSocketConnection->onTimeout();
}
void SocketInitiator::onWrite( SocketConnector& connector, int s )
{
SocketConnections::iterator i = m_connections.find( s );
if ( i == m_connections.end() ) return ;
SocketConnection* pSocketConnection = i->second;
if( pSocketConnection->processQueue() )
pSocketConnection->unsignal();
}
bool SocketInitiator::onData( SocketConnector& connector, int s )
{
SocketConnections::iterator i = m_connections.find( s );
if ( i == m_connections.end() ) return false;
SocketConnection* pSocketConnection = i->second;
return pSocketConnection->read( connector );
}
void SocketInitiator::onDisconnect( SocketConnector&, int s )
{
SocketConnections::iterator i = m_connections.find( s );
SocketConnections::iterator j = m_pendingConnections.find( s );
SocketConnection* pSocketConnection = 0;
if( i != m_connections.end() )
pSocketConnection = i->second;
if( j != m_pendingConnections.end() )
pSocketConnection = j->second;
if( !pSocketConnection )
return;
setDisconnected( pSocketConnection->getSession()->getSessionID() );
Session* pSession = pSocketConnection->getSession();
if ( pSession )
{
pSession->disconnect();
setDisconnected( pSession->getSessionID() );
}
delete pSocketConnection;
m_connections.erase( s );
m_pendingConnections.erase( s );
}
void SocketInitiator::onError( SocketConnector& connector )
{
onTimeout( connector );
}
void SocketInitiator::onTimeout( SocketConnector& )
{
time_t now;
::time( &now );
if ( (now - m_lastConnect) >= m_reconnectInterval )
{
connect();
m_lastConnect = now;
}
SocketConnections::iterator i;
for ( i = m_connections.begin(); i != m_connections.end(); ++i )
i->second->onTimeout();
}
void SocketInitiator::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::SocketInitiator::m_rcvBufSize
int m_rcvBufSize
Definition: SocketInitiator.h:76
Settings.h
FIX::SocketInitiator::onConnect
void onConnect(SocketConnector &, int)
Definition: SocketInitiator.cpp:154
FIX::SocketInitiator::onStart
void onStart()
Implemented to start connecting to targets.
Definition: SocketInitiator.cpp:87
FIX::Session::lookupSession
static Session * lookupSession(const SessionID &)
Definition: Session.cpp:1403
FIX::SocketConnection::getSession
Session * getSession() const
Definition: SocketConnection.h:56
FIX::SocketInitiator::~SocketInitiator
virtual ~SocketInitiator()
Definition: SocketInitiator.cpp:55
FIX::Session
Maintains the state and implements the logic of a FIX session.
Definition: Session.h:45
FIX::SocketInitiator::m_pendingConnections
SocketConnections m_pendingConnections
Definition: SocketInitiator.h:70
FIX::SocketInitiator::m_connector
SocketConnector m_connector
Definition: SocketInitiator.h:69
FIX::RECONNECT_INTERVAL
const char RECONNECT_INTERVAL[]
Definition: SessionSettings.h:69
FIX::SocketInitiator::doConnect
void doConnect(const SessionID &, const Dictionary &d)
Implemented to connect a session to its target.
Definition: SocketInitiator.cpp:131
FIX::SocketConnection::processQueue
bool processQueue()
Definition: SocketConnection.cpp:72
FIX::SocketConnector
Connects sockets to remote ports and addresses.
Definition: SocketConnector.h:35
FIX::SOCKET_NODELAY
const char SOCKET_NODELAY[]
Definition: SessionSettings.h:66
FIX::SocketInitiator::m_sessionToHostNum
SessionToHostNum m_sessionToHostNum
Definition: SocketInitiator.h:68
FIX::Initiator::connect
void connect()
Definition: Initiator.cpp:131
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::SocketInitiator::onConfigure
void onConfigure(const SessionSettings &)
Implemented to configure acceptor.
Definition: SocketInitiator.cpp:67
FIX::SocketInitiator::m_noDelay
bool m_noDelay
Definition: SocketInitiator.h:74
FIX::SocketInitiator::onStop
void onStop()
Implemented to stop a running initiator.
Definition: SocketInitiator.cpp:127
FIX::SocketConnection::read
bool read(SocketConnector &s)
Definition: SocketConnection.cpp:106
FIX::Dictionary::has
bool has(const std::string &) const
Check if the dictionary contains a value for key.
Definition: Dictionary.cpp:150
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::SOCKET_RECEIVE_BUFFER_SIZE
const char SOCKET_RECEIVE_BUFFER_SIZE[]
Definition: SessionSettings.h:68
FIX::Log
This interface must be implemented to log messages and events.
Definition: Log.h:81
FIX::SocketConnector::getMonitor
SocketMonitor & getMonitor()
Definition: SocketConnector.h:47
FIX::SocketInitiator::SocketInitiator
SocketInitiator(Application &, MessageStoreFactory &, const SessionSettings &)
Definition: SocketInitiator.cpp:32
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::SocketConnector::block
void block(Strategy &strategy, bool poll=0, double timeout=0.0)
Definition: SocketConnector.cpp:111
FIX::SocketConnection::unsignal
void unsignal()
Definition: SocketConnection.h:69
FIX::RuntimeError
Application encountered serious error during runtime
Definition: Exceptions.h:94
FIX::ConfigError
Application is not configured correctly
Definition: Exceptions.h:87
FIX::SessionSettings
Container for setting dictionaries mapped to sessions.
Definition: SessionSettings.h:130
FIX::Session::disconnect
void disconnect()
Definition: Session.cpp:542
FIX::SocketInitiator::onPoll
bool onPoll(double timeout)
Implemented to connect and poll for events.
Definition: SocketInitiator.cpp:108
FIX::Initiator::start
void start()
Start initiator.
Definition: Initiator.cpp:190
FIX::LogFactory
This interface must be implemented to create a Log.
Definition: Log.h:42
FIX::SocketConnection::onTimeout
void onTimeout()
Definition: SocketConnection.cpp:234
FIX::Session::getSessionID
const SessionID & getSessionID() const
Definition: Session.h:75
FIX::MessageStoreFactory
This interface must be implemented to create a MessageStore.
Definition: MessageStore.h:41
FIX::SocketInitiator::getHost
void getHost(const SessionID &, const Dictionary &, std::string &, short &)
Definition: SocketInitiator.cpp:231
FIX::SocketInitiator::m_connections
SocketConnections m_connections
Definition: SocketInitiator.h:71
FIX::Initiator::isLoggedOn
bool isLoggedOn()
Check to see if any sessions are currently logged on.
Definition: Initiator.cpp:269
FIX::SocketInitiator::onDisconnect
void onDisconnect(SocketConnector &, int)
Definition: SocketInitiator.cpp:183
FIX::SocketInitiator::onData
bool onData(SocketConnector &, int)
Definition: SocketInitiator.cpp:175
FIX::SocketInitiator::m_lastConnect
time_t m_lastConnect
Definition: SocketInitiator.h:72
FIX::Initiator::setConnected
void setConnected(const SessionID &)
Definition: Initiator.cpp:154
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::SocketConnection
Encapsulates a socket file descriptor (single-threaded).
Definition: SocketConnection.h:46
FIX::Dictionary::getString
std::string getString(const std::string &, bool capitalize=false) const
Get a value as a string.
Definition: Dictionary.cpp:32
SocketInitiator.h
FIX::SessionID
Unique session id consists of BeginString, SenderCompID and TargetCompID.
Definition: SessionID.h:30
FIX::SocketInitiator::onInitialize
void onInitialize(const SessionSettings &)
Implemented to initialize initiator.
Definition: SocketInitiator.cpp:82
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
FIX::Dictionary::getInt
int getInt(const std::string &) const
Get a value as a int.
Definition: Dictionary.cpp:45
FIX::SocketInitiator::m_reconnectInterval
int m_reconnectInterval
Definition: SocketInitiator.h:73
FIX::SocketInitiator::onWrite
void onWrite(SocketConnector &, int)
Definition: SocketInitiator.cpp:166
FIX::Initiator::isStopped
bool isStopped()
Definition: Initiator.h:83
Session.h
FIX::SocketInitiator::m_sendBufSize
int m_sendBufSize
Definition: SocketInitiator.h:75
FIX::SocketConnector::connect
int connect(const std::string &address, int port, bool noDelay, int sendBufSize, int rcvBufSize)
Definition: SocketConnector.cpp:85
FIX::SocketInitiator::onError
void onError(SocketConnector &)
Definition: SocketInitiator.cpp:210
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::SocketInitiator::onTimeout
void onTimeout(SocketConnector &)
Definition: SocketInitiator.cpp:215
FIX::Initiator::setDisconnected
void setDisconnected(const SessionID &)
Definition: Initiator.cpp:163