Gateway.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/CME/iLink3/testing/ABI.h>
#include <OnixS/CME/iLink3/SessionSettings.h>

#if defined(ONIXS_ILINK3_HAS_GATEWAY_EMULATOR)

#include <OnixS/CME/iLink3/Messaging.h>
#include <OnixS/CME/iLink3/ValuesEnums.h>
#include <OnixS/CME/iLink3/testing/MessagePtr.h>
#include <OnixS/CME/iLink3/testing/ClientMessageListener.h>

#include <chrono>
#include <atomic>

namespace OnixS {
namespace CME {
namespace iLink3 {
namespace Testing {

// Assume NegotiationResponse501::credentials is empty
constexpr
UInt16 calculateNegotiationResponse501MaxSize()
{
    using Message = NegotiationResponse501;

    return
        Message::blockLength(Message::Schema::Version)
        + MessageHeader::Size
        + Message::getMinimalVariableFieldsSize(Message::Schema::Version);
}

//  Type aliases
using NegotiationResponse501Container = MessageHolder<NegotiationResponse501, calculateNegotiationResponse501MaxSize()>;
using NegotiationReject502Container = MessageHolder<NegotiationReject502>;
using EstablishmentAck504Container = MessageHolder<EstablishmentAck504>;
using EstablishmentReject505Container = MessageHolder<EstablishmentReject505>;
using Terminate507Container = MessageHolder<Terminate507>;
using Sequence506Container = MessageHolder<Sequence506>;

using MessageBasePtr = MessagePtr<SbeMessage>;
using Negotiate500Ptr = MessagePtr<Negotiate500>;
using Establish503Ptr = MessagePtr<Establish503>;
using Terminate507Ptr = MessagePtr<Terminate507>;
using Sequence506Ptr = MessagePtr<Sequence506>;

/// CME iLink 3 Gateway Emulator.
class ONIXS_ILINK3_EXPORTED Gateway
{
public:
    /// Timeout alias.
    using Timeout = std::chrono::milliseconds;

    Gateway(const SessionSettings::LicenseStores& licenseStores, int port, const char* host = "127.0.0.1",
            const Timeout& acceptTimeout = defaultTimeout(),
            const Timeout& sendReceiveTimeout = defaultTimeout());

    ///
    virtual ~Gateway();

    /// Deleted.
    Gateway(const Gateway&) = delete;
    Gateway& operator=(const Gateway&) = delete;

    /// Move semantic.
    Gateway(Gateway&&) noexcept;
    Gateway& operator=(Gateway&&) noexcept;

    /// Swap with other.
    void swap(Gateway&) noexcept;

    /// Accepts an incoming connection.
    /// If the `acceptTimeout` is zero, the `acceptTimeout` provided in the constructor is used.
    Gateway& acceptConnection(const Timeout& acceptTimeout = defaultTimeout());

    /// Accepts an incoming connection.
    /// If the operation is timed out, `false` is returned.
    bool tryAcceptConnection(const Timeout& acceptTimeout = defaultTimeout());

    /// Accepts an incoming iLink 3 session.
    Gateway& acceptSession(SeqNumber outgoingSequenceNumber = 1, UInt64 previousUuid = 0, UInt32 previousSeqNo = 0);

    /// Waits until the Terminate507 message is received.
    ///
    /// If the listener is provided, then receives and reports application-level messages via the corresponding callback.
    /// Responds to the Sequence506 message if received.
    /// \note actual `outSeqNum` must be provided before invocation
    Terminate507Ptr waitUntilTerminate(
        ClientMessageListener* listener = nullptr, ClientSessionMessageListener* sessionMessageListener = nullptr,
        std::atomic<bool>* stopRequested = nullptr);

    /// Sends the Sequence506 message.
    Gateway& sendSequence(UInt32 nextSeqNo, KeepAliveLapsed::Enum keepAliveLapsed = KeepAliveLapsed::NotLapsed, UInt64 uuid = 0);

    /// Closes the accepted connection.
    Gateway& disconnect();

    /// @return listen host.
    const char * host() const noexcept;

    /// \return the listen port.
    ONIXS_ILINK3_NODISCARD
    int port() const noexcept;

    /// Enables logging.
    Gateway& enableLogger(const std::string& logFileName, unsigned int logLevel = 3);

    /// Sends the given message.
    /// \note  does not update `outSeqNum`.
    template <typename SbeMessageType, size_t MaxMessageSize, typename MessageInitializer>
    Gateway& send(iLink3::Messaging::MessageHolder<SbeMessageType, MaxMessageSize, MessageInitializer>& msg);

    /// Sends the given message.
    /// \note  does not update `outSeqNum`.
    template <typename SbeMessageType, size_t MaxMessageSize, typename MessageInitializer>
    Gateway& send(iLink3::Messaging::MessageHolder<SbeMessageType, MaxMessageSize, MessageInitializer>&& msg);

    /// Sets the given sequence number and sends the message.
    /// \note  does not update `outSeqNum`.
    template <typename SbeMessageType, size_t MaxMessageSize, typename MessageInitializer>
    Gateway& send(iLink3::Messaging::MessageHolder<SbeMessageType, MaxMessageSize, MessageInitializer>& msg, SeqNumber num);

    /// Sets the given sequence number and sends the message.
    /// \note  does not update `outSeqNum`.
    template <typename SbeMessageType, size_t MaxMessageSize, typename MessageInitializer>
    Gateway& send(iLink3::Messaging::MessageHolder<SbeMessageType, MaxMessageSize, MessageInitializer>&& msg, SeqNumber num);

    /// Sets the given sequence numbers and sends the messages.
    /// \note  does not update `outSeqNum`.
    template <typename SbeMessageType, size_t MaxMessageSize, typename MessageInitializer>
    Gateway& send(iLink3::Messaging::MessageHolder<SbeMessageType, MaxMessageSize, MessageInitializer>& msg, std::initializer_list<SeqNumber> numbers);

    /// Sets the given sequence numbers and sends the messages.
    /// \note  does not update `outSeqNum`.
    template <typename SbeMessageType, size_t MaxMessageSize, typename MessageInitializer>
    Gateway& send(iLink3::Messaging::MessageHolder<SbeMessageType, MaxMessageSize, MessageInitializer>&& msg, std::initializer_list<SeqNumber> numbers);


    /// Receives the message of the particular type.
    ///
    /// \throw std::logic_error if an unexpected type is received.
    template <typename TMsg>
    MessagePtr<TMsg> receiveTypedMessage();

    /// Accepts an incoming connection and establishes the session.
    ///
    /// - Receives and reports application-level messages via the corresponding callback.
    /// - Sends Sequence messages in response to the received ones.
    Gateway& run(ClientMessageListener& listener);

    /// Accepts an incoming connection and establishes the session.
    ///
    /// - Receives and reports application-level messages via the corresponding callback.
    /// - Receives and reports session-level messages via the corresponding callback.
    /// - Sends Sequence messages.
    Gateway& run(ClientMessageListener& appMsgListener, ClientSessionMessageListener& sessionMsgListener);

    /// \return the sequence number of the next outgoing message.
    ONIXS_ILINK3_NODISCARD
    SeqNumber outSeqNum() const noexcept;

    /// Sets the sequence number of the next outgoing message.
    Gateway& outSeqNum(SeqNumber nextOutSeqNum) noexcept;

    /// \return UUID.
    ONIXS_ILINK3_NODISCARD
    UInt64 uuid() const noexcept
    {
        return uId_;
    }

    /// \return the Fault Tolerance Indicator.
    ONIXS_ILINK3_NODISCARD
    FTI::Enum faultToleranceIndicator() const noexcept
    {
        return faultToleranceIndicator_;
    }

    /// Sets the Fault Tolerance Indicator.
    Gateway& faultToleranceIndicator(FTI::Enum value)
    {
        faultToleranceIndicator_ = value;

        return *this;
    }

    /// Sets the value of the option to improve latency at the expense of message throughput (the `TCP_NODELAY` socket option).
    Gateway& tcpNoDelayOption(bool noDelay);

    /// \return the value of the option to improve latency at the expense of message throughput (the `TCP_NODELAY` socket option).
    ///
    /// The default value is SessionSettings::Default::TcpNoDelayOption.
    ONIXS_ILINK3_NODISCARD
    bool tcpNoDelayOption() const noexcept ;

    /// \return the socket receive buffer size.
    ///
    /// The default value is SessionSettings::Default::SocketReceiveBufferSize.
    ONIXS_ILINK3_NODISCARD
    int socketReceiveBufferSize() const;

    /// Sets the socket receive buffer size.
    ///
    /// \param bufferSize the size of the socket receive buffer size.
    /// If SessionSettings::UseOsDefaultSocketBufferSize then the default operating system value is used.
    Gateway& socketReceiveBufferSize(int bufferSize);

    /// \return the size of the socket send buffer size.
    ///
    /// The default value is SessionSettings::Default::SocketSedndBufferSize.
    ONIXS_ILINK3_NODISCARD
    int socketSendBufferSize() const noexcept;

    /// Sets the size of the socket send buffer size.
    ///
    /// \param bufferSize the size of the socket send buffer size.
    /// If SessionSettings::UseOsDefaultSocketBufferSize then the default operating system value is used.
    Gateway& socketSendBufferSize(int bufferSize);

protected:
    /// Checks whether the connection is closed by the counterparty.
    ONIXS_ILINK3_NODISCARD
    bool isConnectionClosed(const Timeout& timeout) const;

    /// Checks whether the connection is closed by the counterparty
    /// using the default timeout (provided in the constructor).
    ONIXS_ILINK3_NODISCARD
    bool isConnectionClosed() const;

    /// Receives a message.
    MessageBasePtr receive() const;

    /// Receives a message using the provided timeout.
    /// If the operation is timed out, an empty message is returned.
    MessageBasePtr tryReceive(const Timeout& timeout = defaultTimeout()) const;

    /// Receives the Negotiate500 message.
    ///
    /// \throw std::logic_error if an unexpected type is received.
    Negotiate500Ptr receiveNegotiate();

    /// Receives the Negotiate500 and responds with the NegotiationResponse501 message.
    ///
    /// \throw std::logic_error if an unexpected type is received.
    Negotiate500Ptr acceptNegotiate(UInt32 previousSeqNo = 0);

    /// Receives the Negotiate500 message and responds with the NegotiationReject502 message.
    ///
    /// \throw std::logic_error if an unexpected type is received.
    Negotiate500Ptr rejectNegotiate(NegotiationRejectErrorCode::Enum errorCodes = NegotiationRejectErrorCode::Enum(),
        const std::string & reason = std::string());

    /// Receives the Establish503 message.
    ///
    /// \throw std::logic_error if an unexpected type is received.
    Establish503Ptr receiveEstablish();

    /// Receive the Establish503 message and respond with the EstablishmentAck504 message.
    ///
    /// \throw std::logic_error if an unexpected type is received.
    Establish503Ptr acceptEstablish(UInt32 outgoingSequenceNumber = 1, UInt64 previousUuid = 0, UInt32 previousSeqNo = 0);

    /// Receives the Establish503 message and responds with the EstablishmentReject505 message.
    ///
    /// \throw std::logic_error if an unexpected type is received.
    Establish503Ptr rejectEstablish(UInt32 outgoingSequenceNumber = 1, EstablishmentRejectErrorCode::Enum errorCodes = EstablishmentRejectErrorCode::Enum(), const std::string & reason = std::string());

    /// Receive the Terminate507 message.
    ///
    /// \throw std::logic_error if an unexpected type is received.
    Terminate507Ptr receiveTerminate();

    /// Receive the Terminate507 message and responds with the Terminate507 message.
    ///
    /// \throw std::logic_error if an unexpected type is received.
    Terminate507Ptr acceptTerminate();

    /// Sends the Terminate507 message.
    Gateway& sendTerminate(const std::string& reason = {}, UInt16 errCode = TerminateErrorCode::Other);

    /// Sends the Terminate507 message and waits until the response is received.
    Terminate507Ptr terminate(const std::string& reason = {}, UInt16 errCode = TerminateErrorCode::Other);

    /// Creates a NegotiationResponse501 message
    NegotiationResponse501Container createNegotiationResponse(const Negotiate500Ptr& request, UInt32 previousSeqNo);

    /// Creates a NegotiationReject502 message.
    NegotiationReject502Container createNegotiationReject(const Negotiate500Ptr& request, NegotiationRejectErrorCode::Enum errorCodes,
        const std::string & reason);

    /// Creates a EstablishmentAck504 message.
    EstablishmentAck504Container createEstablishmentAck(const Establish503Ptr& request, UInt32 outgoingSequenceNumber);

    /// Creates a EstablishmentReject505 message.
    EstablishmentReject505Container createEstablishmentReject(const Establish503Ptr& request, UInt32 outgoingSequenceNumber,
        EstablishmentRejectErrorCode::Enum errorCodes, const std::string & reason);

    /// Creates a Terminate507 message.
    Terminate507Container createTerminate(const Terminate507Ptr& request);

    /// Creates a Terminate507 message.
    Terminate507Container createTerminate(const std::string& reason, UInt16 errCode);

    /// Creates a Sequence506 message.
    Sequence506Container createSequence(UInt32 nextSeqNo, KeepAliveLapsed::Enum keepAliveLapsed = KeepAliveLapsed::NotLapsed, UInt64 uuid = 0);

    /// Sends the given message.
    Gateway& send(Messaging::SbeMessage, Messaging::SimpleOpenFramingHeader&);

    /// Sends the given data
    Gateway& sendData(const void* data, size_t size);

    /// Close the listening socket
    Gateway& close();

    /// Default timeout
    static Timeout defaultTimeout()
    {
        return Timeout(30000);
    }

    /// UUID.
    UInt64 uId_{0};

private:
    Messaging::FTI::Enum faultToleranceIndicator_{Messaging::FTI::Enum::Primary};

    class Impl;
    Impl* impl_;
};


template <typename TMsg>
MessagePtr<TMsg> Gateway::receiveTypedMessage()
{
    return cast<TMsg>(receive());
}

template <typename SbeMessageType, size_t MaxMessageSize, typename MessageInitializer>
Gateway& Gateway::send(iLink3::Messaging::MessageHolder<SbeMessageType, MaxMessageSize, MessageInitializer>& msg)
{
    const auto messageSize = msg.setHeader();
    auto* const header = msg.header();
    assert(header);

    return send(Messaging::SbeMessage(header + 1, messageSize), *header);
}

template <typename SbeMessageType, size_t MaxMessageSize, typename MessageInitializer>
Gateway& Gateway::send(iLink3::Messaging::MessageHolder<SbeMessageType, MaxMessageSize, MessageInitializer>&& msg)
{
    return send(msg);
}

template <typename Holder>
typename std::enable_if<std::is_base_of<TagBased::Message, typename Holder::MessageType>::value>::type
    setSeqNum(Holder& msg, SeqNumber num)
{
    msg->set(TagBased::Tags::SeqNum, num);
}

template <typename Holder>
typename std::enable_if<!std::is_base_of<TagBased::Message, typename Holder::MessageType>::value>::type
    setSeqNum(Holder& msg, SeqNumber num)
{
    msg->setSeqNum(num);
}

template <typename SbeMessageType, size_t MaxMessageSize, typename MessageInitializer>
Gateway& Gateway::send(iLink3::Messaging::MessageHolder<SbeMessageType, MaxMessageSize, MessageInitializer>& msg, SeqNumber num)
{
    setSeqNum(msg, num);
    return send(msg);
}

template <typename SbeMessageType, size_t MaxMessageSize, typename MessageInitializer>
Gateway& Gateway::send(iLink3::Messaging::MessageHolder<SbeMessageType, MaxMessageSize, MessageInitializer>&& msg, SeqNumber num)
{
    setSeqNum(msg, num);
    return send(msg);
}

template <typename SbeMessageType, size_t MaxMessageSize, typename MessageInitializer>
Gateway& Gateway::send(
    iLink3::Messaging::MessageHolder<SbeMessageType, MaxMessageSize, MessageInitializer>& msg, std::initializer_list<SeqNumber> numbers)
{
    for(auto seqNum : numbers)
    {
        send(msg, seqNum);
    }

    return *this;
}

template <typename SbeMessageType, size_t MaxMessageSize, typename MessageInitializer>
Gateway& Gateway::send(
    iLink3::Messaging::MessageHolder<SbeMessageType, MaxMessageSize, MessageInitializer>&& msg, std::initializer_list<SeqNumber> numbers)
{
    for(auto seqNum : numbers)
    {
        send(msg, seqNum);
    }

    return *this;
}


}}}}


#endif