ThreadedSocketConnection.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 "ThreadedSocketConnection.h"
#include "ThreadedSocketAcceptor.h"
#include "ThreadedSocketInitiator.h"
#include "Session.h"
#include "Utility.h"
namespace FIX
{
ThreadedSocketConnection::ThreadedSocketConnection
( int s, Sessions sessions, Log* pLog )
: 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 );
}
ThreadedSocketConnection::ThreadedSocketConnection
( const SessionID& sessionID, int s,
const std::string& address, short port,
Log* pLog )
: 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 );
}
ThreadedSocketConnection::~ThreadedSocketConnection()
{
if ( m_pSession )
{
m_pSession->setResponder( 0 );
Session::unregisterSession( m_pSession->getSessionID() );
}
}
bool ThreadedSocketConnection::send( const std::string& msg )
{
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;
}
bool ThreadedSocketConnection::connect()
{
return socket_connect(getSocket(), m_address.c_str(), m_port) >= 0;
}
void ThreadedSocketConnection::disconnect()
{
m_disconnect = true;
socket_close( m_socket );
}
bool ThreadedSocketConnection::read()
{
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;
}
}
bool ThreadedSocketConnection::readMessage( std::string& msg )
throw( SocketRecvFailed )
{
try
{
return m_parser.readFixMessage( msg );
}
catch ( MessageParseError& ) {}
return true;
}
void ThreadedSocketConnection::processStream()
{
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;
}
}
}
}
bool ThreadedSocketConnection::setSession( const std::string& msg )
{
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;
}
} // namespace FIX
FIX::Session::lookupSession
static Session * lookupSession(const SessionID &)
Definition: Session.cpp:1403
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::readMessage
bool readMessage(std::string &msg)
Definition: ThreadedSocketConnection.cpp:139
FIX::ThreadedSocketConnection::m_port
int m_port
Definition: ThreadedSocketConnection.h:71
FIX::Session::unregisterSession
static void unregisterSession(const SessionID &)
Definition: Session.cpp:1454
FIX::MessageParseError
Unable to parse message.
Definition: Exceptions.h:73
Utility.h
FIX::Session::next
void next()
Definition: Session.cpp:125
FIX::ThreadedSocketConnection::m_pSession
Session * m_pSession
Definition: ThreadedSocketConnection.h:76
FIX::process_sleep
void process_sleep(double s)
Definition: Utility.cpp:444
FIX::ThreadedSocketConnection::send
bool send(const std::string &)
Definition: ThreadedSocketConnection.cpp:67
FIX::ThreadedSocketConnection::ThreadedSocketConnection
ThreadedSocketConnection(int s, Sessions sessions, Log *pLog)
Definition: ThreadedSocketConnection.cpp:35
FIX::ThreadedSocketConnection::m_socket
int m_socket
Definition: ThreadedSocketConnection.h:67
ThreadedSocketInitiator.h
FIX::Session::isSessionRegistered
static bool isSessionRegistered(const SessionID &)
Definition: Session.cpp:1438
FIX::ThreadedSocketConnection::read
bool read()
Definition: ThreadedSocketConnection.cpp:91
FIX::ThreadedSocketConnection::processStream
void processStream()
Definition: ThreadedSocketConnection.cpp:150
FIX::Log
This interface must be implemented to log messages and events.
Definition: Log.h:81
FIX
Definition: Acceptor.cpp:34
FIX::Session::registerSession
static Session * registerSession(const SessionID &)
Definition: Session.cpp:1444
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::SocketRecvFailed
Socket recv operation failed.
Definition: Exceptions.h:280
FIX::ThreadedSocketConnection::m_pLog
Log * m_pLog
Definition: ThreadedSocketConnection.h:73
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::ThreadedSocketConnection::Sessions
std::set< SessionID > Sessions
Definition: ThreadedSocketConnection.h:47
FIX::Parser::addToStream
void addToStream(const char *str, size_t len)
Definition: Parser.h:48
FIX::Session::getSessionID
const SessionID & getSessionID() const
Definition: Session.h:75
FIX::ThreadedSocketConnection::~ThreadedSocketConnection
virtual ~ThreadedSocketConnection()
Definition: ThreadedSocketConnection.cpp:58
FIX::ThreadedSocketConnection::m_sessions
Sessions m_sessions
Definition: ThreadedSocketConnection.h:75
FIX::ThreadedSocketConnection::setSession
bool setSession(const std::string &msg)
Definition: ThreadedSocketConnection.cpp:175
FIX::ThreadedSocketConnection::disconnect
void disconnect()
Definition: ThreadedSocketConnection.cpp:85
FIX::Log::onIncoming
virtual void onIncoming(const std::string &)=0
FIX::Parser::readFixMessage
bool readFixMessage(std::string &str)
Definition: Parser.cpp:59
FIX::socket_send
ssize_t socket_send(int s, const char *msg, size_t length)
Definition: Utility.cpp:153
FIX::TYPE::UtcTimeStamp
Definition: FieldTypes.h:674
FIX::Session::setResponder
void setResponder(Responder *pR)
Definition: Session.h:197
FIX::ThreadedSocketConnection::m_disconnect
bool m_disconnect
Definition: ThreadedSocketConnection.h:77
FIX::ThreadedSocketConnection::connect
bool connect()
Definition: ThreadedSocketConnection.cpp:80
ThreadedSocketConnection.h
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::InvalidMessage
Not a recognizable message.
Definition: Exceptions.h:80
ThreadedSocketAcceptor.h
FIX::ThreadedSocketConnection::m_address
std::string m_address
Definition: ThreadedSocketConnection.h:70
Session.h
FIX::Session::isLoggedOn
bool isLoggedOn()
Definition: Session.h:65