spark/src/server/sparkd/fix/FixRouter.cpp

#include "FixRouter.h"

#include "IFixSession.h"
#include "SecurityIndex.h"
#include "lib/md/MDConsumerHandle.h"
#include "lib/utils/LockBox.h"

#include <quickfix/FileStore.h>
#include <quickfix/SocketAcceptor.h>
#include <quickfix/SessionSettings.h>
#include <quickfix/FileLog.h>
#include <quickfix/Message.h>
#include <quickfix/SessionID.h>
#include <quickfix/fix44/Reject.h>
#include <quickfix/fix44/BusinessMessageReject.h>

#include <boost/algorithm/string/replace.hpp>
 
#include <string>
#include <iostream>
#include <fstream>
#include <memory>

namespace bts
{
namespace fix
{
      
struct FixRouter::PImpl
{
   utils::LockBox< boost::container::flat_map< FIX::SessionID, std::shared_ptr<IFixSession> > > sessions;
   SecurityIndex index;
   std::mutex lock;
   Factory factory;
   
   PImpl(md::MDClient& md, Factory factory) : 
      sessions(),
      index(md),
      lock(),
      factory(factory)
   {
   }   
};
   
FixRouter::FixRouter(md::MDClient& md, Factory factory) : me(new PImpl(md, factory))
{
}
   
FixRouter::~FixRouter()
{
}
   
void FixRouter::send(FIX::Message& mess, const FIX::SessionID& sess) 
{
   try
   {
      std::lock_guard<std::mutex> guard(me->lock);   
      FIX::Session::sendToTarget(mess, sess);
   }
   catch(FIX::SessionNotFound)
   {
      bts_assert_msg(false, "session not found " + sess.toString());
   }
   catch(std::exception ex)
   {
      LOG(Fix, Error, "session error {} {}", sess.toString(), std::string(ex.what()));
   }
}
   
void FixRouter::send_session_reject(const FIX::SessionID& sess, const std::string& text, const FIX::Message* refmessage)
{
   FIX44::Reject rej;
   if(refmessage && refmessage->getHeader().isSetField(FIX::FIELD::MsgSeqNum))
      rej.setField(FIX::FIELD::RefSeqNum, refmessage->getHeader().getField(FIX::FIELD::MsgSeqNum));
   rej.setField(FIX::FIELD::Text, text);
   send(rej, sess);
}
   
void FixRouter::send_biz_reject(const FIX::SessionID sess, const std::string& text, const FIX::Message& refmessage, int reason_code) 
{
   FIX44::BusinessMessageReject rej;
   rej.setField(FIX::FIELD::RefMsgType, refmessage.getHeader().getField(FIX::FIELD::MsgType));
   rej.setField(FIX::FIELD::RefSeqNum, refmessage.getHeader().getField(FIX::FIELD::MsgSeqNum));
   rej.setField(FIX::FIELD::BusinessRejectReason, std::to_string(reason_code));
   rej.setField(FIX::FIELD::Text, text);
   send(rej, sess);
}
   
void FixRouter::fromAdmin(const FIX::Message& message, const FIX::SessionID& sessionID) 
                          throw(FIX::FieldNotFound, FIX::IncorrectDataFormat, FIX::IncorrectTagValue, FIX::RejectLogon)
{
   try
   {
      session(sessionID).fromAdmin(message, sessionID);
   }
   catch (FIX::FieldNotFound ex) { handle_field_not_found(ex, sessionID, message); }
   catch (FIX::UnsupportedMessageType) { throw; }
   catch (FIX::IncorrectTagValue) { throw; }
   catch (std::exception ex) { handle_session_exception(ex, sessionID, &message); }   
}
   
void FixRouter::fromApp(const FIX::Message &message,const FIX::SessionID &sessionID) 
                        throw(FIX::FieldNotFound, FIX::IncorrectDataFormat, FIX::IncorrectTagValue, FIX::UnsupportedMessageType)
{
   try
   {
      session(sessionID).fromApp(message, sessionID);
   }
   catch (FIX::FieldNotFound ex) { handle_field_not_found(ex, sessionID, message); }
   catch (FIX::UnsupportedMessageType) { throw; }
   catch (FIX::IncorrectTagValue) { throw; }
   catch (std::exception ex) { handle_session_exception(ex, sessionID, &message); }   
}
                     
void FixRouter::onCreate(const FIX::SessionID& sessionID)
{
   try
   {
      session(sessionID).onCreate(sessionID);    
   }
   catch(std::exception ex)
   {
      handle_session_exception(ex, sessionID);
   }       
}
   
void FixRouter::onLogon(const FIX::SessionID& sessionID)
{
   try
   {
      session(sessionID).onLogon(sessionID);
   }
   catch (std::exception ex)
   {
      handle_session_exception(ex, sessionID);
   }   
}
   
void FixRouter::onLogout(const FIX::SessionID& sessionID)
{
   try
   {
      //TODO stackoverflow?
      session(sessionID).onLogout(sessionID);
   }
   catch (std::exception ex)
   {
      handle_session_exception(ex, sessionID);
   }   
}
   
void FixRouter::toAdmin(FIX::Message& message, const FIX::SessionID& sessionID)
{
   try
   {
      session(sessionID).toAdmin(message, sessionID);
   }
   catch (FIX::FieldNotFound ex) { handle_field_not_found(ex, sessionID, message); }
   catch (FIX::UnsupportedMessageType) {  throw; }
   catch (FIX::IncorrectTagValue) { throw; }
   catch (std::exception ex) { handle_session_exception(ex, sessionID, &message); }   
}
   
void FixRouter::toApp(FIX::Message& message, const FIX::SessionID& sessionID) throw(FIX::DoNotSend)
{
   try
   {
      session(sessionID).toApp(message, sessionID);
   }
   catch (FIX::FieldNotFound ex) { handle_field_not_found(ex, sessionID, message); }
   catch (FIX::UnsupportedMessageType) { throw FIX::DoNotSend() ; }
   catch (FIX::IncorrectTagValue) { throw FIX::DoNotSend() ; }
   catch (std::exception ex) { handle_session_exception(ex, sessionID, &message); }   
}

IFixSession& FixRouter::session(const FIX::SessionID& sid)
{
   auto ptr = me->sessions.lock([this, sid] (auto& sessions)
   {
      auto it = sessions.find(sid);
      if(it == sessions.end())
      {
         auto ptr = me->factory(*this, sid);
         sessions[sid] = ptr;
         return ptr;
      }   
      else
         return it->second;      
   });
   return *ptr;
}
   
void FixRouter::handle_session_exception(const std::exception& ex, const FIX::SessionID& sessionID, const FIX::Message* message)
{
   std::string msg = "==ERROR in session " + sessionID.toString() + "== ";
   msg +=  (message ? message->getHeader().getField(FIX::FIELD::MsgType) : "");
   msg +=  std::string(ex.what());

   LOG(Fix, Error, msg.c_str());
   //now shut this fucker down
   try
   {
      auto sess = FIX::Session::lookupSession(sessionID);
      if(sess != nullptr)
         sess->disconnect();
   }
   catch(...) { }   
}

void FixRouter::handle_field_not_found(const FIX::FieldNotFound& ex, const FIX::SessionID& sess, const FIX::Message& message)
{
   if(ex.field > 0)
   {
      std::string text = "Field " + std::to_string(ex.field) + " was expected but not found";
      send_biz_reject(sess, text, message, 5 );
   }
   //QF does not include the tag number....
   //throw new FieldNotFoundException($"Field {ex.Field} was expected but not found", ex);
   else
      throw ex;  
}

std::vector<md::SecurityPtr> FixRouter::get_securities()
{
   return me->index.get_securities();
}
   
const md::SecurityPtr FixRouter::lookup(md::security_key key) 
{
   return me->index.lookup(key);
}
   
const md::SecurityPtr FixRouter::lookup(const std::string& desc)
{
   return me->index.lookup(desc);
}

     
}
}