Replay Sample

The Replay sample feeds previously recorded market data back into the handler without requiring live multicast connectivity. It supports handler logs, PCAP captures, and CME DataMine historical files, and can fall back to matching sample data from ../Data, making it useful for debugging, regression checks, and offline development.

Arguments

Argument	Description	Default
--log [file1 [file2 [..]]]	Replays the specified handler log files. If present without files, sample log data is replayed.	Sample log files from ../Data
--pcap [file1 [file2 [..]]]	Replays the specified PCAP files. If present without files, sample PCAP data is replayed.	Sample PCAP files from ../Data
--datamine [file1 [file2 [..]]]	Replays the specified CME DataMine historical files. If present without files, sample DataMine data is replayed.	Sample .gz files from ../Data
--channel id	Defines which market data channel is replayed.	310
--connectivity <file>	Uses the specified connectivity configuration file.	../Settings/config.xml
--logFile <filename>	Writes replay events to the specified log file.	Replay.log
--logSeverity Important\|Regular\|Debug	Controls event logging verbosity.	Regular
--help	Displays command-line help.	Disabled

Running Examples

Replay the default sample log data for channel 310:

./Replay --channel 310

Replay a specific handler log file:

./Replay --channel 310 --log ../Data/CmeMdpPremiumHandler-Channel_310.log

Replay sample PCAP data:

./Replay --channel 310 --pcap

Replay sample CME DataMine data:

./Replay --channel 310 --datamine

Notes

Only one replay source option can be selected at a time: --log, --pcap, or --datamine. When no replay source option is selected, log replay is used.

Source Code

 
using namespace OnixS::CME::MDH;
 
// Options.
 
class Configuration
    : public ChannelConfiguration
    , public ConnectivityConfiguration
    , public LoggerConfiguration
{
public:
    Configuration(size_t qty, char** args)
        : ConfigurationBase(qty, args)
        , ChannelConfiguration(qty, args, 310)
        , ConnectivityConfiguration(qty, args)
        , LoggerConfiguration(qty, args, "Replay.log")
    {
        const unsigned optionQty = countOption("log") + countOption("pcap") + countOption("datamine");
 
        if (1 < optionQty)
        {
            throw std::runtime_error("Cannot initialize market data replay "
                                     "due to conflict in the input parameters. "
                                     "Multiple replay sources (Log, PCAP, DataMine) "
                                     "selected at the same time. Please, fix the command "
                                     "line arguments to use only one type of market data "
                                     "source (either log or PCAP or DataMine). ");
        }
    }
 
    bool replayPcap() const
    {
        return exist(options(), "pcap");
    }
 
    bool replayDatamine() const
    {
        return exist(options(), "datamine");
    }
 
    bool filesToReplay(FileList& files) const
    {
        const OptionArgs* args = options().args("log");
 
        if (!args)
        {
            args = options().args("pcap");
 
            if (!args)
            {
                args = options().args("datamine");
            }
        }
 
        if (args)
        {
            std::copy(args->begin(), args->end(), std::back_inserter(files));
 
            return true;
        }
 
        return false;
    }
 
private:
    unsigned countOption(const char* name) const
    {
        assert(nullptr != name);
 
        return exist(options(), name) ? 1 : 0;
    }
 
    void showOptions(std::ostream& out) const override
    {
        out << "  --log [file1 [file2 [..]]]" << std::endl
            << "\tTells to replay the given list of log files. " << std::endl
            << "\tIf no files are specified, sample data is replayed. " << std::endl
            << std::endl
            << "  --pcap [file1 [file2 [..]]]" << std::endl
            << "\tTells to replay the given list of PCAP files. " << std::endl
            << "\tIf no files are specified, sample data is replayed. " << std::endl
            << std::endl
            << "  --datamine [file1 [file2 [..]]]" << std::endl
            << "\tTells to replay the given list of CME DataMine (historical) files. " << std::endl
            << "\tIf no files are specified, sample data is replayed. " << std::endl
            << std::endl;
 
        ChannelConfiguration::showOptions(out);
        ConnectivityConfiguration::showOptions(out);
        LoggerConfiguration::showOptions(out);
    }
};
 
class ReplayManager
{
public:
    virtual ~ReplayManager() {}
 
    virtual void gather(FileList&, const std::string&, ChannelId) = 0;
 
    virtual void configure(HandlerSettings&) = 0;
 
    virtual void replay(const FileList&, Handler&) = 0;
 
protected:
    ReplayManager() {}
};
 
struct LogReplayManager : ReplayManager
{
    LogReplayManager() {}
 
    void gather(FileList& list, const std::string& folder, ChannelId channel) override
    {
        gatherLogFiles(list, channel, folder);
    }
 
    void configure(HandlerSettings& /*settings*/) override
    {
        // In the most common way, the Handler grabs all required
        // parameters from the log files it replays. If the data
        // has to be replayed with the different session settings,
        // this is the right place to update Handler's parameters.
        // However, in such case, the replay must be configured
        // additionally.
    }
 
    void replay(const FileList& files, Handler& handler) override
    {
        replayLogFiles(files, handler);
    }
};
 
class PcapReplayManager : public ReplayManager
{
public:
    PcapReplayManager() {}
 
    NetAddressAliases& aliases()
    {
        return supplements_.aliases();
    }
 
    void gather(FileList& files, const std::string& folder, ChannelId) override
    {
        gatherFiles(files, folder, ".pcap");
    }
 
    void configure(HandlerSettings& settings) override
    {
        setSessionToNaturalRefresh(settings.session());
    }
 
    void replay(const FileList& files, Handler& handler) override
    {
        if (supplements_.aliases().empty())
        {
            replayPcapFiles(files, handler);
        }
        else
        {
            replayPcapFiles(files, handler, supplements_);
        }
    }
 
private:
    PcapReplaySettings supplements_;
};
 
class DatamineReplayManager : public ReplayManager
{
public:
    explicit DatamineReplayManager(bool useInstrumentCache)
        : instruments_(useInstrumentCache ? "../Data/SecDef-310.dat" : "")
    {
    }
 
    void gather(FileList& files, const std::string& folder, ChannelId) override
    {
        gatherFiles(files, folder, ".gz");
    }
 
    void configure(HandlerSettings& settings) override
    {
        setSessionToContinueRealtimeProcessingUponIssues(settings.session());
        settings.session().joinRecovery(
            instruments_.empty() ? JoinRecoveryOptions::Disabled : JoinRecoveryOptions::Instruments
        );
        settings.instrumentCache(instruments_);
    }
 
    void replay(const FileList& files, Handler& handler) override
    {
        replayDatamineFiles(files, handler);
    }
 
private:
    std::string instruments_;
};
 
// Spawns market data processing from the previously recorded log file.
class Application
{
public:
    // Initializes the sample according to the arguments.
    Application(const Configuration& configuration)
        : configuration_(configuration)
    {
        // Configures replay manager for the
        // selected data (either log or PCAP).
 
        configuration_.filesToReplay(files_);
 
        if (configuration_.replayPcap())
        {
            ScopedPtr<PcapReplayManager> pcapManager(new PcapReplayManager());
 
            if (files_.empty())
            {
                // The Handler is configured to use connectivity
                // configuration for NRCert environment, whereas
                // recorded packets belong to production environment.
                // The following steps force the PCAP replay machinery
                // to map data from production incremental feed to the
                // certification one.
 
                pcapManager->aliases()[NetFeedConnection("224.0.31.1", 14310)] = NetFeedConnection("224.0.28.1", 14310);
            }
 
            replayManager_.reset(pcapManager.release());
        }
        else if (configuration_.replayDatamine())
        {
            replayManager_.reset(new DatamineReplayManager(files_.empty()));
        }
        else
        {
            replayManager_.reset(new LogReplayManager());
        }
 
        if (files_.empty())
        {
            replayManager_->gather(files_, "../Data", configuration_.channel());
        }
    }
 
    // Allows instances to be destroyed.
    ~Application() {}
 
    // Spawns market data processing from
    // the previously recorded log file.
    void run()
    {
        // Even in replay, handler still logs its events.
        FileLoggerEventTracer loggerTracer;
 
        const ScopedPtr<Logger> logger(constructFileLogger(configuration_, &loggerTracer));
 
        // Constructs handler for replay in regular way.
        Handler handler;
 
        // A bit of configuring for the instance.
        HandlerSettings& settings = handler.settings();
        setLicense(settings);
 
        settings.channel(configuration_.channel()).connectivityConfigurationFile(configuration_.connectivityFile());;
 
        settings.bookManagement().consolidatedBooks().maintain(true);
 
        settings.bookManagement().mboBooks().maintain(true);
 
        // Additional turn-up.
        replayManager_->configure(settings);
 
        // Binds the given logger to the Handler.
        apply(settings.logging(), configuration_, logger.get());
 
        // Associates events processing with handler.
        SecurityEventTracer securityEventTracer;
        securityEventTracer.bind(handler);
 
        MarketDataEventTracer marketDataTracer;
        marketDataTracer.bind(handler);
 
        ServiceEventTracer serviceTracer;
        serviceTracer.bind(handler);
 
        std::cout << "Replaying the recorded market data.. " << std::endl;
 
        replayManager_->replay(files_, handler);
 
        std::cout << std::endl << "Done. " << std::endl << std::endl;
    }
 
    // A bit of information on the sample.
    static void identify()
    {
        std::cout << "Market Data Replay for the CME MDP Premium Market Data Handler, v" << toStr(Version::current())
                  << ". "
                  << std::endl
                  << std::endl;
    }
 
private:
    const Configuration& configuration_;
 
    ScopedPtr<ReplayManager> replayManager_;
 
    FileList files_;
 
    // Prohibits copy construction.
    Application(const Application&);
 
    // No re-assignments for this time.
    Application& operator=(const Application&);
};
 
int main(int qty, char** args)
{
    try
    {
        Application::identify();
 
        const Configuration configuration(qty, args);
 
        if (configuration.show())
        {
            configuration.show(std::cout);
        }
        else
        {
            Application(configuration).run();
        }
 
        return 0;
    }
    catch (const std::exception& ex)
    {
        std::cerr << std::endl << "ERROR: " << ex.what() << std::endl;
 
        return 1;
    }
}