Samples :: Scheduler

Samples :: Scheduler

This module contains four samples: a simple acceptor, a programmatic schedule sample, a generic schedule sample, and a file-based schedule sample.

Simple Acceptor

Description

This sample acts as a simple acceptor for the scheduler samples.

The “SimpleAcceptor” application must be started first.

Directory Contents

Item	Description
conf	configuration directory
src	source code

Usage

Run the sample (scripts are under src/main/script/):
- win: 1-SimpleAcceptor.bat
- linux: 1-SimpleAcceptor.sh
Clean everything (scripts are under samples/src/main/script/ at the repo root):
- win: clean.bat
- linux: clean.sh

Source Code

import biz.onixs.cme.ilink3.testing.Emulator;
import biz.onixs.cme.ilink3.testing.TestUtility;
import biz.onixs.util.settings.PropertyBasedSettings;
import biz.onixs.util.settings.Settings;
import org.junit.jupiter.api.Assertions;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class SimpleAcceptor {
    private static final Logger LOG = LoggerFactory.getLogger(SimpleAcceptor.class);
    private static final String SETTINGS_RESOURCE = "sample/SimpleAcceptor.properties";

    private void run() {
        try {
            //
            LOG.info("Loading settings from: {}", SETTINGS_RESOURCE);
            final Settings settings = new PropertyBasedSettings(SETTINGS_RESOURCE);

            LOG.info("Creates an Emulator object configured as acceptor");
            final Emulator emulator = new Emulator(new TestUtility());
            emulator.setPort(settings.getInteger("ListenPort"));

            LOG.info("Accepts an incoming TCP connection and prepares for message exchange");
            emulator.acceptConnection();

            LOG.info("Accepts a \"Negotiate\" message from the initiator");
            emulator.acceptNegotiation();

            LOG.info("Accepts an \"Establish\" message, finalizing the session setup");
            emulator.acceptEstablishment(1);

            LOG.info("Accepts a \"Terminate\" message, ending the session");
            emulator.acceptTerminate();

            // Asserts that the Emulator's connection is now closed.
            Assertions.assertTrue(emulator.isConnectionClosed());
        } catch (final Exception e) {
            LOG.error(e.getMessage(), e);
        }
    }

    public static void main(final String[] args) {
        try {
            LOG.info("SimpleAcceptor");
            LOG.info("The application is starting...");
            final SimpleAcceptor acceptor = new SimpleAcceptor();
            acceptor.run();
        } catch (Throwable throwable) {
            LOG.error(throwable.getMessage(), throwable);
        } finally {
            LOG.info("The application is stopped.");
        }
    }
}

Program Schedule Sample

Description

This sample demonstrates configuring the Session Scheduler programmatically with a simple daily schedule.

The “SimpleAcceptor” application must be started first.

Directory Contents

Item	Description
conf	configuration directory
src	source code

Usage

Run the sample (scripts are under src/main/script/):
- win: 2-ProgramScheduleSample.bat
- linux: 2-ProgramScheduleSample.sh
Clean everything (scripts are under samples/src/main/script/ at the repo root):
- win: clean.bat
- linux: clean.sh

Source Code


import biz.onixs.cme.ilink3.handler.Handler;
import biz.onixs.cme.ilink3.handler.Session;
import biz.onixs.cme.ilink3.handler.session.SessionSettings;
import biz.onixs.cme.ilink3.scheduler.*;
import biz.onixs.util.settings.PropertyBasedSettings;
import biz.onixs.util.settings.Settings;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.time.DayOfWeek;
import java.time.LocalTime;

public class ProgramScheduleSample implements Runnable {
    private static final Logger LOG = LoggerFactory.getLogger(ProgramScheduleSample.class);
    private static final String SETTINGS_RESOURCE = "sample/ProgramScheduleSample.properties";
    private final DemoSessionStateChangeListener stateChangeListener = new DemoSessionStateChangeListener();
    private final int marketSegmentId;
    private final String host;
    private final int port;
    private final Settings settings;
    private Session session = null;
    private SessionScheduler scheduler = null;

    public ProgramScheduleSample(final int marketSegmentId, final String host, final int port,
                                 final Settings settings) {
        this.marketSegmentId = marketSegmentId;
        this.host = host;
        this.port = port;
        this.settings = settings;
    }

    public void run() {
        try {
            LOG.info("Starting the Handler...");
            Handler.init(settings);
            //
            LOG.info("---! Please make sure 'SimpleAcceptor' app is started !---");
            //
            createAndStartScheduler();
            //
            createInitiatorSession();
            //
            final SessionSchedule schedule = createSchedule();
            final SessionConnection connection = createInitiatorSettings();
            scheduler.register(session, schedule, connection);
            LOG.info("Initiator session is registered in the scheduler.");
            //
            LOG.info("Waiting to connect...");
            stateChangeListener.waitEstablished();
            LOG.info("Connected.");
            //
            LOG.info("Waiting to disconnect...");
            stateChangeListener.waitDisconnected();
            LOG.info("Disconnected.");
            //
            scheduler.unregister(session);
        } catch (final Exception e) {
            LOG.error(e.getMessage(), e);
        } finally {
            if (null != scheduler) {
                try {
                    scheduler.stop();
                } catch (final SessionSchedulerException e) {
                    LOG.error("Scheduler stop error", e);
                }
            }
            LOG.info("Handler shutdown ...");
            if (Handler.isInited()) {
                Handler.getInstance().shutdown();
            }
            LOG.info("The application is stopped.");
        }
    }

    private void createInitiatorSession() {
        final SessionSettings sessionSettings = new SessionSettings();
        sessionSettings.init(settings);
        session = new Session(sessionSettings, marketSegmentId);
        session.addStateChangeListener(stateChangeListener);
    }

    /**
     * Creates and configures the scheduler instance.
     */
    private void createAndStartScheduler() throws SessionSchedulerException {
        scheduler = new SessionScheduler();
        scheduler.addLogoutNotification(1);
        scheduler.addLogoutNotification(2);
        scheduler.getListenerManager().addErrorListener(new DemoErrorListener());
        scheduler.getListenerManager().addLogoutNotificationListener(new DemoLogoutNotificationListener());
        scheduler.start();
    }

    /**
     * Creates and configures the initiator connection settings.
     */
    private SessionConnection createInitiatorSettings() {
        final SessionConnection connection = new SessionConnection();
        final InetSocketAddress address = InetSocketAddress.createUnresolved(host, port);
        connection.addAddress(address);
        return connection;
    }

    /**
     * Creates and configures the simple session schedule.
     * Features:
     * <ul>
     * <li>logon and logout every day from Monday till Sunday</li>
     * <li>logon time is set to 5 seconds from now</li>
     * <li>logout time is set to 10 seconds from now</li>
     * </ul>
     */
    private static SessionSchedule createSchedule() {
        final SingleDayLengthSchedule schedule = new SingleDayLengthSchedule();
        final LocalTime now = LocalTime.now();
        schedule.setLogonTime(now.plusSeconds(5))
                .setLogoutTime(now.plusSeconds(15))
                .setFirstDay(DayOfWeek.MONDAY)
                .setLastDay(DayOfWeek.SUNDAY);
        return schedule;
    }

    private static void configureSettings(final Settings settings) throws IOException {
        settings.setString(SessionSettings.SESSION_PROP, "XXX");
        settings.setString(SessionSettings.FIRM_PROP, "001");
        settings.setString(SessionSettings.ACCESS_KEY_PROP, "dGVzdHRlc3R0ZXN0dA==");
        settings.setString(SessionSettings.SECRET_KEY_PROP, "dGVzdHRlc3R0ZXN0dGVzdHRlc3R0");
        settings.setInteger(SessionSettings.KEEP_ALIVE_INTERVAL_PROP, 50);
        settings.setInteger(SessionSettings.RECONNECT_ATTEMPTS_PROP, 0);
        settings.setString(SessionSettings.TRADING_SYSTEM_VERSION_PROP, "1.1.0");
        settings.setString(SessionSettings.TRADING_SYSTEM_NAME_PROP, "Trading System");
        settings.setString(SessionSettings.TRADING_SYSTEM_VENDOR_PROP, "OnixS");
    }

    public static void main(final String[] args) throws IOException {
        LOG.info("Program Schedule Sample");
        LOG.info("The application is starting...");
        LOG.info("Loading settings from: {}", SETTINGS_RESOURCE);
        final Settings settings = new PropertyBasedSettings(SETTINGS_RESOURCE);
        //
        int marketSegmentId;
        String host;
        int port;
        //
        if (3 > args.length) {
            LOG.info("Emulator is used, remote usage: [MarketSegmentId] [Host] [Port]");
            marketSegmentId = settings.getInteger("MarketSegmentId", 59);
            host = settings.getString("CounterpartyHost");
            port = settings.getInteger("CounterpartyPort");
            configureSettings(settings);
        } else {
            marketSegmentId = Integer.parseInt(args[0]);
            host = args[1];
            port = Integer.parseInt(args[2]);
        }
        final ProgramScheduleSample sample = new ProgramScheduleSample(marketSegmentId, host, port, settings);
        sample.run();
    }
}

Program Generic Schedule Sample

Description

This sample demonstrates building a custom weekly schedule with explicit logon/logout intervals.

The “SimpleAcceptor” application must be started first.

Directory Contents

Item	Description
conf	configuration directory
src	source code

Usage

Run the sample (scripts are under src/main/script/):
- win: 2-ProgramGenericScheduleSample.bat
- linux: 2-ProgramGenericScheduleSample.sh
Clean everything (scripts are under samples/src/main/script/ at the repo root):
- win: clean.bat
- linux: clean.sh

Source Code


import biz.onixs.cme.ilink3.handler.Handler;
import biz.onixs.cme.ilink3.handler.Session;
import biz.onixs.cme.ilink3.handler.session.SessionSettings;
import biz.onixs.cme.ilink3.scheduler.*;
import biz.onixs.util.settings.PropertyBasedSettings;
import biz.onixs.util.settings.Settings;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.time.DayOfWeek;
import java.time.LocalDate;
import java.time.LocalTime;

public class ProgramGenericScheduleSample implements Runnable {
    private static final Logger LOG = LoggerFactory.getLogger(ProgramGenericScheduleSample.class);
    private static final String SETTINGS_RESOURCE = "sample/ProgramGenericScheduleSample.properties";
    private final DemoSessionStateChangeListener stateChangeListener = new DemoSessionStateChangeListener();
    private final int marketSegmentId;
    private final String host;
    private final int port;
    private final Settings settings;
    private Session session = null;
    private SessionScheduler scheduler = null;

    public ProgramGenericScheduleSample(final int marketSegmentId, final String host, final int port,
                                        final Settings settings) {
        this.marketSegmentId = marketSegmentId;
        this.host = host;
        this.port = port;
        this.settings = settings;
    }

    public void run() {
        try {
            LOG.info("Starting the Handler...");
            Handler.init(settings);
            //
            LOG.info("---! Please make sure 'SimpleAcceptor' app is started !---");
            //
            scheduler = createAndStartScheduler();
            //
            createInitiatorSession();
            //
            final SessionSchedule schedule = createSchedule();
            final SessionConnection connection = createInitiatorSettings();
            scheduler.register(session, schedule, connection);
            LOG.info("Initiator session is registered in the scheduler.");
            //
            LOG.info("Waiting to connect...");
            stateChangeListener.waitEstablished();
            LOG.info("Connected.");
            //
            LOG.info("Waiting to disconnect...");
            stateChangeListener.waitDisconnected();
            LOG.info("Disconnected.");
            //
            scheduler.unregister(session);
        } catch (final Exception e) {
            LOG.error(e.getMessage(), e);
        } finally {
            if (null != scheduler) {
                try {
                    scheduler.stop();
                } catch (final SessionSchedulerException e) {
                    LOG.error("Scheduler stop error", e);
                }
            }
            LOG.info("Handler shutdown ...");
            if (Handler.isInited()) {
                Handler.getInstance().shutdown();
            }
            LOG.info("The application is stopped.");
        }
    }

    private void createInitiatorSession() {
        final SessionSettings sessionSettings = new SessionSettings();
        sessionSettings.init(settings);
        session = new Session(sessionSettings, marketSegmentId);
        session.addStateChangeListener(stateChangeListener);
    }

    /**
     * Creates and configures the scheduler instance.
     */
    private static SessionScheduler createAndStartScheduler() throws SessionSchedulerException {
        final SessionScheduler scheduler = new SessionScheduler()
                .addLogoutNotification(1)
                .addLogoutNotification(2);
        scheduler.getListenerManager()
                .addErrorListener(new DemoErrorListener())
                .addLogoutNotificationListener(new DemoLogoutNotificationListener());
        scheduler.start();
        return scheduler;
    }

    /**
     * Creates and configures the initiator connection settings.
     */
    private SessionConnection createInitiatorSettings() {
        final SessionConnection connection = new SessionConnection();
        final InetSocketAddress address = InetSocketAddress.createUnresolved(host, port);
        connection.addAddress(address);
        return connection;
    }

    /**
     * Creates and configures generic weekly session schedule which can consist of several activity intervals.
     */
    private static SessionSchedule createSchedule() throws ScheduleValidationException {
        final GenericWeeklySchedule schedule = new GenericWeeklySchedule();
        //
        final LocalTime now = LocalTime.now();
        final DayOfWeek todayDayOfWeek = LocalDate.now().getDayOfWeek();
        //
        final WeeklyInterval interval = new WeeklyInterval()
                .setLogonDay(todayDayOfWeek)
                .setLogonTime(now.plusSeconds(5))
                .setLogoutDay(todayDayOfWeek)
                .setLogoutTime(now.plusSeconds(10))
                .setResetOnLogon(false);
        schedule.addInterval(interval);
        //
        return schedule;
    }

    private static void configureSettings(final Settings settings) throws IOException {
        settings.setString(SessionSettings.SESSION_PROP, "XXX");
        settings.setString(SessionSettings.FIRM_PROP, "001");
        settings.setString(SessionSettings.ACCESS_KEY_PROP, "dGVzdHRlc3R0ZXN0dA==");
        settings.setString(SessionSettings.SECRET_KEY_PROP, "dGVzdHRlc3R0ZXN0dGVzdHRlc3R0");
        settings.setInteger(SessionSettings.KEEP_ALIVE_INTERVAL_PROP, 50);
        settings.setInteger(SessionSettings.RECONNECT_ATTEMPTS_PROP, 0);
        settings.setString(SessionSettings.TRADING_SYSTEM_VERSION_PROP, "1.1.0");
        settings.setString(SessionSettings.TRADING_SYSTEM_NAME_PROP, "Trading System");
        settings.setString(SessionSettings.TRADING_SYSTEM_VENDOR_PROP, "OnixS");
    }

    public static void main(final String[] args) throws IOException {
        LOG.info("Program Generic Schedule Sample");
        LOG.info("The application is starting...");
        LOG.info("Loading settings from: {}", SETTINGS_RESOURCE);
        final Settings settings = new PropertyBasedSettings(SETTINGS_RESOURCE);
        //
        int marketSegmentId;
        String host;
        int port;
        //
        if (3 > args.length) {
            LOG.info("Emulator is used, remote usage: [MarketSegmentId] [Host] [Port]");
            marketSegmentId = settings.getInteger("MarketSegmentId", 59);
            host = settings.getString("CounterpartyHost");
            port = settings.getInteger("CounterpartyPort");
            configureSettings(settings);
        } else {
            marketSegmentId = Integer.parseInt(args[0]);
            host = args[1];
            port = Integer.parseInt(args[2]);
        }
        final ProgramGenericScheduleSample sample =
                new ProgramGenericScheduleSample(marketSegmentId, host, port, settings);
        sample.run();
    }
}

File Schedule Sample

Description

This sample loads scheduler settings from conf/sample/SchedulerSettings.xml and registers the session using those settings.

The “SimpleAcceptor” application must be started first.

Directory Contents

Item	Description
conf	configuration directory
src	source code

Usage

Run the sample (scripts are under src/main/script/):
- win: 2-FileScheduleSample.bat
- linux: 2-FileScheduleSample.sh
Clean everything (scripts are under samples/src/main/script/ at the repo root):
- win: clean.bat
- linux: clean.sh

Source Code


import biz.onixs.cme.ilink3.handler.Handler;
import biz.onixs.cme.ilink3.handler.Session;
import biz.onixs.cme.ilink3.handler.session.SessionSettings;
import biz.onixs.cme.ilink3.scheduler.*;
import biz.onixs.util.settings.PropertyBasedSettings;
import biz.onixs.util.settings.Settings;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.net.InetSocketAddress;

public class FileScheduleSample {
    private static final Logger LOG = LoggerFactory.getLogger(FileScheduleSample.class);
    private static final String SETTINGS_RESOURCE = "sample/FileScheduleSample.properties";
    private static final String SCHEDULER_SETTINGS_RESOURCE = "sample/SchedulerSettings.xml";
    private final DemoSessionStateChangeListener stateChangeListener = new DemoSessionStateChangeListener();

    private void run(final Settings settings, final int marketSegmentId, final SessionConnection connectionOverride) {
        SessionScheduler scheduler = null;
        try {
            LOG.info("Starting the Handler...");
            Handler.init(settings);
            //
            LOG.info("---! Please make sure 'SimpleAcceptor' app is started !---");
            //
            scheduler = createAndStartScheduler();
            final SchedulerSettings schedulerSettings = loadSchedulerSettings(SCHEDULER_SETTINGS_RESOURCE);
            final Session initiator = createInitiatorSession(settings, marketSegmentId);
            if (null == connectionOverride) {
                scheduler.register(initiator, schedulerSettings, "Schedule1", "Initiator1");
            } else {
                scheduler.register(initiator, schedulerSettings, "Schedule1", connectionOverride);
            }
            LOG.info("Initiator session is registered in the scheduler.");
            //
            LOG.info("Waiting to connect...");
            stateChangeListener.waitEstablished();
            LOG.info("Connected.");
            //
            scheduler.unregister(initiator);
            //
            initiator.terminate();
        } catch (final Exception e) {
            LOG.error(e.getMessage(), e);
        } finally {
            if (null != scheduler) {
                try {
                    scheduler.stop();
                } catch (final SessionSchedulerException e) {
                    LOG.error("Scheduler stop error", e);
                }
            }
            LOG.info("Handler shutdown ...");
            if (Handler.isInited()) {
                Handler.getInstance().shutdown();
            }
            LOG.info("The application is stopped.");
        }
    }

    /**
     * Creates and configures the initiator session.
     */
    private Session createInitiatorSession(final Settings settings, final int marketSegmentId) {
        final SessionSettings sessionSettings = new SessionSettings();
        sessionSettings.init(settings);
        final Session session = new Session(sessionSettings, marketSegmentId);
        session.addStateChangeListener(stateChangeListener);
        return session;
    }

    /**
     * Creates and configures the scheduler instance.
     */
    private static SessionScheduler createAndStartScheduler() throws SessionSchedulerException {
        final SessionScheduler scheduler = new SessionScheduler();
        scheduler.getListenerManager().addErrorListener(new DemoErrorListener());
        scheduler.start();
        return scheduler;
    }

    private static SchedulerSettings loadSchedulerSettings(final String resource)
            throws SchedulerSettingsLoaderException {
        final SchedulerSettingsLoader loader = new SchedulerSettingsLoader();
        return loader.load(resource);
    }

    private static void configureSettings(final Settings settings) throws IOException {
        settings.setString(SessionSettings.SESSION_PROP, "XXX");
        settings.setString(SessionSettings.FIRM_PROP, "001");
        settings.setString(SessionSettings.ACCESS_KEY_PROP, "dGVzdHRlc3R0ZXN0dA==");
        settings.setString(SessionSettings.SECRET_KEY_PROP, "dGVzdHRlc3R0ZXN0dGVzdHRlc3R0");
        settings.setInteger(SessionSettings.KEEP_ALIVE_INTERVAL_PROP, 50);
        settings.setInteger(SessionSettings.RECONNECT_ATTEMPTS_PROP, 0);
        settings.setString(SessionSettings.TRADING_SYSTEM_VERSION_PROP, "1.1.0");
        settings.setString(SessionSettings.TRADING_SYSTEM_NAME_PROP, "Trading System");
        settings.setString(SessionSettings.TRADING_SYSTEM_VENDOR_PROP, "OnixS");
    }

    private static SessionConnection createInitiatorSettings(final String host, final int port) {
        final SessionConnection connection = new SessionConnection();
        final InetSocketAddress address = InetSocketAddress.createUnresolved(host, port);
        connection.addAddress(address);
        return connection;
    }

    public static void main(final String[] args) throws IOException {
        LOG.info("File Schedule Sample");
        LOG.info("The application is starting...");
        LOG.info("Loading settings from: {}", SETTINGS_RESOURCE);
        final Settings settings = new PropertyBasedSettings(SETTINGS_RESOURCE);
        //
        final int marketSegmentId;
        final SessionConnection connectionOverride;
        if (3 > args.length) {
            LOG.info("Emulator is used, remote usage: [MarketSegmentId] [Host] [Port]");
            marketSegmentId = settings.getInteger("MarketSegmentId", 59);
            connectionOverride = null;
            configureSettings(settings);
        } else {
            marketSegmentId = Integer.parseInt(args[0]);
            connectionOverride = createInitiatorSettings(args[1], Integer.parseInt(args[2]));
        }
        final FileScheduleSample fileScheduleSample = new FileScheduleSample();
        fileScheduleSample.run(settings, marketSegmentId, connectionOverride);
    }
}

Source Codes Used In Samples

import biz.onixs.cme.ilink3.scheduler.ErrorListener;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class DemoErrorListener implements ErrorListener {
    private static final Logger LOG = LoggerFactory.getLogger(DemoErrorListener.class);

    @Override
    public void onError(final ErrorArgs args) {
        LOG.error("onError(): {}", args);
    }

    /**
     * Logon error event callback.
     *
     * @param args event arguments
     */
    @Override
    public void onLogonError(final LogonErrorArgs args) {
        LOG.error("onLogonError(): {}", args);
    }
}

import biz.onixs.cme.ilink3.scheduler.LogoutNotification;
import biz.onixs.cme.ilink3.scheduler.LogoutNotificationListener;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * Demo logout notification listener implementation.
 */
public class DemoLogoutNotificationListener implements LogoutNotificationListener {
    private static final Logger LOG = LoggerFactory.getLogger(DemoLogoutNotificationListener.class);

    @Override
    public void onLogoutNotification(final LogoutNotification notification) {
        LOG.info("{}: logout notification: {} seconds left", notification.getSession(), notification.getInterval());
    }
}

import biz.onixs.cme.ilink3.handler.SessionState;
import biz.onixs.cme.ilink3.handler.session.StateChangeArgs;
import biz.onixs.cme.ilink3.handler.session.StateChangeListener;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.concurrent.Semaphore;

public class DemoSessionStateChangeListener implements StateChangeListener {
    private static final Logger LOG = LoggerFactory.getLogger(DemoSessionStateChangeListener.class);
    private final Semaphore semaphoreIsEstablished = new Semaphore(0);
    private final Semaphore semaphoreIsDisconnected = new Semaphore(0);

    @Override
    public void onStateChange(final Object sender, final StateChangeArgs args) {
        final SessionState prevState = args.getPrevState();
        final SessionState newState = args.getNewState();
        if (prevState != SessionState.TERMINATED && SessionState.TERMINATED == newState) {
            semaphoreIsDisconnected.release();
        } else if (SessionState.ESTABLISHED == newState) {
            semaphoreIsEstablished.release();
        }
    }

    public void waitEstablished() throws InterruptedException {
        semaphoreIsEstablished.acquire();
    }

    public void waitDisconnected() throws InterruptedException {
        semaphoreIsDisconnected.acquire();
    }
}