Session Scheduler Sample

This sample demonstrates all aspects of the session scheduler functionality.

Source code

• SessionScheduler.cpp
• SchedulerConfiguration.xml

SessionScheduler.cpp

#include <OnixS/FIXEngine.h>
#include <OnixS/FIXEngine/Threading/Guard.h>
 
#include "../../Common/Helpers.h"
#include "../../Common/Settings.h"
 
using namespace Settings;
using namespace OnixS::FIX;
using namespace OnixS::FIX::Scheduling;
 
namespace {
class MySessionListener : public ISessionListener
{
public:
    void onInboundApplicationMsg(Message &, Session *) final {}
};
};
 
struct SchedulingTimeSpan {
    class Seconds
    {
    public:
        Seconds(int span)
            : span_(span) {
        }
 
        int span() const {
            return span_;
        }
 
    private:
        int span_;
    };
};
 
TimeOfDay operator + (TimeOfDay time, SchedulingTimeSpan::Seconds seconds)
{
    constexpr int SecondsPerMinute = 60;
    constexpr int MinutesPerHour = 60;
    constexpr int HoursPerDay = 24;
    constexpr int SecondsPerHour = SecondsPerMinute * MinutesPerHour;
    constexpr int SecondsPerDay = HoursPerDay * SecondsPerHour;
 
    // Normalize both positive and negative spans into the current day.
    int sinceMidnight = (time.sinceMidnight() + seconds.span()) % SecondsPerDay;
 
    if(sinceMidnight < 0)
        sinceMidnight += SecondsPerDay;
 
    return TimeOfDay(
               sinceMidnight / SecondsPerHour,
               (sinceMidnight / SecondsPerMinute) % MinutesPerHour,
               sinceMidnight % SecondsPerMinute);
}
 
class SessionStateChangeTracer : public ISessionListener
{
    void onStateChange(SessionState::Enum newState, SessionState::Enum ,
                               Session * session) final {
        std::cout    << static_cast<const std::string &>(*session) << " changed its state to " <<
                SessionState::toString(newState) << '.' << std::endl;
    }
 
    void onInboundApplicationMsg(Message &, Session *) final {
        ;
    }
};
 
class SchedulingIssueDetector : public SessionSchedulerListener
{
public:
    SchedulingIssueDetector() : failed_(false)
    {
    }
 
    bool failed() const
    {
        OnixS::Threading::MutexGuard guard(failureMutex_);
        return failed_;
    }
 
    void onLoggingOut(const SessionScheduler & /*scheduler*/, Session * , bool * /*allowLogout*/) final
    {}
 
    void onInitiatorConnecting(const SessionScheduler & /*scheduler*/, Session * , const Counterparty & /*counterparty*/) final
    {}
 
    void onWarning(const SessionScheduler & /*scheduler*/, Session * session, const std::string & warningReason) final
    {
        std::cout << "Scheduler reported a warning for the session " << static_cast<const std::string &>(*session) << ": " << warningReason;
    }
 
    void onError(const SessionScheduler & /*scheduler*/, Session * session, const std::string & errorReason) final
    {
        recordFailure();
        std::cout << "Error occurred while scheduling session " << static_cast<const std::string &>(*session) << ": " << errorReason;
    }
 
private:
    void recordFailure()
    {
        OnixS::Threading::MutexGuard guard(failureMutex_);
        failed_ = true;
    }
 
    mutable OnixS::Threading::Mutex failureMutex_;
    bool failed_;
};
 
class Sample
{
public:
    explicit Sample(SchedulerThreadingModel::Enum model):
        threadingModel_(model),
        reactor_(threadingModel_ == SchedulerThreadingModel::ExternalThread ? new TCPStandard::Stack() : nullptr),
        engineShutdownGuard_(reactor_.get()),
        shutdownState_(ShutdownRequired)
    {
        representYourself();
 
        initializeFixEngine();
 
        constructSessions();
 
        constructScheduler();
    }
 
    ~Sample() noexcept
    {
        if(shutdownState_ == ShutdownComplete)
            return;
 
        shutdownForException();
    }
 
    void shutdown()
    {
        shutdownComponents();
 
        shutdownState_ = ShutdownComplete;
    }
 
    void run() {
        acceptor_->resetLocalSequenceNumbers();
        initiator_->resetLocalSequenceNumbers();
 
        acceptor_->logonAsAcceptor();
 
        std::cout << "Scheduling Initiator session " << static_cast<const std::string &>
             (*initiator_) << " for automatic connection." << std::endl;
 
        SessionSchedule initiatorSchedule = constructShortTimeActivitySchedule();
 
        InitiatorConnectionSettings initiatorConnectivity("localhost", ListenPort);
 
        scheduler_->add(initiator_.get(), initiatorSchedule, initiatorConnectivity);
 
        std::cout << "Waiting for activity on scheduled session " << static_cast<const std::string &>
             (*initiator_) << '.' << std::endl << std::endl;
 
        waitUntilState(initiator_.get(), SessionState::Active);
 
        waitUntilState(initiator_.get(), SessionState::Disconnected);
 
        std::cout << std::endl << "Removing Initiator session " << static_cast<const std::string &>
             (*initiator_) << " from scheduling service." << std::endl;
 
        // Session had to 'pulse' connection till this time, so scheduling can be destroyed.
        scheduler_->remove(initiator_.get());
 
        // Clean-up.
        acceptor_->logout();
    }
 
    void useConfigurationFile() {
        initiator_->resetLocalSequenceNumbers();
 
        SessionSchedulerOptions schedulerOptions;
 
        schedulerOptions.configurationFile("SchedulerConfiguration.xml");
 
        SessionScheduler scheduler(schedulerOptions);
 
        scheduler.add(initiator_.get(), "ScheduleId", "ConnectionId");
 
        OnixS::Threading::ThisThread::sleep(1000 * sessionActivityTimeInSeconds());
 
        scheduler.remove(initiator_.get());
    }
 
private:
 
    enum ShutdownState
    {
        ShutdownRequired,
        ShutdownComplete
    };
 
    SchedulerThreadingModel::Enum threadingModel_;
 
    OnixS::PtrTraits<TCPStandard::Stack>::UniquePtr reactor_;
    EngineShutdownGuard engineShutdownGuard_;
    ShutdownState shutdownState_;
 
    MySessionListener acceptorListener_;
    OnixS::PtrTraits<Session>::UniquePtr acceptor_;
 
    SessionStateChangeTracer initiatorStateChangeTracer_;
    OnixS::PtrTraits<Session>::UniquePtr initiator_;
 
    SchedulingIssueDetector schedulingIssueDetector_;
    OnixS::PtrTraits<SessionScheduler>::UniquePtr scheduler_;
 
    void shutdownComponents()
    {
        // Stop all scheduler workers before releasing the sessions they reference.
        scheduler_.reset();
 
        initiator_->shutdown();
        initiator_.reset();
 
        acceptor_->shutdown();
        acceptor_.reset();
    }
 
    static void shutdownSessionForException(OnixS::PtrTraits<Session>::UniquePtr & session) noexcept
    {
        if(session.get() == nullptr)
            return;
 
        try
        {
            session->shutdown();
        }
        catch(const std::exception & ex)
        {
            std::clog << "Session shutdown failed during exception cleanup: " << ex.what() << std::endl;
        }
        catch(...)
        {
            std::clog << "Session shutdown failed during exception cleanup with an unknown error." << std::endl;
        }
 
        // unique_ptr::reset is noexcept, so destruction cannot be recovered here.
        session.reset();
    }
 
    void shutdownForException() noexcept
    {
        // Stop state changers before releasing the borrowed session pointers.
        scheduler_.reset();
 
        shutdownSessionForException(initiator_);
        shutdownSessionForException(acceptor_);
    }
 
    void constructScheduler() {
        SessionSchedulerOptions schedulerOptions;
        schedulerOptions.eventListener(&schedulingIssueDetector_);
        schedulerOptions.threadingModel(threadingModel_);
 
        scheduler_.reset(new SessionScheduler(schedulerOptions));
    }
 
    void constructSessions() {
        acceptor_.reset(new Session(TargetCompId, SenderCompId, FixProtocolVersion, &acceptorListener_));
 
        initiator_.reset(new Session(reactor_.get(), SenderCompId, TargetCompId, FixProtocolVersion, &initiatorStateChangeTracer_));
    }
 
    static SessionSchedule constructShortTimeActivitySchedule() {
        TimeOfDay now = TimeOfDay::now();
 
        TimeOfDay logonTime = now + SchedulingTimeSpan::Seconds(5);
 
        TimeOfDay logoutTime = logonTime + SchedulingTimeSpan::Seconds(sessionActivityTimeInSeconds());
 
        if(logoutTime.sinceMidnight() < logonTime.sinceMidnight())
        {
            // Keep the short run within one day so every activity-week day has a logon.
            logonTime = TimeOfDay::midnight();
            logoutTime = logonTime + SchedulingTimeSpan::Seconds(sessionActivityTimeInSeconds());
        }
 
        return SessionSchedule(
                   DayOfWeek::Monday,
                   DayOfWeek::Sunday,
                   logonTime,
                   logoutTime,
                   SessionDuration::Day,
                   SequenceNumberResetPolicy::Never);
    }
 
    void waitUntilState(Session * session, SessionState::Enum state) {
        const unsigned oneSecondPause = 1000;
        const int spinWaitMicrosecondPause = 10;
 
        while(session->state() != state)
        {
            if(schedulingIssueDetector_.failed())
                throw std::runtime_error("Session scheduling failed.");
 
            if(threadingModel_ == SchedulerThreadingModel::DedicatedThreads)
                OnixS::Threading::ThisThread::sleep(oneSecondPause);
            else
            {
                scheduler_->dispatchEvents();
                reactor_->dispatchEvents();
                OnixS::Threading::ThisThread::spinWait(spinWaitMicrosecondPause);
            }
        }
    }
 
    static void initializeFixEngine() {
        EngineSettings settings;
 
        settings.licenseStore(LicenseStore)
                .listenPort(ListenPort);
 
        Engine::init(settings);
    }
 
    static int sessionActivityTimeInSeconds() {
        return 30;
    }
 
    void representYourself() {
        std::cout << "OnixS C++ FIX Engine Session Scheduling Sample with " << SchedulerThreadingModel::toString(threadingModel_) << "." << std::endl << std::endl;
        std::cout << "Usage: SessionScheduler threadingModel[DedicatedThreads][ExternalThread]" << std::endl << std::endl;
    }
};
 
int main(int argc, char * argv[])
{
    try {
 
        const std::string ThreadingModel = (argc > 1 ? argv[1] : "DedicatedThreads");
 
        Sample sample(ThreadingModel == "ExternalThread" ? SchedulerThreadingModel::ExternalThread : SchedulerThreadingModel::DedicatedThreads);
 
        sample.run();
        sample.shutdown();
 
        std::cout << std::endl << "Done." << std::endl;
 
        return 0;
    }
    catch(const std::exception & ex) {
        processSampleException(ex.what());
 
        return 1;
    }
 
}
 

SchedulerConfiguration.xml

The useConfigurationFile() step loads schedule and connection settings from this file.

<?xml version="1.0" encoding="utf-8" ?>
 
<SchedulerSettings
  xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
  xsi:schemaLocation="https://ref.onixs.biz/fix/scheduler https://ref.onixs.biz/fix/scheduler/scheduler-settings-1.8.xsd"
  xmlns="https://ref.onixs.biz/fix/scheduler"
  >
  <Schedules>
    <Schedule
      id="ScheduleId"
      firstDay="Sunday"
      logonTime="17:00"
 
      lastDay="Friday"
      logoutTime="16:00"
 
      sessionDuration="week"
      resetSessionNumber="weekly"
      />
  </Schedules>
 
  <Connections>
    <Connection id="ConnectionId" role="initiator">
      <Counterparty host="localhost" port="4500"/>
    </Connection>
  </Connections>
 
</SchedulerSettings>