Session Scheduler Sample

This sample demonstrates all aspects of the session scheduler functionality.

Source code:

#include <OnixS/FIXEngine.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 *) ONIXS_FIXENGINE_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)
{
    const int SecondsPerMinute = 60;
    const int MinutesPerHour = 60;
    const int SecondsPerHour = SecondsPerMinute * MinutesPerHour;
 
    int sinceMidnight = time.sinceMidnight() + seconds.span();
 
    return TimeOfDay(
               sinceMidnight / SecondsPerHour,
               (sinceMidnight / SecondsPerMinute) % MinutesPerHour,
               sinceMidnight % SecondsPerMinute);
}
 
class SessionStateChangeTracer : public ISessionListener
{
    void onStateChange(SessionState::Enum newState, SessionState::Enum /*prevState*/,
                               Session * session) ONIXS_FIXENGINE_FINAL {
        std::cout    << static_cast<const std::string &>(*session) << " changed its state to " <<
                SessionState::toString(newState) << '.' << std::endl;
    }
 
    void onInboundApplicationMsg(Message &, Session *) ONIXS_FIXENGINE_FINAL {
        ;
    }
};
 
class SchedulingIssueDetector : public SessionSchedulerListener
{
public:
    void onLoggingOut(const SessionScheduler & /*scheduler*/, Session * /*session*/, bool * /*allowLogout*/) ONIXS_FIXENGINE_FINAL
    {}
 
    void onInitiatorConnecting(const SessionScheduler & /*scheduler*/, Session * /*session*/, const Counterparty & /*counterparty*/) ONIXS_FIXENGINE_FINAL
    {}
 
    void onWarning(const SessionScheduler & /*scheduler*/, Session * session, const std::string & warningReason) ONIXS_FIXENGINE_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) ONIXS_FIXENGINE_FINAL
    {
        std::cout << "Error occurred while scheduling session " << static_cast<const std::string &>(*session) << ": " << errorReason;
    }
};
 
class Sample
{
public:
    explicit Sample(SchedulerThreadingModel::Enum model):
        threadingModel_(model), reactor_(threadingModel_ == SchedulerThreadingModel::ExternalThread ? new TCPStandard::Stack() : ONIXS_FIXENGINE_NULLPTR)
    {
        representYourself();
 
        initializeFixEngine();
 
        constructSessions();
 
        constructScheduler();
    }
 
    ~Sample() {
        initiator_->shutdown();
        initiator_.reset();
 
        acceptor_->shutdown();
        acceptor_.reset();
 
        scheduler_.reset();
 
        Engine::shutdown();
 
        if(threadingModel_ == SchedulerThreadingModel::ExternalThread)
        {
            while(!reactor_->isQuiescent())
                reactor_->dispatchEvents();
        }
    }
 
    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:
 
    SchedulerThreadingModel::Enum threadingModel_;
 
    OnixS::PtrTraits<Session>::UniquePtr acceptor_;
    MySessionListener acceptorListener_;
 
    OnixS::PtrTraits<Session>::UniquePtr initiator_;
    SessionStateChangeTracer initiatorStateChangeTracer_;
 
    OnixS::PtrTraits<SessionScheduler>::UniquePtr scheduler_;
    SchedulingIssueDetector schedulingIssueDetector_;
 
    OnixS::PtrTraits<TCPStandard::Stack>::UniquePtr reactor_;
 
    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());
 
        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(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();
 
        std::cout << std::endl << "Done." << std::endl;
 
        return 0;
    }
    catch(const std::exception & ex) {
        processSampleException(ex.what());
 
        return 1;
    }
 
}