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/ConflatedTCP/ABI.h>
 
#if defined(ONIXS_CONFLATEDTCP_HAS_GATEWAY_EMULATOR)
 
#include <OnixS/CME/ConflatedTCP/Messaging.h>
#include <OnixS/CME/ConflatedTCP/testing/MessagePtr.h>
#include <OnixS/CME/ConflatedTCP/testing/ClientMessageListener.h>
 
#include <chrono>
 
namespace OnixS {
namespace CME {
namespace ConflatedTCP {
namespace Testing {
 
// Assume NegotiationResponse202::credentials is empty
constexpr
UInt16 calculateNegotiationResponse202MaxSize()
{
    using Message = NegotiationResponse202;
 
    return
        Message::blockLength(Message::Schema::Version)
        + MessageHeader::Size
        + Message::getMinimalVariableFieldsSize(Message::Schema::Version);
}
 
//  Type aliases
using NegotiationResponse202Container = MessageHolder<NegotiationResponse202, calculateNegotiationResponse202MaxSize()>;
using NegotiationReject201Container = MessageHolder<NegotiationReject201>;
using Terminate203Container = MessageHolder<Terminate203>;
using RequestAck206Container = MessageHolder<RequestAck206>;
using RequestReject207Container = MessageHolder<RequestReject207>;
 
using MessageBasePtr = MessagePtr<SbeMessage>;
using Negotiate200Ptr = MessagePtr<Negotiate200>;
using Terminate203Ptr = MessagePtr<Terminate203>;
using RequestAck206Ptr = MessagePtr<RequestAck206>;
using RequestReject207Ptr = MessagePtr<RequestReject207>;

class ONIXS_CONFLATEDTCP_EXPORTED Gateway
{
public:
    explicit
    Gateway(int port, const char* host = "127.0.0.1", const std::chrono::seconds& timeout = std::chrono::seconds(10));

    virtual ~Gateway();

    Gateway(const Gateway&) = delete;
    Gateway& operator=(const Gateway&) = delete;

    Gateway(Gateway&&) noexcept;
    Gateway& operator=(Gateway&&) noexcept;

    void swap(Gateway&) noexcept;

    Gateway& acceptConnection();

    Gateway& acceptSession(SeqNumber outgoingSequenceNumber = 1, UInt64 previousUuid = 0, UInt32 previousSeqNo = 0);

    Terminate203Ptr waitUntilTerminate(ClientMessageListener* listener = nullptr);

    Gateway& disconnect();

    ONIXS_CONFLATEDTCP_NODISCARD
    int port() const noexcept;

    Gateway& enableLogger(const std::string& logFileName);

    template <typename SbeMessageType, size_t MaxMessageSize, typename MessageInitializer>
    Gateway& send(ConflatedTCP::Messaging::MessageHolder<SbeMessageType, MaxMessageSize, MessageInitializer>& msg);

    template <typename SbeMessageType, size_t MaxMessageSize, typename MessageInitializer>
    Gateway& send(ConflatedTCP::Messaging::MessageHolder<SbeMessageType, MaxMessageSize, MessageInitializer>&& msg);

    template <typename SbeMessageType, size_t MaxMessageSize, typename MessageInitializer>
    Gateway& send(ConflatedTCP::Messaging::MessageHolder<SbeMessageType, MaxMessageSize, MessageInitializer>& msg, SeqNumber num);

    template <typename SbeMessageType, size_t MaxMessageSize, typename MessageInitializer>
    Gateway& send(ConflatedTCP::Messaging::MessageHolder<SbeMessageType, MaxMessageSize, MessageInitializer>&& msg, SeqNumber num);

    template <typename TMsg>
    MessagePtr<TMsg> receiveTypedMessage();

    Gateway& run(ClientMessageListener& listener);

    ONIXS_CONFLATEDTCP_NODISCARD
    SeqNumber outSeqNum() const;

    Gateway& outSeqNum(SeqNumber nextOutSeqNum);

    ONIXS_CONFLATEDTCP_NODISCARD
    UInt64 uuid() const noexcept
    {
        return uId_;
    }

    Gateway& tcpNoDelayOption(bool noDelay);

    ONIXS_CONFLATEDTCP_NODISCARD
    bool tcpNoDelayOption() const;

    ONIXS_CONFLATEDTCP_NODISCARD
    int socketReceiveBufferSize() const;

    Gateway& socketReceiveBufferSize(int bufferSize);

    ONIXS_CONFLATEDTCP_NODISCARD
    int socketSendBufferSize() const;

    Gateway& socketSendBufferSize(int bufferSize);
 
protected:
    ONIXS_CONFLATEDTCP_NODISCARD
    bool isConnectionClosed(const std::chrono::seconds& timeout) const;

    ONIXS_CONFLATEDTCP_NODISCARD
    bool isConnectionClosed() const;

    MessageBasePtr receive() const;

    Negotiate200Ptr receiveNegotiate();

    Negotiate200Ptr acceptNegotiate(UInt32 previousSeqNo = 0);

    Negotiate200Ptr rejectNegotiate(ErrorCodes::Enum errorCodes = ErrorCodes::Enum(), const std::string & reason = std::string());

    Terminate203Ptr receiveTerminate();

    Terminate203Ptr acceptTerminate();

    Gateway& sendTerminate(const std::string& reason = {}, ErrorCodes::Enum errCode = ErrorCodes::Other);

    Terminate203Ptr terminate(const std::string& reason = {}, ErrorCodes::Enum errCode = ErrorCodes::Other);

    NegotiationResponse202Container createNegotiationResponse(const Negotiate200Ptr& request, UInt32 previousSeqNo);

    NegotiationReject201Container createNegotiationReject(const Negotiate200Ptr& request, ErrorCodes::Enum errorCodes, const std::string & reason);

    template <typename TRequestMsg>
    RequestAck206Container createRequestAck(const TRequestMsg& request);

    template <typename TRequestMsg>
    RequestReject207Container createRequestReject(const TRequestMsg& request);

    Terminate203Container createTerminate(const Terminate203Ptr& request);

    Terminate203Container createTerminate(const std::string& reason, ErrorCodes::Enum errCode);

    Gateway& send(Messaging::SbeMessage, Messaging::SimpleOpenFramingHeader&);

    Gateway& sendData(const void* data, size_t size);

    UInt64 uId_{0};
 
private:
    template<typename MessageType>
    void invokeCallback(void(ClientMessageListener::*Callback)(const MessageType &, Gateway *), ClientMessageListener * listener, const MessageBasePtr & msg)
    {
        if(listener)
            (listener->*Callback)(cast<MessageType>(msg), this);
    }
 
    class Impl;
    Impl* impl_;
};
 
template <typename TRequestMsg>
RequestAck206Container Gateway::createRequestAck(const TRequestMsg& request)
{
    MessageHolder<RequestAck206> response;
    
    response.message()
        .setReqId(request->reqId())
        .setSubscriptionReqType(request->subscriptionReqType())
        .setReqIdStatus(RequestIDStatus::FullAck);
    
    return response;
}
 
template <typename TRequestMsg>
RequestReject207Container Gateway::createRequestReject(const TRequestMsg& request)
{
    MessageHolder<RequestReject207> response;
 
    response.message()
        .setReqId(request->reqId())
        .setReqRejReason(ReqRejReason::Other);
 
    return response;
}
 
template <typename TMsg>
MessagePtr<TMsg> Gateway::receiveTypedMessage()
{
    return cast<TMsg>(receive());
}
 
template <typename SbeMessageType, size_t MaxMessageSize, typename MessageInitializer>
Gateway& Gateway::send(ConflatedTCP::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(ConflatedTCP::Messaging::MessageHolder<SbeMessageType, MaxMessageSize, MessageInitializer>&& msg)
{
    return send(msg);
}
 
template <typename SbeMessageType, size_t MaxMessageSize, typename MessageInitializer>
Gateway& Gateway::send(ConflatedTCP::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(ConflatedTCP::Messaging::MessageHolder<SbeMessageType, MaxMessageSize, MessageInitializer>&& msg, SeqNumber num)
{
    setSeqNum(msg, num);
    return send(msg);
}
 
}}}}
 
#endif