TCPDirect Getting Started Sample

This sample shows how to connect TCPDirect session to the pre-defined host and port.

When the session is established, the NewOrderSngle SME message is sent to the counterparty.

Using TCPDirect Network Stack

TCPDirect is part of Solarflare’s suite of network acceleration technologies.

The TCP loopback is not supported by TCPDirect technology, so the emulator should be started on a remote host (or on a different physically connected card with a configured route to it).

For developing/debugging purposes OnixS iLink-3 provides TCPDirect emulation for Windows over usual sockets, so TCPDirectGettingStarted sample can be run in loopback mode under Windows.

Source code

#include <OnixS/CME/iLink3/TcpDirectStack.h>

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

using namespace Samples;
using namespace Threading;

int securityId;
Int64 priceMantissa;

int main(int argc, char * argv[]) {
    clog << "CME iLink 3 TCPDirectGettingStarted Sample." << endl << endl;

    int marketSegmentId = 99;
    // Use the IP address belonging to the TcpDirectStack interface in emulator mode.
    string host = "192.168.0.1";
    Port port = 49252;

    bool useEmulator = false;

    if(argc < 4) {
#if defined(ONIXS_ILINK3_HAS_GATEWAY_EMULATOR)
        useEmulator = true;
#else
        std::cerr << "usage: [MarketSegmentId] [Host] [Port] (SecurityId) (PriceMantissa)" << std::endl;
        return 1;
#endif
    }
    else {
        marketSegmentId = atoi(argv[1]);
        host = argv[2];
        port = atoi(argv[3]);
    }

    securityId = argc > 4 ? atoi(argv[4]) : DefaultSecurityId;
    priceMantissa = argc > 5 ? fromStr<Int64>(argv[5]) : DefaultPriceMantissa;

#if defined(ONIXS_ILINK3_HAS_GATEWAY_EMULATOR)
    std::unique_ptr<GatewayEmulatorThread> gateway;
#endif

    int result = 0;

    try {
        SignalHelper::manageLinuxSignals();

        const SessionSettings settings = fillSettings(useEmulator);

#if defined(ONIXS_ILINK3_HAS_GATEWAY_EMULATOR)
        if(useEmulator)
            gateway.reset(new GatewayEmulatorThread(settings.licenseStores(), host, port));
#endif

        TcpDirectAttr attr;

        // The default interface name could be provided by the environment variable
        //     ZF_ATTR="interface=<interface-name>", see TCPDirect specs,
        //
        // or could be set manually:
        //     attr.set("interface", "interface-name");

        TcpDirectStack tcpDirectStack(attr);

        bool finished = false;

        class TcpDirectListener : public Listener
        {
        public:
            explicit TcpDirectListener(bool & finished)
                : finished_(finished) {}

            void onStateChange(SessionStateId::Enum newState, SessionStateId::Enum prevState, Session * session) ONIXS_ILINK3_OVERRIDE
            {
                Listener::onStateChange(newState, prevState, session);

                switch(newState) {
                case SessionStateId::Established:
                {
                    NewOrderSingle order;
                    Helper::setOrderFields(order, PartyDetailsListReqID, securityId, priceMantissa);

                    // The TCPDirect technology is NOT thread-safe.
                    // Therefore, the event dispatching and message sending should be performed from the same thread.
                    session->send(order);
                    clog << "\nThe order was sent." << endl;

                    break;
                }

                case SessionStateId::Disconnected:
                    finished_ = true;
                    break;

                default:
                    break;
                }
            }

            void onError(
                SessionErrorReason::Enum reason, const std::string & text, Session * session, Messaging::SbeMessage msg) ONIXS_ILINK3_OVERRIDE
            {
                Listener::onError(reason, text, session, msg);
                finished_ = true;
            }

        private:
            bool & finished_;
        }
        listener(finished);

        Session session(tcpDirectStack, settings, marketSegmentId, &listener);

        if(session.negotiated())
            session.reset(true);

        session.connectAsync(host, port);

        while(!finished)
        {
            // The TCPDirect technology is NOT thread-safe.
            // Therefore, the event dispatching and message sending should be performed from the same thread.
            tcpDirectStack.dispatchEvents();

            if(Helper::keyPressed())
                break;
        }

        session.disconnectAsync();

        while(!tcpDirectStack.isQuiescent())
            tcpDirectStack.dispatchEvents();
    }
    catch(const std::exception & ex) {
        cerr << "\nEXCEPTION: " << ex.what() << endl;
        result = 1;
    }

#if defined(ONIXS_ILINK3_HAS_GATEWAY_EMULATOR)
    if(gateway) {
        try {
            gateway->wait();
        }
        catch(...) {
            // Emulator exception if any) is already reported from Emulator's thread
        }
    }
#endif

    return result;
}