Emulator.h

Go to the documentation of this file.

/*
 * Copyright Onix Solutions Limited [OnixS]. All rights reserved.
 *
 * This software owned by Onix Solutions Limited [OnixS] and is protected by copyright law
 * and international copyright treaties.
 *
 * Access to and use of the software is governed by the terms of the applicable ONIXS Software
 * Services Agreement (the Agreement) and Customer end user license agreements granting
 * a non-assignable, non-transferable and non-exclusive license to use the software
 * for it's own data processing purposes under the terms defined in the Agreement.
 *
 * Except as otherwise granted within the terms of the Agreement, copying or reproduction of any part
 * of this source code or associated reference material to any other location for further reproduction
 * or redistribution, and any amendments to this copyright notice, are expressly prohibited.
 *
 * Any reproduction or redistribution for sale or hiring of the Software not in accordance with
 * the terms of the Agreement is a violation of copyright law.
 */
#pragma once
 
#include <OnixS/ICE/BOE/IceBoe.h>
#include <OnixS/ICE/BOE/testing/Gateway.h>
 
#include "../Common/Tools.h"
 
#include <thread>
#include <future>
#include <functional>
#include <chrono>
#include <iostream>
 
namespace Samples {
 
using namespace ONIXS_ICEBOE_NAMESPACE;
using namespace ONIXS_ICEBOE_MESSAGING_NAMESPACE;
 
class BusSessionGatewayListener final : public ONIXS_ICEBOE_NAMESPACE::Testing::ClientMessageListener
{
public:
    BusSessionGatewayListener(const std::string& host, Port port)
        : host_(host)
        , port_(port)
    {
    }
 
    void onIPRequest(const IPRequest request, ONIXS_ICEBOE_NAMESPACE::Testing::Gateway* gateway) override
    {
        assert(!request.users().empty());
 
        auto report = gateway->createIpReport(request.users()[0].userId(), toStrRef(host_), port_, toStrRef("emulator_token"));
 
        gateway->send(report, gateway->outSeqNum());
    }
 
private:
    const std::string host_{};
    const Port port_{};
};
 
class GatewayListener final : public ONIXS_ICEBOE_NAMESPACE::Testing::ClientMessageListener
{
public:
    void onNewOrderRequest(const NewOrderRequest order, ONIXS_ICEBOE_NAMESPACE::Testing::Gateway* gateway) override
    {
        updateReport(order);
 
        gateway->send(report_, gateway->outSeqNum());
    }
 
 
protected:
    void updateReport(const NewOrderRequest& order)
    {
        report_->clOrdId(order.clOrdId())
            .originatorUserId(order.originatorUserId())
            .side(order.side())
            .symbol(order.symbol())
            .ordType(order.ordType())
            .side(order.side())
            .leavesQty(0)
            .meAcceptanceTime(UtcWatch::now())
            .timePriority(UtcWatch::now())
            .orderStatus(OrderStatusEnum::Fill)
            .execId(OrderExecID{1, 1, UtcWatch::now()})
            .memo("HelloFromEmulator")
            ;
    }
 
    MessageHolder<ExecutionReport_New> report_;
};
 
template <class Listener>
class GatewayEmulatorThread
{
public:
    GatewayEmulatorThread(const SessionSettings::LicenseStores& licenseStores, const std::string & host, Port port, std::unique_ptr<Listener> clientMessageListener)
        : gateway_(licenseStores, port, host.c_str(), std::chrono::seconds{10}, std::chrono::seconds{10})
        , clientMessageListener_(std::move(clientMessageListener))
    {
        std::promise<void> emulatorTaskPromise;
        emulatorTaskDone_ = emulatorTaskPromise.get_future();
 
        std::thread(
            std::bind(
                [&](std::promise<void>& p)
                {
                    try
                    {
                        gateway_.run(*clientMessageListener_, &stopRequested_);
                        p.set_value();
                    }
                    catch (const std::exception& ex)
                    {
                        std::cerr << "Exception in Emulator's thread: " << ex.what() << '.' << std::endl;
                        p.set_exception(std::current_exception());
                    }
                    catch (...)
                    {
                        std::cerr << "UNKNOWN Exception in Emulator's thread." << std::endl;
                        p.set_exception(std::current_exception());
                    }
                },
                std::move(emulatorTaskPromise)))
            .detach();
    }
 
    ~GatewayEmulatorThread()
    {
        stopRequested_.store(true);
 
        if(emulatorTaskDone_.valid())
            emulatorTaskDone_.wait_for(std::chrono::seconds(10));
    }
 
    void wait(std::chrono::seconds timeout = std::chrono::seconds(10))
    {
        if(emulatorTaskDone_.valid())
        {
            std::future<void> f = std::move(emulatorTaskDone_);
            f.wait_for(timeout);
            f.get();
        }
    }
 
    const Testing::Gateway& gateway() const noexcept
    {
        return gateway_;
    }
 
private:
    std::atomic<bool> stopRequested_ {false};
    Testing::Gateway gateway_;
    std::unique_ptr<Listener> clientMessageListener_;
    std::future<void> emulatorTaskDone_;
};
 
 
inline
std::pair<std::unique_ptr<GatewayEmulatorThread<BusSessionGatewayListener>>, std::unique_ptr<GatewayEmulatorThread<GatewayListener>>> createEmulator(const SessionSettings& settings, const ConnectivityConfiguration& cfg, bool tcpDirect = false)
{
    if(cfg.useEmulator())
    {
#if !defined (_WIN32)
        if(tcpDirect)
            throw std::runtime_error("Emulator is not supported for TCPDirect.");
#endif
 
        std::clog << "!Emulator is used!" << std::endl;
 
        auto bgwSessionGatewayListener = makeUnique<GatewayListener>();
        auto bgwSessionEmulatorThread = makeUnique<GatewayEmulatorThread<GatewayListener>>(settings.licenseStores(), "127.0.0.1", 0, std::move(bgwSessionGatewayListener));
 
        auto busSessionGatewayListener = makeUnique<BusSessionGatewayListener>(bgwSessionEmulatorThread->gateway().host(), bgwSessionEmulatorThread->gateway().port());
        auto busSessionEmulatorThread = makeUnique<GatewayEmulatorThread<BusSessionGatewayListener>>(settings.licenseStores(), "127.0.0.1", cfg.port(), std::move(busSessionGatewayListener));
 
        return std::make_pair(std::move(busSessionEmulatorThread), std::move(bgwSessionEmulatorThread));
    }
 
    return {};
}
 
}