Testing Scenarios with Pipelines

This page presents feed and pipeline examples based on the repository's complex order-book tests. The first example shows a complete three-feed orchestration. The remaining fragments omit handler construction, listeners, waits, and assertions so that the ordering and protocol state carried by the test messages remain visible.

In an integration test, wait for the handler to join each feed and for the stated recovery transition before sending the next phase. The multicast sequence number passed to send is independent of the per-security rptSeq encoded in market-data entries.

Run an end-to-end three-feed test

The flagship workflow queues an incremental during initial recovery, supplies the instrument definition, completes market recovery from a snapshot, and then sends book, trade-summary, and statistics messages in one incremental packet.

Harness represents the application's test fixture rather than a testing API type. It must provide startHandler, stopHandler, waitForFeed(NetFeedType::Enum), waitForInstrumentRecovery, waitForMarketRecovery, and waitForListenerNotification operations. The supplied FeedSettings must configure Instrument, Incremental, and Snapshot feeds that match the handler.

template <typename Harness>
void typicalThreeFeedTest(Harness& harness, const FeedSettings& feedSettings)
{
    using namespace ONIXS_CMEMDH_NAMESPACE;
    using namespace ONIXS_CMEMDH_MESSAGING_NAMESPACE;
    using namespace ONIXS_CME_MDH_TESTING_NAMESPACE;
 
    const auto securityId = SecurityId{300500};
    const auto snapshotBoundary = SequenceNumber{100};
 
    MulticastFeed incrementalFeed(NetFeedType::Incremental, feedSettings);
    MulticastFeed instrumentFeed(NetFeedType::Instrument, feedSettings);
    MulticastFeed snapshotFeed(NetFeedType::Snapshot, feedSettings);
 
    harness.startHandler();
    harness.waitForFeed(NetFeedType::Incremental);
 
    incrementalFeed.send(
            snapshotBoundary,
            initialize<IncrementalRefreshBook>(EventFlags::EndOfRealQuotes | EventFlags::EndOfEvent)
                | ONIXS_CMEMDH_ENTRY(New, securityId, EntryTypeBook::Bid, 11, 2, Decimal9(99), 10, 2)
        ).get();
 
    harness.waitForFeed(NetFeedType::Instrument);
 
    instrumentFeed.send(
            1,
            initialize<InstrumentDefinitionFuture, SecurityUpdateAction::Add>(1, securityId)
        ).get();
 
    harness.waitForInstrumentRecovery();
    harness.waitForFeed(NetFeedType::Snapshot);
 
    snapshotFeed.send(
            1,
            initialize<SnapshotFullRefresh>(1, securityId, snapshotBoundary, 11)
                | ONIXS_CMEMDH_ENTRY(New, EntryType::Bid, 1, Decimal9(100), 10, 1)
                | ONIXS_CMEMDH_ENTRY(New, EntryType::Bid, 2, Decimal9(99), 10, 2)
        ).get();
 
    harness.waitForMarketRecovery();
 
    // All three incremental messages are encoded into the same packet in argument order.
    incrementalFeed.send(
            snapshotBoundary + 1,
            initialize<IncrementalRefreshBook>(EventFlags::EndOfRealQuotes)
                | ONIXS_CMEMDH_ENTRY(Delete, securityId, EntryTypeBook::Bid, 12, 2)
                | ONIXS_CMEMDH_ENTRY(New, securityId, EntryTypeBook::Offer, 13, 1, Decimal9(101), 12, 1),
            initialize<IncrementalRefreshTradeSummary>(EventFlags::EndOfTrades)
                | ONIXS_CMEMDH_ENTRY(New, securityId, 14, Decimal9(100), 5, 2)
                | ONIXS_CMEMDH_TRADE_ENTRY(9001, 3)
                | ONIXS_CMEMDH_TRADE_ENTRY(9002, 2),
            initialize<IncrementalRefreshSessionStatistics>(EventFlags::EndOfStatistics | EventFlags::EndOfEvent)
                | ONIXS_CMEMDH_ENTRY(New, securityId, EntryTypeStatistics::HighestBid, 15, Decimal9(100))
                | ONIXS_CMEMDH_ENTRY(New, securityId, EntryTypeStatistics::LowestOffer, 16, Decimal9(101))
        ).get();
 
    harness.waitForListenerNotification();
    harness.stopHandler();
}

Each readiness and recovery wait is significant. Sending the next phase before the handler reaches the corresponding state can make an integration test timing-dependent.

Build a book during pre-open

In pre-opening mode, a test can publish the security status and definition before adding book levels. Each incremental pipeline sets both the quote boundary and event boundary.

using namespace ONIXS_CMEMDH_NAMESPACE;
using namespace ONIXS_CMEMDH_MESSAGING_NAMESPACE;
using namespace ONIXS_CME_MDH_TESTING_NAMESPACE;
 
const auto securityId = SecurityId{100500};
 
incrementalFeed.send(
        1, initialize<SecurityStatus>(securityId, SecurityTradingStatus::PreOpen)
    ).get();
 
incrementalFeed.send(
        2, initialize<InstrumentDefinitionFuture, SecurityUpdateAction::Add>(1, securityId)
    ).get();
 
incrementalFeed.send(
        3,
        initialize<IncrementalRefreshBook>(EventFlags::EndOfRealQuotes | EventFlags::EndOfEvent)
            | ONIXS_CMEMDH_ENTRY(New, securityId, EntryTypeBook::Bid, 1, 1, Decimal9(100), 10, 2)
            | ONIXS_CMEMDH_ENTRY(New, securityId, EntryTypeBook::Bid, 2, 2, Decimal9(99), 10, 2)
    ).get();

This sequence creates two bid levels for one security. Change the entry type to Offer and invert the price ordering so that prices ascend with the level to build the other side.

Complete late-join market recovery

A late-join test first sends an incremental that the handler queues, supplies the instrument definition, then sends a snapshot whose lastMsgSeqNumProcessed identifies the last incremental incorporated into that snapshot. The next incremental continues with both the packet sequence and the per-security report sequence advanced.

using namespace ONIXS_CMEMDH_NAMESPACE;
using namespace ONIXS_CMEMDH_MESSAGING_NAMESPACE;
using namespace ONIXS_CME_MDH_TESTING_NAMESPACE;
 
const auto securityId = SecurityId{300500};
const auto snapshotBoundary = SequenceNumber{100};
 
// The handler queues this packet while initial recovery is in progress.
incrementalFeed.send(
        snapshotBoundary,
        initialize<IncrementalRefreshBook>(EventFlags::EndOfRealQuotes | EventFlags::EndOfEvent)
            | ONIXS_CMEMDH_ENTRY(New, securityId, EntryTypeBook::Bid, 11, 2, Decimal9(99), 10, 2)
    ).get();
 
instrumentFeed.send(1, initialize<InstrumentDefinitionFuture, SecurityUpdateAction::Add>(1, securityId)).get();
 
snapshotFeed.send(
        1,
        initialize<SnapshotFullRefresh>(1, securityId, snapshotBoundary, 11)
            | ONIXS_CMEMDH_ENTRY(New, EntryType::Bid, 1, Decimal9(100), 10, 1)
            | ONIXS_CMEMDH_ENTRY(New, EntryType::Bid, 2, Decimal9(99), 10, 2)
    ).get();
 
incrementalFeed.send(
        snapshotBoundary + 1,
        initialize<IncrementalRefreshBook>(EventFlags::EndOfRealQuotes | EventFlags::EndOfEvent)
            | ONIXS_CMEMDH_ENTRY(Delete, securityId, EntryTypeBook::Bid, 12, 2)
    ).get();

The test harness should wait for instrument recovery to stop before sending snapshots, and wait for market recovery to stop before asserting the resulting book.

Once the snapshot is applied, the queued incremental is replayed and discarded as outdated; it does not contribute to the resulting book.

Send a complete multi-security snapshot cycle

The instrument definition's totNumReports is the number of definitions in the instrument-recovery loop. The snapshot's totNumReports is the number of reports in the current snapshot cycle; these are distinct quantities and need not be equal. Each security may have its own rptSeq, but all reports in one coherent cycle normally refer to the same lastMsgSeqNumProcessed.

using namespace ONIXS_CMEMDH_NAMESPACE;
using namespace ONIXS_CMEMDH_MESSAGING_NAMESPACE;
using namespace ONIXS_CME_MDH_TESTING_NAMESPACE;
 
const auto firstSecurityId = SecurityId{500501};
const auto secondSecurityId = SecurityId{500502};
const auto definitionsInRecoveryLoop = UInt32{2};
const auto reportsInSnapshotCycle = UInt32{2};
const auto snapshotBoundary = SequenceNumber{100};
 
instrumentFeed.send(
        1,
        initialize<InstrumentDefinitionFuture, SecurityUpdateAction::Add>(definitionsInRecoveryLoop, firstSecurityId)
    ).get();
 
instrumentFeed.send(
        2,
        initialize<InstrumentDefinitionFuture, SecurityUpdateAction::Add>(definitionsInRecoveryLoop, secondSecurityId)
    ).get();
 
snapshotFeed.send(
        1,
        initialize<SnapshotFullRefresh>(reportsInSnapshotCycle, firstSecurityId, snapshotBoundary, 11)
            | ONIXS_CMEMDH_ENTRY(New, EntryType::Bid, 1, Decimal9(100), 10, 1)
            | ONIXS_CMEMDH_ENTRY(New, EntryType::Bid, 2, Decimal9(99), 12, 2)
    ).get();
 
snapshotFeed.send(
        2,
        initialize<SnapshotFullRefresh>(reportsInSnapshotCycle, secondSecurityId, snapshotBoundary, 21)
            | ONIXS_CMEMDH_ENTRY(New, EntryType::Offer, 1, Decimal9(101), 11, 1)
            | ONIXS_CMEMDH_ENTRY(New, EntryType::Offer, 2, Decimal9(102), 13, 2)
    ).get();

After the cycle completes, incrementals must continue after lastMsgSeqNumProcessed and use the next report sequence for each affected security.

Accumulate reports during snapshot recovery

During one MBP late-join recovery attempt, the handler retains each recovered security until it has collected the expected number of reports. Publish the instrument definitions and wait for instrument recovery to complete first, as in the previous scenario; during normal initial recovery the handler joins the snapshot feed only after instrument recovery succeeds. Repeating an already processed snapshot packet sequence number does not clear retained reports. In this example, the duplicate report for the first security is ignored and the report for the second security completes recovery using the first report already collected.

using namespace ONIXS_CMEMDH_NAMESPACE;
using namespace ONIXS_CMEMDH_MESSAGING_NAMESPACE;
using namespace ONIXS_CME_MDH_TESTING_NAMESPACE;
 
const auto firstSecurityId = SecurityId{800501};
const auto secondSecurityId = SecurityId{800502};
const auto reportsInCycle = UInt32{2};
const auto snapshotBoundary = SequenceNumber{100};
 
// Recovery retains this report while waiting for the second security.
snapshotFeed.send(
        1,
        initialize<SnapshotFullRefresh>(reportsInCycle, firstSecurityId, snapshotBoundary, 11)
            | ONIXS_CMEMDH_ENTRY(New, EntryType::Bid, 1, Decimal9(100), 10, 1)
    ).get();
 
// Reusing packet sequence 1 does not reset the reports already collected.
// This duplicate is discarded without its contents being read.
snapshotFeed.send(
        1,
        initialize<SnapshotFullRefresh>(reportsInCycle, firstSecurityId, snapshotBoundary, 99)
            | ONIXS_CMEMDH_ENTRY(New, EntryType::Offer, 1, Decimal9(200), 20, 2)
    ).get();
 
snapshotFeed.send(
        2,
        initialize<SnapshotFullRefresh>(reportsInCycle, secondSecurityId, snapshotBoundary, 22)
            | ONIXS_CMEMDH_ENTRY(New, EntryType::Bid, 1, Decimal9(101), 12, 1)
            | ONIXS_CMEMDH_ENTRY(New, EntryType::Bid, 2, Decimal9(100), 14, 2)
    ).get();

Duplicate reports for an already recovered security are discarded without their contents being read. To test a genuinely new snapshot cycle, start a new recovery attempt so that the handler clears its recovery state.

Group multiple securities into an event

A single incremental message may update several securities. EventFlags::EndOfRealQuotes closes the real-quote portion; EventFlags::EndOfEvent closes the entire event. Listener notification behavior then depends on the handler's configured book-update strategy.

using namespace ONIXS_CMEMDH_NAMESPACE;
using namespace ONIXS_CMEMDH_MESSAGING_NAMESPACE;
using namespace ONIXS_CME_MDH_TESTING_NAMESPACE;
 
const auto firstSecurityId = SecurityId{100500};
const auto secondSecurityId = SecurityId{100501};
 
incrementalFeed.send(
        10,
        initialize<IncrementalRefreshBook>(EventFlags::EndOfRealQuotes | EventFlags::EndOfEvent)
            | ONIXS_CMEMDH_ENTRY(New, firstSecurityId, EntryTypeBook::Offer, 1, 1, Decimal9(101), 10, 1)
            | ONIXS_CMEMDH_ENTRY(New, secondSecurityId, EntryTypeBook::Offer, 1, 1, Decimal9(102), 12, 1)
            | ONIXS_CMEMDH_ENTRY(New, firstSecurityId, EntryTypeBook::Bid, 2, 1, Decimal9(100), 11, 1)
            | ONIXS_CMEMDH_ENTRY(New, secondSecurityId, EntryTypeBook::Bid, 2, 1, Decimal9(99), 13, 1)
    ).get();

Use monotonically increasing rptSeq values independently for each security even when their entries share a packet.

Separate direct and implied updates

Direct and implied quote boundaries can occur in different packets. The initial message below seeds the direct level required by strict MBP processing. The next message ends real quotes but leaves the event open; the final message ends implied quotes and the event.

using namespace ONIXS_CMEMDH_NAMESPACE;
using namespace ONIXS_CMEMDH_MESSAGING_NAMESPACE;
using namespace ONIXS_CME_MDH_TESTING_NAMESPACE;
 
const auto directSecurityId = SecurityId{100500};
const auto impliedSecurityId = SecurityId{100501};
 
// A strict MBP change requires the direct price level to already exist.
incrementalFeed.send(
        19,
        initialize<IncrementalRefreshBook>(EventFlags::EndOfRealQuotes | EventFlags::EndOfEvent)
            | ONIXS_CMEMDH_ENTRY(New, directSecurityId, EntryTypeBook::Bid, 1, 1, Decimal9(100), 10, 1)
    ).get();
 
incrementalFeed.send(
        20,
        initialize<IncrementalRefreshBook>(EventFlags::EndOfRealQuotes)
            | ONIXS_CMEMDH_ENTRY(Change, directSecurityId, EntryTypeBook::Bid, 2, 1, Decimal9(100), 11, 2)
    ).get();
 
incrementalFeed.send(
        21,
        initialize<IncrementalRefreshBook>(EventFlags::EndOfImpliedQuotes | EventFlags::EndOfEvent)
            | ONIXS_CMEMDH_ENTRY(New, impliedSecurityId, EntryTypeBook::ImpliedBid, 1, 1, Decimal9(99), 1, 0)
    ).get();

This sequence is useful when verifying consolidated-book notifications and end-of-event behavior.

Link order-ID data to an MBP entry

An order-ID entry uses a one-based referenceId to identify the related market-data entry in the same incremental message. The order-ID stage must therefore follow the referenced MBP entry.

using namespace ONIXS_CMEMDH_NAMESPACE;
using namespace ONIXS_CMEMDH_MESSAGING_NAMESPACE;
using namespace ONIXS_CME_MDH_TESTING_NAMESPACE;
 
const auto securityId = SecurityId{100510};
 
incrementalFeed.send(
    30,
    initialize<IncrementalRefreshBook>(EventFlags::EndOfRealQuotes)
        | ONIXS_CMEMDH_ENTRY(New, securityId, EntryTypeBook::Bid, 1, 1, Decimal9(100), 20, 1)
        // referenceId 1 links this order to the first market-data entry.
        | ONIXS_CMEMDH_ORDER_ID_ENTRY(New, 9001, 100, 20, 1)
    ).get();

Multiple market-data and order-ID stages may be combined in one recipe. Keep each referenceId aligned with the one-based position of its associated market-data entry.

Repeat a book update

The multicast packet sequence number passed to send and the per-security rptSeq encoded in each entry advance independently. Recipe values are captured when the recipe is built, so construct a fresh recipe for each logical update that advances rptSeq, or factor the common fields into a helper.

using namespace ONIXS_CMEMDH_NAMESPACE;
using namespace ONIXS_CMEMDH_MESSAGING_NAMESPACE;
using namespace ONIXS_CME_MDH_TESTING_NAMESPACE;
 
const auto securityId = SecurityId{100500};
 
const auto makeUpdate = [securityId](UInt32 rptSeq) {
    return initialize<IncrementalRefreshBook>(EventFlags::EndOfRealQuotes | EventFlags::EndOfEvent)
        | ONIXS_CMEMDH_ENTRY(New, securityId, EntryTypeBook::Bid, rptSeq, 1, Decimal9(100), 10, 1);
};
 
incrementalFeed.send(40, makeUpdate(1)).get();
incrementalFeed.send(41, makeUpdate(2)).get();

Reusing a fully materialized recipe for the same security resends the captured rptSeq; use that only when testing a duplicate or outdated update.