Log Replay Sample

Replays a previously recorded Handler log file. It implements a custom LogReplayInputStream that reads the log, feeds it to the Handler via the input stream setting, and controls the replay speed with logPlayerSpeedMultiplier (0.0 for maximum speed).

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 <cstddef>
 
#define USER_DEFINED_LOG_INPUT_STREAM
 
#ifdef _WIN32
#  include <windows.h>
#elif defined(__linux__)
#  include <unistd.h>
#endif
 
#include <cstdlib>
#include <fstream>
#include <stdexcept>
 
namespace md = OnixS::ICE::iMpact::MarketData;
 
namespace {
#ifdef USER_DEFINED_LOG_INPUT_STREAM
class UserDefinedLogInputStream : public md::LogReplayInputStream
{
public:
    static const std::size_t READ_LINE_SLEEP_MILLISECONDS = 0;

    UserDefinedLogInputStream(const char* const logName)
    {
        ifs.open(logName, std::ifstream::in);
 
        if (ifs.bad())
        {
            throw std::runtime_error("Can't open log file");
        }
    }

    std::size_t read(char* const data, std::size_t size)
    {
        if (READ_LINE_SLEEP_MILLISECONDS)
        {
            sleep(READ_LINE_SLEEP_MILLISECONDS);
        }
 
        if (ifs.eof())
        {
            return 0;
        }
 
        ifs.read(data, size);
 
        return static_cast<std::size_t>(ifs.gcount());
    }
 
private:
    UserDefinedLogInputStream(const UserDefinedLogInputStream&);
    UserDefinedLogInputStream& operator=(const UserDefinedLogInputStream&);
 
    static void sleep(const int milliseconds)
    {
#  if defined(_WIN32)
        ::Sleep(milliseconds);
#  else
        timespec tv;
 
        tv.tv_sec = milliseconds / 1000L;
        tv.tv_nsec = milliseconds % 1000L * 1000000L;
 
        nanosleep(&tv, 0);
#  endif
    }
 
private:
    std::ifstream ifs;
};
#endif // USER_DEFINED_LOG_INPUT_STREAM
} // anonymous namespace
 
int main(int argc, char* argv[])
{
    try
    {
        introduceMyself("Log Replay");
 
        md::HandlerSettings handlerSettings;
 
        const char* const logToReplay = argc > 1 ? argv[1] : "replay/Brent_Futures_TOP5PL.log";
 
        std::cout << "Try to replay log file: '" << logToReplay << "'." << std::endl;
 
#ifdef USER_DEFINED_LOG_INPUT_STREAM
        UserDefinedLogInputStream inputStream(logToReplay);
 
        handlerSettings.inputStream = &inputStream;
#else
        handlerSettings.logs.push_back(logToReplay);
#endif
        // 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 = AP1_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.
        md::MarketSubscriptions subscriptions;
 
        md::MarketSubscription brentFutures(
            md::KnownMarketTypes::IpeBrentFutures,
            md::MarketSubType::NonImplied,
            md::SecurityType::Futures,
            md::BookDepth::TOP5PL
        );
 
        subscriptions.insert(brentFutures);
 
        // Now starts the subscription.
        handler.start(subscriptions);
 
        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;
}