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 <OnixS/CME/MDH/Integral.h>
#include <OnixS/CME/MDH/Time.h>

#include <OnixS/CME/MDH/Domain.h>
#include <OnixS/CME/MDH/FeedAttributes.h>
#include <OnixS/CME/MDH/FeedConnection.h>
#include <OnixS/CME/MDH/SessionSettings.h>

ONIXS_CMEMDH_NAMESPACE_BEGIN

// A bit of forward declarations.
ONIXS_CMEMDH_EXPORTED_CLASS_DECL(Handler);

/// Ordered list of logs to be replayed.
typedef std::vector<std::string> FileList;

/// Gathers log files logs which are available for
/// given channel and are stored in a given folder.
/// Gathering assumes log files are named by log file
/// naming routine exposed by given SDK.
ONIXS_CMEMDH_EXPORTED
void gatherLogFiles(FileList&, ChannelId, const std::string&);

/// Gathers log files logs which are available for
/// given channel and are stored in a given folder.
/// Gathering assumes log files are named by log file
/// naming routine exposed by given SDK. In contrast
/// to other overloads, given one allows to define
/// prefix pattern for logs to be gathered.
ONIXS_CMEMDH_EXPORTED
void gatherLogFiles(FileList&, ChannelId, const std::string&, const std::string&);

/// Gathers files which are stored in the given folder with
/// the given extension. Gathered files are sorted by name.
///
/// In contrast to the gatherLogFiles function gathering log
/// files, which collects log files in exact order the files
/// were produced by a logger, the given one represents a generic
/// routine finding files according to the given pattern (extension).
ONIXS_CMEMDH_EXPORTED
void gatherFiles(FileList&, const std::string&, const std::string&);

#if !defined(ONIXS_CMEMDH_NO_DEPRECATED)

/// Gathers files which are stored in the given folder with
/// the given extension. Gathered files are sorted by name.
///
/// \warning The given function is deprecated.
/// Consider using `gatherFiles` instead.
inline void gatherPcapFiles(FileList& files, const std::string& location, const std::string& ext)
{
    gatherFiles(files, location, ext);
}

#endif // !ONIXS_CMEMDH_NO_DEPRECATED

/// Defines whether processing and other settings
/// must be extracted from logs and used during the
/// replay, or the original settings as they were
/// defined by replay invoker must be used.
struct ONIXS_CMEMDH_LTWT HandlerSettingsUse
{
    enum Enum
    {
        /// Handler settings aren't touched and
        /// used as they are during the replay.
        AsIs,

        /// Settings affecting the way the data
        /// is processed must be extracted from
        /// the logs and applied to the Handlers
        /// before replay starts.
        Suggested
    };
};

/// Defines range of log entries to be replayed.
class ONIXS_CMEMDH_LTWT ReplaySpan
{
public:
    /// Initializes empty span.
    ReplaySpan()
        : begin_()
        , end_()
    {
    }

    /// Initializes the instance
    /// according to the given bounds.
    ReplaySpan(const Timestamp& begin, const Timestamp& end)
        : begin_(begin)
        , end_(end)
    {
    }

    /// Initializes the instance
    /// as a copy of the other one.
    ReplaySpan(const ReplaySpan& other)
        : begin_(other.begin_)
        , end_(other.end_)
    {
    }

    /// Indicates whether span is empty or not.
    bool empty() const
    {
        return (begin_ >= end_);
    }

    /// Indicates whether the given timestamp
    /// belongs to the given time span.
    bool contain(const Timestamp& timestamp) const
    {
        return (begin_ <= timestamp && timestamp < end_);
    }

    /// Indicates the beginning of the range.
    const Timestamp& begin() const
    {
        return begin_;
    }

    /// Updates the beginning of the range.
    void begin(const Timestamp& value)
    {
        begin_ = value;
    }

    /// Indicates the end of the range.
    const Timestamp& end() const
    {
        return end_;
    }

    /// Indicates the end of the range.
    void end(const Timestamp& value)
    {
        end_ = value;
    }

    /// Re-initializes as a copy of the other one.
    ReplaySpan& operator=(const ReplaySpan& other)
    {
        begin_ = other.begin_;
        end_ = other.end_;

        return *this;
    }

private:
    Timestamp begin_;
    Timestamp end_;
};

/// Controls speed of market data replay.
struct ONIXS_CMEMDH_LTWT ReplaySpeed
{
    /// Controls speed of market data replay.
    enum Enum
    {
        /// Replays run at maximal speed.
        NoDelay = 0,

        /// Data is replayed with the original speed.
        Original = 1,

        /// Data is replayed two times faster in
        /// compare to the original speed at which
        /// the Handler was receiving that data.
        X2 = 2
    };
};

/// Controls the replay delay.
struct ONIXS_CMEMDH_EXPORTED ReplayDelayer
{
    /// Implements the delay between packets.
    virtual void onDelay(const TimeSpan& delay) = 0;
};

/// Implements the replay delay using the `sleep` system call.
struct ONIXS_CMEMDH_EXPORTED SleepReplayDelayer : public ReplayDelayer
{
  /// Implements the delay between packets.
    void onDelay(const TimeSpan& delay) ONIXS_CMEMDH_OVERRIDE;

    /// Returns the instance reference.
    static ReplayDelayer& service();
};

/// Implements the replay delay by polling in the userspace
struct ONIXS_CMEMDH_EXPORTED PollReplayDelayer : public ReplayDelayer
{
    /// Implements the delay between packets.
    void onDelay(const TimeSpan& delay) ONIXS_CMEMDH_OVERRIDE;

    /// Returns the instance reference.
    static ReplayDelayer& service();
};

/// Various supplemental settings affecting
/// the way the logged data is replayed.
template <class DataSource, class DataSourceLess = std::less<DataSource> >
class ReplaySettings
{
public:
    /// The table of data source aliases.
    typedef std::map<DataSource, DataSource, DataSourceLess> Aliases;

    /// Initializes the settings with the default values.
    ReplaySettings()
        : delayer_(&SleepReplayDelayer::service())
        , settingsUse_(HandlerSettingsUse::Suggested)
        , aliases_()
        , timeSpan_()
        , speed_(ReplaySpeed::NoDelay)
    {
    }

    /// Initializes as a copy of the other one.
    ReplaySettings(const ReplaySettings& other)
        : delayer_(other.delayer_)
        , settingsUse_(other.settingsUse_)
        , aliases_(other.aliases_)
        , timeSpan_(other.timeSpan_)
        , speed_(other.speed_)
    {
    }

    /// Handler settings use policy.
    HandlerSettingsUse::Enum settingsUse() const
    {
        return settingsUse_;
    }

    /// Defines handler settings use policy.
    void settingsUse(HandlerSettingsUse::Enum policy)
    {
        settingsUse_ = policy;
    }

    /// Aliases to be used during the replay.
    ///
    /// Aliases allow redirecting data from one source
    /// (feed, multicast group, channel, etc.) to the other
    /// one.
    const Aliases& aliases() const
    {
        return aliases_;
    }

    /// Aliases to be used during the replay.
    ///
    /// Aliases allow redirecting data from one source
    /// (feed, multicast group, channel, etc.) to the other
    /// one.
    Aliases& aliases()
    {
        return aliases_;
    }

    /// Time span for which entries are to be processed.
    const ReplaySpan& timeSpan() const
    {
        return timeSpan_;
    }

    /// Time span for which entries are to be processed.
    ReplaySpan& timeSpan()
    {
        return timeSpan_;
    }

    /// Indicates processing speed policy.
    ReplaySpeed::Enum speed() const
    {
        return speed_;
    }

    /// Defines processing speed.
    void speed(ReplaySpeed::Enum policy)
    {
        speed_ = policy;
    }

    /// Sets a policy for implementing replay delay.
    void delayPolicy(ReplayDelayer& delayer)
    {
        delayer_ = &delayer;
    }

    /// Indicates policy for implementing replay delay.
    ReplayDelayer& delayPolicy() const
    {
        assert(delayer_);
        return *delayer_;
    }

    /// Re-initializes as a copy of the other one.
    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_;
};

/// Replay supplements for log replay functionality.
typedef ReplaySettings<NetFeedId> LogReplaySettings;

/// Collection of logged data sources (feed
/// identifiers) to be used by the Log Replay
/// to replace logged sources with the sources
/// used by Handlers involved into replay.
typedef LogReplaySettings::Aliases FeedIdAliases;

/// Replay supplements for the PCAP replay functionality.
class ONIXS_CMEMDH_LTWT PcapReplaySettings : public ReplaySettings<NetFeedConnection, NetFeedConnectionLess>
{
public:
    /// Initializes settings with the default values.
    PcapReplaySettings()
    {
        settingsUse(HandlerSettingsUse::AsIs);
    }

    /// Settings aren't stored in PCAPs
    /// thus constant value is returned.
    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;
};

/// Collection of data sources (feed identifiers)
/// to be used by the PCAP file replay to replace
/// captured sources with the sources used by Handlers
/// involved into replay.
typedef PcapReplaySettings::Aliases NetAddressAliases;

/// Replay supplements for the CME DataMine
/// historical data replay functionality.
class ONIXS_CMEMDH_LTWT DatamineReplaySettings : public ReplaySettings<ChannelId>
{
public:
    /// Initializes settings with the default values.
    DatamineReplaySettings()
    {
        settingsUse(HandlerSettingsUse::AsIs);
    }

    /// Settings aren't stored in Datamine
    /// thus constant value is returned.
    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;
};

/// Collection of channels to be used by the CME Datamine
/// historical file replay to replace captured sources
/// with the sources used by Handlers involved into replay.
typedef DatamineReplaySettings::Aliases ChannelIdAliases;

/// Extracts market data stored in the given
/// log files and pushes it to the given handlers
/// for further processing.
ONIXS_CMEMDH_EXPORTED
void replayLogFiles(const FileList&, Handler**, size_t, const LogReplaySettings&);

/// Extracts market data stored in the given
/// log files and pushes it to the given handler
/// for further processing.
inline void replayLogFiles(const FileList& logs, Handler& handler, const LogReplaySettings& supplements)
{
    Handler* handlers[] = {&handler};

    replayLogFiles(logs, handlers, 1, supplements);
}

/// Processes market data stored in the log
/// files according to specified settings.
inline void replayLogFiles(const FileList& logs, Handler** handlers, size_t handlerQty)
{
    replayLogFiles(logs, handlers, handlerQty, LogReplaySettings());
}

/// Processes market data stored in the log
/// files according to specified settings.
inline void replayLogFiles(const FileList& logs, Handler& handler)
{
    Handler* handlers[] = {&handler};

    replayLogFiles(logs, handlers, 1);
}

/// A Marker instructing the log replay procedure to
/// use Handler settings as they are without applying
/// any modifications based on logged information.
struct ONIXS_CMEMDH_LTWT UseHandlerSettingsAsIs
{};

/// Processes market data stored in the log
/// files according to specified settings.
inline void replayLogFiles(const FileList& logs, Handler** handlers, size_t handlerQty, const UseHandlerSettingsAsIs&)
{
    LogReplaySettings supplements;

    supplements.settingsUse(HandlerSettingsUse::AsIs);

    replayLogFiles(logs, handlers, handlerQty, supplements);
}

/// Processes market data stored in the log
/// files according to specified settings.
inline void replayLogFiles(const FileList& logs, Handler& handler, const UseHandlerSettingsAsIs& marker)
{
    Handler* handlers[] = {&handler};

    replayLogFiles(logs, handlers, 1, marker);
}

/// Replays the given list of PCAP files for the
/// given Handlers according to the given settings.
ONIXS_CMEMDH_EXPORTED
void replayPcapFiles(const FileList&, Handler**, size_t, const PcapReplaySettings&);

/// Applies the given list of PCAP files with snapshots then
/// replays the given list of PCAP files with incrementals for the
/// given Handlers according to the given settings.
/// @note snapshots are being applied with no delay
/// @note only the first snapshot cycle from the given snapshots is applied, the rest are skipped
ONIXS_CMEMDH_EXPORTED
void replayPcapFiles(
    const FileList& snapshots,
    MarketRecoveryOptions::Enum snapshotsApplyOptions,
    const FileList& incrementals,
    Handler** handlers,
    size_t handlerQty,
    const PcapReplaySettings& supplements
);

/// Applies the given list of PCAP files with snapshots then
/// replays the given list of PCAP files with incrementals for the
/// given Handler according to the given settings.
/// @note snapshots are being applied with no delay
/// @note only the first snapshot cycle from the given snapshots is applied, the rest are skipped
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);
}

/// Replays the given list of PCAP files for the
/// given Handler according to the given settings.
inline void replayPcapFiles(const FileList& logs, Handler& handler, const PcapReplaySettings& supplements)
{
    Handler* handlers[] = {&handler};

    replayPcapFiles(logs, handlers, 1, supplements);
}

/// Replays the given list of PCAP files for the
/// given Handler according to the default settings.
inline void replayPcapFiles(const FileList& logs, Handler& handler)
{
    replayPcapFiles(logs, handler, PcapReplaySettings());
}

/// Merges the given Datamine files into a single one. The output file is gzipped.
ONIXS_CMEMDH_EXPORTED
void mergeDatamineFiles(const FileList& inFileNames, std::string outFileName, const ReplaySpan& timeSpan = ReplaySpan());

/// Replays the given list of Datamine files for the
/// given Handlers according to the given settings.
ONIXS_CMEMDH_EXPORTED
void replayDatamineFiles(const FileList&, Handler**, size_t, const DatamineReplaySettings&);

/// Replays the given list of Datamine files for the
/// given Handlers according to the given settings.
inline void replayDatamineFiles(const FileList& logs, Handler& handler, const DatamineReplaySettings& supplements)
{
    Handler* handlers[] = {&handler};

    replayDatamineFiles(logs, handlers, 1, supplements);
}

/// Replays the given list of historical data for the
/// given Handler according to the default settings.
inline void replayDatamineFiles(const FileList& logs, Handler& handler)
{
    replayDatamineFiles(logs, handler, DatamineReplaySettings());
}

ONIXS_CMEMDH_NAMESPACE_END