Listener.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/SessionListener.h>
#include <OnixS/ICE/BOE/threading/Event.h>
 
#include <functional>
#include <future>
 
#include "../Common/Helpers.h"
#include "../Common/Tools.h"
 
namespace Samples {
 
using namespace ONIXS_ICEBOE_NAMESPACE;
using namespace ONIXS_ICEBOE_MESSAGING_NAMESPACE;
using namespace ONIXS_ICEBOE_NAMESPACE::Threading;
 
struct MessagePrinter
{
    template <typename MessageType>
    void operator()(const MessageType msg) const
    {
        std::clog << msg << std::endl;
    }
 
    static void printMsg(const Messaging::SbeMessage msg)
    {
        if (!processTypified(msg, MessagePrinter()))
            std::clog << "Unknown message type";
    }
};
 
class Listener : public SessionListener
{
public:
    ~Listener() override = default;
 
    void onMessageSending(char* bytes, size_t size, Session*) override
    {
        std::clog << "\nOutbound message:\n";
        MessagePrinter::printMsg(SbeMessage(bytes, static_cast<Messaging::MessageSize>(size)));
    }
 
    void onLogonReport(const Messaging::LogonReport msg, Session*) override
    {
        std::clog << "\nReceived:\n" << msg << std::endl;
    }
 
    void onLogoutReport(const Messaging::LogoutReport msg, Session*) override
    {
        std::clog << "\nReceived:\n" << msg << std::endl;
    }
 
    void onHeartBeat(const Messaging::HeartBeat msg, Session*) override
    {
        std::clog << "\nReceived:\n" << msg << std::endl;
    }
 
    void onError(const Messaging::Error msg, Session*) override
    {
        std::clog << "\nReceived:\n" << msg << std::endl;
    }
 
    void onNews(const Messaging::News msg, Session*) override
    {
        std::clog << "\nReceived:\n" << msg << std::endl;
    }
 
    void onSecurityDefinitionReport_Strategy(const Messaging::SecurityDefinitionReport_Strategy msg, Session*) override
    {
        std::clog << "\nReceived:\n" << msg << std::endl;
    }
 
    void onSecurityDefinitionReport(const Messaging::SecurityDefinitionReport msg, Session*) override
    {
        std::clog << "\nReceived:\n" << msg << std::endl;
    }
 
    void onSecurityDefinitionReject(const Messaging::SecurityDefinitionReject msg, Session*) override
    {
        std::clog << "\nReceived:\n" << msg << std::endl;
    }
 
    void onSecurityDefinitionReport_Product(const Messaging::SecurityDefinitionReport_Product msg, Session*) override
    {
        std::clog << "\nReceived:\n" << msg << std::endl;
    }
 
    void onIPReport(const Messaging::IPReport msg, Session*) override
    {
        std::clog << "\nReceived:\n" << msg << std::endl;
    }
 
    void onTraderLogonReport(const Messaging::TraderLogonReport msg, Session*) override
    {
        std::clog << "\nReceived:\n" << msg << std::endl;
 
        // `text` is optional on the wire and capped at 100 characters, so it need not carry the
        // values the code's entry asks for. Let `Messaging::format` describe such a frame instead
        // of dereferencing an absent `text` and throwing out of this callback, which would leave
        // `traderLoggedOn` unsatisfied and every waiter on it blocked until its timeout.
        if(msg.exchangeCode())
        {
            traderLoggedOn.set_exception(std::make_exception_ptr(std::runtime_error(Messaging::format(msg))));
            return;
        }
 
        traderLoggedOn.set_value();
    }
 
    void onTraderLogoutReport(const Messaging::TraderLogoutReport msg, Session*) override
    {
        std::clog << "\nReceived:\n" << msg << std::endl;
    }
 
    void onExecutionReport_New(const Messaging::ExecutionReport_New msg, Session*) override
    {
        std::clog << "\nReceived:\n" << msg << std::endl;
    }
 
    void onExecutionReport_Modify(const Messaging::ExecutionReport_Modify msg, Session*) override
    {
        std::clog << "\nReceived:\n" << msg << std::endl;
    }
 
    void onExecutionReport_Cancel(const Messaging::ExecutionReport_Cancel msg, Session*) override
    {
        std::clog << "\nReceived:\n" << msg << std::endl;
    }
 
    void onExecutionReport_Reject(const Messaging::ExecutionReport_Reject msg, Session*) override
    {
        std::clog << "\nReceived:\n" << msg << std::endl;
    }
 
    void onExecutionReport_Snapshot(const Messaging::ExecutionReport_Snapshot msg, Session*) override
    {
        std::clog << "\nReceived:\n" << msg << std::endl;
    }
 
    void onExecutionReport_Trade(const Messaging::ExecutionReport_Trade msg, Session*) override
    {
        std::clog << "\nReceived:\n" << msg << std::endl;
    }
 
    void onQuoteReport(const Messaging::QuoteReport msg, Session*) override
    {
        std::clog << "\nReceived:\n" << msg << std::endl;
    }
 
    void onNewOrderCrossReport(const Messaging::NewOrderCrossReport msg, Session*) override
    {
        std::clog << "\nReceived:\n" << msg << std::endl;
    }
 
    void onOrderMassCancelReport(const Messaging::OrderMassCancelReport msg, Session*) override
    {
        std::clog << "\nReceived:\n" << msg << std::endl;
    }
 
    void onMassQuoteReport(const Messaging::MassQuoteReport msg, Session*) override
    {
        std::clog << "\nReceived:\n" << msg << std::endl;
    }
 
    void onStateChange(SessionStateId::Enum newState, SessionStateId::Enum prevState, Session*) override
    {
        std::clog
            << "\nSession's state is changed, prevState="
            << SessionStateId::toString(prevState)
            << ", newState="
            << SessionStateId::toString(newState)
            << std::endl;
    }
 
    void onError(
        SessionErrorReason::Enum, const std::string& description, Session*, Messaging::SbeMessage msg) override
    {
        std::cerr << "\nSession-level error: " << description;
 
        if(msg.valid())
        {
            std::cerr << ",\nCaused by the message: ";
            MessagePrinter::printMsg(msg);
        }
 
        std::cerr << std::endl;
    }
 
    void onWarning(
        SessionWarningReason::Enum, const std::string& description, Session*, Messaging::SbeMessage msg) override
    {
        std::cerr << "\nSession-level warning: " << description;
 
        if(msg.valid())
        {
            std::cerr << ",\nCaused by the message: ";
            MessagePrinter::printMsg(msg);
        }
 
        std::cerr << std::endl;
    }
 
    std::promise<void> traderLoggedOn;
};
 
template <typename T, typename Stack>
std::future<T> wait(Stack& stack, std::promise<T>& promise, std::chrono::seconds timeout = std::chrono::seconds{30})
{
    auto future = promise.get_future();
 
    const auto deadline = std::chrono::steady_clock::now() + timeout;
    std::uint32_t spins = 0;
 
    for (;;)
    {
        stack.dispatchEvents();
 
        if (future.wait_for(std::chrono::nanoseconds{0}) == std::future_status::ready)
            return future;
 
        if ((++spins & 0xFFu) == 0)  // every 256 iterations
        {
            if (std::chrono::steady_clock::now() >= deadline)
            {
                promise.set_exception(std::make_exception_ptr(std::runtime_error("Operation timed out.")));
                return future;
            }
        }
    }
}
 
template <typename T>
std::future<T> wait(std::promise<T>& promise, std::chrono::seconds timeout = std::chrono::seconds{30})
{
    auto future = promise.get_future();
 
    if(future.wait_for(timeout) != std::future_status::ready)
        promise.set_exception(std::make_exception_ptr(std::runtime_error("Operation timed out.")));
 
    return future;
}
 
// Request Binary Order Gateway credentials from Binary Utility Service Gateway.
inline BgwCredentials receiveBgwCredentials(SessionSettings settings, std::string host, Port port)
{
    settings.threadingModel(ThreadingModel::Dedicated);
 
    Listener listener;
 
    // Create an instance of a `BUS` session.
    BusSession busSession(settings, &listener);
 
    // Connect.
    busSession.connect(host, port);
 
    const auto bgwCredentialsFuture = busSession.getBgwCredentialsAsync(settings.userId());
 
    const auto bgwCredentials = bgwCredentialsFuture.get();
 
    // Disconnect.
    busSession.disconnect();
 
    return bgwCredentials;
}
 
}