Initiator.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 "Initiator.h"
#include "Utility.h"
#include "Session.h"
#include "SessionFactory.h"
#include "HttpServer.h"
#include <algorithm>
#include <fstream>
namespace FIX
{
Initiator::Initiator( Application& application,
MessageStoreFactory& messageStoreFactory,
const SessionSettings& settings ) throw( ConfigError )
: m_threadid( 0 ),
m_application( application ),
m_messageStoreFactory( messageStoreFactory ),
m_settings( settings ),
m_pLogFactory( 0 ),
m_pLog( 0 ),
m_firstPoll( true ),
m_stop( true )
{ initialize(); }
Initiator::Initiator( Application& application,
MessageStoreFactory& messageStoreFactory,
const SessionSettings& settings,
LogFactory& logFactory ) throw( ConfigError )
: m_threadid( 0 ),
m_application( application ),
m_messageStoreFactory( messageStoreFactory ),
m_settings( settings ),
m_pLogFactory( &logFactory ),
m_pLog( logFactory.create() ),
m_firstPoll( true ),
m_stop( true )
{ initialize(); }
void Initiator::initialize() throw ( ConfigError )
{
std::set < SessionID > sessions = m_settings.getSessions();
std::set < SessionID > ::iterator i;
if ( !sessions.size() )
throw ConfigError( "No sessions defined" );
SessionFactory factory( m_application, m_messageStoreFactory,
m_pLogFactory );
for ( i = sessions.begin(); i != sessions.end(); ++i )
{
if ( m_settings.get( *i ).getString( "ConnectionType" ) == "initiator" )
{
m_sessionIDs.insert( *i );
m_sessions[ *i ] = factory.create( *i, m_settings.get( *i ) );
setDisconnected( *i );
}
}
if ( !m_sessions.size() )
throw ConfigError( "No sessions defined for initiator" );
}
Initiator::~Initiator()
{
Sessions::iterator i;
for ( i = m_sessions.begin(); i != m_sessions.end(); ++i )
delete i->second;
if( m_pLogFactory && m_pLog )
m_pLogFactory->destroy( m_pLog );
}
Session* Initiator::getSession( const SessionID& sessionID,
Responder& responder )
{
Sessions::iterator i = m_sessions.find( sessionID );
if ( i != m_sessions.end() )
{
i->second->setResponder( &responder );
return i->second;
}
return 0;
}
Session* Initiator::getSession( const SessionID& sessionID ) const
{
Sessions::const_iterator i = m_sessions.find( sessionID );
if( i != m_sessions.end() )
return i->second;
else
return 0;
}
const Dictionary* const Initiator::getSessionSettings( const SessionID& sessionID ) const
{
try
{
return &m_settings.get( sessionID );
}
catch( ConfigError& )
{
return 0;
}
}
void Initiator::connect()
{
Locker l(m_mutex);
SessionIDs disconnected = m_disconnected;
SessionIDs::iterator i = disconnected.begin();
for ( ; i != disconnected.end(); ++i )
{
Session* pSession = Session::lookupSession( *i );
if ( pSession->isEnabled() && pSession->isSessionTime(UtcTimeStamp()) )
doConnect( *i, m_settings.get( *i ));
}
}
void Initiator::setPending( const SessionID& sessionID )
{
Locker l(m_mutex);
m_pending.insert( sessionID );
m_connected.erase( sessionID );
m_disconnected.erase( sessionID );
}
void Initiator::setConnected( const SessionID& sessionID )
{
Locker l(m_mutex);
m_pending.erase( sessionID );
m_connected.insert( sessionID );
m_disconnected.erase( sessionID );
}
void Initiator::setDisconnected( const SessionID& sessionID )
{
Locker l(m_mutex);
m_pending.erase( sessionID );
m_connected.erase( sessionID );
m_disconnected.insert( sessionID );
}
bool Initiator::isPending( const SessionID& sessionID )
{
Locker l(m_mutex);
return m_pending.find( sessionID ) != m_pending.end();
}
bool Initiator::isConnected( const SessionID& sessionID )
{
Locker l(m_mutex);
return m_connected.find( sessionID ) != m_connected.end();
}
bool Initiator::isDisconnected( const SessionID& sessionID )
{
Locker l(m_mutex);
return m_disconnected.find( sessionID ) != m_disconnected.end();
}
void Initiator::start() throw ( ConfigError, RuntimeError )
{
m_stop = false;
onConfigure( m_settings );
onInitialize( m_settings );
HttpServer::startGlobal( m_settings );
if( !thread_spawn( &startThread, this, m_threadid ) )
throw RuntimeError("Unable to spawn thread");
}
void Initiator::block() throw ( ConfigError, RuntimeError )
{
m_stop = false;
onConfigure( m_settings );
onInitialize( m_settings );
startThread(this);
}
bool Initiator::poll( double timeout ) throw ( ConfigError, RuntimeError )
{
if( m_firstPoll )
{
m_stop = false;
onConfigure( m_settings );
onInitialize( m_settings );
connect();
m_firstPoll = false;
}
return onPoll( timeout );
}
void Initiator::stop( bool force )
{
if( isStopped() ) return;
HttpServer::stopGlobal();
std::vector<Session*> enabledSessions;
SessionIDs connected;
{
Locker l(m_mutex);
connected = m_connected;
}
SessionIDs::iterator i = connected.begin();
for ( ; i != connected.end(); ++i )
{
Session* pSession = Session::lookupSession(*i);
if( pSession && pSession->isEnabled() )
{
enabledSessions.push_back( pSession );
pSession->logout();
}
}
if( !force )
{
for ( int second = 1; second <= 10 && isLoggedOn(); ++second )
process_sleep( 1 );
}
{
Locker l(m_mutex);
for ( i = connected.begin(); i != connected.end(); ++i )
setDisconnected( Session::lookupSession(*i)->getSessionID() );
}
m_stop = true;
onStop();
if( m_threadid )
thread_join( m_threadid );
m_threadid = 0;
std::vector<Session*>::iterator session = enabledSessions.begin();
for( ; session != enabledSessions.end(); ++session )
(*session)->logon();
}
bool Initiator::isLoggedOn()
{
Locker l(m_mutex);
SessionIDs connected = m_connected;
SessionIDs::iterator i = connected.begin();
for ( ; i != connected.end(); ++i )
{
if( Session::lookupSession(*i)->isLoggedOn() )
return true;
}
return false;
}
THREAD_PROC Initiator::startThread( void* p )
{
Initiator * pInitiator = static_cast < Initiator* > ( p );
pInitiator->onStart();
return 0;
}
}
HttpServer.h
Initiator.h
Session.h
SessionFactory.h
Utility.h
THREAD_PROC
#define THREAD_PROC
Definition: Utility.h:184
FIX::Application
This interface must be implemented to define what your FIX application does.
Definition: Application.h:44
FIX::Dictionary
For storage and retrieval of key/value pairs.
Definition: Dictionary.h:37
FIX::Dictionary::getString
std::string getString(const std::string &, bool capitalize=false) const
Get a value as a string.
Definition: Dictionary.cpp:32
FIX::HttpServer::startGlobal
static void startGlobal(const SessionSettings &)
Definition: HttpServer.cpp:37
FIX::HttpServer::stopGlobal
static void stopGlobal()
Definition: HttpServer.cpp:53
FIX::Initiator
Base for classes which act as an initiator for establishing connections.
Definition: Initiator.h:52
FIX::Initiator::getSessionSettings
const Dictionary *const getSessionSettings(const SessionID &sessionID) const
Definition: Initiator.cpp:119
FIX::Initiator::onStart
virtual void onStart()=0
Implemented to start connecting to targets.
FIX::Initiator::m_threadid
thread_id m_threadid
Definition: Initiator.h:135
FIX::Initiator::m_connected
SessionIDs m_connected
Definition: Initiator.h:131
FIX::Initiator::isLoggedOn
bool isLoggedOn()
Check to see if any sessions are currently logged on.
Definition: Initiator.cpp:275
FIX::Initiator::m_mutex
Mutex m_mutex
Definition: Initiator.h:146
FIX::Initiator::setConnected
void setConnected(const SessionID &)
Definition: Initiator.cpp:154
FIX::Initiator::m_pLog
Log * m_pLog
Definition: Initiator.h:142
FIX::Initiator::SessionIDs
std::set< SessionID > SessionIDs
Definition: Initiator.h:124
FIX::Initiator::setPending
void setPending(const SessionID &)
Definition: Initiator.cpp:145
FIX::Initiator::startThread
static THREAD_PROC startThread(void *p)
Definition: Initiator.cpp:289
FIX::Initiator::initialize
void initialize()
Definition: Initiator.cpp:63
FIX::Initiator::stop
void stop(bool force=false)
Stop initiator.
Definition: Initiator.cpp:226
FIX::Initiator::setDisconnected
void setDisconnected(const SessionID &)
Definition: Initiator.cpp:163
FIX::Initiator::connect
void connect()
Definition: Initiator.cpp:131
FIX::Initiator::Initiator
Initiator(Application &, MessageStoreFactory &, const SessionSettings &)
Definition: Initiator.cpp:36
FIX::Initiator::m_pending
SessionIDs m_pending
Definition: Initiator.h:130
FIX::Initiator::start
void start()
Start initiator.
Definition: Initiator.cpp:190
FIX::Initiator::doConnect
virtual void doConnect(const SessionID &, const Dictionary &)=0
Implemented to connect a session to its target.
FIX::Initiator::getSession
Session * getSession(const SessionID &sessionID, Responder &)
Definition: Initiator.cpp:98
FIX::Initiator::m_sessions
Sessions m_sessions
Definition: Initiator.h:128
FIX::Initiator::poll
bool poll(double timeout=0.0)
Poll the initiator.
Definition: Initiator.cpp:212
FIX::Initiator::isDisconnected
bool isDisconnected(const SessionID &)
Definition: Initiator.cpp:184
FIX::Initiator::m_disconnected
SessionIDs m_disconnected
Definition: Initiator.h:132
FIX::Initiator::block
void block()
Block on the initiator.
Definition: Initiator.cpp:203
FIX::Initiator::isConnected
bool isConnected(const SessionID &)
Definition: Initiator.cpp:178
FIX::Initiator::onConfigure
virtual void onConfigure(const SessionSettings &)
Implemented to configure acceptor.
Definition: Initiator.h:110
FIX::Initiator::m_pLogFactory
LogFactory * m_pLogFactory
Definition: Initiator.h:141
FIX::Initiator::m_sessionIDs
SessionIDs m_sessionIDs
Definition: Initiator.h:129
FIX::Initiator::onInitialize
virtual void onInitialize(const SessionSettings &)
Implemented to initialize initiator.
Definition: Initiator.h:112
FIX::Initiator::m_settings
SessionSettings m_settings
Definition: Initiator.h:139
FIX::Initiator::isStopped
bool isStopped()
Definition: Initiator.h:83
FIX::Initiator::isPending
bool isPending(const SessionID &)
Definition: Initiator.cpp:172
FIX::Initiator::~Initiator
virtual ~Initiator()
Definition: Initiator.cpp:88
FIX::Initiator::m_stop
bool m_stop
Definition: Initiator.h:145
FIX::Initiator::m_application
Application & m_application
Definition: Initiator.h:136
FIX::Initiator::m_messageStoreFactory
MessageStoreFactory & m_messageStoreFactory
Definition: Initiator.h:137
FIX::Initiator::onStop
virtual void onStop()=0
Implemented to stop a running initiator.
FIX::Locker
Locks/Unlocks a mutex using RAII.
Definition: Mutex.h:96
FIX::LogFactory
This interface must be implemented to create a Log.
Definition: Log.h:43
FIX::LogFactory::destroy
virtual void destroy(Log *)=0
FIX::MessageStoreFactory
This interface must be implemented to create a MessageStore.
Definition: MessageStore.h:42
FIX::Responder
Interface implements sending on and disconnecting a transport.
Definition: Responder.h:35
FIX::SessionFactory
Responsible for creating Session objects.
Definition: SessionFactory.h:47
FIX::SessionFactory::create
Session * create(const SessionID &sessionID, const Dictionary &settings)
Definition: SessionFactory.cpp:41
FIX::Session
Maintains the state and implements the logic of a FIX session.
Definition: Session.h:46
FIX::Session::logout
void logout(const std::string &reason="")
Definition: Session.h:57
FIX::Session::getSessionID
const SessionID & getSessionID() const
Definition: Session.h:75
FIX::Session::isEnabled
bool isEnabled()
Definition: Session.h:59
FIX::Session::isSessionTime
bool isSessionTime(const UtcTimeStamp &time)
Definition: Session.h:108
FIX::Session::lookupSession
static Session * lookupSession(const SessionID &)
Definition: Session.cpp:1496
FIX::Session::isLoggedOn
bool isLoggedOn()
Definition: Session.h:65
FIX::SessionID
Unique session id consists of BeginString, SenderCompID and TargetCompID.
Definition: SessionID.h:31
FIX::SessionSettings
Container for setting dictionaries mapped to sessions.
Definition: SessionSettings.h:221
FIX::SessionSettings::getSessions
std::set< SessionID > getSessions() const
Definition: SessionSettings.cpp:160
FIX::SessionSettings::get
const Dictionary & get(const SessionID &) const
Get a dictionary for a session.
Definition: SessionSettings.cpp:127
FIX::UtcTimeStamp
Date and Time represented in UTC.
Definition: FieldTypes.h:583
FIX
Definition: Acceptor.cpp:35
FIX::thread_spawn
bool thread_spawn(THREAD_START_ROUTINE func, void *var, thread_id &thread)
Definition: Utility.cpp:416
FIX::process_sleep
void process_sleep(double s)
Definition: Utility.cpp:466
FIX::thread_join
void thread_join(thread_id thread)
Definition: Utility.cpp:437
FIX::ConfigError
Application is not configured correctly
Definition: Exceptions.h:88
FIX::RuntimeError
Application encountered serious error during runtime
Definition: Exceptions.h:95