FIX Messaging Sample

The FIX Messaging sample demonstrates the tag-value view of decoded CME MDP market data. It is useful when an application wants to consume handler events through the FIX messaging API and inspect fields, repeating groups, timestamps, enumerations, and numeric values without binding directly to each SBE message type.

Arguments

Argument	Description	Default
--channel id	Defines which market data channel is processed.	310
--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
--logFile <filename>	Writes events to the specified log file.	FixMessaging.log
--logSeverity Important\|Regular\|Debug	Controls event logging verbosity.	Regular
--help	Displays command-line help.	Disabled

Running Examples

Process channel 310 and print decoded FIX market data:

./FixMessaging --channel 310

Use an explicit connectivity file and feed interface:

./FixMessaging --channel 310 --connectivity ../Settings/config.xml --ifA eth0

Write debug-level logs:

./FixMessaging --channel 310 --logSeverity Debug

Notes

The sample uses the feed engine in external processing mode and runs until interrupted with Ctrl+C.

Source Code

 
using namespace OnixS::CME::MDH;
 
// Command-line parameters affecting behavior of the sample.
 
class Configuration
    : public ChannelConfiguration
    , public ConnectivityConfiguration
    , public FeedConfiguration
    , public LoggerConfiguration
{
public:
    Configuration(size_t qty, char** args)
        : ConfigurationBase(qty, args)
        , ChannelConfiguration(qty, args, 310)
        , ConnectivityConfiguration(qty, args)
        , FeedConfiguration(qty, args)
        , LoggerConfiguration(qty, args, "FixMessaging.log")
    {
    }
 
private:
    void showOptions(std::ostream& out) const override
    {
        ChannelConfiguration::showOptions(out);
 
        ConnectivityConfiguration::showOptions(out);
        FeedConfiguration::showOptions(out);
 
        LoggerConfiguration::showOptions(out);
    }
};
 
// FIX message manipulations.
 
void trace(std::ostream& output, const Messaging::FIX::Message& message)
{
    using namespace OnixS::CME::MDH::Messaging::FIX;
 
    if (message.type() == "X")
    {
        const Field transact = message[Tags::TransactTime];
 
        const Field indicator = message[Tags::MatchEventIndicator];
 
        // Any time-stamp field can be successfully converted into
        // unsigned 64-bit integer representing ticks since the epoch.
 
        // Bit-set fields can be accessed either as raw integers or
        // as structured entities as they are exposed by SBE messages.
 
        output << "Transact time: " << Messaging::toStr(transact.cast<Messaging::Timestamp>()) << "." << std::endl
               << "Transact time (Ticks from Epoch): " << Messaging::toStr(transact.cast<Messaging::UInt64>()) << "." << std::endl
               << "Event Indicator: " << Messaging::toStr(indicator.cast<Messaging::MatchEventIndicator>()) << "." << std::endl
               << "Event Indicator (Raw Numeric Value): " << Messaging::toStr(indicator.cast<Messaging::UInt32>()) << "." << std::endl;
 
        const Group entries = message.group(Tags::Entries);
 
        if (entries)
        {
            for (Group::Size index = 0; index != entries.size(); ++index)
            {
                const GroupEntry entry = entries[index];
 
                std::cout << "[" << index << "] {Price=";
 
                if (const Field px = entry[Tags::EntryPx])
                {
                    std::cout << Messaging::toStr(px.cast<Messaging::Decimal>());
                }
 
                Messaging::Int32 size;
 
                // If side is not available then
                // the casting operation will fail.
 
                if (entry[Tags::EntrySize].toNumber(size))
                {
                    std::cout << ",Size=" << size;
                }
 
                const Field side = entry[Tags::EntryType];
 
                std::cout << ",Side=" << Messaging::toStr(side.cast<Messaging::EntryType>())
                          << ",Side(Raw)=" << Messaging::toStr(side.cast<Messaging::EntryType::Base>());
 
                const Field action = entry[Tags::UpdateAction];
 
                // In addition to accessing the field value in a raw FIX
                // format, it's possible to cast the value to the corresponding
                // enumeration type. In such case, the field value can be
                // compared with enumeration constants.
 
                std::cout << ",Action=" << Messaging::toStr(action.cast<Messaging::UpdateAction>())
                          << ",Action(Raw)=" << Messaging::toStr(action.cast<Messaging::UpdateAction::Base>());
 
                std::cout << "}" << std::endl;
            }
        }
    }
    else if (message.type() == "d")
    {
        Messaging::MaturityMonthYear mmy;
 
        if (message[Tags::MaturityMonthYear].toMaturityMonthYear(mmy))
        {
            const SecurityId securityId = message[Tags::SecurityId].cast<SecurityId>();
 
            std::cout << "Maturity Month-Year for $" << securityId << ": " << Messaging::toStr(mmy) << std::endl;
        }
    }
}
 
//
 
class FixMessageTracer : Messaging::FIX::MessageListener
{
public:
    void bind(Handler& handler)
    {
        handler.settings().listeners().marketData(this);
    }
 
private:
    void onMessage(Handler&, const Messaging::FIX::MessageArgs& args) override
    {
        trace(std::cout, args.message());
    }
};
 
// Traces issues from various sources (Handler, Feed Engine, Logger, etc.).
 
class IssueTracer
    : public HandlerListener
    , public FeedEngineThreadPoolListener
    , public FileLoggerListener
{
public:
    IssueTracer()
        : out_(std::cout)
    {
    }
 
    IssueTracer(std::ostream& output)
        : out_(output)
    {
    }
 
    ~IssueTracer() {}
 
    void bind(Handler& handler)
    {
        handler.settings().listeners().handler(this);
    }
 
private:
    // Handler issues.
 
    void onWarning(Handler& handler, const WarningArgs& warning) override
    {
        std::stringstream message;
 
        message << '[' << handler.settings().channel() << "] WARNING: " << warning << std::endl;
 
        out_ << message.str();
    }
 
    void onError(Handler& handler, const ErrorArgs& error) override
    {
        std::stringstream message;
 
        message << '[' << handler.settings().channel() << "] ERROR: " << error << std::endl;
 
        out_ << message.str();
    }
 
    // Feed Engine issues.
 
    void onFeedEngineThreadIssue(const FeedEngineThreadPool&, const Messaging::Char* issue) override
    {
        std::stringstream message;
 
        message << "ALERT!!! "
                   "Issue occurred in working thread. "
                << issue << std::endl;
 
        out_ << message.str();
    }
 
    // File Logger issues.
 
    void onFileLoggerIssue(const FileLogger&, const Messaging::Char* issue) override
    {
        std::stringstream message;
 
        message << "ALERT!!! "
                   "Issue occurred while logging data. "
                << issue << std::endl;
 
        out_ << message.str();
    }
 
    std::ostream& out_;
};
 
class Application
{
public:
    // Constructs handler for the given channel.
    Application(const Configuration& configuration)
        : configuration_(configuration)
    {
        try
        {
            logger_.reset(constructFileLogger(configuration_, &issueTracer_));
 
            feedEngine_.reset(constructFeedEngine(logger_.get()));
 
            handler_.reset(new Handler());
 
            HandlerSettings& settings = handler_->settings();
            setLicense(settings);
 
            settings.channel(configuration_.channel()).connectivityConfigurationFile(configuration_.connectivityFile());
 
            settings.feeds().engine(feedEngine_.get());
 
            apply(settings.feeds(), configuration);
 
            applyDefaults(settings.session());
 
            apply(settings.logging(), configuration, logger_.get());
 
            fixMessageTracer_.bind(*handler_);
            issueTracer_.bind(*handler_);
        }
        catch (...)
        {
            handler_.reset();
            feedEngine_.reset();
            logger_.reset();
 
            throw;
        }
    }
 
    // Utilizes all the handlers.
    ~Application() {}
 
    // Spawns market data processing and waits until user interrupts it.
    void run()
    {
        std::cout << "Starting processing market data.. " << std::endl << "Press Ctrl+C to stop exit. " << std::endl;
 
        handler_->start();
 
        while (!InterruptDetector::instance().detected())
        {
            try
            {
                feedEngine_->process();
            }
            catch (const std::exception& ex)
            {
                std::cerr << std::endl
                          << "WARNING! Feed engine raised an "
                             "issue while processing incoming data. "
                          << ex.what() << std::endl;
            }
        }
 
        std::cout << "Stopping processing market data.. " << std::endl;
 
        handler_->stop();
 
        std::cout << std::endl << std::endl << "Done. " << std::endl << std::endl;
    }
 
    static void identify()
    {
        std::cout << "FIX Messaging for CME MDP Premium Market Data Handler, v" << toStr(Version::current()) << ". "
                  << std::endl
                  << std::endl;
    }
 
private:
    const Configuration& configuration_;
 
    ScopedPtr<Logger> logger_;
 
    ScopedPtr<NetFeedEngine> feedEngine_;
 
    ScopedPtr<Handler> handler_;
 
    FixMessageTracer fixMessageTracer_;
 
    IssueTracer issueTracer_;
 
    // 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
        {
            checkNetworkSettings(configuration);
 
            Application app(configuration);
 
            app.run();
        }
 
        return 0;
    }
    catch (const std::exception& ex)
    {
        std::cerr << std::endl << "ERROR: " << ex.what() << std::endl;
 
        return 1;
    }
}