Thread Pool Sample

This sample demonstrates how to use the ThreadPool mode.

Source code:

#include <OnixS/FIXEngine.h>

#include "../../Common/Helpers.h"
#include "../../Common/Settings.h"

using namespace Settings;
using namespace OnixS::FIX;
using namespace OnixS::FIX::FIX44;

namespace
{
    class Listener : public OnixS::FIX::ISessionListener
    {
        const std::string name_;

    public:

        explicit Listener(const std::string & name):name_(name){}

        /// Is called when the application-level message is received from the counterparty.
        void onInboundApplicationMsg(Message& msg, Session* sn) ONIXS_FIXENGINE_FINAL
        {
            std::clog << "\n[" << name_ << "] - Incoming application-level message:\n" << msg << std::endl;

            try
            {
                if(msg.type() == Values::MsgType::NewOrderSingle)
                {
                    Message execReport(Values::MsgType::ExecutionReport, FixProtocolVersion);

                    execReport.set(Tags::OrderID, msg.get(Tags::ClOrdID))
                              .set(Tags::ExecID, "ExecID")
                              .set(Tags::ExecType, "0")
                              .set(Tags::ExecTransType, "0")
                              .set(Tags::OrdStatus,  "0")
                              .set(Tags::Symbol, msg.get(Tags::Symbol))
                              .set(Tags::Side, msg.get(Tags::Side))
                              .set(Tags::CumQty, msg.get(Tags::OrderQty))
                              .set(Tags::AvgPx, 100.0)
                              .set(Tags::LeavesQty, msg.get(Tags::OrderQty));

                    sn->send(&execReport);
                }
            }
            catch(const std::exception & ex)
            {
                std::clog << "Exception during the processing of incoming message: " << ex.what() << std::endl;
            }
        }

        /// Is called when the session-level message is received from the counterparty.
        void onInboundSessionMsg(Message& msg, Session*) ONIXS_FIXENGINE_FINAL
        {
            std::clog << "\n[" << name_ << "] - Incoming session-level message:\n" << msg << std::endl;
        }

        void onStateChange(SessionState::Enum newState, SessionState::Enum prevState, Session*) ONIXS_FIXENGINE_FINAL
        {
            std::clog
                << "\n[" << name_ << "] - Session's state is changed, prevState="
                << SessionState::toString(prevState)
                << ", newState="
                << SessionState::toString(newState)
                << std::endl;
        }

        void onError(ErrorReason::Enum, const std::string& description, Session*) ONIXS_FIXENGINE_FINAL
        {
            std::cerr << "\n[" << name_ << "] - Session-level error:" << description << std::endl;
        }

        void onWarning(WarningReason::Enum, const std::string& description, Session*) ONIXS_FIXENGINE_FINAL
        {
            std::cerr << "\n[" << name_ << "] - Session-level warning:" << description << std::endl;
        }
    };
}

int main()
{
    std::clog << "ThreadPool Sample." << std::endl << std::endl;

    // The default value of thread pool size equals the CPUs number, so it works fine in most cases.
    // However, if needs, one can set it to a particular value.
    const int ThreadPoolSize = 4;

    try
    {
        EngineSettings settings;

        settings.listenPort(ListenPort)
                .licenseStore(LicenseStore)
                .threadingModel(ThreadingModel::ThreadPool)
                .threadPoolSize(ThreadPoolSize);

        Engine::init(settings);

        Listener acceptorListener(TargetCompId);
        Session acceptor(TargetCompId, SenderCompId, FixProtocolVersion, &acceptorListener);

        acceptor.logonAsAcceptor();

        Listener initiatorListener(SenderCompId);
        Session initiator(SenderCompId, TargetCompId, FixProtocolVersion, &initiatorListener);

        initiator.logonAsInitiator("localhost", ListenPort);

        waitUntilEnterKey("Press any key to send the order from the initiator to the acceptor...");

        Message order(Values::MsgType::NewOrderSingle, FixProtocolVersion);

        order.set(Tags::ClOrdID, "Unique identifier for Order")
             .set(Tags::Symbol, "IBM")
             .set(Tags::Side, Values::Side::Buy)
             .set(Tags::OrderQty, 1000)
             .set(Tags::OrdType, Values::OrdType::Market)
             .set(Tags::TransactTime, Timestamp::utc(), TimestampFormat::YYYYMMDDHHMMSSMsec);

        initiator.send(&order);

        std::clog << "The order " << std::endl << order << std::endl << " was sent" << std::endl;

        waitUntilEnterKey("Press any key to disconnect sessions and terminate the application.");

        initiator.logout("The session is disconnected by BuySide").shutdown();
        acceptor.logout().shutdown();

        Engine::shutdown();
    }
    catch(const std::exception & ex)
    {
        processSampleException(ex.what());
        return 1;
    }

    return 0;
}