Multicast Replay Sample

This sample demonstrates how to replay CME MDP Premium Market Data Handler log files through multicast feed settings.

The sample can replay an explicit list of log files or, when no log files are provided, gather matching sample data for the selected channel from the ../Data folder. It also allows replay delay control, feed A/feed B interface overrides, primary feed selection, and arbitrage simulation for all multicast feed targets.

Argument	Description	Default
--channel id	Defines which market data channel is replayed.	310
--connectivity <file>	Uses the specified connectivity configuration file.	../Settings/config.xml
--log [file1 [file2 [..]]]	Replays the specified log files. If omitted, sample data is replayed.	Matching files from ../Data
--send-delay <natural\|microseconds>	Defines the delay before sending another packet to the same source. Use natural to preserve delays from the log file.	natural
--source-switch-delay <microseconds>	Defines the delay when target NetFeedType changes.	1000
--update-sending-time <true\|false>	Updates SendingTime in the Binary Packet Header before sending each packet.	false
--primary-feed <A\|B>	Defines the feed used as the primary feed when sending data. Without arbitrage, only this feed is used.	A
--simulate-arbitrage	Sends data through both feeds for all configured multicast feed targets.	Disabled
--arbitrage-delay <microseconds>	Defines the delay before sending the same packet to the secondary feed when arbitrage simulation is enabled.	0
--iface-A <interface>	Specifies the network interface to use for feed A.	Not overridden
--iface-B <interface>	Specifies the network interface to use for feed B.	Not overridden
--packets-count <count>	Stops replay after the specified number of packets sent to incremental feed targets. Historical packets remapped to incremental are included.	Unlimited

Example:

./MulticastReplay --channel 310 --connectivity ../Settings/config.xml --log ../Data/CmeMdpPremiumHandler-Channel_310.log

 
using namespace OnixS::CME::MDH;
 
namespace {
 
    const std::string SampleDataLocation = "../Data";
 
    // Options.
 
    class MulticastReplayConfiguration : public virtual ConfigurationBase
    {
    public:
        bool filesToReplay(FileList& files) const
        {
            const Messaging::Char* const option = "log";
            const auto args = options().args(option);
 
            if (args)
            {
                const auto argQty = args->size();
 
                if (0 == argQty)
                    throwBadOptionArgQty(option, argQty, 1, static_cast<size_t>(-1));
 
                std::copy(args->begin(), args->end(), std::back_inserter(files));
 
                return true;
            }
 
            return false;
        }
 
        LogPlayerSettings& apply(LogPlayerSettings& settings) const
        {
            applySendDelay(settings);
 
            settings.delayPolicy(CombinedReplayDelayer::service());
            settings.sourceSwitchDelay(sourceSwitchDelay());
            settings.updateSendingTime(updateSendingTime());
            settings.incrementalPacketsCount(packetsCount());
 
            return settings;
        }
 
        FeedSettings& apply(FeedSettings& settings) const
        {
            applyFeedLayout(settings);
 
            if (const auto args = argsFor("iface-A", 1, 1))
                settings.feedANetworkInterfaces((*args)[0]);
 
            if (const auto args = argsFor("iface-B", 1, 1))
                settings.feedBNetworkInterfaces((*args)[0]);
 
            return settings;
        }
 
        Messaging::TimeSpan arbitrageDelay() const
        {
            const Messaging::Char* const option = "arbitrage-delay";
            const auto args = argsFor(option, 1, 1);
            const auto delay = args ? convertUInt64(option, (*args)[0]) : 0;
 
            return makeMicrosecondSpan(option, delay);
        }
 
    protected:
        MulticastReplayConfiguration(size_t qty, char** args)
            : ConfigurationBase(qty, args)
        {
        }
 
        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 omitted, sample data is replayed. " << std::endl
                << std::endl
                << "  --send-delay <natural|microseconds>" << std::endl
                << "\tDefines delay before sending another packet to the same source. " << std::endl
                << "\tUse natural to preserve delays from the log file. Default is natural. " << std::endl
                << std::endl
                << "  --source-switch-delay <microseconds>" << std::endl
                << "\tDefines delay when target NetFeedType changes. Default is 1000. " << std::endl
                << std::endl
                << "  --update-sending-time <true|false>" << std::endl
                << "\tUpdates SendingTime in the Binary Packet Header before sending each packet. "
                << "Default is false. " << std::endl
                << std::endl
                << "  --primary-feed <A|B>" << std::endl
                << "\tDefines feed used as the primary feed when sending data. "
                << "Without arbitrage, only this feed is used. Default is A. " << std::endl
                << std::endl
                << "  --simulate-arbitrage" << std::endl
                << "\tSends data through both feeds to simulate arbitrage. " << std::endl
                << std::endl
                << "  --arbitrage-delay <microseconds>" << std::endl
                << "\tDefines delay before sending the same packet to the secondary feed. "
                << "Default is 0. " << std::endl
                << std::endl
                << "  --iface-A <interface>" << std::endl
                << "\tSpecifies network interface to use for feed A. " << std::endl
                << std::endl
                << "  --iface-B <interface>" << std::endl
                << "\tSpecifies network interface to use for feed B. " << std::endl
                << std::endl
                << "  --packets-count <count>" << std::endl
                << "\tStops replay after the specified number of packets sent to incremental "
                << "feed targets. Historical packets remapped to incremental are included. " << std::endl
                << std::endl;
        }
 
    private:
        const OptionArgs* argsFor(const Messaging::Char* option, size_t minQty, size_t maxQty) const
        {
            const auto args = options().args(option);
 
            if (args)
            {
                const auto argQty = args->size();
 
                if (argQty < minQty || maxQty < argQty)
                    throwBadOptionArgQty(option, argQty, minQty, maxQty);
            }
 
            return args;
        }
 
        static Messaging::TimeSpan makeMicrosecondSpan(const Messaging::Char* option, Messaging::UInt64 microseconds)
        {
            typedef Messaging::TimeSpan::Microseconds Microseconds;
 
            if (microseconds > static_cast<Messaging::UInt64>((std::numeric_limits<Microseconds>::max)()))
                throwOptionArgConversionFailed(option);
 
            return Messaging::MicrosecondSpan(static_cast<Microseconds>(microseconds));
        }
 
        static Messaging::UInt64 convertUInt64(const Messaging::Char* option, const std::string& value)
        {
            return OptionArgConverter<Messaging::UInt64>(option)(value);
        }
 
        void applySendDelay(LogPlayerSettings& settings) const
        {
            const Messaging::Char* const option = "send-delay";
            const auto args = argsFor(option, 1, 1);
 
            if (!args || (*args)[0] == "natural")
            {
                settings.useNaturalSendDelays(true);
                return;
            }
 
            settings.useNaturalSendDelays(false);
            settings.sendDelay(makeMicrosecondSpan(option, convertUInt64(option, (*args)[0])));
        }
 
        Messaging::TimeSpan sourceSwitchDelay() const
        {
            const Messaging::Char* const option = "source-switch-delay";
            const auto args = argsFor(option, 1, 1);
            const auto delay = args ? convertUInt64(option, (*args)[0]) : 1000;
 
            return makeMicrosecondSpan(option, delay);
        }
 
        bool updateSendingTime() const
        {
            const Messaging::Char* const option = "update-sending-time";
            const auto args = argsFor(option, 1, 1);
 
            return args ? OptionArgConverter<bool>(option)((*args)[0]) : false;
        }
 
        NetFeedRole::Enum primaryFeed() const
        {
            const Messaging::Char* const option = "primary-feed";
            const auto args = argsFor(option, 1, 1);
 
            return args ? OptionArgConverter<NetFeedRole::Enum>(option)((*args)[0]) : NetFeedRole::A;
        }
 
        bool simulateArbitrage() const
        {
            const Messaging::Char* const option = "simulate-arbitrage";
 
            return nullptr != argsFor(option, 0, 0);
        }
 
        static RecoveryFeedLayout::Enum recoveryFeedLayout(NetFeedRole::Enum primaryFeed, bool arbitrage)
        {
            switch (primaryFeed)
            {
                case NetFeedRole::A:
                    return arbitrage ? RecoveryFeedLayout::FeedAWithFailoverToFeedB : RecoveryFeedLayout::FeedAOnly;
 
                case NetFeedRole::B:
                    return arbitrage ? RecoveryFeedLayout::FeedBWithFailoverToFeedA : RecoveryFeedLayout::FeedBOnly;
 
                default:
                    throwOptionArgConversionFailed("primary-feed");
                    return RecoveryFeedLayout::FeedAOnly;
            }
        }
 
        static RealtimeFeedLayout::Enum realtimeFeedLayout(NetFeedRole::Enum primaryFeed, bool arbitrage)
        {
            switch (primaryFeed)
            {
                case NetFeedRole::A:
                    return arbitrage ? RealtimeFeedLayout::FeedAWithFailoverToFeedB : RealtimeFeedLayout::FeedAOnly;
 
                case NetFeedRole::B:
                    return arbitrage ? RealtimeFeedLayout::FeedBWithFailoverToFeedA : RealtimeFeedLayout::FeedBOnly;
 
                default:
                    throwOptionArgConversionFailed("primary-feed");
                    return RealtimeFeedLayout::FeedAOnly;
            }
        }
 
        void applyFeedLayout(FeedSettings& settings) const
        {
            const auto primary = primaryFeed();
            const auto arbitrage = simulateArbitrage();
            const auto recoveryLayout = recoveryFeedLayout(primary, arbitrage);
 
            settings.instrumentFeeds().layout(recoveryLayout);
            settings.snapshotFeeds().layout(recoveryLayout);
            settings.mboSnapshotFeeds().layout(recoveryLayout);
            settings.incrementalFeeds().layout(realtimeFeedLayout(primary, arbitrage));
        }
 
        size_t packetsCount() const
        {
            const Messaging::Char* const option = "packets-count";
            const auto args = argsFor(option, 1, 1);
 
            if (!args)
                return (std::numeric_limits<size_t>::max)();
 
            const auto count = convertUInt64(option, (*args)[0]);
 
            if (count > static_cast<Messaging::UInt64>((std::numeric_limits<size_t>::max)()))
                throwOptionArgConversionFailed(option);
 
            return static_cast<size_t>(count);
        }
    };
 
    class Configuration
        : public ChannelConfiguration
        , public ConnectivityConfiguration
        , public MulticastReplayConfiguration
    {
    public:
        Configuration(size_t qty, char** args)
            : ConfigurationBase(qty, args)
            , ChannelConfiguration(qty, args, 310)
            , ConnectivityConfiguration(qty, args)
            , MulticastReplayConfiguration(qty, args)
        {
        }
 
    private:
        void showOptions(std::ostream& out) const override
        {
            ChannelConfiguration::showOptions(out);
            ConnectivityConfiguration::showOptions(out);
            MulticastReplayConfiguration::showOptions(out);
        }
    };
 
    void identify()
    {
        std::cout << "Multicast Replay Sample for CME MDP Premium Market Data Handler, v" << toStr(Version::current()) << ". "
                  << std::endl
                  << std::endl;
    }
 
    FileList resolveReplayLogs(const Configuration& configuration)
    {
        FileList logs;
        configuration.filesToReplay(logs);
 
        if (logs.empty())
            gatherLogFiles(logs, configuration.channel(), SampleDataLocation);
 
        return logs;
    }
}
 
int main(int argc, char** argv)
{
    identify();
 
    try
    {
        const Configuration configuration(argc, argv);
 
        if (configuration.show())
        {
            configuration.show(std::cout);
            return 0;
        }
 
        LogPlayerSettings logPlayerSettings(configuration.channel(), resolveReplayLogs(configuration));
        configuration.apply(logPlayerSettings);
 
        FeedSettings feedSettings;
        connectivityFromConfiguration(feedSettings, configuration.connectivityFile(), configuration.channel());
        configuration.apply(feedSettings);
 
        Test::MulticastSender sender{feedSettings, configuration.arbitrageDelay(), logPlayerSettings.delayPolicy()};
        replayLogFiles(logPlayerSettings, sender);
    }
    catch (const std::exception& e)
    {
        std::cerr << e.what() << std::endl;
        return 1;
    }
 
    return 0;
}