Event listeners

The handler uses a concept of events and event listeners to notify the client code about various aspects of market data processing and deliver data to the user (e.g., receiving a security definition or a direct book update).

The SDK groups events and provides classes with virtual members (event listeners) for each group. Client code may override any or all listener members to implement the processing of a particular event.

Associating an event listener with an instance of the OnixS::CME::MDH::Handler class is called subscribing to events.

Note

Associating the listener with the OnixS::CME::MDH::Handler instance must be performed while the handler is in the stopped state. Once the market data processing is started, assigning event listeners is not allowed and leads to throwing an exception.

Callback contract

Ownership and lifetime

OnixS::CME::MDH::HandlerListeners stores pointers to application-owned listeners; it does not copy or delete them. Keep each listener alive while registered and until all callbacks using it have returned. A simple arrangement is to construct listeners before the Handler, stop the Handler from the control thread, then destroy the Handler before its listeners. For explicit engines, also follow Explicit engine lifetime.

Treat callback arguments as borrowed data for the duration of the call. Books and securities belong to the handler; packets, messages, entries, iterators and text views can refer to reused storage. Copy required values into owned storage before handing them to another thread. Copying a view does not copy its underlying bytes. Books are not thread-safe; a worker must not read a handler-owned book while processing continues. See Copying order book content and Network Packets and Messages.

Execution and concurrency

Callbacks execute inline on the thread performing the operation that raises them:

• start() invokes onStarted, startup error callbacks and initial connection notifications on its calling thread. stop() invokes shutdown callbacks on its calling thread.
• Feed-driven callbacks run on a thread calling OnixS::CME::MDH::NetFeedEngine::process: an application thread, an explicit processing pool worker, or an internal default-engine worker.
• The built-in Socket and Solarflare engines serialize data processing and callbacks for a single Handler. Successive calls can use different workers; there is no fixed callback thread. Multiple Handlers can invoke the same listener concurrently, even when sharing an engine.
• Socket reception for that channel waits for its processing and callbacks. Solarflare's ef_vi path can poll and queue received buffers on another worker while a callback runs; delivery to the Handler remains serialized. Its socket fallback also reads under the Handler processing lock.

These built-in engine rules do not establish threading guarantees for a custom engine. Replay callbacks follow the replay driver thread; see Replaying market data. Synchronize application state shared with control threads, workers or other Handlers. Inline calls can also nest: for example, an exception in a callback can immediately invoke onWarning on the same thread.

Exceptions

Catch application failures inside callbacks and return normally. The handler catches exceptions escaping the handler event listeners described below and continues its own processing without rolling back application work or requesting recovery. It attempts to report an OnixS::CME::MDH::Issue::UserException warning through OnixS::CME::MDH::HandlerListener::onWarning before continuing processing. Exceptions raised while reporting the callback failure are suppressed.

Exceptions escaping onWarning and onError are suppressed without another listener notification. Throwing from a callback is therefore not a way to fail start() or stop the handler.

Feed engine thread callbacks have a separate exception reporting policy; see OnixS::CME::MDH::FeedEngineThreadPoolListener.

Operations from callbacks

Inspect callback arguments, read the session's unchanged settings, and copy or enqueue owned data. OnixS::CME::MDH::Handler::selectInstruments is also supported inside OnixS::CME::MDH::HandlerListener::onEndOfInstrumentRecovery when its status is OnixS::CME::MDH::RecoveryCompletionStatus::Succeeded; see Selecting securities of interest.

Defer lifecycle calls (start(), stop() and destruction), listener replacement, settings changes and instrument-cache cleanup to the control thread after the callback and any enclosing lifecycle call have returned. Reentering lifecycle operations can invalidate objects still in use by the invoking code. See Observing startup for lifecycle notification ordering. Do not recursively call the engine's process() or stop/join its worker pool from a callback. Do not wait for a worker that needs the current callback to return.

Handing data to a worker

This example uses Boost.Lockfree's single-producer/single-consumer queue with one Handler and one application worker. Boost.Lockfree is an optional example dependency; you can replace it with an equivalent bounded, thread-safe queue.

It copies only the security ID and best bid into preallocated, bounded storage. If the queue is full, the callback drops the new sample and increments a counter. The worker calls take() and processes accepted samples in FIFO order. This loss policy suits optional telemetry; consumers requiring every update need their own overflow response and must not silently apply this policy to a sequence of book deltas.

#include <atomic>
#include <boost/lockfree/spsc_queue.hpp>
 
constexpr std::size_t BidQueueCapacity = 1024;
 
struct OwnedBid
{
    SecurityId securityId{};
    ImpliedPriceLevel bestBid; // Owns presence, price and aggregate quantity.
};
 
class BidHandoff : public SecurityListener
{
    boost::lockfree::spsc_queue<OwnedBid, boost::lockfree::capacity<BidQueueCapacity>> queue_;
    std::atomic<unsigned> dropped_{0};
 
public:
    void onBookUpdate(Handler&, const Security& security, const DirectBook& book) override
    {
        OwnedBid sample;
        sample.securityId = security.id();
 
        // An absent best bid is a value too; never retain a reference to the book.
        if (!book.bids().empty())
            sample.bestBid = book.bids().front();
 
        // Optional telemetry: drop the newest sample on overflow and count the loss.
        if (!queue_.push(sample))
            ++dropped_;
    }
 
    bool take(OwnedBid& sample)
    {
        return queue_.pop(sample);
    }
 
    unsigned dropped() const
    {
        return dropped_.load();
    }
};

Keep the queue and listener alive until the Handler is stopped and the application worker is joined. The example illustrates ownership and handoff only; book readiness still follows Interpreting a book update. For callback costs and ordering, see Keeping callbacks responsive.

Events groups

The sections below describe event groups and show the correspondence among events, listeners, and the handler settings methods used to subscribe to them:

OnixS::CME::MDH::HandlerListener

Method: OnixS::CME::MDH::HandlerListeners::handler

Usage example:

handler.settings().listeners().handler(&listener);

The handler invokes the event listener members during market data processing. Events reflect processing session stages like entering or leaving the recovery cycle, recovery completion status (OnixS::CME::MDH::RecoveryCompletionArgs), and resuming real-time data processing.

OnixS::CME::MDH::MarketDataListener

Method: OnixS::CME::MDH::HandlerListeners::marketData

Usage example:

handler.settings().listeners().marketData(&listener);

The handler invokes the event listener members when it begins or ends processing various market data entities (e.g., a packet or a message).

Events are triggered only for market data accepted by the handler and according to CME's rules. The handler does not invoke members of the listener for any duplicated or out-of-order packets.

OnixS::CME::MDH::SecurityListener

Method: OnixS::CME::MDH::HandlerListeners::security

Usage example:

handler.settings().listeners().security(&listener);

Members of the given listener are invoked to reflect security-related events like the receiving of trade, security status, and order book updates.

By default, members of this listener are invoked for each security available on the channel. However, it is possible to restrict callback invocation for a set of securities using the OnixS::CME::MDH::Handler::selectInstruments member. See Selecting securities of interest for more information.

OnixS::CME::MDH::ChannelFeedListener

Method: OnixS::CME::MDH::HandlerListeners::feeds

Usage example:

handler.settings().listeners().feeds(&listener);

The OnixS::CME::MDH::ChannelFeedListener members are invoked each time a feed-related event occurs.

Example

The sample below demonstrates listening to Handler's events:

struct
    : public HandlerListener
    , public MarketDataListener
    , public SecurityListener
    , public ChannelFeedListener
{
    void onStarted(Handler&) override
    {
        std::cout << "Handler started." << std::endl;
    }
 
    void onStopped(Handler&) override
    {
        std::cout << "Handler stopped." << std::endl;
    }
 
    void onPacket(Handler&, const PacketArgs& args) override
    {
        std::cout << "A packet with sequence number " << args.seqNumber() << " is received."  << std::endl;
    }
 
    void onBookUpdate(Handler&, const Security&, const ConsolidatedBook& book) override
    {
        std::cout << "A book is updated: " << brief(book) << std::endl;
    }
 
    void onFeedConnected(Handler&, const NetFeed& feed) override
    {
        std::cout << "Feed " << feed.id() << " is connected." << std::endl;
    }
 
    void onFeedDisconnected(Handler&, const NetFeed& feed) override
    {
        std::cout << "Feed " << feed.id() << " is disconnected." << std::endl;
    }
}
listener;
 
// Declare the Handler after its listener so it is destroyed first.
Handler handler;
handler.settings().listeners()
    .handler(&listener)
    .marketData(&listener)
    .security(&listener)
    .feeds(&listener);