Pluggable Storage Sample

Memory-based session storage is typically used to maintain a high-performance FIX session when persisting of a sessions state and messages to the file system is not required. This sample demonstrates how to use pluggable session storage.

Source code

• MySessionStorage.h
• MySessionStorage.cpp
• PluggableStorage.cpp

MySessionStorage.h

#include <OnixS/FIXEngine/FIX/ISessionStorage.h>
 
#include <map>
 
class MySessionStorage : public OnixS::FIX::ISessionStorage
{
public:
    MySessionStorage();
 
    void clear() override;
 
    void close(bool keepSequenceNumbers, bool doBackup) override;
 
    void getOutbound(OnixS::FIX::SequenceNumber beginSequenceNumber, OnixS::FIX::SequenceNumber endSequenceNumber, OnixS::FIX::ISessionStorage::ISessionStorageListener * listener) override;
 
    OnixS::FIX::SequenceNumber inSeqNum() override;
 
    void inSeqNum(OnixS::FIX::SequenceNumber messageSequenceNumber) override;
 
    void storeInbound(const OnixS::FIX::Message & message, OnixS::FIX::SequenceNumber messageSequenceNumber, const OnixS::FIX::ISessionStorage::RawMessagePointer& pointer, bool logMessage) override;
 
    void storeInbound(const OnixS::FIX::FlatMessage & message, OnixS::FIX::SequenceNumber messageSequenceNumber, const OnixS::FIX::ISessionStorage::RawMessagePointer & pointer, bool logMessage) override;
 
    using RawMessage = std::vector<char>;
 
    using MessageSequenceNumber2RawMessage = std::map<OnixS::FIX::SequenceNumber, RawMessage>;
 
    MessageSequenceNumber2RawMessage outgoingMessages_;
 
    void storeOutbound(const OnixS::FIX::Message &, const OnixS::FIX::ISessionStorage::RawMessagePointer & pointer, bool logMessage) override;
 
    void storeOutbound(const OnixS::FIX::FlatMessage &, OnixS::FIX::SequenceNumber sequenceNumber, const OnixS::FIX::ISessionStorage::RawMessagePointer & pointer, bool logMessage) override;
 
    void setSessionTerminationFlag(bool terminated) override;
 
    OnixS::FIX::SequenceNumber outSeqNum() override;
 
    void outSeqNum(OnixS::FIX::SequenceNumber messageSequenceNumber) override;
 
    void sessionCreationTime(OnixS::FIX::Timestamp) override;
 
    OnixS::FIX::Timestamp sessionCreationTime() override;
 
    void flush() override {}
 
    size_t resendingQueueSize() const override { return 1000; };
 
    void resendingQueueSize(size_t) override {}
 
private:
 
    bool isSessionTerminated_;
    OnixS::FIX::SequenceNumber inSeqNum_;
    OnixS::FIX::SequenceNumber outSeqNum_;
    OnixS::FIX::Timestamp sessionCreationTime_;
};
 

MySessionStorage.cpp

 
#include "MySessionStorage.h"
 
#include <iostream>
#include <string>
 
using namespace OnixS::FIX;
 
MySessionStorage::MySessionStorage()
    : isSessionTerminated_(false),
      inSeqNum_(0),
      outSeqNum_(0)
{}
 
void MySessionStorage::clear()
{
    std::clog << "Clear the session storage." << std::endl;
 
    inSeqNum_ = 0;
    outSeqNum_ = 0;
    outgoingMessages_.clear();
}
 
void MySessionStorage::close(bool keepSequenceNumbers, bool doBackup)
{
    if(doBackup)
    {
        // Back up the stored data.
        std::clog << "Back up the stored data." << std::endl;
    }
 
    if(!keepSequenceNumbers)
        clear();
}
 
void MySessionStorage::getOutbound(SequenceNumber beginSequenceNumber, SequenceNumber endSequenceNumber, ISessionStorageListener * listener)
{
    std::clog << "getOutbound(" << beginSequenceNumber << ", " << endSequenceNumber << ",..)" << std::endl;
 
    for(SequenceNumber i = beginSequenceNumber; i <= endSequenceNumber; ++i)
    {
        const auto it = outgoingMessages_.find(i);
 
        if(it != outgoingMessages_.end())
            listener->onReplayedMessage(ISessionStorage::RawMessagePointer(&it->second[0], it->second.size()));
    }
}
 
SequenceNumber MySessionStorage::inSeqNum()
{
    return inSeqNum_;
}
 
void MySessionStorage::inSeqNum(SequenceNumber messageSequenceNumber)
{
    std::clog << "inSeqNum(" << messageSequenceNumber << ")" << std::endl;
 
    inSeqNum_ = messageSequenceNumber;
}
 
void MySessionStorage::storeInbound(const Message &, SequenceNumber messageSequenceNumber, const RawMessagePointer& pointer, bool logMessage)
{
    inSeqNum_ = messageSequenceNumber;
 
    if (logMessage)
    {
 
        std::clog << "storeInbound(" << messageSequenceNumber << ", " << std::string(
            reinterpret_cast<const char*>(pointer.buffer_), pointer.length_)
            << ')' << std::endl;
    }
}
 
void MySessionStorage::storeInbound(const FlatMessage &, SequenceNumber messageSequenceNumber, const RawMessagePointer & pointer, bool logMessage)
{
    inSeqNum_ = messageSequenceNumber;
 
    if(logMessage)
    {
 
        std::clog << "storeInbound(" << messageSequenceNumber << ", " << std::string(
            reinterpret_cast<const char*>(pointer.buffer_), pointer.length_)
            << ')' << std::endl;
    }
}
 
void MySessionStorage::storeOutbound(const Message & message, const RawMessagePointer & pointer, bool logMessage)
{
    OnixS::FIX::Int32 messageSequenceNumber = message.seqNum();
    outSeqNum_ = messageSequenceNumber;
 
    if(logMessage)
    {
        std::clog << "storeOutbound(" << messageSequenceNumber << ", " << std::string(
            reinterpret_cast<const char*>(pointer.buffer_), pointer.length_)
            << ')' << std::endl;
 
        outgoingMessages_.insert(make_pair(messageSequenceNumber, RawMessage(pointer.buffer_,
            pointer.buffer_ + pointer.length_)));
    }
}
 
void MySessionStorage::storeOutbound(const FlatMessage &, SequenceNumber sequenceNumber, const RawMessagePointer & pointer, bool logMessage)
{
    outSeqNum_ = sequenceNumber;
 
    if(logMessage)
    {
        std::clog << "storeOutbound(" << sequenceNumber << ", " << std::string(
            reinterpret_cast<const char*>(pointer.buffer_), pointer.length_)
            << ')' << std::endl;
 
        outgoingMessages_.insert(make_pair(sequenceNumber, RawMessage(pointer.buffer_,
            pointer.buffer_ + pointer.length_)));
    }
}
 
void MySessionStorage::setSessionTerminationFlag(bool terminated)
{
    std::clog << "setSessionTerminationFlag(" << terminated << ")" << std::endl;
 
    isSessionTerminated_ = terminated;
}
 
SequenceNumber MySessionStorage::outSeqNum()
{
    return outSeqNum_;
}
 
void MySessionStorage::outSeqNum(SequenceNumber messageSequenceNumber)
{
    std::clog << "outSeqNum(" << messageSequenceNumber << ")" << std::endl;
 
    outSeqNum_ = messageSequenceNumber;
}
 
void MySessionStorage::sessionCreationTime(Timestamp timestamp)
{
    std::clog << "sessionCreationTime(" << timestamp.toString() <<  ")" << std::endl;
 
    sessionCreationTime_ = timestamp;
}
 
Timestamp MySessionStorage::sessionCreationTime()
{
    return sessionCreationTime_;
}
 

PluggableStorage.cpp

#include <OnixS/FIXEngine.h>
 
#include "MySessionStorage.h"
 
#include "../../Common/Helpers.h"
#include "../../Common/Settings.h"
 
using namespace Settings;
using namespace OnixS::FIX;
using namespace OnixS::FIX::FIX44;
 
namespace {
class MySessionListener : public ISessionListener
{
public:
    void onInboundApplicationMsg(Message &, Session *) override {}
};
};
 
using MessagePtr = OnixS::PtrTraits<Message>::UniquePtr;
 
int main()
{
    try {
        EngineSettings settings;
        settings.licenseStore(LicenseStore);
 
        Engine::init(settings);
        EngineShutdownGuard engineShutdownGuard;
 
        MySessionListener listener;
        MySessionStorage storage;
 
        Session session("MySessionStorage_Initiator", "MySessionStorage_Acceptor", FixProtocolVersion, true, &listener, SessionStorageType::Pluggable, &storage);
 
        Message order(Values::MsgType::NewOrderSingle, FixProtocolVersion);
 
        session.send(&order);
 
        MessagePtr foundMessage(session.findSentMessage(1));
 
        if(!foundMessage.get())
            throw std::runtime_error("Stored message was not found.");
 
        std::clog << std::endl << "Message found: " << *foundMessage << std::endl;
 
        session.resetLocalSequenceNumbers();
    }
    catch(const std::exception & ex) {
        processSampleException(ex.what());
        return 1;
    }
 
    return 0;
}