MessageRepository.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 "MessageRepository.h"

#include <OnixS/SgxTitan/MarketData/Itch/Exception.h>
#include <OnixS/SgxTitan/MarketData/Itch/Messages.h>

#include <OnixS/HandlerCore/OrderBook.h>

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

#include "ListenerHolder.h"
#include "HandlerLogger.h"

#include "Formatting.h"
#include "OrderBookHolder.h"
#include "GlimpseService.Messages.h"
#include "Formatting.Messages.h"

#include <boost/bind.hpp>

#include <util/String.h>

using namespace OnixS::ItchCore;

ONIXS_HANDLER_NAMESPACE_BEGIN

    template <typename MessageType>
    bool isMarketOrder(const MessageType* message, const std::set<OrderId>& marketOrderIds)
    {
        BOOST_ASSERT(message != nullptr);

        return marketOrderIds.find(message->orderId()) != marketOrderIds.end();
    }

    bool isMarketOrder(const AddOrderMsg* message)
    {
        BOOST_ASSERT(message != nullptr);

        return message->price().isNull();
    }

    MessageRepository::MessageRepository(
            ListenerHolder& listenerHolder,
            const HandlerSettings& settings,
            const Logging::LogFacility* parent,
            HandlerCore::Common::HandlerLogger* logger,
            OrderBookAllocator* bookAllocator) :
        Logging::LogFacility("MessageRepository", parent, OnixS::Logging::LOG_LEVEL_DEBUG),
        settings_(settings),
        listenerHolder_(listenerHolder),
        buildOrderBooks_(settings.buildInternalOrderBooks),
        logger_(logger),
        expectedMessageSeqNum_(1),
        skipGaps_(false),
        orderBookHolder_(buildOrderBooks_ ? new OrderBookHolder(boost::bind(&MessageRepository::onOrderBookUpdated, this, _1), bookAllocator, 10) : nullptr)
    {
        BOOST_ASSERT(logger_ != nullptr);
        log(ONIXS_LOG_DEBUG[this] << "Skip gapped messages: " << (skipGaps_ ? "yes" : "no"));

        checkOrderBookListener();
    }

    MessageRepository::~MessageRepository()
    {
    }

    template <typename MessageType>
    bool MessageRepository::filterByInstrument(const MessageType* message, const std::set<OrderBookId>& filter)
    {
        BOOST_ASSERT(message != nullptr);

        if (filter.empty())
            return false;

        if (filter.find(message->orderBookId()) == filter.end())
        {
            log(ONIXS_LOG_DEBUG[this] << "Message filtered by Instrument");
            return true;
        }

        return false;
    }

    void MessageRepository::setOrderBookIdFilter(const std::set<OrderBookId>& orderBookIdFilter)
    {
        orderBookIdFilter_.clear();
        orderBookIdFilter_.insert(orderBookIdFilter.begin(), orderBookIdFilter.end());
    }

    void MessageRepository::checkOrderBookListener()
    {
        if (buildOrderBooks_ && !listenerHolder_.hasOrderBookListener())
            listenerHolder_.invokeWarning("Order books are being built, but Order Book Listener is not registered");

        if (listenerHolder_.hasOrderBookListener() && !buildOrderBooks_)
            listenerHolder_.invokeWarning("Order Book Listener is registered, but order books are not being built");

        if (skipGaps_ && buildOrderBooks_)
            listenerHolder_.invokeWarning("impossible to build order books because gapped messages being skipped");
    }

    template <typename MessageType>
    inline const MessageType* MessageRepository::checkDowncastMsg(const IncomingMessage* message)
    {
        BOOST_ASSERT(message != nullptr);

        MessageType::validateSize(message->binarySize());

        const MessageType* const msg = static_cast<const MessageType*>(message);

        try
        {
            log(ONIXS_LOG_DEBUG << *msg << " received");
        }
        catch (const std::exception& ex)
        {
            const std::string what = std::string("Exception during message serialisation: ") + ex.what();
            listenerHolder_.invokeWarning(what);
            log(ONIXS_LOG_WARN[this] << what);
        }
        catch (...)
        {
            const std::string what = "Unknown exception during message serialisation.";
            listenerHolder_.invokeWarning(what);
            log(ONIXS_LOG_WARN[this] << what);
        }

        return msg;
    }

    void MessageRepository::reset()
    {
        log(ONIXS_LOG_DEBUG[this] << "reset");

        Guard guard(messageQueueLock_);

        packetQueue_.clear();
        expectedMessageSeqNum_ = 1;

        resetMarketState();
    }

    void MessageRepository::resetMarketState()
    {
        if (buildOrderBooks_)
            orderBookHolder_->reset();
        
        marketOrderIds_.clear();
    }

    void MessageRepository::onSnapshotRecoveryStarted()
    {
        reset();

        if (buildOrderBooks_)
        {
            orderBookHolder_->onBookOutOfDate(
                    boost::bind(&ListenerHolder::invokeOrderBookOutOfDate, &listenerHolder_, _1));

            orderBookHolder_->snapshotRecoveryStarted();
        }

        listenerHolder_.invokeSnapshotRecoveryStarted();
    }

    void MessageRepository::onSnapshotRecoveryRestarted()
    {
        resetMarketState();

        if (buildOrderBooks_)
            orderBookHolder_->snapshotRecoveryStarted();

        listenerHolder_.invokeSnapshotRecoveryStarted();
    }

    void MessageRepository::onPacket(DataSource& dataSource,
                                     OnixS::HandlerCore::MarketData::FE::PacketContainer& packetContainer,
                                     bool runProcessCachedMessages)
    {
        BOOST_ASSERT((dataSource.origin == DataSource::replay) || (dataSource.origin == DataSource::multicast));

        Guard guard(messageQueueLock_);

        const MoldUDP64::PacketHeader* packetHeader = &MoldUDP64::PacketHelper::header(packetContainer);
        
        if (packetHeader->sequenceNumber() == expectedMessageSeqNum_)
        {
            const char* data = static_cast<char*>(packetContainer.data());
            size_t offset = sizeof(MoldUDP64::PacketHeader);

            for (unsigned i = 0; i < packetHeader->messageCount(); ++i)
            {
                dataSource.packetMessageNumber = static_cast<UInt8>(i);
                dataSource.messageSeqNum = packetHeader->sequenceNumber() + i;

                const MoldUDP64::MessageBlockHeader* const messageBlockHeader = reinterpret_cast<const MoldUDP64::MessageBlockHeader*>(data + offset);
                offset += sizeof(MoldUDP64::MessageBlockHeader);

                IncomingMessage message = IncomingMessage(data + offset, messageBlockHeader->length());

                processMessage(dataSource, &message);
                ++expectedMessageSeqNum_;

                offset += messageBlockHeader->length();
            }

            if(runProcessCachedMessages)
                processCachedMessages();
        }
        else if (packetHeader->sequenceNumber() > expectedMessageSeqNum_)
        {
            log(ONIXS_LOG_DEBUG[this] << packetHeader->sequenceNumber() << " packet cached, expected: " << expectedMessageSeqNum_);
            packetQueue_.push(dataSource, packetContainer);
        }
    }

    void MessageRepository::processCachedMessages()
    {
        if (!packetQueue_.empty())
        {
            log(ONIXS_LOG_DEBUG[this] << "Processing cached messages");
            
            while (!packetQueue_.empty())
            {
                DataSource& dataSource = packetQueue_.front().context();

                if (dataSource.messageSeqNum == expectedMessageSeqNum_)
                {
                    log(ONIXS_LOG_DEBUG[this] << dataSource.messageSeqNum << " packet processed from cache");

                    onPacket(dataSource, packetQueue_.front().packetContainer(), false);

                    packetQueue_.front().packetContainer().release();
                    packetQueue_.pop();
                }
                else if (dataSource.messageSeqNum > expectedMessageSeqNum_)
                    return;
                else
                {
                    log(ONIXS_LOG_DEBUG[this] << dataSource.messageSeqNum << " Unexpected packet - dropped, expected: " << expectedMessageSeqNum_);

                    packetQueue_.front().packetContainer().release();
                    packetQueue_.pop();
                }
            }

            BOOST_ASSERT(packetQueue_.empty());
            packetQueue_.clear();
        }
    }

    void MessageRepository::processMessage(const DataSource& dataSource, const IncomingMessage* message)
    {
        Guard guard(messageQueueLock_);

        const Byte messageType = message->type();

        try
        {
            if (messageType == MessageType::Seconds)
            {
                listenerHolder_.invokeTime(checkDowncastMsg<SecondsMsg>(message), dataSource);
            }
            else if (messageType == MessageType::OrderBookDirectory)
            {
                const OrderBookDirectoryMsg* const msg = checkDowncastMsg<OrderBookDirectoryMsg>(message);

                if(filterByInstrument(msg, orderBookIdFilter_))
                    return;

                listenerHolder_.invokeOrderBookDirectory(msg, dataSource);
            }
            else if (messageType == MessageType::CombinationOrderBookLeg)
            {
                const CombinationOrderBookLegMsg* const msg = checkDowncastMsg<CombinationOrderBookLegMsg>(message);

                if(filterByInstrument(msg, orderBookIdFilter_))
                    return;

                listenerHolder_.invokeCombinationOrderBookLeg(msg, dataSource);
            }
            else if (messageType == MessageType::TickSizeTableEntry)
            {
                const TickSizeTableEntryMsg* const msg = checkDowncastMsg<TickSizeTableEntryMsg>(message);

                if(filterByInstrument(msg, orderBookIdFilter_))
                    return;

                listenerHolder_.invokeTickSizeTableEntry(msg, dataSource);
            }
            else if (messageType == MessageType::SystemEvent)
            {
                const SystemEventMsg* const msg = checkDowncastMsg<SystemEventMsg>(message);

                listenerHolder_.invokeSystemEvent(msg, dataSource);
            }
            else if (messageType == MessageType::OrderBookState)
            {
                const OrderBookStateMsg* const msg = checkDowncastMsg<OrderBookStateMsg>(message);

                if(filterByInstrument(msg, orderBookIdFilter_))
                    return;

                listenerHolder_.invokeOrderBookState(msg, dataSource);
            }
            else if (messageType == MessageType::AddOrder)
            {
                const AddOrderMsg* const msg = checkDowncastMsg<AddOrderMsg>(message);

                if(filterByInstrument(msg, orderBookIdFilter_))
                    return;

                listenerHolder_.invokeAddOrder(msg, dataSource);

                if (buildOrderBooks_)
                {
                    if (isMarketOrder(msg))
                        marketOrderIds_.insert(msg->orderId());
                    else
                        orderBookHolder_->onOrderAdd(*msg);
                }
            }
            else if (messageType == MessageType::OrderExecuted)
            {
                BOOST_ASSERT(dataSource.origin != DataSource::glimpse);

                const OrderExecutedMsg* const msg = checkDowncastMsg<OrderExecutedMsg>(message);

                if(filterByInstrument(msg, orderBookIdFilter_))
                    return;

                listenerHolder_.invokeOrderExecuted(msg, dataSource);

                if (buildOrderBooks_)
                {
                    if (isMarketOrder(msg, marketOrderIds_))
                        marketOrderIds_.erase(msg->orderId());
                    else
                        orderBookHolder_->onOrderExecuted(*msg);
                }
            }
            else if (messageType == MessageType::OrderExecutedWithPrice)
            {
                BOOST_ASSERT(dataSource.origin != DataSource::glimpse);

                const OrderExecutedWithPriceMsg* const msg = checkDowncastMsg<OrderExecutedWithPriceMsg>(message);

                if(filterByInstrument(msg, orderBookIdFilter_))
                    return;

                listenerHolder_.invokeOrderExecutedWithPrice(msg, dataSource);

                if (buildOrderBooks_)
                {
                    if (isMarketOrder(msg, marketOrderIds_))
                    {
                        //skipped, can see a few OrderExecutedWithPrice for the market orders
                        //marketOrderIds_.erase(msg->orderId());
                    }
                    else
                        orderBookHolder_->onOrderExecutedWithPrice(*msg);
                }
            }
            else if (messageType == MessageType::OrderReplace)
            {
                BOOST_ASSERT(dataSource.origin != DataSource::glimpse);

                const OrderReplaceMsg* const msg = checkDowncastMsg<OrderReplaceMsg>(message);

                if(filterByInstrument(msg, orderBookIdFilter_))
                    return;

                listenerHolder_.invokeOrderReplace(msg, dataSource);

                if (buildOrderBooks_ && !isMarketOrder(msg, marketOrderIds_))
                    orderBookHolder_->onOrderReplace(*msg);
            }
            else if (messageType == MessageType::OrderDelete)
            {
                BOOST_ASSERT(dataSource.origin != DataSource::glimpse);

                const OrderDeleteMsg* const msg = checkDowncastMsg<OrderDeleteMsg>(message);

                if(filterByInstrument(msg, orderBookIdFilter_))
                    return;

                listenerHolder_.invokeOrderDelete(msg, dataSource);

                if (buildOrderBooks_)
                {
                    if (isMarketOrder(msg, marketOrderIds_))
                        marketOrderIds_.erase(msg->orderId());
                    else
                        orderBookHolder_->onOrderDelete(*msg);
                }
            }
            else if (messageType == MessageType::Trade)
            {
                const TradeMsg* const msg = checkDowncastMsg<TradeMsg>(message);

                if(filterByInstrument(msg, orderBookIdFilter_))
                    return;

                listenerHolder_.invokeTrade(msg, dataSource);
            }
            else if (messageType == MessageType::EquilibriumPriceUpdate)
            {
                const EquilibriumPriceUpdateMsg* const msg = checkDowncastMsg<EquilibriumPriceUpdateMsg>(message);

                if(filterByInstrument(msg, orderBookIdFilter_))
                    return;

                listenerHolder_.invokeEquilibriumPriceUpdate(msg, dataSource);
            }
            else if (static_cast<GlimpseMessageType::Enum>(messageType) == (GlimpseMessageType::EndOfSnapshot))
            {
                const EndOfSnapshotMsg* const msg = checkDowncastMsg<EndOfSnapshotMsg>(message);

                if(buildOrderBooks_)
                    orderBookHolder_->snapshotRecoveryFinished();

                listenerHolder_.invokeSnapshotRecoveryFinished();

                expectedMessageSeqNum_ = msg->sequenceNumber();
                processCachedMessages();
            }
            else
                throw OperationException(__FUNCTION__, OnixS::Util::format("Unsupported message type (messageType=%d).", messageType).c_str());
        }
        catch (const std::exception& ex)
        {
            listenerHolder_.invokeWarning(ex.what());
            log(ONIXS_LOG_WARN[this] << ex.what());
        }
        catch (...)
        {
            const std::string what = "Unhandled exception in processMessage method";
            listenerHolder_.invokeWarning(what);
            log(ONIXS_LOG_WARN[this] << what);
        }
    }

    void MessageRepository::onGap(SequenceNumber expectedSeqNum, SequenceNumber receivedSeqNum)
    {
        BOOST_ASSERT(expectedSeqNum < receivedSeqNum);

        if (skipGaps_)
            skipPackets(expectedSeqNum, receivedSeqNum);

        listenerHolder_.invokeGap(expectedSeqNum, receivedSeqNum);
    }

    void MessageRepository::skipPackets(SequenceNumber seqNum1, SequenceNumber seqNum2)
    {
        BOOST_ASSERT(seqNum1 < seqNum2);
        BOOST_ASSERT(seqNum1 != 0);

        BOOST_ASSERT(seqNum1 == expectedMessageSeqNum_);

        Guard guard(messageQueueLock_);

        if (expectedMessageSeqNum_ < seqNum2)
            expectedMessageSeqNum_ = seqNum2;

        log(ONIXS_LOG_DEBUG[this] << "Packets " << seqNum1 << "-" << seqNum2 << " is skipped");
        processCachedMessages();
    }

    void MessageRepository::onInactivity()
    {
        listenerHolder_.invokeInactivity();

        if (buildOrderBooks_)
            orderBookHolder_->onBookOutOfDate(boost::bind(&ListenerHolder::invokeOrderBookOutOfDate, &listenerHolder_, _1));
    }

    void MessageRepository::onOrderBookUpdated(const OrderBookInternal& book)
    {
        if (settings_.logSettings & LogSettings::LogBooks)
            log(ONIXS_LOG_DEBUG[this] << "Order book updated:" << "\n" << book);

        listenerHolder_.invokeOrderBookUpdated(book);
    }

    void MessageRepository::doNotSkipGaps()
    {
        skipGaps_ = false;
    }
    
ONIXS_HANDLER_NAMESPACE_END