MessageRepository.h

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. 
*/
#pragma once

#include <map>

#include <boost/function.hpp>

#include <OnixS/HandlerCore/HandlerLogger.h>
#include <OnixS/HandlerCore/PacketQueue.h>

#include <OnixS/SgxTitan/MarketData/Itch/HandlerSettings.h>
#include <OnixS/SgxTitan/MarketData/Itch/Enumerations.h>
#include <OnixS/SgxTitan/MarketData/Itch/String.h>

#include "NamespaceHelper.h"

ONIXS_HANDLER_NAMESPACE_BEGIN

class IncomingMessage;

class ListenerHolder;
class OrderBookHolder;
class OrderBookInternal;
class OrderBookAllocator;
class OrderBookHolder;
class OrderBookInternal;

class MessageRepository : public Logging::LogFacility
{
public:
    MessageRepository(
            ListenerHolder& listenerHolder,
            const HandlerSettings& settings,
            const Logging::LogFacility* parent,
            HandlerCore::Common::HandlerLogger* logger,
            OrderBookAllocator* bookAllocator);
    
    virtual ~MessageRepository();

    void onPacket(
            DataSource& dataSource,
            OnixS::HandlerCore::MarketData::FE::PacketContainer& packetContainer,
            bool runProcessCachedMessages = true);

    void processMessage(const DataSource& dataSource, const IncomingMessage* message);

    void onGap(SequenceNumber expectedSeqNum, SequenceNumber receivedSeqNum);
    void onInactivity();
    void reset();
    void resetMarketState();
    void skipPackets(SequenceNumber seqNum1, SequenceNumber seqNum2);

    void doNotSkipGaps();

    void onSnapshotRecoveryStarted();
    void onSnapshotRecoveryRestarted();

    /// Log helper.
    template<typename TMsg>
    void log (const TMsg & msg) { if(logger_) logger_->log (msg); }

    void setOrderBookIdFilter(const std::set<OrderBookId>& orderBookIdFilter);

    template <typename MessageType>
    bool filterByInstrument(const MessageType* message, const std::set<OrderBookId>& filter);

private:
    template <typename MessageType>
    const MessageType* checkDowncastMsg(const IncomingMessage* message);

    void onOrderBookUpdated(const OrderBookInternal& book);

    void checkOrderBookListener();
    void processCachedMessages();
        
private:
    const HandlerSettings& settings_;
    ListenerHolder& listenerHolder_;
    const bool buildOrderBooks_;

    HandlerCore::Common::HandlerLogger* logger_;
    SequenceNumber expectedMessageSeqNum_;
    HandlerCore::Common::PacketQueue<DataSource> packetQueue_;
    bool skipGaps_;

    typedef OnixS::Concurrency::RecursiveMutex Mutex;
    typedef OnixS::Concurrency::Guard<Mutex> Guard;
    mutable Mutex messageQueueLock_;

    boost::scoped_ptr<OrderBookHolder> orderBookHolder_;

    std::set<OrderBookId> orderBookIdFilter_;

    std::set<OrderId> marketOrderIds_;
};


ONIXS_HANDLER_NAMESPACE_END