Benchmark Sample

This sample demonstrates how to measure the handler latency.

Source code

/*
* 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 <iostream>
 
#include <OnixS/Eurex/MarketData/EmdiHandler.h>
#include <OnixS/Eurex/MarketData/RdiHandler.h>
#include <OnixS/Eurex/MarketData/EobiHandler.h>
#include <OnixS/Eurex/MarketData/Exception.h>
#include <OnixS/Eurex/MarketData/FeedEngine.h>
 
#include "RdiListener.h"
#include "EmdiListener.h"
#include "EobiListener.h"
 
#include "../Common/Signal.h"
#include "../Common/Options.h"
#include "../Common/Settings.h"
#include "../Common/Utils.h"
 
using namespace OnixS::Eurex::MarketData;
using namespace OnixS::Eurex::MarketData::EOBI;
 
template <typename Configuration> void process(const EmdiDescriptors& descriptors, FeedEngine& feedEngine, const Configuration& cfg);
template <typename Configuration> void process(const EobiDescriptors& descriptors, FeedEngine& feedEngine, const Configuration& cfg);
 

int main(int argc, char* argv[])
{
    // `--help` to show options.
    const Samples::AppConfiguration<
            Samples::NetworkInterfaceConfiguration
            , Samples::EnvironmentConfiguration
            , Samples::LogDirectoryConfiguration
            , Samples::AffinityConfiguration
            , Samples::MarketSegmentConfiguration
            , Samples::ProductConfiguration
            , Samples::PacketCountConfiguration
    > cfg{"Benchmark", argc, argv};
 
    try
    {
        Samples::SignalHelper::manageSignals();
 
        ThisThread::affinity(cfg.cpu());
        SocketFeedEngine feedEngine(0);
 
        auto rdiSettings = Samples::makeSettings<RdiHandlerSettings>(cfg);
        fillEnvironment(rdiSettings, cfg);
 
        RdiListener rdiListener;
        RdiHandler rdiHandler(rdiSettings);
        rdiHandler.bindFeedEngine(feedEngine);
 
        rdiHandler.registerErrorListener(&rdiListener);
        rdiHandler.registerWarningListener(&rdiListener);
        rdiHandler.registerReferenceDataListener(&rdiListener);
 
        std::clog << "Will start the RDI Handler ..." << std::endl;
        std::clog << "Please press Ctrl+C key to stop..." << std::endl;
 
        rdiHandler.start();
 
        while(!rdiListener.referenceDataReceived() && !Samples::SignalHelper::interruptDetected())
            process(feedEngine);
 
        std::clog << "Will stop the RDI Handler ..." << std::endl;
        rdiHandler.stop();
 
        while(process(feedEngine));
 
        if(Samples::SignalHelper::interruptDetected())
            return 0;
 
        const MarketSegments mktSeg {cfg.marketSegment()};
 
        switch (cfg.product())
        {
            case Samples::Product::Emdi:
                process(rdiHandler.findEmdiDescriptors(mktSeg), feedEngine, cfg);
                break;
 
            case Samples::Product::Eobi:
                process(rdiHandler.findEobiDescriptors(mktSeg), feedEngine, cfg);
                break;
        }
    }
    catch(const std::exception& ex)
    {
        std::cerr << "EXCEPTION: " << ex.what() << std::endl;
    }
 
    return 0;
}
 
template <typename Settings, typename Configuration, typename Descriptor>
Settings makeBmSettings(const Configuration& cfg, const Descriptor& descriptor)
{
    auto settings = Samples::makeSettings<Settings>(cfg);
 
    settings.buildInternalOrderBooks = false;
    settings.logLevel = LogLevel::Fatal;
    settings.logSettings  = LogSettings::TraceToFile;
    settings.interfaceDescriptor.incrementalFeed = descriptor.incrementalFeed;
    settings.interfaceDescriptor.snapshotFeed = descriptor.snapshotFeed;
 
    return settings;
}
 
template <typename Configuration>
void process(const EmdiDescriptors& descriptors, FeedEngine& feedEngine, const Configuration& cfg)
{
    if(descriptors.empty())
        throw std::runtime_error("No products found.");
 
    const auto& descriptor = descriptors[0];
 
    EmdiListener listener{cfg.packetsCount()};
 
    const auto settings = makeBmSettings<EmdiHandlerSettings>(cfg, descriptor);
    auto handler = Samples::makeUnique<EmdiHandler>(settings);
 
    handler->bindFeedEngine(feedEngine);
 
    handler->setPartitionIdFilters(descriptor.partitionIdFilters);
    handler->setMarketSegmentIdFilters(descriptor.marketSegmentIdFilters);
 
    listener.subscribe(*handler);
 
    handler->start();
 
    while(!Samples::SignalHelper::interruptDetected() || listener.done())
        process(feedEngine);
 
    handler->stop();
 
    process(feedEngine);
 
    listener.saveLatencies("emdi-results.csv");
    listener.processLatencies();
}
 
template <typename Configuration>
void process(const EobiDescriptors& descriptors, FeedEngine& feedEngine, const Configuration& cfg)
{
    if(descriptors.empty())
        throw std::runtime_error("No products found.");
 
    const auto& descriptor = descriptors[0];
 
    EobiListener listener{cfg.packetsCount()};
 
    const auto settings = makeBmSettings<EobiHandlerSettings>(cfg, descriptor);
    auto handler = Samples::makeUnique<EobiHandler>(settings);
 
    handler->bindFeedEngine(feedEngine);
 
    handler->setPartitionIdFilters(descriptor.partitionIdFilters);
    handler->setMarketSegmentIdFilters(descriptor.marketSegmentIdFilters);
 
    listener.subscribe(*handler);
 
    handler->start();
 
    while(!Samples::SignalHelper::interruptDetected() || listener.done())
        process(feedEngine);
 
    handler->stop();
 
    process(feedEngine);
 
    listener.saveLatencies("eobi-results.csv");
    listener.processLatencies();
}

/*
* 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 <OnixS/Eurex/MarketData/ErrorListener.h>
#include <OnixS/Eurex/MarketData/WarningListener.h>
#include <OnixS/Eurex/MarketData/ReferenceDataListener.h>
#include <OnixS/Eurex/MarketData/Semaphore.h>
 
 
class RdiListener final
    : public OnixS::Eurex::MarketData::ErrorListener
    , public OnixS::Eurex::MarketData::WarningListener
    , public OnixS::Eurex::MarketData::ReferenceDataListener
{
public:
    RdiListener();

    ~RdiListener() override;

    void onError(OnixS::Eurex::MarketData::ErrorCode::Enum code, const std::string& description) override;

    void onWarning(const std::string& description) override;

    void onSnapshotCycleStart() override;

    void onProduct(const OnixS::Eurex::MarketData::ProductSnapshot& productSnapshot, const OnixS::Eurex::MarketData::DataSource& dataSource) override;

    void onInstrument(const OnixS::Eurex::MarketData::InstrumentSnapshot& msg, const OnixS::Eurex::MarketData::DataSource& dataSource) override;

    void onInstrumentIncremental(const OnixS::Eurex::MarketData::InstrumentIncremental& msg, const OnixS::Eurex::MarketData::DataSource& dataSource) override;

    void onVarianceFuturesStatus(const OnixS::Eurex::MarketData::VarianceFuturesStatus& msg, const OnixS::Eurex::MarketData::DataSource& dataSource) override;

    void onTotalReturnFuturesStatus(const OnixS::Eurex::MarketData::TotalReturnFuturesStatus& msg, const OnixS::Eurex::MarketData::DataSource& dataSource) override;

    void onTradeAtReferencePriceStatus(const OnixS::Eurex::MarketData::TradeAtReferencePriceStatus& msg, const OnixS::Eurex::MarketData::DataSource& dataSource) override;

    void onSnapshotCycleEnd() override;

    bool referenceDataReceived() const;
 
private:
    bool referenceDataReady_;
};

/*
* 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 <iostream>
#include <sstream>
 
#include "RdiListener.h"
 
 
using namespace OnixS::Eurex::MarketData;
 
/*
    RdiListener
*/
 
RdiListener::RdiListener()
    : referenceDataReady_(false)
{
}
 
RdiListener::~RdiListener() = default;
 
void RdiListener::onError(ErrorCode::Enum code, const std::string& description)
{
    std::clog << "Error occurred, errorCode = " << enumToString(code) << ". Description: '" << description << "'" << std::endl;
}
 
void RdiListener::onWarning(const std::string& description)
{
    std::stringstream ss;
    std::clog << "Warning occurred. Description: '" << description << "'" << std::endl;
}
 
void RdiListener::onSnapshotCycleStart()
{
}
 
void RdiListener::onProduct(const ProductSnapshot&, const DataSource&)
{
}
 
void RdiListener::onInstrument(const InstrumentSnapshot&, const DataSource&)
{
}
 
void RdiListener::onInstrumentIncremental(const InstrumentIncremental&, const DataSource&)
{
}
 
void RdiListener::onVarianceFuturesStatus(const VarianceFuturesStatus&, const DataSource&)
{
}
 
void RdiListener::onTotalReturnFuturesStatus(const TotalReturnFuturesStatus&, const DataSource&)
{
}
 
void RdiListener::onTradeAtReferencePriceStatus(const TradeAtReferencePriceStatus&, const DataSource&)
{
}
 
void RdiListener::onSnapshotCycleEnd()
{
    referenceDataReady_ = true;
}
 
bool RdiListener::referenceDataReceived() const
{
    return referenceDataReady_;
}

/*
* 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 <OnixS/Eurex/MarketData/ErrorListener.h>
#include <OnixS/Eurex/MarketData/WarningListener.h>
#include <OnixS/Eurex/MarketData/ReferenceDataListener.h>
#include <OnixS/Eurex/MarketData/DepthListener.h>
#include <OnixS/Eurex/MarketData/TopOfBookImpliedListener.h>
#include <OnixS/Eurex/MarketData/ProductStateChangeListener.h>
#include <OnixS/Eurex/MarketData/MassInstrumentStateChangeListener.h>
#include <OnixS/Eurex/MarketData/InstrumentStateChangeListener.h>
#include <OnixS/Eurex/MarketData/QuoteRequestListener.h>
#include <OnixS/Eurex/MarketData/CrossRequestListener.h>
#include <OnixS/Eurex/MarketData/ComplexInstrumentUpdateListener.h>
#include <OnixS/Eurex/MarketData/FlexibleInstrumentUpdateListener.h>
#include <OnixS/Eurex/MarketData/EmdiHandler.h>
 
#include "ListenerBase.h"
 
class EmdiListener
    : public ListenerBase
    , public OnixS::Eurex::MarketData::ErrorListener
    , public OnixS::Eurex::MarketData::WarningListener
    , public OnixS::Eurex::MarketData::DepthListener
    , public OnixS::Eurex::MarketData::TopOfBookImpliedListener
    , public OnixS::Eurex::MarketData::ProductStateChangeListener
    , public OnixS::Eurex::MarketData::MassInstrumentStateChangeListener
    , public OnixS::Eurex::MarketData::InstrumentStateChangeListener
    , public OnixS::Eurex::MarketData::QuoteRequestListener
    , public OnixS::Eurex::MarketData::CrossRequestListener
    , public OnixS::Eurex::MarketData::ComplexInstrumentUpdateListener
    , public OnixS::Eurex::MarketData::FlexibleInstrumentUpdateListener
{
public:
    explicit EmdiListener(size_t count);

    ~EmdiListener() override;

    void subscribe(OnixS::Eurex::MarketData::EmdiHandler& handler);

    void onError(OnixS::Eurex::MarketData::ErrorCode::Enum code, const std::string& description) override;

    void onWarning(const std::string& description) override;

    void onDepthSnapshot(const OnixS::Eurex::MarketData::DepthSnapshot& snapshot, const OnixS::Eurex::MarketData::DataSource& dataSource) override;

    void onDepthIncremental(const OnixS::Eurex::MarketData::DepthIncremental& incremental, const OnixS::Eurex::MarketData::DataSource& dataSource) override;

    void onDepthReset(OnixS::Eurex::MarketData::MarketSegmentId marketSegmentId) override;

    void onDepthOutOfDate(OnixS::Eurex::MarketData::MarketSegmentId marketSegmentId) override;

    void onTopOfBookImplied(const  OnixS::Eurex::MarketData::TopOfBookImplied& msg, const  OnixS::Eurex::MarketData::DataSource& dataSource) override;

    void onProductStateChange(const OnixS::Eurex::MarketData::ProductStateChange& change, const OnixS::Eurex::MarketData::DataSource& dataSource) override;

    void onMassInstrumentStateChange(const OnixS::Eurex::MarketData::MassInstrumentStateChange& change, const OnixS::Eurex::MarketData::DataSource& dataSource) override;

    void onInstrumentStateChange(const OnixS::Eurex::MarketData::InstrumentStateChange& change, const OnixS::Eurex::MarketData::DataSource& dataSource) override;

    void onQuoteRequest(const OnixS::Eurex::MarketData::QuoteRequest& request, const OnixS::Eurex::MarketData::DataSource& dataSource) override;

    void onCrossRequest(const OnixS::Eurex::MarketData::CrossRequest& request, const OnixS::Eurex::MarketData::DataSource& dataSource) override;

    void onComplexInstrumentUpdate(const OnixS::Eurex::MarketData::ComplexInstrumentUpdate& update, const OnixS::Eurex::MarketData::DataSource& dataSource) override;

    void onFlexibleInstrumentUpdate(const OnixS::Eurex::MarketData::FlexibleInstrumentUpdate& update, const OnixS::Eurex::MarketData::DataSource& dataSource) override;
};

/*
* 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 <map>
#include <climits>
#include <iostream>
#include <sstream>
 
#include "EmdiListener.h"
 
 
using namespace OnixS::Eurex::MarketData;
 
/*
    EmdiListener
*/
 
EmdiListener::EmdiListener(size_t count)
    : ListenerBase(count)
{
}
 
EmdiListener::~EmdiListener() = default;
 
void EmdiListener::subscribe(OnixS::Eurex::MarketData::EmdiHandler& handler)
{
    handler.registerErrorListener(this);
    handler.registerWarningListener(this);
    handler.registerDepthListener(this);
    handler.registerTopOfBookImpliedListener(this);
    handler.registerProductStateChangeListener(this);
    handler.registerMassInstrumentStateChangeListener(this);
    handler.registerInstrumentStateChangeListener(this);
    handler.registerQuoteRequestListener(this);
    handler.registerCrossRequestListener(this);
    handler.registerComplexInstrumentUpdateListener(this);
    handler.registerFlexibleInstrumentUpdateListener(this);
}
 
void EmdiListener::onError(ErrorCode::Enum code, const std::string& description)
{
    std::clog << "Error occurred, errorCode = " << enumToString(code) << ". Description: '" << description << "'" << std::endl;
}
 
void EmdiListener::onWarning(const std::string& description)
{
    std::clog << "Warning occurred. Description: '" << description << "'" << std::endl;
}
 
void EmdiListener::onDepthSnapshot(const DepthSnapshot&, const DataSource&)
{
}
 
void EmdiListener::onDepthIncremental(const DepthIncremental&, const DataSource& dataSource)
{
    if(dataSource.packetMessageSeqNum == 2 && !dataSource.cached)
        addLatency(dataSource.packetReceptionTime);
}
 
void EmdiListener::onDepthReset(MarketSegmentId)
{
}
 
void EmdiListener::onDepthOutOfDate(MarketSegmentId)
{
}
 
void EmdiListener::onTopOfBookImplied(const TopOfBookImplied&, const DataSource& dataSource)
{
    if(dataSource.packetMessageSeqNum == 2 && !dataSource.cached)
        addLatency(dataSource.packetReceptionTime);
}
 
void EmdiListener::onProductStateChange(const ProductStateChange&, const DataSource& dataSource)
{
    if(dataSource.packetMessageSeqNum == 2 && !dataSource.cached)
        addLatency(dataSource.packetReceptionTime);
}
 
void EmdiListener::onMassInstrumentStateChange(const MassInstrumentStateChange&, const DataSource& dataSource)
{
    if(dataSource.packetMessageSeqNum == 2 && !dataSource.cached)
        addLatency(dataSource.packetReceptionTime);
}
 
void EmdiListener::onInstrumentStateChange(const InstrumentStateChange&, const DataSource& dataSource)
{
    if(dataSource.packetMessageSeqNum == 2 && !dataSource.cached)
        addLatency(dataSource.packetReceptionTime);
}
 
void EmdiListener::onQuoteRequest(const QuoteRequest&, const DataSource& dataSource)
{
    if(dataSource.packetMessageSeqNum == 2 && !dataSource.cached )
        addLatency(dataSource.packetReceptionTime);
}
 
void EmdiListener::onCrossRequest(const CrossRequest&, const DataSource& dataSource)
{
    if(dataSource.packetMessageSeqNum == 2 && !dataSource.cached )
        addLatency(dataSource.packetReceptionTime);
}
 
void EmdiListener::onComplexInstrumentUpdate(const ComplexInstrumentUpdate&, const DataSource& dataSource)
{
    if(dataSource.packetMessageSeqNum == 2 && !dataSource.cached)
        addLatency(dataSource.packetReceptionTime);
}
 
void EmdiListener::onFlexibleInstrumentUpdate(const FlexibleInstrumentUpdate&, const DataSource& dataSource)
{
    if(dataSource.packetMessageSeqNum == 2 && !dataSource.cached)
        addLatency(dataSource.packetReceptionTime);
}

/*
* 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 <OnixS/Eurex/MarketData/Compiler.h>
#include <OnixS/Eurex/MarketData/ErrorListener.h>
#include <OnixS/Eurex/MarketData/WarningListener.h>
#include <OnixS/Eurex/MarketData/EOBI/OrderDataListener.h>
#include <OnixS/Eurex/MarketData/EOBI/TradeDataListener.h>
#include <OnixS/Eurex/MarketData/EOBI/StateChangeListener.h>
#include <OnixS/Eurex/MarketData/EOBI/ReferenceDataListener.h>
#include <OnixS/Eurex/MarketData/EOBI/SnapshotListener.h>
#include <OnixS/Eurex/MarketData/EOBI/MessageListener.h>
#include <OnixS/Eurex/MarketData/EOBI/OrderBookListener.h>
#include <OnixS/Eurex/MarketData/EobiHandler.h>
 
#include "ListenerBase.h"
 
 
class EobiListener final
    : public ListenerBase
    , public OnixS::Eurex::MarketData::ErrorListener
    , public OnixS::Eurex::MarketData::WarningListener
    , public OnixS::Eurex::MarketData::EOBI::OrderDataListener
    , public OnixS::Eurex::MarketData::EOBI::TradeDataListener
    , public OnixS::Eurex::MarketData::EOBI::StateChangeListener
    , public OnixS::Eurex::MarketData::EOBI::ReferenceDataListener
{
public:
    explicit EobiListener(size_t count);

    ~EobiListener() override;

    void subscribe(OnixS::Eurex::MarketData::EOBI::EobiHandler& handler);

    void onError(OnixS::Eurex::MarketData::ErrorCode::Enum code, const std::string& description) override;

    void onWarning(const std::string& description) override;
 
    void onOrderAdd(const OnixS::Eurex::MarketData::EOBI::OrderAdd& orderAdd) override;
 
    void onOrderModify(const OnixS::Eurex::MarketData::EOBI::OrderModify& orderModify) override;
 
    void onOrderModifySamePriority(const OnixS::Eurex::MarketData::EOBI::OrderModifySamePriority& orderModify) override;
 
    void onOrderDelete(const OnixS::Eurex::MarketData::EOBI::OrderDelete& orderDelete) override;
 
    void onOrderMassDelete(const OnixS::Eurex::MarketData::EOBI::OrderMassDelete& orderMassDelete) override;
 
    void onPartialOrderExecution(const OnixS::Eurex::MarketData::EOBI::PartialOrderExecution& partialOrderExecution) override;
 
    void onFullOrderExecution(const OnixS::Eurex::MarketData::EOBI::FullOrderExecution& fullOrderExecution) override;
 
    void onAuctionBestBidOffer(const OnixS::Eurex::MarketData::EOBI::AuctionBestBidOffer& auctionBestBidOffer) override;
 
    void onAuctionClearingPrice(const OnixS::Eurex::MarketData::EOBI::AuctionClearingPrice& auctionClearingPrice) override;
 
    void onTopOfBook(const OnixS::Eurex::MarketData::EOBI::TopOfBook& topOfBook) override;
 
    void onExecutionSummary(const OnixS::Eurex::MarketData::EOBI::ExecutionSummary& executionSummary) override;
 
    void onQuoteRequest(const OnixS::Eurex::MarketData::EOBI::QuoteRequest& quoteRequest) override;
 
    void onCrossRequest(const OnixS::Eurex::MarketData::EOBI::CrossRequest& crossRequest) override;
 
    void onTradeReport(const OnixS::Eurex::MarketData::EOBI::TradeReport& tradeReport) override;
 
    void onTradeReversal(const OnixS::Eurex::MarketData::EOBI::TradeReversal& tradeReversal) override;
 
    void onProductStateChange(const OnixS::Eurex::MarketData::EOBI::ProductStateChange& productStateChange) override;
 
    void onInstrumentStateChange(const OnixS::Eurex::MarketData::EOBI::InstrumentStateChange& instrumentStateChange) override;
 
    void onMassInstrumentStateChange(const OnixS::Eurex::MarketData::EOBI::MassInstrumentStateChange& instrumentStateChange) override;
 
    void onAddComplexInstrument(const OnixS::Eurex::MarketData::EOBI::AddComplexInstrument& addComplexInstrument) override;
 
    void onAddFlexibleInstrument(const OnixS::Eurex::MarketData::EOBI::AddFlexibleInstrument& addFlexibleInstrument) override;
 
    void onTESTradeReport(const OnixS::Eurex::MarketData::EOBI::TESTradeReport& tradeReport) override;
 
    void onAddScaledSimpleInstrument(const OnixS::Eurex::MarketData::EOBI::AddScaledSimpleInstrument& addFlexibleInstrument) override;
};

/*
* 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 <map>
#include <iostream>
#include <sstream>
 
#include <OnixS/Eurex/MarketData/EOBI/Messages.h>
 
#include "EobiListener.h"
 
using namespace OnixS::Eurex::MarketData;
using namespace OnixS::Eurex::MarketData::EOBI;
 
/*
    EobiListener
*/
 
EobiListener::EobiListener(size_t count)
    : ListenerBase(count)
{
}
 
EobiListener::~EobiListener() = default;
 
void EobiListener::subscribe(OnixS::Eurex::MarketData::EOBI::EobiHandler& handler)
{
    handler.registerErrorListener(this);
    handler.registerWarningListener(this);
    handler.registerOrderDataListener(this);
    handler.registerTradeDataListener(this);
    handler.registerStateChangeListener(this);
    handler.registerReferenceDataListener(this);
}
 
void EobiListener::onError(ErrorCode::Enum code, const std::string& description)
{
    std::clog << "Error occurred, errorCode = " << enumToString(code) << ". Description: '" << description << "'" << std::endl;
}
 
void EobiListener::onWarning(const std::string& description)
{
    std::clog << "Warning occurred. Description: '" << description << "'" << std::endl;
}
 
void EobiListener::onOrderAdd(const OrderAdd& orderAdd)
{
    const DataSource& dataSource = orderAdd.dataSource();
    if(dataSource.packetMessageSeqNum == 1 && dataSource.completionIndicator && !dataSource.cached)
        addLatency(dataSource.packetReceptionTime);
}
 
void EobiListener::onOrderModify(const OrderModify& orderModify)
{
    const DataSource& dataSource = orderModify.dataSource();
    if(dataSource.packetMessageSeqNum == 1 && dataSource.completionIndicator && !dataSource.cached)
        addLatency(dataSource.packetReceptionTime);
}
 
void EobiListener::onOrderModifySamePriority(const OrderModifySamePriority& orderModify)
{
    const DataSource& dataSource = orderModify.dataSource();
    if(dataSource.packetMessageSeqNum == 1 && dataSource.completionIndicator && !dataSource.cached)
        addLatency(dataSource.packetReceptionTime);
}
 
void EobiListener::onOrderDelete(const OrderDelete& orderDelete)
{
    const DataSource& dataSource = orderDelete.dataSource();
    if(dataSource.packetMessageSeqNum == 1 && dataSource.completionIndicator && !dataSource.cached)
        addLatency(dataSource.packetReceptionTime);
}
 
void EobiListener::onOrderMassDelete(const OrderMassDelete& orderMassDelete)
{
    const DataSource& dataSource = orderMassDelete.dataSource();
    if(dataSource.packetMessageSeqNum == 1 && dataSource.completionIndicator && !dataSource.cached)
        addLatency(dataSource.packetReceptionTime);
}
 
void EobiListener::onPartialOrderExecution(const PartialOrderExecution& partialOrderExecution)
{
    const DataSource& dataSource = partialOrderExecution.dataSource();
    if(dataSource.packetMessageSeqNum == 1 && dataSource.completionIndicator && !dataSource.cached)
        addLatency(dataSource.packetReceptionTime);
}
 
void EobiListener::onFullOrderExecution(const FullOrderExecution& fullOrderExecution)
{
    const DataSource& dataSource = fullOrderExecution.dataSource();
    if(dataSource.packetMessageSeqNum == 1 && dataSource.completionIndicator && !dataSource.cached)
        addLatency(dataSource.packetReceptionTime);
}
 
void EobiListener::onAuctionBestBidOffer(const AuctionBestBidOffer& auctionBestBidOffer)
{
    const DataSource& dataSource = auctionBestBidOffer.dataSource();
    if(dataSource.packetMessageSeqNum == 1 && dataSource.completionIndicator && !dataSource.cached)
        addLatency (dataSource.packetReceptionTime);
}
 
void EobiListener::onAuctionClearingPrice(const AuctionClearingPrice& auctionClearingPrice)
{
    const DataSource& dataSource = auctionClearingPrice.dataSource();
    if (dataSource.packetMessageSeqNum == 1 && dataSource.completionIndicator && !dataSource.cached)
        addLatency (dataSource.packetReceptionTime);
}
 
void EobiListener::onTopOfBook(const TopOfBook& topOfBook)
{
    const DataSource& dataSource = topOfBook.dataSource();
    if (dataSource.packetMessageSeqNum == 1 && dataSource.completionIndicator && !dataSource.cached)
        addLatency (dataSource.packetReceptionTime);
}
 
void EobiListener::onExecutionSummary(const ExecutionSummary& executionSummary)
{
    const DataSource& dataSource = executionSummary.dataSource();
    if (dataSource.packetMessageSeqNum == 1 && dataSource.completionIndicator && !dataSource.cached)
        addLatency (dataSource.packetReceptionTime);
}
 
void EobiListener::onQuoteRequest(const EOBI::QuoteRequest& quoteRequest)
{
    const DataSource& dataSource = quoteRequest.dataSource();
    if (dataSource.packetMessageSeqNum == 1 && dataSource.completionIndicator && !dataSource.cached)
        addLatency (dataSource.packetReceptionTime);
}
 
void EobiListener::onCrossRequest(const EOBI::CrossRequest& crossRequest)
{
    const DataSource& dataSource = crossRequest.dataSource();
    if (dataSource.packetMessageSeqNum == 1 && dataSource.completionIndicator && !dataSource.cached)
        addLatency (dataSource.packetReceptionTime);
}
 
void EobiListener::onTradeReport(const TradeReport& tradeReport)
{
    const DataSource& dataSource = tradeReport.dataSource();
    if (dataSource.packetMessageSeqNum == 1 && dataSource.completionIndicator && !dataSource.cached)
        addLatency (dataSource.packetReceptionTime);
}
 
void EobiListener::onTradeReversal(const TradeReversal& tradeReversal)
{
    const DataSource& dataSource = tradeReversal.dataSource();
    if (dataSource.packetMessageSeqNum == 1 && dataSource.completionIndicator && !dataSource.cached)
        addLatency (dataSource.packetReceptionTime);
}
 
void EobiListener::onProductStateChange(const EOBI::ProductStateChange& productStateChange)
{
    const DataSource& dataSource = productStateChange.dataSource();
    if (dataSource.packetMessageSeqNum == 1 && dataSource.completionIndicator && !dataSource.cached)
        addLatency (dataSource.packetReceptionTime);
}
 
void EobiListener::onInstrumentStateChange(const EOBI::InstrumentStateChange& instrumentStateChange)
{
    const DataSource& dataSource = instrumentStateChange.dataSource();
    if (dataSource.packetMessageSeqNum == 1 && dataSource.completionIndicator && !dataSource.cached)
        addLatency (dataSource.packetReceptionTime);
}
 
void EobiListener::onMassInstrumentStateChange(
    const OnixS::Eurex::MarketData::EOBI::MassInstrumentStateChange& instrumentStateChange)
{
    const DataSource& dataSource = instrumentStateChange.dataSource();
    if (dataSource.packetMessageSeqNum == 1 && dataSource.completionIndicator && !dataSource.cached)
        addLatency (dataSource.packetReceptionTime);
}
 
void EobiListener::onAddComplexInstrument(const AddComplexInstrument& addComplexInstrument)
{
    const DataSource& dataSource = addComplexInstrument.dataSource();
    if (dataSource.packetMessageSeqNum == 1 && dataSource.completionIndicator && !dataSource.cached)
        addLatency (dataSource.packetReceptionTime);
}
 
void EobiListener::onAddFlexibleInstrument (const OnixS::Eurex::MarketData::EOBI::AddFlexibleInstrument& addFlexibleInstrument)
{
    const DataSource& dataSource = addFlexibleInstrument.dataSource();
    if (dataSource.packetMessageSeqNum == 1 && dataSource.completionIndicator && !dataSource.cached)
        addLatency (dataSource.packetReceptionTime);
}
 
void EobiListener::onTESTradeReport(const OnixS::Eurex::MarketData::EOBI::TESTradeReport& tradeReport)
{
    const DataSource& dataSource = tradeReport.dataSource();
    if (dataSource.packetMessageSeqNum == 1 && dataSource.completionIndicator && !dataSource.cached)
        addLatency (dataSource.packetReceptionTime);
}
 
void EobiListener::onAddScaledSimpleInstrument (const OnixS::Eurex::MarketData::EOBI::AddScaledSimpleInstrument& addFlexibleInstrument)
{
    const DataSource& dataSource = addFlexibleInstrument.dataSource();
    if (dataSource.packetMessageSeqNum == 1 && dataSource.completionIndicator && !dataSource.cached)
        addLatency (dataSource.packetReceptionTime);
}

/*
* 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 <vector>
 
#include <OnixS/Eurex/MarketData/TimeSpan.h>
#include <OnixS/Eurex/MarketData/Timestamp.h>
 
 
class ListenerBase
{
public:
    virtual ~ListenerBase();

    void processLatencies();

    void saveLatencies(const std::string& filename);

    bool done() const noexcept
    {
        return latencies_.size() == maxCount_;
    }
 
protected:
    explicit ListenerBase(size_t count);

    void addLatency(const OnixS::Eurex::MarketData::Timestamp&);
 
    typedef unsigned long long Latency;
    typedef std::vector<Latency> Latencies;

    const size_t maxCount_;

    Latencies latencies_;
 
private:
    static Latency calculateAdjustment();
};

/*
* 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 <map>
#include <climits>
#include <iostream>
#include <fstream>
#include <cmath>
#include <algorithm>
 
#include "ListenerBase.h"
 
 
using namespace OnixS::Eurex::MarketData;
 
/*
    EobiListener
*/
 
ListenerBase::ListenerBase(size_t count)
    : maxCount_(count)
{
    latencies_.reserve(maxCount_);
}
 
ListenerBase::~ListenerBase() = default;
 
void ListenerBase::addLatency(const OnixS::Eurex::MarketData::Timestamp& receivingTime)
{
    if ONIXS_EUREX_EMDI_UNLIKELY(latencies_.size() >= maxCount_)
        return;
 
    const TimeSpan latency = OnixS::Eurex::MarketData::Timestamp::utcNow() - receivingTime;
    latencies_.push_back(latency.ticks());
}
 
ListenerBase::Latency ListenerBase::calculateAdjustment()
{
    const int iterations = 10000;
    Latencies latencies;
 
    latencies.reserve(iterations);
 
    for(int i = 0; i < iterations; ++i)
    {
        const TimeSpan latency = Timestamp::utcNow() - Timestamp::utcNow();
        latencies.push_back(latency.ticks());
    }
 
    std::sort(latencies.begin(), latencies.end());
 
    return(latencies[latencies.size() / 2]);
}
 
void ListenerBase::processLatencies()
{
    size_t count = latencies_.size();
 
    if(count == 0)
    {
        std::clog << "Nothing to process(latencies list is empty)." << std::endl;
        return;
    }
 
    std::sort(latencies_.begin(), latencies_.end());
 
    const Latency adjustment = calculateAdjustment();
 
    const long double medianLatency =
        static_cast<double>((latencies_[latencies_.size() / 2]) - adjustment);
 
    std::clog << "Results:" << std::endl;
    std::clog << "Latency(nanoseconds):" << std::endl;
    std::clog << "Count: " << count << std::endl;
    std::clog << "Median: " << medianLatency << std::endl;
}
 
void ListenerBase::saveLatencies(const std::string& filename)
{
    std::ofstream csvFile;
 
    csvFile.open(filename.c_str());
 
    for(auto && latency : latencies_)
        csvFile << latency << std::endl;
 
    csvFile.close();
}