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

class ONIXS_B3_BOE_EXPORTED Session
{
public:
    static const UInt64 UndefinedSessionVerID = 0;

    Session(const SessionSettings &settings, SessionListener *listener, SessionStorageType::Enum storageType = SessionStorageType::FileBased,
        SessionStorage *storage = ONIXS_B3_BOE_NULLPTR, UInt64 sessionVerId = UndefinedSessionVerID, const std::string &customKey = "");
 
    Session(SessionReactor & stack, const SessionSettings & settings, SessionListener * listener, SessionStorageType::Enum storageType = SessionStorageType::FileBased,
        SessionStorage * storage = ONIXS_B3_BOE_NULLPTR, UInt64 sessionVerId = UndefinedSessionVerID, const std::string &customKey = "");
 
    ~Session();

    UInt64 sessionVerId() const ONIXS_B3_BOE_NOTHROW;

    std::string id() const;

    Messaging::SessionID sessionId() const ONIXS_B3_BOE_NOTHROW;

    std::string customKey() const;

    bool negotiated() const;

    Session& negotiated(bool negotiated);

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

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

    Session& disconnect();

    Threading::SharedFuture<void> disconnectAsync();

    template <typename SbeMessageType, size_t MaxMessageSize, typename MessageInitializer>
    Session& send(
        Messaging::MessageHolder<SbeMessageType, MaxMessageSize, MessageInitializer> &msg,
        Messaging::Timestamp sendingTime = UtcWatch::now());
 
#if defined (ONIXS_B3_BOE_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 B3BOEMaxPacketSize = 1420;
 
#if defined (ONIXS_B3_BOE_CXX11)

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

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

    Session& sendSequenceMessage();

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

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

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

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

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

    Session& receiveSpinningTimeout(unsigned timeoutInUs);

    unsigned sendSpinningTimeout() const ONIXS_B3_BOE_NOTHROW;

    Session& sendSpinningTimeout(unsigned timeoutInUs);

    bool reportNewMessagesWhileWaitingForMissedMessages() const;

    Session& reportNewMessagesWhileWaitingForMissedMessages(bool value);

    unsigned messageGrouping() const ONIXS_B3_BOE_NOTHROW;

    Session& messageGrouping(unsigned numberOfMessagesToGroup);

    SessionStorageType::Enum storageType() const ONIXS_B3_BOE_NOTHROW;

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

    SeqNumber outSeqNum() const ONIXS_B3_BOE_NOTHROW;

    Session& outSeqNum(SeqNumber nextOutSeqNum);

    SeqNumber inSeqNum() const;

    Session& inSeqNum(SeqNumber nextExpectedInSeqNum);

    Session& reset();

    SessionStateId::Enum state() const;

    Messaging::Timestamp creationTime() const ONIXS_B3_BOE_NOTHROW;

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

    static const Handle InvalidSocketHandle;

    Handle socketHandle();

    size_t outboundQueueBytes();

    bool getTcpInfo(TcpInfo&);

    static const std::string& version() ONIXS_B3_BOE_NOTHROW;
 
private:
    Session(const Session& );
    Session& operator=(const Session& );
 
    void send(NetworkMessage msg, Messaging::Timestamp ts);
    void send(NetMessages& msgs, size_t maxPacketSize, Messaging::Timestamp ts);
    void warmUp(NetworkMessage msg, int warmupFlags, Messaging::Timestamp ts);
    void warmUp(NetMessages& msgs, 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 sessionVerId, SeqNumber from, SeqNumber to);
 
    struct Impl;
    Impl *const impl_;
 
friend struct AdHoc::SessionHelper;
};
 
template <typename SbeMessageType, size_t MaxMessageSize, typename MessageInitializer>
ONIXS_B3_BOE_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_B3_BOE_HOTPATH
Session& Session::warmUp(
    Messaging::MessageHolder<SbeMessageType, MaxMessageSize> &msg,
    Messaging::Timestamp sendingTime,
    int warmupFlags)
{
    warmUp(msg.toNetworkMessage(), warmupFlags, sendingTime);
 
    return *this;
}
 
#if defined (ONIXS_B3_BOE_CXX11)
 
template <typename SbeMessageType, size_t MaxMessageSize, typename MessageInitializer>
ONIXS_B3_BOE_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_B3_BOE_CXX11)
 
template <typename SbeMessageType, size_t MaxMessageSize, typename MessageInitializer>
ONIXS_B3_BOE_HOTPATH
Session& Session::send(
    Messaging::MessageBatch<SbeMessageType, MaxMessageSize, MessageInitializer> & msgs,
    Messaging::Timestamp ts,
    size_t maxPacketSize)
{
    assert(ts != Messaging::Timestamp());
 
    send(msgs.netMsgs_, maxPacketSize, ts);
    return *this;
}
 
template <typename SbeMessageType, size_t MaxMessageSize, typename MessageInitializer>
ONIXS_B3_BOE_HOTPATH
Session& Session::warmUp(
    Messaging::MessageBatch<SbeMessageType, MaxMessageSize, MessageInitializer> & msgs,
    Messaging::Timestamp ts, size_t maxPacketSize, int flags)
{
    assert(ts != Messaging::Timestamp());
 
    warmUp(msgs.netMsgs_, 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
{
    validateVersion(holder->version());
}
 
}
}
}