Session.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 <limits.h>
 
#include <OnixS/CME/iLink3/ABI.h>
#include <OnixS/CME/iLink3/Defines.h>
#include <OnixS/CME/iLink3/SessionListener.h>
#include <OnixS/CME/iLink3/SessionSettings.h>
#include <OnixS/CME/iLink3/SessionStorage.h>
#include <OnixS/CME/iLink3/SessionReactor.h>
#include <OnixS/CME/iLink3/TcpInfo.h>
#include <OnixS/CME/iLink3/messaging/SbeMessage.h>
#include <OnixS/CME/iLink3/threading/Future.h>
 
namespace OnixS
{
namespace CME
{
namespace iLink3
{
// Forward declarations
class TcpDirectStack;
namespace AdHoc { struct SessionHelper; }

class ONIXS_ILINK3_EXPORTED Session
{
public:
    static const UInt64 UndefinedUuid = 0;

    Session(const SessionSettings &settings, int marketSegmentId,
        SessionListener *listener, SessionStorageType::Enum storageType = SessionStorageType::FileBased,
        SessionStorage *storage = nullptr, UInt64 uuid = UndefinedUuid, const std::string &customKey = "");
 
    Session(SessionReactor & stack, const SessionSettings & settings, int marketSegmentId,
        SessionListener * listener, SessionStorageType::Enum storageType = SessionStorageType::FileBased,
        SessionStorage * storage = nullptr, UInt64 uuid = UndefinedUuid, const std::string &customKey = "");
 
    ~Session();

    UInt64 uuid() const noexcept;

    std::string id() const;

    int marketSegmentId() const noexcept;

    std::string customKey() const;

    bool negotiated() const;

    Session& negotiated(bool negotiated);

    Messaging::FTI::Enum faultToleranceIndicator() const;

    Session& faultToleranceIndicator(Messaging::FTI::Enum value);

    Session& connect(const std::string &host, Port port);

    Threading::SharedFuture<void> connectAsync(const std::string & host, Port port);

    Session& disconnect(const std::string &reason = "");

    Threading::SharedFuture<void> disconnectAsync(const std::string & reason = "");

    Session& breakConnection();

    Threading::SharedFuture<void> breakConnectionAsync();

    template <typename SbeMessageType, size_t MaxMessageSize, typename MessageInitializer>
    Session& send(
        Messaging::MessageHolder<SbeMessageType, MaxMessageSize, MessageInitializer> &msg,
        Messaging::Timestamp sendingTime = UtcWatch::now());
 
#if defined (ONIXS_ILINK3_CXX11)

    template <typename SbeMessageType, size_t MaxMessageSize, typename MessageInitializer>
    Session& send(
        Messaging::MessageHolder<SbeMessageType, MaxMessageSize, MessageInitializer> &&msg,
        Messaging::Timestamp sendingTime = UtcWatch::now());
 
#endif

    static const size_t CmeMaxPacketSize = 1420;
 
#if defined (ONIXS_ILINK3_CXX11)

    template <typename SbeMessageType, size_t MaxMessageSize, typename MessageInitializer>
    Session& send(
        MessageBatch<SbeMessageType, MaxMessageSize, MessageInitializer> & msgs,
        Messaging::Timestamp sendingTime = UtcWatch::now(),
        size_t maxPacketSize = CmeMaxPacketSize);

    Session& send(
        MessageBatchCombiner & msgs,
        Messaging::Timestamp sendingTime = UtcWatch::now(),
        size_t maxPacketSize = CmeMaxPacketSize);
 
#endif

    Session& send(
            NetworkMessage* begin, NetworkMessage* end,
            Messaging::Timestamp sendingTime = UtcWatch::now(),
            size_t maxPacketSize = CmeMaxPacketSize) ONIXS_ILINK3_NONULL;

    Session& send(
            NetMessagesRange range,
            Messaging::Timestamp sendingTime = UtcWatch::now(),
            size_t maxPacketSize = CmeMaxPacketSize);

    Session& sendSequenceMessage(Messaging::KeepAliveLapsed::Enum keepAliveLapsed = Messaging::KeepAliveLapsed::Lapsed);

    template <typename SbeMessageType, size_t MaxMessageSize>
    Session& warmUp(
        Messaging::MessageHolder<SbeMessageType, MaxMessageSize> &msg,
        Messaging::Timestamp sendingTime = UtcWatch::now(),
        int warmupFlags = 0);
 
 
#if defined (ONIXS_ILINK3_CXX11)

    template <typename SbeMessageType, size_t MaxMessageSize, typename MessageInitializer>
    Session& warmUp(
        MessageBatch<SbeMessageType, MaxMessageSize, MessageInitializer> & msgs,
        Messaging::Timestamp sendingTime = UtcWatch::now(),
        size_t maxPacketSize = CmeMaxPacketSize,
        int warmupFlags = 0);

    Session& warmUp(
        MessageBatchCombiner & msgs, Messaging::Timestamp sendingTime = UtcWatch::now(),
        size_t maxPacketSize = CmeMaxPacketSize, int warmupFlags = 0);
#endif

    Session& warmUp(
            NetworkMessage* begin, NetworkMessage* end, Messaging::Timestamp sendingTime = UtcWatch::now(),
            size_t maxPacketSize = CmeMaxPacketSize, int warmupFlags = 0) ONIXS_ILINK3_NONULL;

    Session& warmUp(
            NetMessagesRange range, Messaging::Timestamp sendingTime = UtcWatch::now(),
            size_t maxPacketSize = CmeMaxPacketSize, int warmupFlags = 0);
 

    Session& throttle();

    size_t tryThrottle();

    Session & throttlingLimit(size_t messagesCount, size_t intervalInMs = 1000);

    std::string remoteHost() const;

    std::string remoteIpAddress() const;

    Port remotePort() const;

    std::pair<Port, Port> localPortRange() const noexcept;

    Session& localPortRange(std::pair<Port, Port> portRange);

    std::string localNetworkInterface() const;

    Session& localNetworkInterface(const std::string &value);

    Port localPort() const;

    bool tcpNoDelayOption() const;

    Session& tcpNoDelayOption(bool value);

    const SocketOptions &socketOptions() const;

    Session& socketOptions(const SocketOptions &options);
 
    static const int UndefinedAffinity = -1;

    Session& receivingThreadAffinity(CpuIndex cpuIndex);

    Session& receivingThreadAffinity(const CpuIndexes &cpuIndexes);

    const CpuIndexes &receivingThreadAffinity() const;

    Session& sendingThreadAffinity(CpuIndex cpuIndex);

    Session& sendingThreadAffinity(const CpuIndexes &cpuIndexes);

    const CpuIndexes &sendingThreadAffinity() const;

    int receivingThreadPriority() const;

    Session& receivingThreadPriority(int priority);

    int sendingThreadPriority() const;

    Session& sendingThreadPriority(int priority);

    int receivingThreadPolicy() const;

    Session& receivingThreadPolicy(int policy);

    int sendingThreadPolicy() const;

    Session& sendingThreadPolicy(int policy);

    unsigned receiveSpinningTimeout() const noexcept;

    Session& receiveSpinningTimeout(unsigned timeoutInUs);

    unsigned sendSpinningTimeout() const noexcept;

    Session& sendSpinningTimeout(unsigned timeoutInUs);

    bool reportNewMessagesWhileWaitingForMissedMessages() const;

    Session& reportNewMessagesWhileWaitingForMissedMessages(bool value);

    unsigned messageGrouping() const noexcept;

    Session& messageGrouping(unsigned numberOfMessagesToGroup);

    ThreadingModel::Enum threadingModel() const;

    SessionStorageType::Enum storageType() const noexcept;

    const std::string &storageId() const noexcept;

    SeqNumber outSeqNum() const noexcept;

    Session& outSeqNum(SeqNumber nextOutSeqNum);

    SeqNumber inSeqNum() const;

    Session& inSeqNum(SeqNumber nextExpectedInSeqNum);

    SeqNumber previousSeqNo() const noexcept;

    Session& previousSeqNo(SeqNumber value);

    Messaging::UInt64 previousUuid() const noexcept;

    Session& previousUuid(Messaging::UInt64 value);

    Session& reset(bool startOfWeek = false);

    SessionStateId::Enum state() const;

    Messaging::Timestamp creationTime() const noexcept;

    void flushSessionStorage();

    unsigned reconnectAttempts() const;

    Session& reconnectAttempts(unsigned reconnectAttempts);

    unsigned reconnectInterval() const;

    Session& reconnectInterval(unsigned seconds);

    bool localTimeUsage() const;

    Session& localTimeUsage(bool useLocalTime);

    const std::string &storageDirectory() const;

    bool logBeforeSending() const;

    Session& logBeforeSending(bool value);

    size_t incomingMessageGapQueueMaximumSize() const;

    Session& incomingMessageGapQueueMaximumSize(size_t value);

    const Session& log(const std::string &message) const;

    std::string toString() const;

    Messaging::Timestamp licenseExpirationDate() const;

    Messaging::SchemaVersion messagingVersion() const noexcept;

    bool detectMessageGapForPreviousUuid() const;

    Session& detectMessageGapForPreviousUuid(bool detectGap);

    static const Handle InvalidSocketHandle;

    Handle socketHandle();

    size_t outboundQueueBytes();

    bool getTcpInfo(TcpInfo&);

    static std::string version();
 
private:
    Session(const Session& );
    Session& operator=(const Session& );
 
    void send(NetworkMessage msg, Messaging::Timestamp ts);
    void send(NetworkMessage* begin, NetworkMessage* end, size_t maxPacketSize, Messaging::Timestamp ts);
    void warmUp(NetworkMessage msg, int warmupFlags, Messaging::Timestamp ts);
    void warmUp(NetworkMessage* begin, NetworkMessage* end, size_t maxPacketSize, int warmupFlags, Messaging::Timestamp ts);
 
    template <typename SbeMessageType, size_t MaxMessageSize, typename MessageInitializer>
    void validate(const Messaging::MessageHolder<SbeMessageType, MaxMessageSize, MessageInitializer> &msg) const;
 
    void validateVersion(Messaging::SchemaVersion version) const;
 
    Session& sendRetransmitRequest(UInt64 uuid, SeqNumber from, SeqNumber to);
 
    struct Impl;
    Impl *const impl_;
 
friend struct AdHoc::SessionHelper;
};

class ONIXS_ILINK3_EXPORTED CgwSession : public Session
{
public:
    CgwSession(const SessionSettings &settings,
               SessionListener *listener, SessionStorageType::Enum storageType = SessionStorageType::FileBased,
               SessionStorage *storage = nullptr, UInt64 uuid = UndefinedUuid, const std::string &customKey = "");

    CgwSession(SessionReactor & stack, const SessionSettings & settings,
               SessionListener * listener, SessionStorageType::Enum storageType = SessionStorageType::FileBased,
               SessionStorage * storage = nullptr, UInt64 uuid = UndefinedUuid, const std::string &customKey = "");
 
private:
    int marketSegmentId() const noexcept;
    static const int marketSegmentId_ = INT_MAX;
};
 
template <typename SbeMessageType, size_t MaxMessageSize, typename MessageInitializer>
ONIXS_ILINK3_HOTPATH
Session& Session::send(Messaging::MessageHolder<SbeMessageType, MaxMessageSize, MessageInitializer> &msg, Messaging::Timestamp ts)
{
    assert(ts != Messaging::Timestamp());
 
#ifndef NDEBUG
    validate(msg);
#endif
 
    send(msg.toNetworkMessage(), ts);
 
    return *this;
}
 
template <typename SbeMessageType, size_t MaxMessageSize>
ONIXS_ILINK3_HOTPATH
Session& Session::warmUp(
    Messaging::MessageHolder<SbeMessageType, MaxMessageSize> &msg,
    Messaging::Timestamp sendingTime,
    int warmupFlags)
{
    warmUp(msg.toNetworkMessage(), warmupFlags, sendingTime);
    return *this;
}
 
#if defined (ONIXS_ILINK3_CXX11)
 
template <typename SbeMessageType, size_t MaxMessageSize, typename MessageInitializer>
ONIXS_ILINK3_HOTPATH
Session& Session::send(Messaging::MessageHolder<SbeMessageType, MaxMessageSize, MessageInitializer> &&msg, Messaging::Timestamp ts)
{
    assert(ts != Messaging::Timestamp());
 
    send(msg, ts);
    return *this;
}
 
#endif
 
#if defined (ONIXS_ILINK3_CXX11)
 
template <typename SbeMessageType, size_t MaxMessageSize, typename MessageInitializer>
ONIXS_ILINK3_HOTPATH
Session& Session::send(
    Messaging::MessageBatch<SbeMessageType, MaxMessageSize, MessageInitializer> & msgs,
    Messaging::Timestamp ts,
    size_t maxPacketSize)
{
    assert(ts != Messaging::Timestamp());
 
    if ONIXS_ILINK3_UNLIKELY(msgs.netMsgs_.empty()) {
        return *this;
    }
 
    NetworkMessage* const begin = &*msgs.netMsgs_.begin();
    NetworkMessage* const end = begin + msgs.netMsgs_.size();
 
    send(begin, end, maxPacketSize, ts);
    return *this;
}
 
template <typename SbeMessageType, size_t MaxMessageSize, typename MessageInitializer>
ONIXS_ILINK3_HOTPATH
Session& Session::warmUp(
    Messaging::MessageBatch<SbeMessageType, MaxMessageSize, MessageInitializer> & msgs,
    Messaging::Timestamp ts, size_t maxPacketSize, int flags)
{
    assert(ts != Messaging::Timestamp());
 
    if ONIXS_ILINK3_UNLIKELY(msgs.netMsgs_.empty())
    {
        return *this;
    }
 
    NetworkMessage* const begin = &*msgs.netMsgs_.begin();
    NetworkMessage* const end = begin + msgs.netMsgs_.size();
 
    warmUp(begin, end, maxPacketSize, flags, ts);
    return *this;
}
 
#endif
 
template <typename SbeMessageType, size_t MaxMessageSize, typename MessageInitializer>
void Session::validate(const Messaging::MessageHolder<SbeMessageType, MaxMessageSize, MessageInitializer> &holder) const
{
    Messaging::validate(holder);
    validateVersion(holder->version());
}
 
}
}
}