HandlerImpl.cpp

Go to the documentation of this file.

/*
* Copyright Onix Solutions Limited [OnixS]. All rights reserved.
*
* This software owned by Onix Solutions Limited [OnixS] and is protected by copyright law
* and international copyright treaties.
*
* Access to and use of the software is governed by the terms of the applicable ONIXS Software
* Services Agreement (the Agreement) and Customer end user license agreements granting
* a non-assignable, non-transferable and non-exclusive license to use the software
* for it's own data processing purposes under the terms defined in the Agreement.
*
* Except as otherwise granted within the terms of the Agreement, copying or reproduction of any part
* of this source code or associated reference material to any other location for further reproduction
* or redistribution, and any amendments to this copyright notice, are expressly prohibited.
*
* Any reproduction or redistribution for sale or hiring of the Software not in accordance with
* the terms of the Agreement is a violation of copyright law.
*/
#include <boost/bind.hpp>
#include <boost/current_function.hpp>
#include <boost/format.hpp>

#include <system/Directory.h>
#include <util/Base64Encoding.h>
#include <util/HexEncoding.h>
#include <licensing/LicenseManager.h>

#include <OnixS/HandlerCore/HandlerLogger.h>
#include <OnixS/HandlerCore/Utils.h>
#include <OnixS/HandlerCore/Compiler.h>

#include <OnixS/SgxTitan/Trading/OUCH/Exception.h>

#include "ListenerHolder.h"
#include "Version.h"
#include "ProtocolFormatting.h"
#include "AuditTrailDumper.h"

#include "HandlerImpl.h"

using namespace OnixS::System;
using namespace OnixS::SgxTitan::Trading::Ouch;
using namespace OnixS::SgxTitan::Trading::Ouch::Implementation;


const OnixS::Licensing::ProductId productId = 153;

HandlerImpl::HandlerImpl(const HandlerSettings& settings)
    : Licensing::LicenseChecker(settings.licenseDirectory, productId)
    , base("Handler", nullptr, static_cast<Logging::LogLevel>(settings.logLevel))
    , settings_(settings)
    , logger_(new OnixS::HandlerCore::Common::HandlerLogger(OnixS::HandlerCore::Common::convertLogSettings(settings)))
    , listenerHolder_(new ListenerHolder(this, *logger_))
    , externalState_(HandlerState::Disconnected)
    , session_(nullptr)
{
    LogFacility facility("Handler", nullptr, Logging::LOG_LEVEL_DEBUG);

    forceLog(ONIXS_LOG_INFO[&facility]
                     << "OnixS C++ SGX TITAN OUCH Trading Handler, version "
                     << OnixS::SgxTitan::Trading::Ouch::Implementation::version()
                     << ", "
                     << os()
                     );

    forceLog(ONIXS_LOG_INFO[&facility] << settings_);

    runAuditTrail();
}

void HandlerImpl::onError(OnixS::ItchCore::SoupBinTCP::ReconnectableSession&, const std::string& what)
{
    BOOST_ASSERT(listenerHolder_);
    listenerHolder_->invokeError(ErrorCode::General, what);
}

void HandlerImpl::onWarning(OnixS::ItchCore::SoupBinTCP::ReconnectableSession&, const std::string& what)
{
    BOOST_ASSERT(listenerHolder_);
    listenerHolder_->invokeWarning(what);
}

void HandlerImpl::onStateChange(
        OnixS::ItchCore::SoupBinTCP::ReconnectableSession&,
        OnixS::ItchCore::SoupBinTCP::ReconnectableSessionState::Enum /*oldState*/,
        OnixS::ItchCore::SoupBinTCP::ReconnectableSessionState::Enum newState)
{
    externalState_.change(static_cast<HandlerState::Enum>(newState));

    BOOST_ASSERT(listenerHolder_);
    listenerHolder_->invokeStateChanged(externalState_.read());
}

void HandlerImpl::onData(OnixS::ItchCore::SoupBinTCP::ReconnectableSession&, const void* data, UInt16 size)
{
    BOOST_ASSERT(listenerHolder_);
    listenerHolder_->invokeReceivedBytes(static_cast<const char*>(data), size);

    processData(data, size);
}

UInt64 HandlerImpl::nextExpectedSequenceNumber() const
{
    checkState(HandlerState::Connected);

    Guard guard (sessionLock_);

    BOOST_ASSERT(session_);
    return session_->receivingSeqNo();
}

std::string HandlerImpl::sessionId() const
{
    checkState(HandlerState::Connected);

    Guard guard (sessionLock_);

    BOOST_ASSERT(session_);
    return session_->sessionId();
}

void HandlerImpl::runAuditTrail()
{
    if (!settings_.auditTrailFile.empty())
    {
        std::string auditTrailDir = settings_.auditTrailDirectory;

        if (auditTrailDir.find_last_of("\\/") != auditTrailDir.length() - 1)
            auditTrailDir.append("/");

        if (!OnixS::System::Directory::exists(auditTrailDir))
            OnixS::System::Directory::create(auditTrailDir);

        try
        {
            auditTrailDumper_.reset(
                    new AuditTrailDumper(
                            auditTrailDir + settings_.auditTrailFile + ".txt",
                            this,
                            *logger_,
                            settings_.auditTrailThreadAffinity,
                            settings_.auditTrailFileMaxSize));
        }
        catch (const std::exception& ex)
        {
            throw OperationException(BOOST_CURRENT_FUNCTION, OnixS::Util::format("Can't create Audit Trail file: %s", ex.what()).c_str());
        }
        catch (...)
        {
            throw OperationException(BOOST_CURRENT_FUNCTION, "Can't create Audit Trail file.");
        }
    }
}

HandlerImpl::~HandlerImpl()
{
    try
    {
        if(externalState_.read() == HandlerState::Connected)
            disconnect();
    }
    catch (...) {}

    try
    {
        if (auditTrailDumper_)
            auditTrailDumper_->stop();
    }
    catch (...) {}
}

HandlerState::Enum HandlerImpl::getState()
{
    return externalState_.read();
}

void HandlerImpl::commonLog(int logLevel, const char * msg, size_t length)
{
    BOOST_ASSERT(msg != nullptr);
    BOOST_ASSERT(length != 0);

    if (logLevel < OnixS::Logging::MIN_LOG_LEVEL || logLevel > OnixS::Logging::MAX_LOG_LEVEL)
        throw OperationException (
                BOOST_CURRENT_FUNCTION,
                (boost::format("An error occurred during log operation. Invalid log level (logLevel=%d).")
                    % logLevel).str().c_str()
        );

    static const std::string userLogFacilityName = "USER";

    log (
            OnixS::Logging::LogMsg(
                    OnixS::Logging::LogLevel(logLevel))[userLogFacilityName] << OnixS::Util::ValuePtr(msg, length)
            );
}

void HandlerImpl::connect(const LogonSettings& logonSettings,
    const std::string& hostName,
    const UInt16       port,
    const std::string& localNetworkInterface)
{
    ItchCore::SoupBinTCP::ReconnectableSessionSettings settings;
    settings.receiveTimeoutMs = settings_.receiveTimeoutMs;
    settings.sendTimeoutMs = settings_.sendTimeoutMs;
    settings.reconnectIntervalMs = settings_.connectionRetries.intervalSec * 1000;
    settings.reconnectAttempts = settings_.connectionRetries.number;

    ItchCore::SoupBinTCP::ConnectivitySettings connectivitySettings;
    connectivitySettings.connectionA = ItchCore::SoupBinTCP::ValidatedNetFeedConnection(hostName, port, localNetworkInterface);
    connectivitySettings.authenticationSettings.userName = logonSettings.userName;
    connectivitySettings.authenticationSettings.password = logonSettings.password;
    connectivitySettings.authenticationSettings.session = logonSettings.session;
    connectivitySettings.initialSequenceNumber = logonSettings.sequenceNumber;

    {
        Guard guard (sessionLock_);

        session_.reset(
                new ItchCore::SoupBinTCP::ReconnectableSession(settings, connectivitySettings, this, this)
        );

        session_->setLogger(logger_.get());

        session_->connect();
    }

    checkState(HandlerState::Connected);
}

void HandlerImpl::disconnect()
{
    checkState(HandlerState::Connected);

    Guard guard (sessionLock_);
    
    if (session_)
    {
        session_->disconnect(false);
        session_.reset();
    }
}

void HandlerImpl::disconnectAsync()
{
    checkState(HandlerState::Connected);

    Guard guard (sessionLock_);

    if (session_)
    {
        session_->disconnect(true);
    }
}

void HandlerImpl::send(const OutgoingMessage& msg, bool warmupMode)
{
    checkState(HandlerState::Connected);

    if (auditTrailDumper_)
        auditTrailDumper_->onOutboundMessage(msg, warmupMode);
    
    if (warmupMode)
        return;

    log(ONIXS_LOG_DEBUG[this] << "Sending " << msg);

    Guard guard (sessionLock_);

    BOOST_ASSERT(session_);
    session_->sendData(msg.binary(), msg.binarySize());
}

void HandlerImpl::checkState(HandlerState::Enum expectedState) const
{
    const HandlerState::Enum state = externalState_.read();
    if (state!= expectedState)
        throw OnixS::DomainException(
                (boost::format("Wrong state = %s, expected = %s.") % toStr(state) % toStr(expectedState)).str()
        );
}

std::string HandlerImpl::os()
{
#if BOOST_OS_WINDOWS
    return "Windows";
#elif BOOST_OS_UNIX
    std::string res = "Linux";
    res += ", ";
    res += __VERSION__;
    return res;
#else
#   error "Unknown OS"
#endif
}