Replay.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 <string>
#include <vector>
#include <map>
#include <functional>
#include <limits>
 
#include <OnixS/CME/MDH/messaging/Integral.h>
#include <OnixS/CME/MDH/messaging/Time.h>
 
#include <OnixS/CME/MDH/Auxiliary/DeprecatedHeaderWarnings.Disable.h>
#include <OnixS/CME/MDH/Integral.h>
#include <OnixS/CME/MDH/Time.h>
#include <OnixS/CME/MDH/Auxiliary/DeprecatedHeaderWarnings.Restore.h>
 
#include <OnixS/CME/MDH/Domain.h>
#include <OnixS/CME/MDH/FeedAttributes.h>
#include <OnixS/CME/MDH/FeedConnection.h>
#include <OnixS/CME/MDH/ReplayDelayer.h>
#include <OnixS/CME/MDH/SessionSettings.h>
 
ONIXS_CME_MDH_TESTING_NAMESPACE_BEGIN
class ONIXS_CMEMDH_EXPORTED MulticastSender;
ONIXS_CME_MDH_TESTING_NAMESPACE_END
 
ONIXS_CMEMDH_NAMESPACE_BEGIN
 
// A bit of forward declarations.
ONIXS_CMEMDH_EXPORTED_CLASS_DECL(Handler);

typedef std::vector<std::string> FileList;

ONIXS_CMEMDH_EXPORTED
void gatherLogFiles(FileList&, ChannelId, const std::string&);

ONIXS_CMEMDH_EXPORTED
void gatherLogFiles(FileList&, ChannelId, const std::string&, const std::string&);

ONIXS_CMEMDH_EXPORTED
void gatherFiles(FileList&, const std::string&, const std::string&);
 
#if !defined(ONIXS_CMEMDH_NO_DEPRECATED)

inline void gatherPcapFiles(FileList& files, const std::string& location, const std::string& ext)
{
    gatherFiles(files, location, ext);
}
 
#endif // !ONIXS_CMEMDH_NO_DEPRECATED

struct ONIXS_CMEMDH_LTWT HandlerSettingsUse
{
    enum Enum
    {
        AsIs,

        Suggested
    };
};

class ONIXS_CMEMDH_LTWT ReplaySpan
{
public:
    ReplaySpan()
        : begin_()
        , end_()
    {
    }

    ReplaySpan(const Messaging::Timestamp& begin, const Messaging::Timestamp& end)
        : begin_(begin)
        , end_(end)
    {
    }

    ReplaySpan(const ReplaySpan& other)
        : begin_(other.begin_)
        , end_(other.end_)
    {
    }

    bool empty() const
    {
        return (begin_ >= end_);
    }

    bool contain(const Messaging::Timestamp& timestamp) const
    {
        return (begin_ <= timestamp && timestamp < end_);
    }

    const Messaging::Timestamp& begin() const
    {
        return begin_;
    }

    void begin(const Messaging::Timestamp& value)
    {
        begin_ = value;
    }

    const Messaging::Timestamp& end() const
    {
        return end_;
    }

    void end(const Messaging::Timestamp& value)
    {
        end_ = value;
    }

    ReplaySpan& operator=(const ReplaySpan& other)
    {
        begin_ = other.begin_;
        end_ = other.end_;
 
        return *this;
    }
 
private:
    Messaging::Timestamp begin_;
    Messaging::Timestamp end_;
};

struct ONIXS_CMEMDH_LTWT ReplaySpeed
{
    enum Enum
    {
        NoDelay = 0,

        Original = 1,

        X2 = 2
    };
};

template <class DataSource, class DataSourceLess = std::less<DataSource> >
class ReplaySettings
{
public:
    typedef std::map<DataSource, DataSource, DataSourceLess> Aliases;

    ReplaySettings()
        : delayer_(&SleepReplayDelayer::service())
        , settingsUse_(HandlerSettingsUse::Suggested)
        , aliases_()
        , timeSpan_()
        , speed_(ReplaySpeed::NoDelay)
    {
    }

    ReplaySettings(const ReplaySettings& other)
        : delayer_(other.delayer_)
        , settingsUse_(other.settingsUse_)
        , aliases_(other.aliases_)
        , timeSpan_(other.timeSpan_)
        , speed_(other.speed_)
    {
    }

    HandlerSettingsUse::Enum settingsUse() const
    {
        return settingsUse_;
    }

    void settingsUse(HandlerSettingsUse::Enum policy)
    {
        settingsUse_ = policy;
    }

    const Aliases& aliases() const
    {
        return aliases_;
    }

    Aliases& aliases()
    {
        return aliases_;
    }

    const ReplaySpan& timeSpan() const
    {
        return timeSpan_;
    }

    ReplaySpan& timeSpan()
    {
        return timeSpan_;
    }

    ReplaySpeed::Enum speed() const
    {
        return speed_;
    }

    void speed(ReplaySpeed::Enum policy)
    {
        speed_ = policy;
    }

    void delayPolicy(ReplayDelayer& delayer)
    {
        delayer_ = &delayer;
    }

    ReplayDelayer& delayPolicy() const
    {
        assert(delayer_);
        return *delayer_;
    }

    ReplaySettings& operator=(const ReplaySettings& other)
    {
        delayer_ = other.delayer_;
 
        settingsUse_ = other.settingsUse_;
 
        aliases_ = other.aliases_;
 
        timeSpan_ = other.timeSpan_;
 
        speed_ = other.speed_;
 
        return *this;
    }
 
private:
    ReplayDelayer* delayer_;
    HandlerSettingsUse::Enum settingsUse_;
    Aliases aliases_;
 
    ReplaySpan timeSpan_;
    ReplaySpeed::Enum speed_;
};

typedef ReplaySettings<NetFeedId> LogReplaySettings;

typedef LogReplaySettings::Aliases FeedIdAliases;

class ONIXS_CMEMDH_LTWT PcapReplaySettings : public ReplaySettings<NetFeedConnection, NetFeedConnectionLess>
{
public:
    PcapReplaySettings()
    {
        settingsUse(HandlerSettingsUse::AsIs);
    }

    HandlerSettingsUse::Enum settingsUse() const
    {
        return HandlerSettingsUse::AsIs;
    }
 
private:
    typedef ReplaySettings<NetFeedConnection, NetFeedConnectionLess> Base;
 
    // Not applicable for the PCAP replay as no
    // parameters are stored in the captured files.
    using Base::settingsUse;
};

typedef PcapReplaySettings::Aliases NetAddressAliases;

class ONIXS_CMEMDH_LTWT DatamineReplaySettings : public ReplaySettings<ChannelId>
{
public:
    DatamineReplaySettings()
    {
        settingsUse(HandlerSettingsUse::AsIs);
    }

    HandlerSettingsUse::Enum settingsUse() const
    {
        return HandlerSettingsUse::AsIs;
    }
 
private:
    typedef ReplaySettings<ChannelId> Base;
 
    // Not applicable for the Datamine replay as no
    // parameters are stored in the captured files.
    using Base::settingsUse;
};

typedef DatamineReplaySettings::Aliases ChannelIdAliases;

ONIXS_CMEMDH_EXPORTED
void replayLogFiles(const FileList& logs, Handler** handlers, size_t handlerQty, const LogReplaySettings& supplements);

inline void replayLogFiles(const FileList& logs, Handler& handler, const LogReplaySettings& supplements)
{
    Handler* handlers[] = {&handler};
 
    replayLogFiles(logs, handlers, 1, supplements);
}

inline void replayLogFiles(const FileList& logs, Handler** handlers, size_t handlerQty)
{
    replayLogFiles(logs, handlers, handlerQty, LogReplaySettings());
}

inline void replayLogFiles(const FileList& logs, Handler& handler)
{
    Handler* handlers[] = {&handler};
 
    replayLogFiles(logs, handlers, 1);
}

struct ONIXS_CMEMDH_LTWT UseHandlerSettingsAsIs
{};

inline void replayLogFiles(const FileList& logs, Handler** handlers, size_t handlerQty, const UseHandlerSettingsAsIs&)
{
    LogReplaySettings supplements;
 
    supplements.settingsUse(HandlerSettingsUse::AsIs);
 
    replayLogFiles(logs, handlers, handlerQty, supplements);
}

inline void replayLogFiles(const FileList& logs, Handler& handler, const UseHandlerSettingsAsIs& marker)
{
    Handler* handlers[] = {&handler};
 
    replayLogFiles(logs, handlers, 1, marker);
}

ONIXS_CMEMDH_EXPORTED
void replayPcapFiles(const FileList&, Handler**, size_t, const PcapReplaySettings&);

ONIXS_CMEMDH_EXPORTED
void replayPcapFiles(
    const FileList& snapshots,
    MarketRecoveryOptions::Enum snapshotsApplyOptions,
    const FileList& incrementals,
    Handler** handlers,
    size_t handlerQty,
    const PcapReplaySettings& supplements
);

inline void replayPcapFiles(
    const FileList& snapshots,
    MarketRecoveryOptions::Enum snapshotsApplyOptions,
    const FileList& incrementals,
    Handler& handler,
    const PcapReplaySettings& supplements = PcapReplaySettings()
)
{
    Handler* handlers[] = {&handler};
 
    replayPcapFiles(snapshots, snapshotsApplyOptions, incrementals, handlers, 1, supplements);
}

inline void replayPcapFiles(const FileList& logs, Handler& handler, const PcapReplaySettings& supplements)
{
    Handler* handlers[] = {&handler};
 
    replayPcapFiles(logs, handlers, 1, supplements);
}

inline void replayPcapFiles(const FileList& logs, Handler& handler)
{
    replayPcapFiles(logs, handler, PcapReplaySettings());
}

ONIXS_CMEMDH_EXPORTED
void mergeDatamineFiles(const FileList& inFileNames, std::string outFileName, const ReplaySpan& timeSpan = ReplaySpan());

ONIXS_CMEMDH_EXPORTED
void replayDatamineFiles(const FileList&, Handler**, size_t, const DatamineReplaySettings&);

inline void replayDatamineFiles(const FileList& logs, Handler& handler, const DatamineReplaySettings& supplements)
{
    Handler* handlers[] = {&handler};
 
    replayDatamineFiles(logs, handlers, 1, supplements);
}

inline void replayDatamineFiles(const FileList& logs, Handler& handler)
{
    replayDatamineFiles(logs, handler, DatamineReplaySettings());
}

class LogPlayerSettings
{
public:
    LogPlayerSettings(ChannelId channel, FileList logs) noexcept
        : logs_(std::move(logs))
        , channel_(channel)
    {}

    const FileList& logs() const noexcept
    {
        return logs_;
    }

    LogPlayerSettings& logs(FileList logs) noexcept
    {
        logs_ = std::move(logs);
        return *this;
    }

    ChannelId channel() const noexcept
    {
        return channel_;
    }

    LogPlayerSettings& channel(ChannelId value) noexcept
    {
        channel_ = value;
        return *this;
    }

    Messaging::TimeSpan sendDelay() const noexcept
    {
        return sendDelay_;
    }

    LogPlayerSettings& sendDelay(Messaging::TimeSpan value) noexcept
    {
        sendDelay_ = value;
        return *this;
    }

    bool useNaturalSendDelays() const noexcept
    {
        return useNaturalSendDelays_;
    }

    LogPlayerSettings& useNaturalSendDelays(bool value) noexcept
    {
        useNaturalSendDelays_ = value;
        return *this;
    }

    bool updateSendingTime() const noexcept
    {
        return updateSendingTime_;
    }

    LogPlayerSettings& updateSendingTime(bool value) noexcept
    {
        updateSendingTime_ = value;
        return *this;
    }

    Messaging::TimeSpan sourceSwitchDelay() const noexcept
    {
        return sourceSwitchDelay_;
    }

    LogPlayerSettings& sourceSwitchDelay(Messaging::TimeSpan value) noexcept
    {
        sourceSwitchDelay_ = value;
        return *this;
    }

    LogPlayerSettings& delayPolicy(ReplayDelayer& delayer) noexcept
    {
        delayer_ = &delayer;
        return *this;
    }

    ReplayDelayer& delayPolicy() const noexcept
    {
        assert(delayer_);
        return *delayer_;
    }

    size_t incrementalPacketsCount() const noexcept
    {
        return incrementalPacketsCount_;
    }

    LogPlayerSettings& incrementalPacketsCount(size_t value) noexcept
    {
        incrementalPacketsCount_ = value;
        return *this;
    }
 
private:
    FileList logs_;
    ChannelId channel_ = 310;
    Messaging::TimeSpan sendDelay_ = Messaging::MicrosecondSpan(100);
    bool useNaturalSendDelays_ = false;
    bool updateSendingTime_ = false;
    Messaging::TimeSpan sourceSwitchDelay_ = Messaging::MillisecondSpan(1);
    size_t incrementalPacketsCount_ = (std::numeric_limits<size_t>::max)();
    ReplayDelayer* delayer_ = &SleepReplayDelayer::service();
};

ONIXS_CMEMDH_EXPORTED
void replayLogFiles(const LogPlayerSettings& settings, Test::MulticastSender** targets, size_t targetsQty);

inline void replayLogFiles(const LogPlayerSettings& settings, Test::MulticastSender& target)
{
    Test::MulticastSender* targets[] = {&target};
    replayLogFiles(settings, targets, 1);
}
 
ONIXS_CMEMDH_NAMESPACE_END