PCAP Replay Sample

Replays market data captured in PCAP files instead of connecting to a live feed. It creates a PcapDataProvider for a given provider (e.g. IDS), feeds it to the Handler as a custom input stream, uses a PcapDataAnalyser to derive the market subscriptions and missing snapshots present in the capture, and processes the data as if received live.

Source code

#include "../Common/Defaults.h"
#include "../Common/ExchangeEventTracer.h"
#include "../Common/MarketDataEventTracer.h"
#include "../Common/ServiceEventTracer.h"
#include "../Common/Utils.h"
 
#include <OnixS/ICE/iMpact/MarketData.h>
 
#include <cstdlib>
#include <fstream>
#include <stdexcept>
 
namespace md = OnixS::ICE::iMpact::MarketData;
 
int main(int argc, char* argv[])
{
    try
    {
        introduceMyself("PCAP Replay");
 
        md::HandlerSettings handlerSettings;
 
        const std::string HELP_ARG = "--help";
 
        if (argc == 2 && HELP_ARG == argv[1])
        {
            std::cout << "Usage: PcapReplay <provider> <dir>" << std::endl;
            return EXIT_SUCCESS;
        }
 
        const md::PcapDataProviderId providerId = argc > 1 ? md::KnownPcapDataProviderIds::deserialize(argv[1])
                                                           : md::KnownPcapDataProviderIds::IDS;
 
        const std::string inputDir = argc > 2 ? argv[2] : "replay";
 
        md::PcapDataProviderSettings pcapDataProviderSettings;
 
        pcapDataProviderSettings.logDirectory = LOG_DIRECTORY;
        pcapDataProviderSettings.logLevel = md::LogLevels::Debug;
        pcapDataProviderSettings.pcapDirectory = inputDir;
        pcapDataProviderSettings.connectivityConfiguration = PROD_C10Y_FILE;
 
        RAII<md::PcapDataProvider> pcapDataProvider(
            md::PcapDataProviderFactory::create(providerId, pcapDataProviderSettings)
        );
 
        // Set custom log input stream.
        handlerSettings.inputStream = pcapDataProvider.get();
 
        // 0.0 == replay w/ maximum speed
        // 1.0 == replay w/ original speed -- i.e. take in account time delta
        // between adjacent messages in log file
        handlerSettings.logPlayerSpeedMultiplier = 0.0;
 
        // Let's activate logging to get data for support.
        handlerSettings.logLevel = md::LogLevels::Debug;
        handlerSettings.logDirectory = LOG_DIRECTORY;
        handlerSettings.logFileNamePrefix = LOG_FILE_NAME_PREFIX;
 
        // By default, license is stored in lib folder of the distribute
        // package. Instead of copying, let's use original file.
        handlerSettings.licenseDirectory = LICENSE_DIRECTORY;
 
        // Primary settings - connectivity configuration.
        handlerSettings.connectivityConfiguration = PROD_C10Y_FILE;
 
        // Constructs handler with custom license root and enabled logging.
        md::Handler handler(handlerSettings);
 
        // Traces all service events coming from handlers.
        ServiceEventTracer serviceEventTracer;
 
        serviceEventTracer.attachTo(handler);
 
        // Traces all exchange events coming from handlers.
        ExchangeEventTracer exchangeEventTracer;
 
        exchangeEventTracer.attachTo(handler);
 
        // Traces all market data events coming from handlers.
        MarketDataEventTracer marketDataEventTracer;
 
        marketDataEventTracer.attachTo(handler);
 
        // Making subscriptions.
        RAII<md::PcapDataAnalyser> pcapDataAnalyser(
            md::PcapDataAnalyserFactory::create(providerId, inputDir, PROD_C10Y_FILE)
        );
 
        const md::MarketSubscriptions& marketSubscriptions = pcapDataAnalyser->marketSubscriptions();
 
        for (
            md::MarketSubscriptions::const_iterator it = marketSubscriptions.begin(); it != marketSubscriptions.end();
            ++it
        )
        {
            exchangeEventTracer.addFilterForProductDefinitions(
                it->marketType,
                it->securityType,
                pcapDataAnalyser->missingMarketSnapshotIds(it->marketType, it->securityType)
            );
        }
 
        // Now starts the subscription.
        handler.start(marketSubscriptions);
 
        readUserResponse();
 
        std::cout << "Shutdown the Handler..." << std::endl;
 
        // Stops multicast data processing.
        handler.stop();
 
        std::cout << "Done." << std::endl;
    }
    catch (const std::exception& ex)
    {
        std::cout << "Error: " << ex.what() << std::endl;
        return EXIT_FAILURE;
    }
    catch (...)
    {
        std::cout << "Unknown exception." << std::endl;
        return EXIT_FAILURE;
    }
 
    readUserResponse();
 
    return EXIT_SUCCESS;
}