SocketConnection.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 "SocketConnection.h"
#include "SocketAcceptor.h"
#include "SocketConnector.h"
#include "SocketInitiator.h"
#include "Session.h"
#include "Utility.h"
namespace FIX
{
SocketConnection::SocketConnection( int s, Sessions sessions,
SocketMonitor* pMonitor )
: m_socket( s ), m_sendLength( 0 ),
m_sessions(sessions), m_pSession( 0 ), m_pMonitor( pMonitor )
{
FD_ZERO( &m_fds );
FD_SET( m_socket, &m_fds );
}
SocketConnection::SocketConnection( SocketInitiator& i,
const SessionID& sessionID, int s,
SocketMonitor* pMonitor )
: m_socket( s ), m_sendLength( 0 ),
m_pSession( i.getSession( sessionID, *this ) ),
m_pMonitor( pMonitor )
{
FD_ZERO( &m_fds );
FD_SET( m_socket, &m_fds );
m_sessions.insert( sessionID );
}
SocketConnection::~SocketConnection()
{
if ( m_pSession )
Session::unregisterSession( m_pSession->getSessionID() );
}
bool SocketConnection::send( const std::string& msg )
{
Locker l( m_mutex );
m_sendQueue.push_back( msg );
processQueue();
signal();
return true;
}
bool SocketConnection::processQueue()
{
Locker l( m_mutex );
if( !m_sendQueue.size() ) return true;
struct timeval timeout = { 0, 0 };
fd_set writeset = m_fds;
if( select( 1 + m_socket, 0, &writeset, 0, &timeout ) <= 0 )
return false;
const std::string& msg = m_sendQueue.front();
ssize_t result = socket_send
( m_socket, msg.c_str() + m_sendLength, msg.length() - m_sendLength );
if( result > 0 )
m_sendLength += result;
if( m_sendLength == msg.length() )
{
m_sendLength = 0;
m_sendQueue.pop_front();
}
return !m_sendQueue.size();
}
void SocketConnection::disconnect()
{
if ( m_pMonitor )
m_pMonitor->drop( m_socket );
}
bool SocketConnection::read( SocketConnector& s )
{
if ( !m_pSession ) return false;
try
{
readFromSocket();
readMessages( s.getMonitor() );
}
catch( SocketRecvFailed& e )
{
m_pSession->getLog()->onEvent( e.what() );
return false;
}
return true;
}
bool SocketConnection::read( SocketAcceptor& a, SocketServer& s )
{
std::string msg;
try
{
if ( !m_pSession )
{
struct timeval timeout = { 1, 0 };
fd_set readset = m_fds;
while( !readMessage( msg ) )
{
int result = select( 1 + m_socket, &readset, 0, 0, &timeout );
if( result > 0 )
readFromSocket();
else if( result == 0 )
return false;
else if( result < 0 )
return false;
}
m_pSession = Session::lookupSession( msg, true );
if( !isValidSession() )
{
m_pSession = 0;
if( a.getLog() )
{
a.getLog()->onEvent( "Session not found for incoming message: " + msg );
a.getLog()->onIncoming( msg );
}
}
if( m_pSession )
m_pSession = a.getSession( msg, *this );
if( m_pSession )
m_pSession->next( msg, UtcTimeStamp() );
if( !m_pSession )
{
s.getMonitor().drop( m_socket );
return false;
}
Session::registerSession( m_pSession->getSessionID() );
return true;
}
else
{
readFromSocket();
readMessages( s.getMonitor() );
return true;
}
}
catch ( SocketRecvFailed& e )
{
if( m_pSession )
m_pSession->getLog()->onEvent( e.what() );
s.getMonitor().drop( m_socket );
}
catch ( InvalidMessage& )
{
s.getMonitor().drop( m_socket );
}
return false;
}
bool SocketConnection::isValidSession()
{
if( m_pSession == 0 )
return false;
SessionID sessionID = m_pSession->getSessionID();
if( Session::isSessionRegistered(sessionID) )
return false;
return !( m_sessions.find(sessionID) == m_sessions.end() );
}
void SocketConnection::readFromSocket()
throw( SocketRecvFailed )
{
ssize_t size = socket_recv( m_socket, m_buffer, sizeof(m_buffer) );
if( size <= 0 ) throw SocketRecvFailed( size );
m_parser.addToStream( m_buffer, size );
}
bool SocketConnection::readMessage( std::string& msg )
{
try
{
return m_parser.readFixMessage( msg );
}
catch ( MessageParseError& ) {}
return true;
}
void SocketConnection::readMessages( SocketMonitor& s )
{
if( !m_pSession ) return;
std::string msg;
while( readMessage( msg ) )
{
try
{
m_pSession->next( msg, UtcTimeStamp() );
}
catch ( InvalidMessage& )
{
if( !m_pSession->isLoggedOn() )
s.drop( m_socket );
}
}
}
void SocketConnection::onTimeout()
{
if ( m_pSession ) m_pSession->next();
}
} // namespace FIX
Session.h
SocketAcceptor.h
SocketConnection.h
SocketConnector.h
SocketInitiator.h
Utility.h
FIX::Acceptor::getSession
Session * getSession(const std::string &msg, Responder &)
Definition Acceptor.cpp:104
FIX::Acceptor::getLog
Log * getLog()
Definition Acceptor.h:59
FIX::Locker
Locks/Unlocks a mutex using RAII.
Definition Mutex.h:96
FIX::Log::onIncoming
virtual void onIncoming(const std::string &)=0
FIX::Log::onEvent
virtual void onEvent(const std::string &)=0
FIX::Parser::addToStream
void addToStream(const char *str, size_t len)
Definition Parser.h:48
FIX::Parser::readFixMessage
bool readFixMessage(std::string &str)
Definition Parser.cpp:59
FIX::Session::registerSession
static Session * registerSession(const SessionID &)
Definition Session.cpp:1537
FIX::Session::unregisterSession
static void unregisterSession(const SessionID &)
Definition Session.cpp:1547
FIX::Session::lookupSession
static Session * lookupSession(const SessionID &)
Definition Session.cpp:1496
FIX::Session::getSessionID
const SessionID & getSessionID() const
Definition Session.h:75
FIX::Session::getLog
Log * getLog()
Definition Session.h:227
FIX::Session::next
void next()
Definition Session.cpp:125
FIX::Session::isSessionRegistered
static bool isSessionRegistered(const SessionID &)
Definition Session.cpp:1531
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::SocketAcceptor
Socket implementation of Acceptor.
Definition SocketAcceptor.h:37
FIX::SocketConnection::m_buffer
char m_buffer[BUFSIZ]
Definition SocketConnection.h:90
FIX::SocketConnection::m_pSession
Session * m_pSession
Definition SocketConnection.h:96
FIX::SocketConnection::~SocketConnection
virtual ~SocketConnection()
Definition SocketConnection.cpp:56
FIX::SocketConnection::readMessages
void readMessages(SocketMonitor &s)
Definition SocketConnection.cpp:215
FIX::SocketConnection::m_fds
fd_set m_fds
Definition SocketConnection.h:99
FIX::SocketConnection::signal
void signal()
Definition SocketConnection.h:62
FIX::SocketConnection::m_socket
int m_socket
Definition SocketConnection.h:89
FIX::SocketConnection::m_mutex
Mutex m_mutex
Definition SocketConnection.h:98
FIX::SocketConnection::m_pMonitor
SocketMonitor * m_pMonitor
Definition SocketConnection.h:97
FIX::SocketConnection::processQueue
bool processQueue()
Definition SocketConnection.cpp:72
FIX::SocketConnection::m_sessions
Sessions m_sessions
Definition SocketConnection.h:95
FIX::SocketConnection::m_sendQueue
Queue m_sendQueue
Definition SocketConnection.h:93
FIX::SocketConnection::onTimeout
void onTimeout()
Definition SocketConnection.cpp:234
FIX::SocketConnection::send
bool send(const std::string &)
Definition SocketConnection.cpp:62
FIX::SocketConnection::readMessage
bool readMessage(std::string &msg)
Definition SocketConnection.cpp:205
FIX::SocketConnection::isValidSession
bool isValidSession()
Definition SocketConnection.cpp:187
FIX::SocketConnection::m_sendLength
unsigned m_sendLength
Definition SocketConnection.h:94
FIX::SocketConnection::m_parser
Parser m_parser
Definition SocketConnection.h:92
FIX::SocketConnection::disconnect
void disconnect()
Definition SocketConnection.cpp:100
FIX::SocketConnection::Sessions
std::set< SessionID > Sessions
Definition SocketConnection.h:49
FIX::SocketConnection::readFromSocket
void readFromSocket()
Definition SocketConnection.cpp:197
FIX::SocketConnection::read
bool read(SocketConnector &s)
Definition SocketConnection.cpp:106
FIX::SocketConnection::SocketConnection
SocketConnection(int s, Sessions sessions, SocketMonitor *pMonitor)
Definition SocketConnection.cpp:35
FIX::SocketConnector
Connects sockets to remote ports and addresses.
Definition SocketConnector.h:36
FIX::SocketConnector::getMonitor
SocketMonitor & getMonitor()
Definition SocketConnector.h:48
FIX::SocketInitiator
Socket implementation of Initiator.
Definition SocketInitiator.h:37
FIX::SocketMonitor
Monitors events on a collection of sockets.
Definition SocketMonitor.h:48
FIX::SocketMonitor::drop
bool drop(int socket)
Definition SocketMonitor.cpp:98
FIX::SocketServer
Listens for and accepts incoming socket connections on a port.
Definition SocketServer.h:57
FIX::SocketServer::getMonitor
SocketMonitor & getMonitor()
Definition SocketServer.h:70
FIX::UtcTimeStamp
Date and Time represented in UTC.
Definition FieldTypes.h:583
FIX
Definition Acceptor.cpp:35
FIX::socket_recv
ssize_t socket_recv(int s, char *buf, size_t length)
Definition Utility.cpp:170
FIX::socket_send
ssize_t socket_send(int s, const char *msg, size_t length)
Definition Utility.cpp:175
FIX::InvalidMessage
Not a recognizable message.
Definition Exceptions.h:81
FIX::MessageParseError
Unable to parse message.
Definition Exceptions.h:74
FIX::SocketRecvFailed
Socket recv operation failed.
Definition Exceptions.h:279