EMA Consumer - Posting data to Contribution Channel
|Download tutorial source code||Click here to download|
|Last update||May 2020|
|Compilers||JDK 1.7 or greater|
|Prerequisite||Downloaded, installed, compiled and ran an EMA consumer example|
In this tutorial we will explain and use the EMA Java SDK, to contribute data to the Refinitiv Contribution Channel (RCC) directly, without using the TREP infrastructure. The reader must have an understanding of an Elektron Message API and be familiar with consuming OMM Market Price data. You should also have a basic understanding of EMA Configuration. To understand the contribution concepts, please refer to this article.
Contributing Data to Refinitiv Elektron Feed
Currently clients wishing to Contribute data to Refinitiv, typically do so using on-site contribution systems such as MarketLinkIP. In order to meet the evolving needs of the market, Refinitiv have developed a new Contribution service - the Refinitiv Contribution Channel (RCC). Contributions Channel is provided through a cluster of virtual servers that can provide data optimization & conflation capabilities as well as a much higher rate of content onto Elektron.
There are two key mechanisms through which an API developer can contribute data to Refinitiv head-end. Once contributed, this data can be delivered to other customers who have been permissioned to view it. This service is commonly used by banks and funds to make their data available to trading community.
The first option is to use and configure the Refinitiv Enterprise Platform (TREP), to take care of contribution specific connection and login details.
The second option, which is discussed in this tutorial, is for clients who do not have an onsite TREP installation and will connect to RCC through the Internet. To contribute directly to RCC, an EMA application undertakes following four steps:
- Establish an encrypted connection to the server
- Open a Tunnelstream
- Perform a client login
- Contribute data using post messages
Lets look at the EMA Java code starting with application configuration. This tutorial source code is based on EMA sample: series400\example440__System__TunnelStream which ships with the SDK distribution, in the Examples directory.
EMA SDK undertakes all the work required to establish an encrypted connection. All an application developer needs to do, is define the
<Consumer> and set its
<Channel> type to RSSL_ENCRYPTED, in the EMA configuration file EmaConfig.xml. The hostname and the port of the destination RCC server will have to be specified in the configuration as well.
<Channel> <ChannelType value="ChannelType::RSSL_ENCRYPTED"/> <Host value="TRCC_HOST_NAME"/> <Port value="TRCC_PORT"/> </Channel>
Now, when an
OMMConsumer is created with this configuration, EMA will start an encrypted connection. EMA Java uses security and encryption framework provided by the Java SDK, and uses the Sun JDK's default TLS. The Java standard way is to store the security certificate and private keys required for encryption, in a password protected keystore file. This filename and password has to be specified, when creating the
// Create an OMM consumer OmmConsumer consumer = EmaFactory.createOmmConsumer(EmaFactory.createOmmConsumerConfig() .tunnelingKeyStoreFile("KEYSTORE_FILE_NAME") .tunnelingKeyStorePasswd("KEYSTORE_PASSWORD"));
After establishing an encrypted connection, open the tunnel stream to the service which allows contribution.
// Create a request for a tunnel stream TunnelStreamRequest tsr = EmaFactory.createTunnelStreamRequest() .classOfService(cos) .domainType(EmaRdm.MMT_SYSTEM) .name("TUNNEL1") .serviceName("CONTRIBUTION_SERVICE_NAME"); // Send the request and register for events from tunnel stream long tunnelStreamHandle = consumer.registerClient(tsr, appClient);
Here, we used the
OMMConsumer defined earlier (which is connected to TRCC server by now), and open a tunnel stream to contribution service using the system domain. The tunnel stream handle will be used in the next step to login.
For production applications, it is recommended to establish two simultaneous connections, to two different servers. An application should only contribute to a single stream at any given time. The second stream is for redundancy purposes, and should only be used if there are issues with the first one.
A successful opening of tunnel stream is followed by a status event callback:
onStatusMsg, with state
OmmState.StreamState.OPEN. The application should then perform a client login using the provided contribution user's username and password. The password field is no longer required to be obfuscated before sending.
// create a login request message ElementList elementList = EmaFactory.createElementList(); elementList.add(EmaFactory.createElementEntry().ascii("Password", "PASSWORD")); ReqMsg rMsg = EmaFactory.createReqMsg() .domainType(EmaRdm.MMT_LOGIN) .name("LOGIN_USERNAME") .attrib(elementList) .serviceId(statusMsg.serviceId()) .streamId(statusMsg.streamId()); // get events from login substream _subStreamHandle = _ommConsumer.registerClient(rMsg, this, 1, _tunnelStreamHandle);
If the login is accepted by the TRCC server, application will receive a refresh callback:
onRefreshMsg with state
OmmState.StreamState.OPEN. At this point, the application should store the stream handle, and is ready to contribute data.
TRCE:MaxMessagesPerSecond is present in the login response message, it indicates the maximum message posting rate allowed for this application, and should not be exceeded.
To contribute, application can use off-stream posting - i.e. send post messages on the login stream. Application begins by creating the field list for the FID's it intends to contribute to; packs the field list into an update message, and sends this update message as a PostMessage payload. Following snippet shows how to send an update message containing a BID (FID 22) and an ASK (FID 25).
// populate the contributed FIDs and values FieldList nestedFieldList = EmaFactory.createFieldList(); nestedFieldList.add(EmaFactory.createFieldEntry().real(22, _bid++, OmmReal.MagnitudeType.EXPONENT_NEG_1)); nestedFieldList.add(EmaFactory.createFieldEntry().real(25, _ask++, OmmReal.MagnitudeType.EXPONENT_NEG_1)); // create an update message for our item UpdateMsg nestedUpdateMsg = EmaFactory.createUpdateMsg() .streamId(_postStreamID) .name("CONTRIBUTION_RIC") .payload(nestedFieldList); // post this market price message _ommConsumer.submit(EmaFactory.createPostMsg() .streamId(_postStreamID) .postId(_postID++) .domainType(EmaRdm.MMT_MARKET_PRICE) .solicitAck(true) .complete(true) .payload(nestedUpdateMsg), _subStreamHandle);
Here, the application is also requesting, that the contributions be acknowledged by the system, by setting
solicitAck(true). Setting this flag will cause the application to receive acknowledgment callbacks:
onAckMsg. The callback acknowledgment message
AckMsg, will contain attributes
text reason if the contribution is denied.
AckMsg streamId="5" domain="MarketPrice Domain" ackId="1" nackCode="SymbolUnknown" text="Symbol posted has not been recognised" AckMsgEnd
Since this tutorial is based on EMA Example: series400\example440__System__TunnelStream, follow the same build process, to compile this source code.
- The sample EmaConfig.xml provided, assumes that the data dictionary files are present in the local directory. So copy the RDMFieldDictionary and enumtype.def from EMA /etc directory.
- Provide your existing keystore file or create a new one. Use the keytool provided with Java SDK to create a new keystore file in the JKS format:
keytool -genkey -alias mydomain -keyalg RSA -keystore KeyStore.jks -keysize 2048
- Add the Comodo certificate authority to the keystore, if its not trusted. Refer to troubleshooting section on how to enable Java SSL logging and add Comodo CA.
- Modify the source code to reflect this keystore file name and the keystore password.
- Run output:
Contributing to Refinitiv Contributions Channel Starting encrypted connection... Starting tunnel stream... ----- Status message ---- StatusMsg streamId="5" domain="System Domain" privateStream state="Open / Ok / None / ''" name="TUNNEL1" serviceId="10" serviceName="DDS_TRCE" StatusMsgEnd Sending client login request... ----- Refresh message ---- RefreshMsg streamId="5" domain="Login Domain" solicited RefreshComplete privateStream state="Open / Ok / None / 'Login accepted by host c226fmrdbcata'" itemGroup="00 00" name="******" serviceId="10" Attrib dataType="ElementList" ElementList ElementEntry name="Password" dataType="Ascii" value="******" ElementListEnd AttribEnd RefreshMsgEnd Login accepted, starting posting... ----- Ack message ---- AckMsg streamId="5" domain="MarketPrice Domain" ackId="1" AckMsgEnd Continue posting... ----- Ack message ---- AckMsg streamId="5" domain="MarketPrice Domain" ackId="2" AckMsgEnd Continue posting... ----- Ack message ---- AckMsg streamId="5" domain="MarketPrice Domain" ackId="3" AckMsgEnd
Data which is contributed to the production RCC, can be viewed by subscribing to that RIC in either Eikon or any other consumer application, like EMA Consumer sample. If however, you are using the development network for testing purposes, the contributed data is only available by configuring an Elektron Connect to connect to development feed, and then using an EMA consumer application. Please contact your account manager to get the EZD application, license and connection parameters.
Sample logging.properties file for storing log files in /log subdirectory:
.level=FINEST # write logs in console and in file handlers=java.util.logging.FileHandler, java.util.logging.ConsoleHandler # dump warning messages on console java.util.logging.ConsoleHandler.level=WARNING java.util.logging.ConsoleHandler.formatter=java.util.logging.SimpleFormatter # capture debug trace in the file java.util.logging.FileHandler.level=FINEST java.util.logging.FileHandler.pattern=logs/emaj.log # Write 100000 bytes before rotating this file java.util.logging.FileHandler.limit=50000000 # Number of rotating files to be used java.util.logging.FileHandler.count=20 java.util.logging.FileHandler.formatter=java.util.logging.SimpleFormatter # Format timestamp as date/time with millisecond java.util.logging.SimpleFormatter.format=%1$tY-%1$tm-%1$td %1$tH:%1$tM:%1$tS.%1$tL %4$-7s %2$s %n%5$s
EMA Java logging can be enabled, by passing above properties file to JVM at run time. In addition, it might be helpful to turn on the JDK network logging, to view the security and encryption handshake messages, between the API and the server. Following command line parameters to JVM will enable verbose network and EMA logs.
set LOGGING_CONFIG=-Djava.util.logging.config.file=logs\logging.properties set NET_DEBUGGING=-Djavax.net.debug=all java -cp %CLASSPATH% %LOGGING_CONFIG% %NET_DEBUGGING% Contributor
Verify Comodo CA
For an encrypted tunneling connection type, you might encounter trust issues with Comodo certificates. If you encounter this issue, manually download and import the certificate to your keystore file.
- Verify that your keystore does not have this certificate using command
keytool -v -list -keystore KeyStore.jksand look for Comodo.
Alias name: comodo Creation date: 23-Oct-2018 Entry type: trustedCertEntry Owner: CN=COMODO RSA Organization Validation Secure Server CA, O=COMODO CA Limited, L=Salford, ST=Greater Manchester, C=GB Issuer: CN=COMODO RSA Certification Authority, O=COMODO CA Limited, L=Salford, ST=Greater Manchester, C=GB Serial number: 36825e7fb5a481937ef6d1736bb93ca6 Valid from: Tue Feb 11 19:00:00 EST 2014 until: Sun Feb 11 18:59:59 EST 2029 Certificate fingerprints: MD5: 91:29:73:32:12:D5:F5:65:18:64:D8:6B:7B:AE:0B:98 SHA1: 10:4C:63:D2:54:6B:80:21:DD:10:5E:9F:BA:5A:8D:78:16:9F:6B:32 SHA256: 11:10:06:37:8A:FB:E8:E9:9B:B0:2B:A8:73:90:CA:42:9F:CA:27:73:F7:4D:7F:7E:B5:74:4F:5D:DF:68:01:4B Signature algorithm name: SHA384withRSA Subject Public Key Algorithm: 2048-bit RSA key Version: 3
- Backup your keystore file in case there is a problem importing the certificate.
- Download the Comodo certificate: http://crt.comodoca.com/COMODORSAOrganizationValidationSecureServerCA.crt
- Import into your keystore file:
keytool -import -alias comodo -keystore KeyStore.jks -file COMODORSAOrganizationValidationSecureServerCA.crt
For more information on the use of keystores or understanding encryption in Java language, please refer to this article.