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/ICE/BOE/ABI.h>
#include <OnixS/ICE/BOE/Defines.h>
#include <OnixS/ICE/BOE/SessionListener.h>
#include <OnixS/ICE/BOE/BgwCredentialsProvider.h>
#include <OnixS/ICE/BOE/SessionSettings.h>
#include <OnixS/ICE/BOE/SessionStorage.h>
#include <OnixS/ICE/BOE/SessionReactor.h>
#include <OnixS/ICE/BOE/ServiceFactory.h>
#include <OnixS/ICE/BOE/TcpInfo.h>
#include <OnixS/ICE/BOE/messaging/SbeMessage.h>
#include <OnixS/ICE/BOE/threading/Future.h>
#include <OnixS/ICE/BOE/Tools.h>
 
ONIXS_ICEBOE_NAMESPACE_BEGIN
 
// Forward declarations
class TcpDirectStack;
class ServiceFactory;
 
namespace AdHoc { struct SessionHelper; }

struct SessionType
{
    enum Enum
    {
        BUS,

        BGW,
    };

    ONIXS_ICEBOE_EXPORTED static const std::string toString(Enum state) noexcept;
};
 
inline std::ostream& operator<<(std::ostream& o, SessionType::Enum type)
{
    return o << SessionType::toString(type);
}

class ONIXS_ICEBOE_EXPORTED Session
{
public:
    Session(SessionType::Enum sessionType, const SessionSettings &settings,
        SessionListener *listener, SessionStorageType::Enum storageType = SessionStorageType::FileBased,
        SessionStorage *storage = nullptr, const std::string &customKey = {}, const Utils::BuildInfo & = Utils::consumerBuildInfo());

    Session(SessionReactor & stack, SessionType::Enum sessionType, const SessionSettings & settings,
        SessionListener * listener, SessionStorageType::Enum storageType = SessionStorageType::FileBased,
        SessionStorage * storage = nullptr, const std::string &customKey = {}, const Utils::BuildInfo & = Utils::consumerBuildInfo());
 
    Session(ServiceFactory & service, SessionType::Enum sessionType, const SessionSettings & settings,
        SessionListener * listener, SessionStorageType::Enum storageType = SessionStorageType::FileBased,
        SessionStorage * storage = nullptr, const std::string &customKey = {}, const Utils::BuildInfo & = Utils::consumerBuildInfo());
 
    ~Session();

    std::string id() const;

    int clientId() const noexcept;

    std::string userId() const noexcept;

    std::string customKey() const;

    Session& connect(const std::string& host, Port port, const std::string& ipSessionToken = {});

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

    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());

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

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

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

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

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

    Session& sendSequenceMessage();

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

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

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

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

    Session& warmUp(
            NetMessagesRange range, Messaging::Timestamp sendingTime = UtcWatch::now(),
            size_t maxPacketSize = MaxPacketSize, 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);

    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);

    Session& reset();

    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;

    static const Handle InvalidSocketHandle;

    Handle socketHandle();

    size_t outboundQueueBytes();

    bool getTcpInfo(TcpInfo&);

    static const char* version() noexcept;
 
protected:
    Threading::SharedFuture<BgwCredentials> getBgwCredentialsAsync(const std::string& userId, bool forceRequest);
 
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(SeqNumber from, SeqNumber to);
 
    struct Impl;
    Impl *const impl_;
 
friend struct AdHoc::SessionHelper;
friend struct SessionHelper;
};
 
template <typename SbeMessageType, size_t MaxMessageSize, typename MessageInitializer>
ONIXS_ICEBOE_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_ICEBOE_HOTPATH
Session& Session::warmUp(
    Messaging::MessageHolder<SbeMessageType, MaxMessageSize> &msg,
    Messaging::Timestamp sendingTime,
    int warmupFlags)
{
    warmUp(msg.toNetworkMessage(), warmupFlags, sendingTime);
    return *this;
}
 
template <typename SbeMessageType, size_t MaxMessageSize, typename MessageInitializer>
ONIXS_ICEBOE_HOTPATH
Session& Session::send(Messaging::MessageHolder<SbeMessageType, MaxMessageSize, MessageInitializer> &&msg, Messaging::Timestamp ts)
{
    assert(ts != Messaging::Timestamp());
 
    send(msg, ts);
    return *this;
}
 
template <typename SbeMessageType, size_t MaxMessageSize, typename MessageInitializer>
ONIXS_ICEBOE_HOTPATH
Session& Session::send(
    Messaging::MessageBatch<SbeMessageType, MaxMessageSize, MessageInitializer> & msgs,
    Messaging::Timestamp ts,
    size_t maxPacketSize)
{
    assert(ts != Messaging::Timestamp());
 
    if ONIXS_ICEBOE_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_ICEBOE_HOTPATH
Session& Session::warmUp(
    Messaging::MessageBatch<SbeMessageType, MaxMessageSize, MessageInitializer> & msgs,
    Messaging::Timestamp ts, size_t maxPacketSize, int flags)
{
    assert(ts != Messaging::Timestamp());
 
    if ONIXS_ICEBOE_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;
}
 
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());
}
 
ONIXS_ICEBOE_NAMESPACE_END