EobiPacketProcessor.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 <util/TextBuilder.h>

#include "../PacketHeader.h"
#include <OnixS/Eurex/MarketData/SequenceNumber.h>
#include <boost/function.hpp>

#include <OnixS/HandlerCore/FeedEngine/SharedReceiver.h>
#include <OnixS/HandlerCore/FeedEngine/PacketContainer.h>
#include <OnixS/HandlerCore/FeedEngine/SortedPackets.h>
#include <OnixS/HandlerCore/MarketData/MulticastFeedHandler.h>


#include "../Utils.h"

namespace OnixS {
namespace Eurex {
namespace MarketData {
namespace EOBI {
namespace Implementation {

using namespace HandlerCore::MarketData;

struct PacketHelper
{
    typedef SequenceNumber SequenceNumber_;
    typedef MarketData::Implementation::PacketHeaderForEobi PacketHeader;

    static
    const PacketHeader* getPacketHeader(const FE::PacketContainer& container);

    static
    SequenceNumber_ seqNumber(const FE::PacketContainer& container);

    static
    SequenceNumber_ nextSeqNumber(const FE::PacketContainer& container);

    static
    bool isHeartbeat(const FE::PacketContainer&);

    static
    std::string toString(const FE::PacketContainer& container);

    static
    size_t headerSize();
};


class EobiPacketProcessor : public LegacyLogPlayerPacketProcessor
{
    typedef MarketData::Implementation::PacketHeaderForEobi PacketHeader;

public:
    typedef boost::function<void(DataSource& dataSource, const char* data, size_t length)> OnPacketData;
    typedef boost::function<void()> OnGap;
    typedef boost::function<void()> OnInactivity;
    typedef boost::function<void(FE::NetFeedRole::Enum)> OnFeedInactivity;
    typedef boost::function<void()> OnApplSeqReset;

    EobiPacketProcessor(
        unsigned int maxPacketWaitingTimeInMs, 
        unsigned int inactivityWaitingTimeInSec,
        unsigned int outOfOrderMaxInterval,
        Concurrency::RecursiveMutex* lock);

    virtual ~EobiPacketProcessor() {}

    void reset() override;
    bool tryAcquire() override;
    void release() override;
    void onPacket(FE::PacketContainer&) override;
    void onTimeout(const FE::NetFeed&) override;
    void onFailure(const FE::NetFeed&, const char*) override {}
    void processHoldOnPackets(const FE::Timestamp& now);

    void subscribeOnPacketData(const OnPacketData callback);
    void subscribeOnGap(const OnGap callback);
    void subscribeOnApplSeqReset(const OnApplSeqReset callback);
    void subscribeOnInactivity(const OnInactivity callback);
    void subscribeOnFeedInactivity(const OnFeedInactivity callback);

protected:
    DataSource createDataSource(FE::PacketContainer& packetContainer, SequenceNumber packetSeqNum);

protected:
    OnPacketData onPacketData_;
    OnGap onGap_;
    OnApplSeqReset onApplSeqReset_;
    OnInactivity onInactivity_;
    OnFeedInactivity onFeedInactivity_;

    
    SequenceNumber nextPacketSeqNum_;

    const FE::TimeSpan maxPacketWaitingTimeSpan_;
    const FE::TimeSpan inactivityWaitingTimeSpan_;
    const UInt32 outOfOrderMaxInterval_;

    bool inactivityFlag_;
    bool inactivityFlagA_;
    bool inactivityFlagB_;
    bool applSeqResetting_;

    FE::Timestamp lastReceptionTimeFe_;
    FE::Timestamp lastReceptionTimeFeA_;
    FE::Timestamp lastReceptionTimeFeB_;

    FE::SortedPackets<PacketHelper> outOfOrder_;
    OnixS::Concurrency::RecursiveMutex* lock_;
};

class EobiSnapshotPacketProcessor : public EobiPacketProcessor
{
    typedef MarketData::Implementation::PacketHeaderForEobi PacketHeader;

public:
    EobiSnapshotPacketProcessor(
        unsigned int maxPacketWaitingTimeInMs, 
        unsigned int inactivityWaitingTimeInSec,
        unsigned int outOfOrderMaxInterval,
        Concurrency::RecursiveMutex* lock);

public:
    void onPacket(FE::PacketContainer&) override;
};

}}}}}