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/regex.hpp>
#include <boost/scope_exit.hpp>
#include <boost/format.hpp>

#include <util/TextBuilder.h>

#include <OnixS/HandlerCore/FeedEngine/PacketContainer.h>
#include <OnixS/HandlerCore/TimeHelper.h>

#include <OnixS/SgxTitan/MarketData/Itch/Exception.h>
#include <OnixS/SgxTitan/MarketData/Itch/Replay.h>
#include <OnixS/SgxTitan/MarketData/Itch/BinaryMessage.h>
#include <OnixS/SgxTitan/MarketData/Itch/IncomingMessage.h>

#include <OnixS/ItchCore/MoldUDP64/PacketHeader.h>
#include <OnixS/ItchCore/MoldUDP64/RetransmissionService.h>

#include "FeedEngineImpl.h"
#include "ListenerHolder.h"
#include "MessageRepository.h"
#include "Version.h"
#include "GlimpseService.h"

#include "HandlerLogger.h"
#include "Formatting.h"

#include "HandlerImpl.h"

ONIXS_HANDLER_NAMESPACE_BEGIN

using namespace std;
using namespace HandlerCore::Common;
using namespace HandlerCore::MarketData;

HandlerImpl::HandlerImpl(const HandlerSettings& settings)
    : Licensing::LicenseChecker(settings.licenseDirectory, HANDLER_NAMESPACE::productId())
    , base (HANDLER_NAMESPACE::projectName(), nullptr, settings.maxPacketSize)
    , settings_(settings)
    , bookAllocator_(settings_.buildInternalOrderBooks ? new OrderBookAllocator(settings_.maxBooksObjectAmount) : nullptr)
    , logger_(nullptr)
    , listenerHolder_(nullptr)
    , realtimeFeeds_(nullptr)
    , gapResponseFeeds_(nullptr)
    , realtimePacketProcessor_(nullptr)
    , messageRepository_(nullptr)
    , retransmissionService_(nullptr)
    , glimpseService_(nullptr)
    , backtestingMode_(false)
{
    initLogger(settings);

    forceLog (ONIXS_LOG_INFO[this] 
            << HANDLER_NAMESPACE::projectDescription() <<
            ", version " << HANDLER_NAMESPACE::version()
            << ", " 
            << os()
        );

    forceLog(ONIXS_LOG_INFO[this] << settings.toString());

    try
    {
        listenerHolder_.reset(new ListenerHolder(this, logger_.get()));

        realtimePacketProcessor_.reset(
                new RealtimePacketProcessor(
                        settings_.lostPacketWaitTime,
                        settings_.outOfOrderPacketMaxInterval,
                        settings_.heartbeatInterval,
                        logger_.get(),
                        this,
                        lock(),
                        this
                )
        );

        messageRepository_.reset(new MessageRepository(*listenerHolder_.get(), settings_, this, logger_.get(), bookAllocator_.get()));
    }
    catch(const std::exception& ex)
    {
        log(ONIXS_LOG_ERROR[this] << "Exception in HandlerImpl constructor: " << ex.what());
        throw;
    }
    catch(...)
    {
        log(ONIXS_LOG_WARN [this] << "Unknown exception in HandlerImpl constructor");
        throw;
    }
}

HandlerImpl::~HandlerImpl()
{
    stop(true);
}

void HandlerImpl::setOrderBookIdFilter(const std::set<OrderBookId>& orderBookIdFilter)
{
    BOOST_ASSERT(messageRepository_.get() != nullptr);
    messageRepository_->setOrderBookIdFilter(orderBookIdFilter);
}

void HandlerImpl::bindFeedEngine(FeedEngine& feedEngine)
{
    setFeedEngine(FeedEngineImpl::getFeedEngine(feedEngine));

    forceLog(ONIXS_LOG_INFO[this] << "Bound FeedEngine");
    forceLog(ONIXS_LOG_INFO[this] << feedEngine.settings().toString());
}

void HandlerImpl::invokeError(const std::string& description)
{ 
    BOOST_ASSERT(listenerHolder_.get() != nullptr);
    listenerHolder_->invokeError(ErrorCode::General, description);

    log(ONIXS_LOG_ERROR[this] << description);
}

void HandlerImpl::invokeWarning (const std::string& description) 
{ 
    BOOST_ASSERT(listenerHolder_.get() != nullptr);
    listenerHolder_->invokeWarning(description);

    log(ONIXS_LOG_WARN[this] << description);
}

ListenerHolder& HandlerImpl::listenerHolder() 
{ 
    BOOST_ASSERT(listenerHolder_.get() != nullptr);
    return *listenerHolder_; 
}

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
}
    
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()
        );

    log (OnixS::Logging::LogMsg (OnixS::Logging::LogLevel (logLevel))[std::string("USER")] << OnixS::Util::ValuePtr (msg, length));
}

void HandlerImpl::start()
{
    BOOST_ASSERT(lock());

    lock()->acquire();

    BOOST_SCOPE_EXIT(this_){
            this_->lock()->release();
        } BOOST_SCOPE_EXIT_END


    if (settings_.glimpseFeed.serviceA.valid() || settings_.glimpseFeed.serviceB.valid())
    {
        glimpseService_.reset(new GlimpseService(settings_, this, logger_.get()));
        glimpseService_->subscribeOnFailure(boost::bind(&HandlerImpl::onGlimpseFailure, this, _1));
        glimpseService_->subscribeOnWarning(boost::bind(&HandlerImpl::onGlimpseWarning, this, _1));

        glimpseService_->subscribeOnMessage(boost::bind(&HandlerImpl::onGlimpseMessage, this, _1, _2));
        glimpseService_->subscribeOnRestarted(boost::bind(&HandlerImpl::onGlimpseRestarted, this));
    }

    base::start();
}

void HandlerImpl::stop(bool wait)
{
    BOOST_ASSERT(lock());

    if (retransmissionService_.get() != nullptr)
        retransmissionService_->stop();
    
    if (glimpseService_.get() && glimpseService_->inProgress())
        glimpseService_->stop(wait);

    lock()->acquire();

    BOOST_SCOPE_EXIT(this_) {
            this_->lock()->release();
        } BOOST_SCOPE_EXIT_END

    base::stop();
}

FE::MulticastFeedLayout::Enum getRetransmissionFeedLayout(const HandlerSettings& settings)
{
    if (settings.retransmissionFeed.serviceA.valid() && settings.retransmissionFeed.serviceB.valid())
        return FE::MulticastFeedLayout::AWithFailoverToB;

    if (settings.retransmissionFeed.serviceA.valid())
        return FE::MulticastFeedLayout::AOnly;

    if (settings.retransmissionFeed.serviceA.valid())
        return FE::MulticastFeedLayout::BOnly;

    throw std::invalid_argument("Retransmission Server is not set up");
}

void HandlerImpl::initRetransmissionService()
{
    try
    {
        const FE::MulticastFeedLayout::Enum layout = getRetransmissionFeedLayout(settings_);
        FE::ConnectionInfoList incrementalConnections;
        FE::NifsByFeedRole nifs;

        if (settings_.retransmissionFeed.serviceA.valid())
        {
            FE::ConnectionInfo connectionA;
            connectionA.role(FE::NetFeedRole::A);

            nifs.A(settings_.networkInterfaceA.empty() ? settings_.networkInterface : settings_.networkInterfaceA);

            connectionA.type(FE::NetFeedType::Historical);
            connectionA.id("RetransmissionA");
            connectionA.ip(settings_.retransmissionFeed.serviceA.address);
            connectionA.port(settings_.retransmissionFeed.serviceA.port);

            incrementalConnections.push_back(connectionA);
        }

        if (settings_.retransmissionFeed.serviceB.valid())
        {
            FE::ConnectionInfo connectionB;
            connectionB.role(FE::NetFeedRole::B);
            
            nifs.B(settings_.networkInterfaceA.empty() ? settings_.networkInterface : settings_.networkInterfaceB);

            connectionB.type(FE::NetFeedType::Historical);
            connectionB.id("RetransmissionB");
            connectionB.ip(settings_.retransmissionFeed.serviceB.address);
            connectionB.port(settings_.retransmissionFeed.serviceB.port);

            incrementalConnections.push_back(connectionB);
        }

        gapResponseFeeds_.reset(
                constructUnicastCluster(
                        FE::NetFeedType::Historical,
                        layout,
                        incrementalConnections,
                        nifs,
                        FE::TimeSpan((FE::TimeSpan::Ticks(settings_.lostPacketWaitTime) * FE::TimeTraits::nanosecondsPerMicrosecond()) / 10),
                        OnixS::ItchCore::MoldUDP64::getDummySharedReceiver()
                )
        );

        retransmissionService_.reset(
                new ItchCore::MoldUDP64::RetransmissionService
                        (
                                *gapResponseFeeds_,
                                settings_.maxRetransmissionResponseTime,
                                settings_.lostPacketWaitTime,
                                logger_.get(),
                                this,
                                (layout == FE::MulticastFeedLayout::AWithFailoverToB) || (layout == FE::MulticastFeedLayout::BWithFailoverToA)
                        ));

        retransmissionService_->subscribeOnFailure(boost::bind(&HandlerImpl::onReplayFailure, this, _1, _2, _3));
        retransmissionService_->subscribeOnPacket(boost::bind(&HandlerImpl::onReplayPacket, this, _1));
        retransmissionService_->subscribeOnWarning(boost::bind(&HandlerImpl::invokeWarning, this, _1));
    }
    catch (const std::invalid_argument&)
    {
        retransmissionService_.reset();
        gapResponseFeeds_.reset();
    }
}

void HandlerImpl::onStarting()
{
    realtimePacketProcessor_->reset();
    messageRepository_->reset();

    {
    FE::ConnectionInfo connectionA;
    FE::ConnectionInfo connectionB;

    connectionA.role(FE::NetFeedRole::A);
    connectionB.role(FE::NetFeedRole::B);

    FE::MulticastFeedLayout::Enum layout = FE::MulticastFeedLayout::Both;
    if (!settings_.useFeedA)
        layout = FE::MulticastFeedLayout::BOnly;
    else if (!settings_.useFeedB)
        layout = FE::MulticastFeedLayout::AOnly;

    FE::NifsByFeedRole nifs;
    nifs.A(settings_.networkInterfaceA.empty() ? settings_.networkInterface : settings_.networkInterfaceA);
    nifs.B(settings_.networkInterfaceB.empty() ? settings_.networkInterface : settings_.networkInterfaceB);

    ///
    connectionA.type(FE::NetFeedType::Incremental);
    connectionB.type(FE::NetFeedType::Incremental);

    connectionA.id("MulticastA");
    connectionB.id("MulticastB");

    connectionA.ip(settings_.itchFeed.serviceA.address);
    connectionB.ip(settings_.itchFeed.serviceB.address);

    connectionA.port(settings_.itchFeed.serviceA.port);
    connectionB.port(settings_.itchFeed.serviceB.port);

    FE::ConnectionInfoList incrementalConnections;
    incrementalConnections.push_back(connectionA);
    incrementalConnections.push_back(connectionB);

    realtimeFeeds_.reset(
        constructCluster(
            FE::NetFeedType::Incremental,
            layout,
            incrementalConnections,
            nifs,
            FE::TimeSpan(0, 0, settings_.heartbeatInterval, 0),
            *realtimePacketProcessor_
        )
    );
    }
    
    initRetransmissionService();

    realtimeFeeds_->connect(FE::CanThrowNow());
}

void HandlerImpl::onStopping()
{
    if(realtimeFeeds_)
        realtimeFeeds_->disconnect();

    if (retransmissionService_)
        gapResponseFeeds_->disconnect();
}

void HandlerImpl::onHeartbeat(FE::PacketContainer& packetContainer)
{
    static Logging::LogFacility logFacility("Realtime", this, Logging::LOG_LEVEL_DEBUG);

    const OnixS::ItchCore::MoldUDP64::PacketHeader* unitHeader = &OnixS::ItchCore::MoldUDP64::PacketHelper::header(packetContainer);

    BOOST_ASSERT(unitHeader != nullptr);

    if(settings_.logSettings & LogSettings::LogPackets)
        log( ONIXS_LOG_INFO[&logFacility] << unitHeader->sequenceNumber() << " Heartbeat received (" << Base64Source(packetContainer.data(), packetContainer.size()) << ").");
}

void HandlerImpl::onPacket(FE::PacketContainer& packetContainer, bool fromCache)
{
    static Logging::LogFacility logFacility("Realtime", this, Logging::LOG_LEVEL_DEBUG);

    const OnixS::ItchCore::MoldUDP64::PacketHeader* unitHeader = &OnixS::ItchCore::MoldUDP64::PacketHelper::header(packetContainer);

    BOOST_ASSERT(unitHeader->messageCount() != 0);

    if (settings_.logSettings & LogSettings::LogPackets)
        log( ONIXS_LOG_INFO[&logFacility]
                     << unitHeader->sequenceNumber()
                     << ":"
                     << unitHeader->messageCount()
                     << " Packet received ("
                     << Base64Source(packetContainer.data(), packetContainer.size())
                     << ").");

    DataSource dataSource;
    dataSource.origin = DataSource::multicast;
    dataSource.packetMessageCount = unitHeader->messageCount();
    dataSource.messageSeqNum = unitHeader->sequenceNumber();
    dataSource.session = (Byte*)unitHeader->session();

    if(!fromCache)
        dataSource.packetReceptionTime = TimeHelper::convert<Timestamp>(packetContainer.receiveTime());
    else
        dataSource.packetReceptionTime = Timestamp();

    messageRepository_->onPacket(dataSource, packetContainer);
}

void HandlerImpl::onReplayPacket(OnixS::HandlerCore::MarketData::FE::PacketContainer& packetContainer)
{
    const OnixS::ItchCore::MoldUDP64::PacketHeader* unitHeader =
        &OnixS::ItchCore::MoldUDP64::PacketHelper::header(packetContainer);

    static Logging::LogFacility logFacility("RetransmissionService", this, Logging::LOG_LEVEL_DEBUG);

    BOOST_ASSERT(unitHeader->messageCount() != 0);

    if (settings_.logSettings & LogSettings::LogPackets)
        log( ONIXS_LOG_INFO[&logFacility]
                     << unitHeader->sequenceNumber()
                     << ":"
                     << unitHeader->messageCount()
                     << " Packet received ("
                     << Base64Source(packetContainer.data(), packetContainer.size())
                     << ").");


    BOOST_ASSERT(unitHeader->messageCount() != 0);
    
    DataSource dataSource;
    dataSource.packetReceptionTime = HANDLER_NAMESPACE::Timestamp::utcNow();
    dataSource.packetMessageCount = unitHeader->messageCount();
    dataSource.messageSeqNum = unitHeader->sequenceNumber();
    dataSource.session = (Byte*)unitHeader->session();
    dataSource.origin = DataSource::replay;

    messageRepository_->onPacket(dataSource, packetContainer);
}

void HandlerImpl::onMulticastPacketGap(const OnixS::ItchCore::SessionType& session, SequenceNumber expectedSeqNum, SequenceNumber receivedSeqNum)
{
    BOOST_ASSERT(receivedSeqNum != 0);
    BOOST_ASSERT(expectedSeqNum < receivedSeqNum);

    static Logging::LogFacility logFacility("Multicast", this, Logging::LOG_LEVEL_DEBUG);
    log( ONIXS_LOG_DEBUG[&logFacility] << "Packet gap: expected " << expectedSeqNum << ", but received " << receivedSeqNum);
    
    BOOST_ASSERT(messageRepository_.get() != nullptr);
    messageRepository_->onGap(expectedSeqNum, receivedSeqNum);

    if (!backtestingMode_)
    {
        const SequenceNumber missedPacketNumber = receivedSeqNum - expectedSeqNum;

        if (missedPacketNumber <= settings_.retransmissionMaxPacketNumber && retransmissionService_.get() != nullptr)
        {
            retransmissionService_->requestReplay(session, expectedSeqNum, receivedSeqNum - expectedSeqNum);
        }
        else if (glimpseService_.get() != nullptr)
        {
            if (glimpseService_->inProgress())
            {
                invokeWarning("Glimpse service in progress, unable to restore multicast packet gap. Resetting.");

                glimpseService_->stop(true);
            }

            messageRepository_->onSnapshotRecoveryStarted();

            glimpseService_->request(session, receivedSeqNum - 1);
        }
        else
        {
            const std::string message = "Glimpse service is not initialized";

            if (settings_.buildInternalOrderBooks)
            {
                invokeError(message + ", unable to recovery state, stopping.");
                stop(false);
            }
            else
            {
                invokeWarning(message + ", packets will be skipped.");
                messageRepository_->skipPackets(expectedSeqNum, receivedSeqNum);
            }
        }

    }
}

void HandlerImpl::onMulticastInactivity()
{
    BOOST_ASSERT(listenerHolder_.get() != nullptr);
    listenerHolder_->invokeWarning("No multicast data");

    BOOST_ASSERT(messageRepository_.get() != nullptr);
    messageRepository_->onInactivity();
}

void HandlerImpl::initLogger(const HandlerSettings &settings)
{
    logger_.reset(new HandlerLogger(convertLogSettings(settings)));
    setLogger(&logger_->getLogger());
}

void HandlerImpl::start(const ReplayOptions& options)
{
    backtestingMode_ = true;

    LogPlayerLinks links;

    BOOST_ASSERT(realtimePacketProcessor_.get() != nullptr);
    realtimePacketProcessor_->activateImmediateGapReplayMode();
    links.push_back(LogPlayerLink("Realtime", realtimePacketProcessor_.get()));
    
    retransmissionPacketProcessor_.reset(new RetransmissionPacketProcessor(logger_.get(), this, messageRepository_.get()));
    links.push_back(LogPlayerLink("RetransmissionService", retransmissionPacketProcessor_.get()));

    glimpsePacketProcessor_.reset(new GlimpsePacketProcessor(logger_.get(), this, messageRepository_.get()));
    links.push_back(LogPlayerLink("GlimpseService", glimpsePacketProcessor_.get()));

    const std::string parser =
        std::string("(\\S+).*") +
        std::string("\\[") + HANDLER_NAMESPACE::projectName()  + std::string("\\.(\\w+)\\].*") +
        std::string("(Packet received|Heartbeat received|Message received)") + 
        std::string(" \\((.*)\\)\\.?");

    BOOST_ASSERT(options.listener);

    HandlerCore::MarketData::MulticastFeedHandler::startReplay(
        options.logs,
        0,
        0,
        links,
        boost::bind(&ReplayListener::onReplayError, options.listener, _1),
        boost::bind(&ReplayListener::onReplayFinished, options.listener),
        parser,
        0,
        1,
        3
    );
}

void HandlerImpl::onPrepareLogPlayer(LogPlayer* logPlayer)
{
    if (logPlayer)
    {
        logPlayer->setDefaultHandler(boost::bind(&HandlerImpl::parseGlimpseStart, this, _1, _2));
    }
}

bool HandlerImpl::parseGlimpseStart(const std::string& logLine, HandlerCore::Common::LogEntry& /*logEntry*/)
{
    const std::string parser =
        std::string("(\\S+).*") + 
        std::string("\\[") + HANDLER_NAMESPACE::projectName() + std::string("\\.(\\w+)\\].*") +
        std::string("(SnapshotRecoveryRequested|SnapshotRecoveryRestarted)");

    const boost::regex regEx(parser);
    boost::smatch smatch;
    const bool ok = boost::regex_match(logLine, smatch, regEx);

    if (!ok)
        return false;

    std::string action = smatch[3];
    if(action == "SnapshotRecoveryRequested")
        messageRepository_->onSnapshotRecoveryStarted();
    else if (action == "SnapshotRecoveryRestarted")
        messageRepository_->onSnapshotRecoveryRestarted();

    return false;
}

void HandlerImpl::onReplayFailure(const ItchCore::SessionType&, SequenceNumber begin, ItchCore::Binary8 count)
{
    const std::string what =
            (boost::format("Retransmission service failure, required packets: %d - %d") % begin % (begin + count)).str();

    if (settings_.buildInternalOrderBooks)
    {
        invokeError(what + ", unable to recovery state, stopping.");
        stop(false);
    }
    else
    {
        invokeWarning(what + ", packets will be skipped.");

        BOOST_ASSERT(messageRepository_.get() != nullptr);
        messageRepository_->skipPackets(begin, begin + count);
    }
}

void HandlerImpl::onGlimpseFailure(const std::string& what)
{
    const std::string message = "GLIMPSE service failure: " + what;
    invokeError(message);
    stop(false);
}

void HandlerImpl::onGlimpseWarning(const std::string& what)
{
    const std::string message = "GLIMPSE service warning: " + what;
    invokeWarning(message);
}

void HandlerImpl::onGlimpseMessage(const DataSource& dataSource, const IncomingMessage* message)
{
    if (settings_.logSettings & LogSettings::LogPackets)
        log(ONIXS_LOG_INFO[glimpseService_.get()]
            << " Message received ("
            << Base64Source(reinterpret_cast<const char*>(message->binary()), message->binarySize())
            << ").");


    messageRepository_->processMessage(dataSource, message);
}

void HandlerImpl::onGlimpseRestarted()
{
    BOOST_ASSERT(messageRepository_);
    messageRepository_->onSnapshotRecoveryRestarted();
}

void HandlerImpl::onStateChanged(HandlerCore::MarketData::MulticastFeedHandler::State::Enum newState) 
{
    listenerHolder_->invokeStateChanged(static_cast<HandlerState::Enum>(newState));
}

ONIXS_HANDLER_NAMESPACE_END