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 <string>
#include <set>
#include <vector>
#include <utility>
#include <iosfwd>
#include <limits>
 
#include <OnixS/FIXEngine/Compiler.h>
#include <OnixS/FIXEngine/FIX/Engine.h>
#include <OnixS/FIXEngine/Cryptography/SecureString.h>
 
 
#include <OnixS/FIXEngine/Sockets/Definitions.h>
 
#include <OnixS/FIXEngine/FIX/SessionState.h>
#include <OnixS/FIXEngine/FIX/SessionStorageType.h>
#include <OnixS/FIXEngine/FIX/SessionRole.h>
#include <OnixS/FIXEngine/FIX/ThreadingModel.h>
#include <OnixS/FIXEngine/FIX/EncryptionMethod.h>
#include <OnixS/FIXEngine/FIX/SessionSendMode.h>
#include <OnixS/FIXEngine/FIX/SequenceNumber.h>
#include <OnixS/FIXEngine/FIX/ISessionReactor.h>
 
#include <OnixS/FIXEngine/FIX/Timestamp.h>
 
#include <OnixS/FIXEngine/FIX/FlatMessage.h>
 
#include <OnixS/FIXEngine/Threading/Thread.h>
#include <OnixS/FIXEngine/Threading/Future.h>
 
 
namespace OnixS {
namespace FIX {
ONIXS_FIXENGINE_API_DECL(class, Message);
ONIXS_FIXENGINE_API_DECL(class, Dictionary);
ONIXS_FIXENGINE_API_DECL(class, Engine);
ONIXS_FIXENGINE_API_DECL(class, ISessionListener);
ONIXS_FIXENGINE_API_DECL(class, ISessionStorage);

struct MessageMode {
    enum Enum {
        Message = 1,

        FlatMessage
    };
};
 
namespace Core {
namespace Messaging {
class Message;
}
}
 
template<typename MsgType>
struct MsgBatchTraits
{
    typedef typename PtrTraits<MsgType>::UniquePtr MessagePtr;
    typedef std::vector<FlatMessage*> MsgBatchType;
    typedef std::vector<OnixS::FIX::Core::Messaging::Extras::FlatMessage*> CoreMsgBatchType;
};
 
template<>
struct MsgBatchTraits<Message>
{
    typedef PtrTraits<Message>::UniquePtr MessagePtr;
    typedef std::vector<Message*> MsgBatchType;
    typedef std::vector<OnixS::FIX::Core::Messaging::Message*> CoreMsgBatchType;
};

class ONIXS_FIXENGINE_API Session
{
public:
    Session(
        const std::string & senderCompId,
        const std::string & targetCompId,
        const Dictionary & dictionary,
        ISessionListener * listener,
        SessionStorageType::Enum storageType = SessionStorageType::FileBased,
        ISessionStorage * storage = ONIXS_FIXENGINE_NULLPTR);

    Session(
        ISessionReactor * reactor,
        const std::string & senderCompId,
        const std::string & targetCompId,
        const Dictionary & dictionary,
        ISessionListener * listener,
        SessionStorageType::Enum storageType = SessionStorageType::FileBased,
        ISessionStorage * storage = ONIXS_FIXENGINE_NULLPTR);

    Session(
        const std::string & senderCompId,
        const std::string & targetCompId,
        const Dictionary & dictionary,
        bool keepSequenceNumbersAfterLogout,
        OnixS::FIX::ISessionListener * listener,
        SessionStorageType::Enum storageType = SessionStorageType::FileBased,
        ISessionStorage * storage = ONIXS_FIXENGINE_NULLPTR);
 

    Session(
        ISessionReactor * reactor,
        const std::string & senderCompId,
        const std::string & targetCompId,
        const Dictionary & dictionary,
        bool keepSequenceNumbersAfterLogout,
        OnixS::FIX::ISessionListener * listener,
        SessionStorageType::Enum storageType = SessionStorageType::FileBased,
        ISessionStorage * storage = ONIXS_FIXENGINE_NULLPTR);

    Session(
        const std::string & senderCompId,
        const std::string & targetCompId,
        const Dictionary & dictionary,
        bool keepSequenceNumbersAfterLogout,
        OnixS::FIX::ISessionListener * listener,
        const std::string & customSessionKey,
        SessionStorageType::Enum storageType = SessionStorageType::FileBased,
        ISessionStorage * storage = ONIXS_FIXENGINE_NULLPTR);

    Session(
        ISessionReactor * reactor,
        const std::string & senderCompId,
        const std::string & targetCompId,
        const Dictionary & dictionary,
        bool keepSequenceNumbersAfterLogout,
        OnixS::FIX::ISessionListener * listener,
        const std::string & customSessionKey,
        SessionStorageType::Enum storageType = SessionStorageType::FileBased,
        ISessionStorage * storage = ONIXS_FIXENGINE_NULLPTR);

    ~Session();

    Session & clearOutboundQueue(bool removeFragmentedPacket = false);

    SessionState::Enum state() const;

    EncryptionMethod::Enum encryptionMethod() const;

    Session & encryptionMethod(EncryptionMethod::Enum newEncryptionMethod);

    int listenPort() const;

    Session & listenPort(int listenPort);

    std::string localNetworkInterface() const;

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

    unsigned short localPort() const;
    typedef std::pair<unsigned short, unsigned short> LocalPortRange;

    LocalPortRange localPortRange() const;

    Session & localPortRange(LocalPortRange portRange);

    bool tcpNoDelayOption() const;

    Session & tcpNoDelayOption(bool improveLatency = true);

    bool sendLogoutOnException() const;

    Session & sendLogoutOnException(bool sendLogoutOnException);

    bool validateUnknownMessages() const;

    Session & validateUnknownMessages(bool);

    bool validateUnknownFields() const;

    Session & validateUnknownFields(bool);

    bool validateRequiredFields() const;

    Session & validateRequiredFields(bool);

    bool validateFieldValues() const;

    Session & validateFieldValues(bool);

    bool validateEmptyFieldValues() const;

    Session & validateEmptyFieldValues(bool);

    bool validateRepeatingGroupEntryCount() const;

    Session & validateRepeatingGroupEntryCount(bool);

    bool validateRepeatingGroupLeadingTag() const;

    Session & validateRepeatingGroupLeadingTag(bool);

    bool validateDuplicatedField() const;

    Session & validateDuplicatedField(bool);

    bool validateChecksum() const;

    Session & validateChecksum(bool);

    unsigned messageGrouping() const;

    Session & messageGrouping(unsigned numberOfMessagesToGroup);

    ThreadingModel::Enum threadingModel() const;

    Session & threadingModel(ThreadingModel::Enum value);

    bool useSpinLock() const;

    Session & useSpinLock(bool value);

    int reasonableTransmissionTime() const;

    Session & reasonableTransmissionTime(int value);

    Session & reconnectAttempts(int value);

    int reconnectAttempts() const;

    Session & reconnectInterval(int value);

    int reconnectInterval() const;

    Session & receiveSpinningTimeoutUsec(int usec);

    int receiveSpinningTimeout() const;

    Session & sendSpinningTimeoutUsec(int usec);

    int sendSpinningTimeout() const;

    Session & supportNextExpectedMsgSeqNum(bool support);

    bool supportNextExpectedMsgSeqNum() const;

    std::string sslCertificateFile() const;

    Session & sslCertificateFile(const std::string & file);

    std::string sslPrivateKeyFile() const;

    Session & sslPrivateKeyFile(const std::string & file);

    Cryptography::SecureString sslPrivateKeyPassword() const;

    Session & sslPrivateKeyPassword(const Cryptography::SecureString & password);

    Session & sslVerifyPeer(bool verify);

    bool sslVerifyPeer() const;

    Session & sslClientSni(const std::string & hostName);

    const std::string & sslClientSni() const;

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

    std::string sslCaFile() const;

    Session & sslMinProtocolVersion(SslProtocolMinMaxVersion::Enum version);

    SslProtocolMinMaxVersion::Enum sslMinProtocolVersion() const;

    Session & sslMaxProtocolVersion(SslProtocolMinMaxVersion::Enum version);

    SslProtocolMinMaxVersion::Enum sslMaxProtocolVersion() const;

    Session & logonAsAcceptor();

    Session & logonAsInitiator(const std::string & host, int port, bool setResetSeqNumFlag);

    Session & logonAsInitiator(const std::string & host, int port);

    Session & logonAsInitiator(const std::string & host, int port, int heartbeatIntervalSec);

    Session & logonAsInitiator(const std::string & host, int port, int heartbeatIntervalSec, bool setResetSeqNumFlag);

    Session & logonAsInitiator(const std::string & host, int port, int heartbeatIntervalSec, Message * customLogonMsg);

    Session & logonAsInitiator(const std::string & host, int port, int heartbeatIntervalSec, Message * customLogonMsg, bool setResetSeqNumFlag);

    Threading::SharedFuture<void> logonAsInitiatorAsync(const std::string & host, int port, int heartbeatIntervalSec, Message * customLogonMsg, bool setResetSeqNumFlag);

    Session & logout();

    Session & logout(const std::string & text);

    Threading::SharedFuture<void> logoutAsync(const std::string & text);

    Session & logout(Message * customLogoutMessage);
    Session & logout(FlatMessage * customLogoutMessage);

    Threading::SharedFuture<void> logoutAsync(Message * customLogoutMessage);
    Threading::SharedFuture<void> logoutAsync(FlatMessage * customLogoutMessage);

    Session & breakConnection();

    Session & send(Message * msg);
    Session & send(FlatMessage * msg);

    Session & send(FlatMessage * msg, SessionSendMode::Enum mode);

    template<typename MsgType> class MsgBatch;
 
    class ONIXS_FIXENGINE_API MsgBatchHelper
    {
        typedef MsgBatchTraits<Message> MBT;
        typedef MsgBatchTraits<FlatMessage> SBT;
 
        static void init(MBT::CoreMsgBatchType & coreBatch);
        static void init(SBT::CoreMsgBatchType & coreBatch);
        static void add(MBT::MsgBatchType & batch, MBT::CoreMsgBatchType & coreBatch, MBT::MessagePtr & msg);
        static void add(SBT::MsgBatchType & batch, SBT::CoreMsgBatchType & coreBatch, SBT::MessagePtr & msg);
        static void clear(MBT::MsgBatchType & batch);
        static void clear(SBT::MsgBatchType & batch);
 
        template<typename MsgType> friend class MsgBatch;
    };
 
    template<typename MsgType>
    class MsgBatch
    {
        typedef MsgBatchTraits<MsgType> Traits;
 
        friend class Session;
 
        typename Traits::MsgBatchType batch_;
        typename Traits::CoreMsgBatchType coreBatch_;
 
        MsgBatch(const MsgBatch&);
        MsgBatch & operator=(const MsgBatch&);
 
    public:
 
        typedef typename Traits::MessagePtr MessagePtr;
        typedef typename Traits::MsgBatchType::iterator iterator;
        typedef typename Traits::MsgBatchType::const_iterator const_iterator;
 
        MsgBatch() { MsgBatchHelper::init(coreBatch_); }
        ~MsgBatch() { MsgBatchHelper::clear(batch_); }
 
        void add(MessagePtr & msg) { MsgBatchHelper::add(batch_, coreBatch_, msg); }
 
        MsgType & operator[](size_t index) { return *batch_[index]; }
        const MsgType & operator[](size_t index) const { return *batch_[index]; }
 
        size_t size() const { return batch_.size(); }
 
        const_iterator begin() const { return batch_.begin(); }
        iterator begin() { return batch_.begin(); }
 
        const_iterator end() const { return batch_.end(); }
        iterator end() { return batch_.end(); }
    };
 
    typedef MsgBatch<Message> MessageBatch;
    typedef MsgBatch<FlatMessage> FlatMessageBatch;
    typedef FlatMessageBatch SerializedMessageBatch;

    Session & send(MessageBatch & msgs, size_t maxPacketSize = (std::numeric_limits<size_t>::max)());
    Session & send(FlatMessageBatch & msgs, size_t maxPacketSize = (std::numeric_limits<size_t>::max)());

    Session & send(FlatMessageBatch & msgs, SessionSendMode::Enum mode, size_t maxPacketSize = (std::numeric_limits<size_t>::max)());

    Session & sendAsIs(FlatMessage * msg);

    Session & sendAsIs(FlatMessageBatch & msgs, size_t maxPacketSize = (std::numeric_limits<size_t>::max)());

    Session & throttle();

    size_t tryThrottle(bool resetWhenDelay = false);

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

    Session & warmUp(FlatMessage* msg, int warmupFlags = 0);

    static bool checkWarmupFlags(int warmupFlags, unsigned short baseListenPort = 5000, const std::string & localNetworkInterface = "127.0.0.1");

    void preFill(Message & msg) const;

    void preFill(FlatMessage & msg) const;
    void preFill(FlatMessageBatch & msgs) const;

    Session & shutdown();

    operator const std::string & () const;

    const std::string & storageId() const;

    Session & flushSessionStorage(bool syncData = false);

    std::string counterpartyHost() const;

    std::string counterpartyIpAddress() const;

    size_t counterpartyPort() const;

    const std::string & customKey() const;

    size_t outboundQueueBytes() const;

    int heartbeatIntervalSec() const;

    bool logInboundMessages() const;

    Session & logInboundMessages(bool value);

    bool logOutboundMessages() const;

    Session & logOutboundMessages(bool value);

    typedef std::set<std::string> InboundMessageLogFilter;

    InboundMessageLogFilter inboundMessageLogFilter() const;

    Session & inboundMessageLogFilter(const InboundMessageLogFilter & filter);

    typedef std::set<std::string> OutboundMessageLogFilter;

    OutboundMessageLogFilter outboundMessageLogFilter() const;

    Session & outboundMessageLogFilter(const OutboundMessageLogFilter & filter);

    bool logBeforeSending() const;

    Session & logBeforeSending(bool value);

    SequenceNumber inSeqNum() const;

    Session & inSeqNum(SequenceNumber seqNum);

    SequenceNumber outSeqNum() const;

    Session & outSeqNum(SequenceNumber seqNum);

    static const unsigned int ResendRequestMaximumRangeNoLimit;

    Session & resendRequestMaximumRange(int range);

    int resendRequestMaximumRange() const;

    Session & resendingQueueSize(size_t value);

    size_t resendingQueueSize() const;

    Session & resendingBatchSize(size_t value);

    size_t resendingBatchSize() const;

    SessionRole::Enum role() const;

    const std::string & senderCompId() const;

    bool specifyLastMsgSeqNumProcessed() const;

    Session & specifyLastMsgSeqNumProcessed(bool specify);

    bool updateSendingTimeField() const;

    Session & updateSendingTimeField(bool specify);

    Session & sendingTimeFormat(TimestampFormat::Enum format);

    const std::string & targetCompId() const;

    Dictionary dictionary() const;

    const OnixS::Threading::CpuIndexes & receivingThreadAffinity() const;

    Session & receivingThreadAffinity(const OnixS::Threading::CpuIndexes & cpuIndexes);

    Session & receivingThreadAffinity(const OnixS::Threading::CpuIndex cpuIndex);

    const OnixS::Threading::CpuIndexes & sendingThreadAffinity() const;

    Session & sendingThreadAffinity(const OnixS::Threading::CpuIndexes & cpuIndexes);

    Session & sendingThreadAffinity(const OnixS::Threading::CpuIndex cpuIndex);

    static const int UndefinedPriorityAndPolicy;

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

    Message * findSentMessage(SequenceNumber messageSequenceNumber);
    FlatMessage * findSentFlatMessage(SequenceNumber messageSequenceNumber);

    MsgBatchTraits<Message>::MsgBatchType findSentMessages(SequenceNumber beginSequenceNumber, SequenceNumber endSequenceNumber);
    MsgBatchTraits<FlatMessage>::MsgBatchType findSentFlatMessages(SequenceNumber beginSequenceNumber, SequenceNumber endSequenceNumber);

    Session & reportNewMessagesWhileWaitingForMissedMessages(bool report);

    bool reportNewMessagesWhileWaitingForMissedMessages() const;

    Session & sendResendRequestOnEachMessageGap(bool send);

    bool sendResendRequestOnEachMessageGap() const;

    Session & requestOnlyMissedMessages(bool request);

    bool requestOnlyMissedMessages() const;

    Session & considerRejectOnResendRequestAsGapFill(bool consider);

    bool considerRejectOnResendRequestAsGapFill() const;

    Session & ignoreLessThanExpectedSequenceNumber(bool ignore);

    bool ignoreLessThanExpectedSequenceNumber() const;

    Session & resetLocalSequenceNumbersOnLogon(bool reset);

    bool resetLocalSequenceNumbersOnLogon() const;

    Session & incomingMessageGapQueueMaximumSize(size_t maxSize);

    size_t incomingMessageGapQueueMaximumSize() const;

    Session & registerListener(ISessionListener * listener);

    Session & unregisterListener(ISessionListener * listener);

    Session & unregisterAllListeners();

    Session & resetLocalSequenceNumbers();

    Session & resetSeqNumViaLogonExchange();

    std::string senderSubId() const;

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

    std::string targetSubId() const;

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

    std::string senderLocationId() const;

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

    std::string targetLocationId() const;

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

    Session & validateSequenceNumbers(bool value);

    bool validateSequenceNumbers() const;

    std::string threadNameSuffix() const;

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

    Session & sendTestRequest(const std::string & testReqId = "", const TimeSpan & timeout = TimeSpan::zero());

    Session & sendResendRequest(SequenceNumber beginSeqNumber);

    Session & sendReject(SequenceNumber refSeqNumber, const std::string & text);

    Timestamp creationTime() const;

    std::string toString() const;

    static const OnixS::Sockets::Handle InvalidSocketHandle;

    OnixS::Sockets::Handle socketHandle();

    Session & socketOptions(const OnixS::Sockets::SocketOptions & options);

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

    const std::string & logDirectory() const;

    Session & resendTimeout(const TimeSpan & timeout);

    TimeSpan resendTimeout() const;

    typedef std::vector<Tag> ScrambledFields;

    Session & scrambleLogonFields(const ScrambledFields & fields);

    ScrambledFields scrambleLogonFields() const;

    Session & sendQueueMaxSize(size_t size);

    size_t sendQueueMaxSize() const;

    Session & connectTimeout(const TimeSpan & timeout);

    TimeSpan connectTimeout() const;

    int receiveBufferSize() const;

    Session & receiveBufferSize(int value);

    int sendBufferSize() const;

    Session & sendBufferSize(int value);

    bool enableRxTimestamp() const;

    Session & enableRxTimestamp(bool value);

    void scheduleShrinkToFit();

    struct ProxySettings
    {
        ProxySettings()
            :host(), port(), username(), password()
        {}
 
        ProxySettings(const std::string & proxyHost, int proxyPort, const std::string & proxyUsername = std::string(), const std::string & proxyPassword = std::string())
            :host(proxyHost), port(proxyPort), username(proxyUsername), password(proxyPassword)
        {}
 
        std::string host;
        int port;
        std::string username;
        std::string password;
    };

    Session & proxySettings(const ProxySettings & settings);

    ProxySettings proxySettings() const;

    MessageMode::Enum messageMode() const;

    void messageMode(MessageMode::Enum mode);
 
    struct Impl; 
 
private:
    void init(
        const std::string & senderCompId,
        const std::string & targetCompId,
        const Dictionary & dictionary,
        ISessionListener * listener,
        bool keepSequenceNumbersAfterLogout,
        SessionStorageType::Enum storageType,
        ISessionStorage * storage,
        const std::string & customSessionKey,
        ISessionReactor * reactor);
 
    Impl * impl_;
 
    friend class Engine::Impl;
};

ONIXS_FIXENGINE_API
std::ostream & operator << (std::ostream & os, const Session & session);
}
}