Replay Sample

This sample demonstetes how to use replay functional of the Handler.

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(ONIXS_CMEMDH_NULLPTR != name);
 
        return exist(options(), name) ? 1 : 0;
    }
 
    void showOptions(std::ostream& out) const ONIXS_CMEMDH_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) ONIXS_CMEMDH_OVERRIDE
    {
        gatherLogFiles(list, channel, folder);
    }
 
    void configure(HandlerSettings& /*settings*/) ONIXS_CMEMDH_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) ONIXS_CMEMDH_OVERRIDE
    {
        replayLogFiles(files, handler);
    }
};
 
class PcapReplayManager : public ReplayManager
{
public:
    PcapReplayManager() {}
 
    NetAddressAliases& aliases()
    {
        return supplements_.aliases();
    }
 
    void gather(FileList& files, const std::string& folder, ChannelId) ONIXS_CMEMDH_OVERRIDE
    {
        gatherFiles(files, folder, ".pcap");
    }
 
    void configure(HandlerSettings& settings) ONIXS_CMEMDH_OVERRIDE
    {
        setSessionToNaturalRefresh(settings.session());
    }
 
    void replay(const FileList& files, Handler& handler) ONIXS_CMEMDH_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) ONIXS_CMEMDH_OVERRIDE
    {
        gatherFiles(files, folder, ".gz");
    }
 
    void configure(HandlerSettings& settings) ONIXS_CMEMDH_OVERRIDE
    {
        setSessionToContinueRealtimeProcessingUponIssues(settings.session());
        settings.session().joinRecovery(
            instruments_.empty() ? JoinRecoveryOptions::Disabled : JoinRecoveryOptions::Instruments
        );
        settings.instrumentCache(instruments_);
    }
 
    void replay(const FileList& files, Handler& handler) ONIXS_CMEMDH_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;
    }
}