Multicast Replay Sample

The Multicast Replay sample turns recorded handler log data back into multicast traffic with MulticastSender. It can replay explicit log files or automatically select matching sample data for a channel, which makes it useful for lab testing receivers, validating connectivity files, and simulating primary or arbitrated feed delivery.

Arguments

Argument	Description	Default
--channel id	Defines which market data channel is replayed.	310
--connectivity <file>	Uses the specified connectivity configuration file. Outgoing destinations outside 239.0.0.0/8 require ONIXS_CMEMDH_ALLOW_ANY_MULTICAST_DESTINATION to be enabled.	Generated testing connectivity for the selected channel
--log <file1 [file2 [..]]>	Replays the specified log files. If omitted, sample data is replayed.	Matching files from ../Data folder
--send-delay <natural/microseconds>	Defines delay before sending another packet to the same source. Use natural to preserve delays from the log file.	natural
--source-switch-delay <microseconds>	Defines 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 delay before sending the same packet to the secondary feed when arbitrage simulation is enabled.	0
--iface-A <interface>	Specifies network interface to use for feed A.	lo on Linux; 127.0.0.1 on Windows
--iface-B <interface>	Specifies network interface to use for feed B.	lo on Linux; 127.0.0.1 on Windows
--ttl <0..32>	Defines the IPv4 multicast time-to-live. Zero confines outgoing datagrams to the local host. A positive value permits transmission beyond the local host and requires the ONIXS_CMEMDH_ALLOW_MULTICAST_EGRESS environment variable to be enabled.	0
--packets-count <count>	Stops replay after the specified number of packets sent to incremental feed targets. Historical packets remapped to incremental are included.	Unlimited
--help	Displays command-line help.	Disabled

Running Examples

From a sample layout where Data and Settings are sibling directories:

ONIXS_CMEMDH_ALLOW_ANY_MULTICAST_DESTINATION=1 ./MulticastReplay --channel 310 \
  --connectivity ../Settings/config.xml --log ../Data/CmeMdpPremiumHandler-Channel_310.log

The example uses files included with the samples. Omitting --log makes the sample gather matching channel 310 log files from ../Data.

Replay sample data for channel 310 using generated testing connectivity and automatically selected log files:

./MulticastReplay --channel 310

Simulate feed arbitrage:

./MulticastReplay --channel 310 --simulate-arbitrage --arbitrage-delay 100

Replay to receivers on the local network by selecting the sending interface and setting an explicit TTL. Replace eth0 with the required interface name on Linux or interface IPv4 address on Windows:

ONIXS_CMEMDH_ALLOW_MULTICAST_EGRESS=1 ./MulticastReplay --channel 310 --iface-A eth0 --ttl 1

Notes

When --connectivity is omitted, the sample generates multicast endpoints for the selected channel. Use --update-sending-time true when replayed packets must carry the current send time. Use --packets-count to limit replay during connectivity checks. Outgoing destinations outside 239.0.0.0/8 require ONIXS_CMEMDH_ALLOW_ANY_MULTICAST_DESTINATION; positive TTL values independently require ONIXS_CMEMDH_ALLOW_MULTICAST_EGRESS, so both variables are required when both conditions apply. Each variable is interpreted as a boolean: true, on, yes, and 1 enable it, ignoring case and surrounding whitespace. Missing and other values leave the corresponding permission disabled. The default feed interfaces are loopback-only; specify --iface-A and, when feed B is used, --iface-B to transmit through another interface.

Source Code

 
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);
 
            settings.feedANetworkInterfaces(networkInterface("iface-A"));
            settings.feedBNetworkInterfaces(networkInterface("iface-B"));
 
            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);
        }
 
        Test::MulticastTargetOptions multicastTargetOptions() const
        {
            const Messaging::Char* const option = "ttl";
            const auto args = argsFor(option, 1, 1);
            const auto ttl = args ? convertUInt64(option, (*args)[0]) : 0;
 
            Test::MulticastTargetOptions result;
            result.ttl(ttl);
 
            return result;
        }
 
    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. Default is "
                << Test::loopbackNetworkInterface() << ". " << std::endl
                << std::endl
                << "  --iface-B <interface>" << std::endl
                << "\tSpecifies network interface to use for feed B. Default is "
                << Test::loopbackNetworkInterface() << ". " << std::endl
                << std::endl
                << "  --ttl <0..32>" << std::endl
                << "\tDefines the IPv4 multicast time-to-live. Default is 0. " << std::endl
                << "\tA positive value requires " << Test::MulticastEgressEnvironmentVariable << ' '
                << "to be enabled. " << 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:
        std::string networkInterface(const Messaging::Char* option) const
        {
            const auto args = argsFor(option, 1, 1);
 
            return args ? (*args)[0] : Test::loopbackNetworkInterface();
        }
 
        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)
        {
        }
 
        bool connectivityFileSpecified() const
        {
            return exist(options(), "connectivity");
        }
 
    private:
        void showOptions(std::ostream& out) const override
        {
            ChannelConfiguration::showOptions(out);
            out << "  --connectivity <file>" << std::endl
                << "\tTells to use the given connectivity configuration file. " << std::endl
                << "\tIf omitted, generated testing connectivity is used. " << std::endl
                << "\tOutgoing destinations outside 239.0.0.0/8 require "
                << Test::AllowAnyMulticastDestinationEnvironmentVariable << " to be enabled. " << std::endl
                << std::endl;
            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;
 
        if (configuration.connectivityFileSpecified())
            connectivityFromConfiguration(feedSettings, configuration.connectivityFile(), configuration.channel());
        else
            connectivityFromConfigurationString(
                feedSettings, Test::generateConnectivityConfiguration(configuration.channel()), configuration.channel());
 
        configuration.apply(feedSettings);
 
        Test::MulticastSender sender{feedSettings, configuration.arbitrageDelay(), logPlayerSettings.delayPolicy(), configuration.multicastTargetOptions()};
        replayLogFiles(logPlayerSettings, sender);
    }
    catch (const std::exception& e)
    {
        std::cerr << e.what() << std::endl;
        return 1;
    }
 
    return 0;
}