Advanced Sample

The Advanced sample shows how to run several CME MDP Premium Market Data Handler instances in one process while sharing common services such as logging, feed engines, and TCP recovery. Use it as a reference for multi-channel applications that need explicit recovery settings, selectable book types, feed configuration, and event tracing.

Arguments

Argument	Description	Default
--channels id1 [id2 [id3 ..]]	Defines the market data channels to process.	310
--verbose	Prints packet details, instrument events, and formatted contents of enabled order books.	Disabled
--skipStartupRecovery	Skips initial market recovery.	Disabled
--dontRecoverOnGap	Continues processing on gaps and errors instead of recovering market state from snapshots.	Disabled
--withTcpRecovery <user> <password>	Enables TCP Recovery Service for missing packet recovery with the supplied credentials.	Disabled
--directBooks	Maintains direct market-by-price books.	Disabled
--impliedBooks	Maintains implied market-by-price books.	Disabled
--consolidatedBooks	Maintains consolidated market-by-price books.	Disabled
--mboBooks	Maintains market-by-order books.	Disabled
--connectivity <file>	Uses the specified connectivity configuration file.	../Settings/config.xml
--ifA <interface>	Specifies the network interface for primary feeds.	Not overridden
--ifB <interface>	Specifies the network interface for secondary feeds.	Not overridden
--failoverForRecoveryFeeds	Switches to secondary recovery feeds when primary recovery feeds stop providing data.	Disabled
--arbitrateRealtimeFeeds	Enables real-time feed arbitrage.	Disabled
--feedEngine <kind>	Selects the feed engine. Supported values are Socket and Solarflare.	Socket
--ioWait <number>	Defines feed engine I/O wait time in milliseconds when supported by the selected engine.	10
--hardwareTimestamps	Requests hardware timestamps for incoming packets when supported.	Disabled
--logFile <filename>	Writes events to the specified log file.	LogFileForAllHandlers.txt
--logSeverity Important\|Regular\|Debug	Controls event logging verbosity.	Regular
--help	Displays command-line help.	Disabled

Running Examples

Process the default channel:

./Advanced

Process multiple channels and maintain direct books:

./Advanced --channels 310 311 --directBooks --connectivity ../Settings/config.xml

Print detailed market data and direct books:

./Advanced --channels 310 --directBooks --verbose

Enable TCP recovery and write debug logs:

./Advanced --channels 310 --withTcpRecovery CME CME --logSeverity Debug

Use the Solarflare feed engine for multiple channels:

./Advanced --channels 310 311 --feedEngine Solarflare --ifA eth0 --ifB eth1

Notes

By default, console output includes channel start/stop, recovery progress, real-time processing status, warnings, errors, gaps, and worker-thread lifecycle messages. Use --verbose to also print packet headers, match-event flags, instrument definitions, trades, volumes, statistics, quote requests, trading status, and formatted order books.

The --directBooks, --impliedBooks, --consolidatedBooks, and --mboBooks options enable book maintenance independently of --verbose. Book verification and its warnings remain enabled even when book contents are not printed. The --verbose option affects console tracing only; file logging remains controlled by --logFile and --logSeverity.

The sample runs until interrupted with Ctrl+C. Network interface options are usually required on hosts where the default multicast interface is not suitable.

The Solarflare feed engine requires Linux, compatible Solarflare network adapters, and a library built with EF_VI support (WITH_SFFE=ef_vi). Replace eth0 and eth1 in the example with the appropriate Solarflare interfaces. Libraries built with WITH_SFFE=stub report an error when Solarflare is selected.

When using --feedEngine Solarflare with --hardwareTimestamps, you must synchronize the network adapter clock with the system clock before running the sample. Use the Solarflare PTP daemon and wait until it reports that the clocks are synchronized.

Source Code

 
using namespace OnixS::CME::MDH;
 
// Aliases for code clarity.
 
typedef std::vector<ChannelId> ChannelIds;
 
// Application command-line interface.
 
class Configuration
    : public ConnectivityConfiguration
    , public FeedConfiguration
    , public FeedEngineConfiguration
    , public SessionConfiguration
    , public BookManagementConfiguration
    , public LoggerConfiguration
{
public:
    Configuration(size_t qty, char** args)
        : ConfigurationBase(qty, args)
        , ConnectivityConfiguration(qty, args)
        , FeedConfiguration(qty, args)
        , FeedEngineConfiguration(qty, args)
        , SessionConfiguration(qty, args)
        , BookManagementConfiguration(qty, args)
        , LoggerConfiguration(qty, args, "LogFileForAllHandlers.txt")
    {
    }
 
    bool channels(ChannelIds& ids) const
    {
        return args<ChannelId>(options(), "channels", std::back_inserter(ids), 1);
    }
 
    bool verbose() const
    {
        return exist(options(), "verbose");
    }
 
private:
    void showOptions(std::ostream& out) const override
    {
        out << "  --channels id1 [id2 [id3 ..]]" << std::endl
            << "\tDefines channels whose market data is to be processed. " << std::endl
            << "\tChannel 310 is used if nothing specified. " << std::endl
            << std::endl;
 
        out << "  --verbose" << std::endl
            << "\tPrints packet details, instrument events, and enabled order books. " << std::endl
            << "\tBy default, only operational status and diagnostics are printed. " << std::endl
            << std::endl;
 
        SessionConfiguration::showOptions(out);
        BookManagementConfiguration::showOptions(out);
        ConnectivityConfiguration::showOptions(out);
        FeedConfiguration::showOptions(out);
        FeedEngineConfiguration::showOptions(out);
        LoggerConfiguration::showOptions(out);
    }
};
 
class Application
{
public:
    Application(const Configuration& configuration)
        : configuration_(configuration)
        , handlers_(nullptr)
        , handlerQty_(0)
        , securityTracer_(std::cout, configuration.verbose())
    {
        ChannelIds channels;
 
        if (!configuration_.channels(channels))
            channels.push_back(310);
 
        try
        {
            logger_.reset(constructFileLogger(configuration_, &loggerTracer_));
 
            const auto workingThreads = ((10 > channels.size()) ? 1 : ((20 > channels.size()) ? 2 : 4));
 
            feedEngine_.reset(constructFeedEngine(configuration_, logger_.get()));
 
            workingThreads_.reset(constructFeedEngineThreads(*feedEngine_, workingThreads, &threadTracer_));
 
            tcpRecovery_.reset(constructTcpRecovery(configuration_));
 
            constructHandlers(channels);
        }
        catch (...)
        {
            destructEverything();
 
            throw;
        }
    }
 
    ~Application()
    {
        destructEverything();
    }
 
    // Let’s start the market data processing and wait until the user interrupts it.
    void run()
    {
        const InterruptDetector& interruptDetector = InterruptDetector::instance();
 
        std::cout << "Starting processing market data.. " << std::endl
                  << "Press Ctrl+C to stop and exit. " << std::endl;
 
        Handler* const endOfHandlers = handlers_ + handlerQty_;
 
        std::for_each(handlers_, endOfHandlers, [](Handler& handler){handler.start();});
 
        std::cout << std::endl << std::endl;
 
        while (!interruptDetector.detected())
        {
            CurrentThread::sleep(1);
        }
 
        std::cout << "Stopping processing market data.. " << std::endl;
 
        std::for_each(handlers_, endOfHandlers, [](Handler& handler){handler.stop();});
 
        std::cout << std::endl << "Done. " << std::endl << std::endl;
    }
 
    // A bit of information on the sample.
    static void identify()
    {
        std::cout << "Advanced Sample for CME MDP Premium Market Data Handler, v" << toStr(Version::current()) << ". "
                  << std::endl
                  << std::endl;
    }
 
private:
 
    void destructEverything()
    {
        delete[] handlers_;
 
        tcpRecovery_.reset();
 
        workingThreads_.reset();
        feedEngine_.reset();
 
        logger_.reset();
    }
 
    // Constructs handlers for the given channels.
    void constructHandlers(const ChannelIds& channels)
    {
        handlers_ = new Handler[channels.size()]();
 
        handlerQty_ = 0;
 
        for (ChannelIds::const_iterator channel = channels.begin(); channel != channels.end(); ++channel, ++handlerQty_)
        {
            Handler& handler = handlers_[handlerQty_];
 
            HandlerSettings& settings = handler.settings();
            setLicense(settings);
 
            settings.channel(*channel).connectivityConfigurationFile(configuration_.connectivityFile());
 
            // Engine & Connectivity.
 
            settings.feeds().engine(feedEngine_.get());
 
            // Additional steps on feeds.
 
            apply(settings.feeds(), configuration_);
 
            // Processing flow.
 
            apply(settings.session(), configuration_, tcpRecovery_.get());
 
            // Books.
 
            apply(settings.bookManagement(), configuration_);
 
            // Logging.
 
            apply(settings.logging(), configuration_, logger_.get());
 
            // Events.
 
            securityTracer_.bind(handler);
            if (configuration_.verbose())
                marketDataTracer_.bind(handler);
            serviceTracer_.bind(handler);
        }
    }
 
    const Configuration& configuration_;
 
    ScopedPtr<Logger> logger_;
 
    ScopedPtr<NetFeedEngine> feedEngine_;
 
    ScopedPtr<FeedEngineThreadPool> workingThreads_;
 
    ScopedPtr<TcpRecoveryService> tcpRecovery_;
 
    Handler* handlers_;
    size_t handlerQty_;
 
    SecurityEventTracer securityTracer_;
 
    MarketDataEventTracer marketDataTracer_;
 
    ServiceEventTracer serviceTracer_;
 
    FeedEngineThreadPoolEventTracer threadTracer_;
 
    FileLoggerEventTracer loggerTracer_;
 
    // 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();
 
        Configuration configuration(qty, args);
 
        if (configuration.show())
        {
            configuration.show(std::cout);
        }
        else
        {
            checkNetworkSettings(configuration);
 
            Application(configuration).run();
        }
 
        return 0;
    }
    catch (const std::exception& ex)
    {
        std::cerr << std::endl << "ERROR: " << ex.what() << std::endl;
 
        return 1;
    }
}