B3 UMDF Input Adapter

Name: Use this required field to specify or change the name of this instance of this component. The name must be unique within the current EventFlow module. The name can contain alphanumeric characters, underscores, and escaped special characters. Special characters can be escaped as described in Identifier Naming Rules. The first character must be alphabetic or an underscore.

Adapter: A read-only field that shows the formal name of the adapter.

Class name: Shows the fully qualified class name that implements the functionality of this adapter. If you need to reference this class name elsewhere in your application, you can right-click this field and select Copy from the context menu to place the full class name in the system clipboard.

Start options: This field provides a link to the Cluster Aware tab, where you configure the conditions under which this adapter starts.

Enable Error Output Port: Select this checkbox to add an Error Port to this component. In the EventFlow canvas, the Error Port shows as a red output port, always the last port for the component. See Using Error Ports to learn about Error Ports.

Description: Optionally, enter text to briefly describe the purpose and function of the component. In the EventFlow Editor canvas, you can see the description by pressing Ctrl while the component's tooltip is displayed.

Use the Edit Schemas tab to specify the schemas of the output tuples for this adapter.

For general instructions on using the Edit Schemas tab, see the Properties: Edit Schema Tab section of the Defining Input Streams page.

Of the five available output ports exposed by the adapter, only one (the Events output port) has a fixed schema. For every other port you must define its schema in the Edit Schemas tab. Remember that even if the routing map is empty, there is still one default output port for application-level messages. This means you must define at least one schema in this tab. The schemas do not need to be the same for every port, but there is one mandatory field that must be present in each schema. This field is:

Additionally, the Incremental Update output port (see the "Incremental Refreshes Port" section below) may have the following field to help identify the correct order in which updates are to be applied:

Every other field you define in each of the output ports' schema is expected to have a corresponding field in the incoming FAST message directed to that port. When the adapter receives such a message, it iterates through all the fields in your schema, gets the corresponding value in the FAST field, and sets the tuple's field to that value. Any field in the incoming FAST message that has no corresponding field in the schema is ignored and is not passed on to the application. In this way, you can maximize the efficiency of your application by only defining the FAST fields of interest in your schema.

The adapter uses an algorithm to identify a corresponding FAST field, given a StreamBase field. That algorithm is as follows:

Use the settings in this tab to enable this operator or adapter for runtime start and stop conditions in a multi-node cluster. During initial development of the fragment that contains this operator or adapter, and for maximum compatibility with releases before 10.5.0, leave the Cluster start policy control in its default setting, Start with module.

Cluster awareness is an advanced topic that requires an understanding of StreamBase Runtime architecture features, including clusters, quorums, availability zones, and partitions. See Cluster Awareness Tab Settings on the Using Cluster Awareness page for instructions on configuring this tab.

Use the Concurrency tab to specify parallel regions for this instance of this component, or multiplicity options, or both. The Concurrency tab settings are described in Concurrency Options, and dispatch styles are described in Dispatch Styles.

The adapter has five output ports, listed here in the order in which they appear on the adapter:

Of these, only the Events port has a fixed schema, determined by the adapter. The other ports require you to define the schema you wish to use based on the FAST fields which are of interest to your application. The only mandatory field on these ports is ChannelID (string), which identifies which FAST channel originated the message.

The Adapter defines two input ports:

Add an instance of the adapter to a new StreamBase EventFlow application as follows:

In order to know which IP addresses and ports to connect to the various data streams exposed by UMDF, the adapter requires them to be defined in an XML file. This file may define one or more channels, each of which describe how to connect to the Incremental, Instrument and Recovery UDP streams as well as the Replay TCP stream. All four connections must be defined in order for the channel to be properly defined and operated.

The following shows an example of a properly-configured channel definition file. Change this file for your own use by updating the values in bold text to reflect the proper settings for your organization:

<?xml version="1.0"?>
<configuration environment="AutoCert" updated="2011/04/30-21:13:22">
    <channel id="050" label="Futures Interest rates">
        <connections>
            <connection id="AI1"> <!-- ID attribute may have any value but 
                                       should be unique across the file -->
                <type feed-type="I">Incremental</type>
                <protocol>UDP/IP</protocol>
                <ip>233.111.180.113</ip>
                <port>20050</port>
                <feed>A</feed> <!-- Use B instead if this connection represents 
                                    the backup feed of UMDF for this channel -->
            </connection>
            <connection id="BD"> <!-- ID attribute may have any value but should be 
                                      unique across the file -->
                <type feed-type="D">Instrument</type>
                <protocol>UDP/IP</protocol>
                <ip>233.111.180.112</ip>
                <port>10050</port>
                <feed>A</feed> <!-- Use B instead if this connection represents the 
                                    backup feed of UMDF for this channel -->
            </connection>
            <connection id="R"> <!-- ID attribute may have any value but should be 
                                     unique acrYou may aoss the file -->
                <type feed-type="R">Recovery</type>
                <protocol>UDP/IP</protocol>
                <ip>233.111.180.112</ip>
                <port>30050</port>
                <feed>A</feed> <!-- Use B instead if this connection represents 
                                    the backup feed of UMDF for this channel -->
            </connection>
            <connection id="N"> <!-- ID attribute may have any value but should be 
                                     unique across the file -->
                <type feed-type="N">Replay</type>
                <protocol>TCP/IP</protocol>
                <ip>10.244.0.12</ip>
                <port>10050</port>
                <!--Sender target for replay fix session-->
                <sender>Sender</sender>
                <target>Target</target>
                <feed>A</feed> <!-- Use B instead if this connection 
                                    represents the backup feed of UMDF for this channel -->
            </connection>
        </connections>
    </channel>
    <!-- Add more channels here -->
</configuration>

You must also obtain a valid and up-to-date UMDF FAST template definition file from B3. As of this writing, the template file can be obtained from the following locations:

B3 recommends synchronization of the template file at the start of every trading week.

In addition to the above configuration files the adapter also requires the presence of two files named fixengine.properties and fast.properties, which help specify engine-wide settings. Place these files in your project's top-level directory. The following shows an example of a typical fixengine.properties file:

# Name of the custom package. This property is using for the extending the 
# count of admin-commands. By default, package is null, but if custom commands 
# are present, initialize this property. For example:
# (autostart.acceptor.command.package=com.admin.commands)
autostart.acceptor.commands.package=

autostart.acceptor.targetIds=
autostart.acceptor.FIXADMIN.login=usr
autostart.acceptor.FIXADMIN.password=psw
autostart.acceptor.FIXADMIN.ip=*
autostart.acceptor.FIXADMIN.fixServerListener=com.epam.admintool.AdminTool

# autostart.acceptor.targetIds=admin,admin1
autostart.acceptor.admin.login=admin
autostart.acceptor.admin.password=admin
autostart.acceptor.admin.ip=*
autostart.acceptor.admin.fixServerListener=com.epam.admintool.AdminTool
# Storage type, supported values: Transient or Persistent
# Default values: Transient
#autostart.acceptor.admin.storageType=Transient

# Raw tags. List all tags here engine should treat as raw.
# Raw tag may contain SOH symbol inside it
# and it should be preceided by rawTagLength field.
rawTags=96, 91, 213, 349, 351, 353, 355, 357, 359, 361, 363, 365, 446, 619, 622

# Engine's local IP address to send from. It can be used on a multi-homed host
# for a FIX Engine that will only send IP datagrams from one of its addresses.
# If this parameter is commented, the engine sends IP datagrams from all
# local addresses.
#connectAddress=192.168.0.1

# Limits maximum number of messages during the resend request.
resendRequestNumberOfMessagesLimit=1000

# Reset sequences on switch to backup.
resetOnSwitchToBackup=true

# Reset sequences on switch back to primary connection.
resetOnSwitchToPrimary=true

# Sets disconnect timeout in seconds for Logout ack only when waiting for 
# Logout ack from the counterparty caused due incoming sequence number 
# is less than expected.
forcedLogoffTimeout=60

# Specifies delay between autoreconnect attempts in milliseconds.
autoreconnectDelayInMs=5000

# Specifies number of autoreconnect attempts before give up.
# Negative number = no reconnects, 0 - unlimited number of reconnects,
# positive number = number of reconnect attempts.
# Please use 0 wisely - it means reconnect without limits.
autoreconnectAttempts=0

# Maximum message size supported by this FIX engine instance.
# Should be greater than expected by approximately 1-5%.
maxMessageSize=1048576

# Sends reject if user application is not available.
# If false and client application isn't available acts like
# a "black hole" - accepts and ignores all valid messages.
sendRejectIfApplicationIsNotAvailable=true

# Enable/disable message rejecting. Default value: false.
enableMessageRejecting=true

# Sets queue mode. Set to "false" for persistent queue (slower but no 
# messages will be lost), "true" for in-memory queue (faster but less 
# safe, some messages may be lost).
inMemoryQueue=false

# Sets timeout interval after which connected acceptor session is timed out
# and disposed if logon wasn't received for this session.
loginWaitTimeout=30000

# Include last processed sequence 369 tag in every message for FIX versions >4.2.
includeLastProcessed=true

# Maximum number of messages in a queue before we pause pumper thread to 
# let queued message to be sent out. Set rather high for max performance 
# set 1 or pretty low for realtime experience. 
# 0 - disable queue control, do not pause pumper thread.
queueThresholdSize=3

# The maximum number of messages in buffer before we
# write message to transport.
# Default value is 100
#maxMessagesToSendInBatch=100

# Outgoing storage index. True to support resend request,
# false to never resend messages and always send gap fill.
outgoingStorageIndexed=true

# Enable/Disable force sequence number reset.
# Supported values: Always, OneTime, Never. Default value is: Never.
forceSeqNumReset=Always

# Allows user to replace storage factory with user own implementation.
#storageFactory=com.epam.fixengine.storage.InMemoryStorageFactory

# Enable switch to primary connection. Default value: true.
cyclicSwitchBackupConnection=true

# Enable auto switch to backup connection. Default value: true.
enableAutoSwitchToBackupConnection=true

# Storage directory could be either absolute path (like /tmp/logs or c:\fixengine\logs)
# or relative path e.g. logs (this one is relative to the application start directory).
storageDirectory=${fixaj.home}/logs

# This parameter specifies cleaning mode for message storage of closed sessions.
# Valid values: None | Backup | Delete
# Default value: None
storageCleanupMode=None

# This parameter specifies back-up directory for message logs
# of closed sessions when storageCleanupMode=backup.
# Valid values: existent directory name (relative or absolute path)
# Default value not defined
# See ${fixaj.home} description in the Configuration section of the
# FIX Antenna Java User and Developer Manual.
storageBackupDir=${fixaj.home}/logs/backup

# Specifies the time zone and affects on the time stamp prefix in the FIX in/out logs.
# The parameter should has the following format GMT[±]HH[:MM] or full name.
# For sample logFilesTimeZone=GMT+02:00
#logFilesTimeZone=UTC

# Incoming log filename template. {0} is replaced with actual SenderCompID,
# {1} with actual TargetCompID.
incomingLogFile={0}-{1}.in

# Outgoing log filename template. {0} is replaced with actual SenderCompID,
# {1} with actual TargetCompID.
outgoingLogFile={0}-{1}.out

# Info filename template. {0} is replaced with actual SenderCompID,
# {1} with actual TargetCompID.
sessionInfoFile={0}-{1}.properties

# Out queue file template. {0} is replaced with actual SenderCompID,
# {1} with actual TargetCompID.
outgoingQueueFile={0}-{1}.outq

# Backup incoming log filename template. {0} is replaced with actual 
# SenderCompID, {1} with actual TargetCompID.
backupIncomingLogFile={0}-{1}-{2}.in

# Backup outgoing log filename template. {0} is replaced with actual 
# SenderCompID, {1} with actual TargetCompID.
backupOutgoingLogFile={0}-{1}-{2}.out

# Specifies the time zone and affects the timestamp prefix in the 
# FIX in/out logs. The parameter has the following format:
# GMT[±]HH[:MM] or full time zone name, such as America/Los_Angeles.
# For sample logFilesTimeZone=UTC
#logFilesTimeZone=UTC

# This parameter specifies whether to reset sequence number at
# time defined in resetSequenceTime.
# Valid values: Yes | No
# Default value: No
performResetSeqNumTime=No

# This parameter specifies GMT time when the FIX Engine initiates the reset
# of sequence numbers.
# Valid values: HH:MM:SS time format
# Default value: 00:00:00
resetSequenceTime=00:00:00

# Time zone di for resetSequenceTime property.
# Default value: UTC
resetSequenceTimeZone=UTC

# Enables SSL transport for all initiator sessions.
# For individual session control set this parameter in session configuration.
#enableSSL=false

# Enable/disable Nagle's algorithm for TCP sockets.
#enableNagle=false

# Sets FIX engine global/per message handler(s), handler is called on each and 
# every incoming message.
# Handler could prevent further message processing.
# Note: handler numbers should be unique and sequential!
# DO NOT UPDATE UNLESS YOU KNOW FOR SURE WHAT YOU ARE DOING!!!
# DO NOT EXPOSE FOLLOWING system. prefixed params to end user,
# since they all have default values and most users never need to adjust those
#system.messagehandler.3=com.epam.custom.RejectMessageHandler
#system.messagehandler.global.0=com.epam.fixengine.session.messagehandler.global.DeliverToCompIdMessageHandler
#system.messagehandler.global.2=com.epam.fixengine.custom.GlobalMessageHandler2

# User supplied handlers.
#user.messagehandler.global.0=com.epam.fixengine.session.messagehandler.global.DeliverToCompIdMessageHandler
#user.messagehandler.global.1=com.epam.fixengine.custom.GlobalMessageHandler2


checkSendingTimeAccuracy=false
# Sending time accuracy 2 minutes (120000 ms).
reasonableDelayInMs=120000
measurementAccuracyInMs=10


# Toggle on/off validation of incoming messages according to base of custom dictionaries.
# Following parameters with the Validation suffix works only if this property set to true.
validation=false

# Toggle on/off validation values of fields with 8, 9, 35 & 10 tags.
wellformenessValidation=true

# Toggle on/off validation if fields are allowed in the message.
allowedFieldsValidation=true

# Toggle on/off validation if all required fields presents in the message.
requiredFieldsValidation=true

# Toggle on/off validation of order in which fields appears in the message.
fieldOrderValidation=true

# Toggle on/off validation of duplicated fields in the message.
duplicateFieldsValidation=true

# Toggle on/off validation of field values according defined data types.
fieldTypeValidation=true

# Toggle on/off validation of repeating group fields.
groupValidation=true

# Conditional validation is very much time consuming - use with care.
conditionalValidation=true

# This parameter specifies encryption config file name.
# Valid values: existent valid config file name (relative or absolute path)
# Default value not defined
# See ${fixaj.home} description.
#encryptionConfig=${fixaj.home}/encryption/encryption.cfg

# encryptionMode
# This parameter specifies the default value of encryptionMode.
# Valid values: None | Des | PgpDesMd5
# Default value: None
encryptionMode=None

And here is a typical fast.properties file:

# Specifies the size of stream's queue for incoming packages which should
# be later decoded and processed.
# default value - 200
# decodingQueueSize=200

# Define the minimum required missed message to start recovery.
# The value should be grate than possible distance of mixing UDP packets.
# Otherwise mixing will lead to premature starting of recovery process.
# default value - 9
# minHoleOffset=9

# Define the size of socket receive buffer size in Kb. Increasing buffer may allow
# the network implementation to buffer multiple packets when packets arrive faster 
# than are being received. See details for the setReceiveBufferSize method in the 
java.net.DatagramSocket documentation.
# default value - 1024
# socketReceiveBufferSize=1024

# Define the size of buffer for incoming increments before processing per instrument.
# messageCorridorQueueSize=1000

# Define the size of queue for storing and ordering increments by RptSeqNum(83) per instrument
# messageIncrementQueueSize=1000

# Define a count of full cycles on Snapshot feed before define that there are no snapshots for 
# instruments and they are in actual state. If this property is set to a negative value, then 
# it means there is a need to listen to the snapshot feed until all snapshots for instruments 
# are found. (This is not recommended.)
# Default value - 3
# snapshotCyclesToListen=3

# Define maximum size of gap for recovering via TCP. If you need to recover more messages 
# than defined by this property, Market recovery via Snapshot should be used. Sometimes 
# recovering a large number of messages could affect the time for recovering other 
# instruments (requests are processing one by one). To prevent this, Market recovery 
# could be applied for large gaps.
# Default value - -1 (recovery by TCP any gap)
maxGapForTCP = 500