Migrating legacy SFC (RTMDS v3) Java client to EMA

Thanks Wasin.

To answer your questions from before:

• EMA Java version - 3.6.7.3 (Maven)
• Java version - 17
• OS - Both Linux and Windows
• What is the server? ads3.7.0.L1.linux.rrg 64-bit

Turns out that port 14002 was firewalled; I can now connect fine, and log in successfully.

However, as soon as I subscribe to a RIC, the channel immediately goes down and starts reconnecting.

Key events from the log:
- Channel opened successfully on ALC_Primary.
- Login refresh received and accepted by host (Login accepted by host ...).
- Directory refresh received, state open/ok.
- OmmConsumer created successfully.
- Subscribed to RIC: AUCASH=RBAA on service hEDD.
- Immediately after subscription, the status message shows: Received ChannelDownReconnecting event on channel ALC_PrimaryError text: Connection reset

We’re currently using "hEDD" as the service name in the request. This works in our SFC client, and we were told it should also work with the EMA API.

Is hEDD actually a valid service name for EMA subscriptions, or should we be using one of the service names returned in the source directory?

Here's my test client:

package com.yieldbroker.ema;

import com.refinitiv.ema.access.AckMsg;
import com.refinitiv.ema.access.EmaFactory;
import com.refinitiv.ema.access.GenericMsg;
import com.refinitiv.ema.access.Msg;
import com.refinitiv.ema.access.OmmConsumer;
import com.refinitiv.ema.access.OmmConsumerClient;
import com.refinitiv.ema.access.OmmConsumerConfig;
import com.refinitiv.ema.access.OmmConsumerEvent;
import com.refinitiv.ema.access.RefreshMsg;
import com.refinitiv.ema.access.StatusMsg;
import com.refinitiv.ema.access.UpdateMsg;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.atomic.AtomicBoolean;

public class EmaClient implements AutoCloseable {

    private static final Logger LOG = LogManager.getLogger(EmaClient.class);

    // Default values
    private static final String DEFAULT_CONFIG_PATH = "etc/EmaConfig.xml";
    private static final String DEFAULT_CONSUMER_NAME = "PriceConsumer";
    private static final String DEFAULT_USERNAME = "<Replace Me>";
    private static final String DEFAULT_SERVICE_NAME = "hEDD";
    private static final List<String> DEFAULT_RICS = List.of("AUCASH=RBAA");

    private final String consumerConfigPath;
    private final String consumerName;
    private final String username;
    private final List<String> rics;
    private final String serviceName;

    private final AtomicBoolean started = new AtomicBoolean(false);
    private final List<Long> itemHandles = new ArrayList<>();
    private volatile OmmConsumer consumer;

    public EmaClient(String consumerConfigPath, String consumerName, String username, List<String> rics, String serviceName) {
        this.consumerConfigPath = consumerConfigPath;
        this.consumerName = consumerName;
        this.username = username;
        this.rics = rics;
        this.serviceName = serviceName;
    }

    public static void main(String[] args) throws InterruptedException {
        String configPath = args.length > 0 ? args[0] : DEFAULT_CONFIG_PATH;
        String consumerName = args.length > 1 ? args[1] : DEFAULT_CONSUMER_NAME;
        String username = args.length > 2 ? args[2] : DEFAULT_USERNAME;
        List<String> rics = args.length > 3 ? List.of(args[3].split(",")) : DEFAULT_RICS;
        String serviceName = args.length > 4 ? args[4] : DEFAULT_SERVICE_NAME;

        LOG.info("Using configPath={}, consumerName={}, username={}, serviceName={}, RICs={}",
                configPath, consumerName, username, serviceName, rics);

        EmaClient client = new EmaClient(configPath, consumerName, username, rics, serviceName);
        client.start();

        // Keep the client running to receive messages
        Thread.sleep(10_000);

        client.close();
    }

    public synchronized void start() {
        if (!started.compareAndSet(false, true)) {
            return;
        }

        LOG.info("Starting EMA client...");

        try {
            OmmConsumerConfig cfg = EmaFactory.createOmmConsumerConfig(consumerConfigPath)
                    .consumerName(consumerName)
                    .username(username)
                    .operationModel(OmmConsumerConfig.OperationModel.API_DISPATCH);

            consumer = EmaFactory.createOmmConsumer(cfg, new LoggingClient());
            LOG.info("OmmConsumer created successfully");

            // Subscribe to the RICs
            for (String ric : rics) {
                long handle = consumer.registerClient(
                        EmaFactory.createReqMsg().name(ric).serviceName(serviceName),
                        new LoggingClient(),
                        null
                );
                itemHandles.add(handle);
                LOG.info("Subscribed to RIC: {} on service: {}", ric, serviceName);
            }
        } catch (Exception e) {
            LOG.error("Failed to create OmmConsumer or subscribe", e);
        }
    }

    @Override
    public synchronized void close() {
        if (!started.getAndSet(false)) {
            return;
        }

        LOG.info("Closing EMA client...");

        try {
            if (consumer != null) {
                for (Long handle : itemHandles) {
                    consumer.unregister(handle);
                }
                itemHandles.clear();
                LOG.info("All RIC subscriptions unregistered");

                consumer.uninitialize();
            }
        } catch (Exception e) {
            LOG.warn("Error during close", e);
        }

        consumer = null;
        LOG.info("EMA client closed.");
    }

    private static class LoggingClient implements OmmConsumerClient {

        @Override
        public void onRefreshMsg(RefreshMsg refreshMsg, OmmConsumerEvent event) {
            LOG.info("REFRESH Msg - Domain: {}, StreamState: {}, DataState: {}, Text: {}, Msg: {}",
                    refreshMsg.domainType(),
                    refreshMsg.state().streamStateAsString(),
                    refreshMsg.state().dataStateAsString(),
                    refreshMsg.state().statusText(),
                    refreshMsg);
        }

        @Override
        public void onUpdateMsg(UpdateMsg updateMsg, OmmConsumerEvent event) {
            LOG.info("UPDATE Msg - Domain: {}, Msg: {}", updateMsg.domainType(), updateMsg);
        }

        @Override
        public void onStatusMsg(StatusMsg statusMsg, OmmConsumerEvent event) {
            LOG.info("STATUS Msg - Domain: {}, StreamState: {}, DataState: {}, Text: {}, Msg: {}",
                    statusMsg.domainType(),
                    statusMsg.hasState() ? statusMsg.state().streamStateAsString() : "N/A",
                    statusMsg.hasState() ? statusMsg.state().dataStateAsString() : "N/A",
                    statusMsg.hasState() ? statusMsg.state().statusText() : "N/A",
                    statusMsg);
        }

        @Override
        public void onGenericMsg(GenericMsg genericMsg, OmmConsumerEvent event) {
            LOG.info("GENERIC Msg - Domain: {}, Msg: {}", genericMsg.domainType(), genericMsg);
        }

        @Override
        public void onAckMsg(AckMsg ackMsg, OmmConsumerEvent event) {
            LOG.info("ACK Msg - Domain: {}, Msg: {}", ackMsg.domainType(), ackMsg);
        }

        @Override
        public void onAllMsg(Msg msg, OmmConsumerEvent event) {
            LOG.debug("ALL Msg - Class: {}, Domain: {}, Msg: {}", msg.getClass().getSimpleName(), msg.domainType(), msg);
        }
    }
}
 

I'm getting an entry for hEDD in the directory listing so my guess is that the service name isn't the problem:

<mapEntry flags="0x00" action="ADD" key="425" >
<filterList containerType="ELEMENT_LIST" countHint="0" flags="0x00">
<filterEntry id="1" action="SET" flags="0x00" containerType="ELEMENT_LIST">
<elementList flags="0x08 (HAS_STANDARD_DATA)">
<elementEntry name="Name" dataType="ASCII_STRING" data="hEDD"/>
<elementEntry name="SupportsQoSRange" dataType="UINT" data="0"/>
<elementEntry name="QoS" dataType="ARRAY">
<array itemLength="0" primitiveType="QOS">
<arrayEntry Qos: Realtime/TickByTick/Static - timeInfo: 0 - rateInfo: 0/>
</array>
</elementEntry>
<elementEntry name="Capabilities" dataType="ARRAY">
<array itemLength="1" primitiveType="UINT">
<arrayEntry data="5"/>
<arrayEntry data="6"/>
<arrayEntry data="7"/>
<arrayEntry data="8"/>
<arrayEntry data="9"/>
<arrayEntry data="10"/>
<arrayEntry data="11"/>
<arrayEntry data="13"/>
<arrayEntry data="14"/>
<arrayEntry data="18"/>
<arrayEntry data="19"/>
<arrayEntry data="20"/>
<arrayEntry data="21"/>
<arrayEntry data="23"/>
<arrayEntry data="24"/>
<arrayEntry data="25"/>
<arrayEntry data="26"/>
<arrayEntry data="28"/>
<arrayEntry data="29"/>
<arrayEntry data="33"/>
<arrayEntry data="34"/>
<arrayEntry data="35"/>
<arrayEntry data="126"/>
</array>
</elementEntry>
<elementEntry name="DictionariesProvided" dataType="ARRAY">
<array itemLength="0" primitiveType="ASCII_STRING">
<arrayEntry data="RWFFld"/>
<arrayEntry data="RWFEnum"/>
</array>
</elementEntry>
<elementEntry name="DictionariesUsed" dataType="ARRAY">
<array itemLength="0" primitiveType="ASCII_STRING">
<arrayEntry data="RWFFld"/>
<arrayEntry data="RWFEnum"/>
</array>
</elementEntry>
<elementEntry name="Vendor" dataType="ASCII_STRING" data="Thomson Reuters"/>
<elementEntry name="AcceptingConsumerStatus" dataType="UINT" data="1"/>
</elementList>
</filterEntry>
<filterEntry id="2" action="SET" flags="0x00" containerType="ELEMENT_LIST">
<elementList flags="0x08 (HAS_STANDARD_DATA)">
<elementEntry name="ServiceState" dataType="UINT" data="1"/>
<elementEntry name="AcceptingRequests" dataType="UINT" data="1"/>
</elementList>
</filterEntry>
<filterEntry id="4" action="SET" flags="0x00" containerType="ELEMENT_LIST">
<elementList flags="0x08 (HAS_STANDARD_DATA)">
<elementEntry name="OpenLimit" dataType="UINT" data="2500"/>
</elementList>
</filterEntry>
</filterList>
</mapEntry>

Here are the relevant trace logs around the time of the disconnection, am I doing something stupid? I see the same behaviour regardless of the RIC:

<!-- Outgoing Reactor message -->
<!-- java.nio.channels.SocketChannel[connected local=/10.95.210.54:60847 remote=/xxx.xxx.xxx.x:14002] -->
<!-- Tue Sep 16 18:47:44 AEST 2025 -->
 
<!-- rwfMajorVer="14" rwfMinorVer="1" -->
<REQUEST domainType="MARKET_PRICE" streamId="3" containerType="NO_DATA" flags="0x46 (HAS_PRIORITY|STREAMING|HAS_QOS)" Qos: Realtime/TickByTick/Static - timeInfo: 0 - rateInfo: 0 priorityClass="1" priorityCount="1" dataSize="0">
<key flags="0x03 (HAS_SERVICE_ID|HAS_NAME)" serviceId="425" name="AUCASH=RBAA"/>
<dataBody>
</dataBody>
</REQUEST>
 
18:47:44.599 [main] INFO  com.yieldbroker.ema.EmaClient - Subscribed to RIC: AUCASH=RBAA on service: hEDD
18:47:44.611 [pool-3-thread-1] WARN  com.refinitiv.ema.access.OmmConsumerImpl - loggerMsg
    ClientName: ChannelCallbackClient
    Severity: Warning
    Text:    Received ChannelDownReconnecting event on channel ALC_Primary
	RsslReactor @3bffddff
	RsslChannel @7609ccd2
	Error Id 0
	Internal sysError 0
	Error Location null
	Error text Connection reset
loggerMsgEnd

BTW, here's my EMA config, I forgot to post this before:

<EmaConfig>
    <ConsumerGroup>
        <DefaultConsumer value="PriceConsumer"/>
        <ConsumerList>
            <Consumer>
                <Name value="PriceConsumer"/>
                <Channel value="ALC_Primary"/>
                <Dictionary value="Dictionary_1"/>
                <XmlTraceToStdout value="1"/>
            </Consumer>
        </ConsumerList>
    </ConsumerGroup>

    <ChannelGroup>
        <ChannelList>
            <Channel>
                <Name value="ALC_Primary"/>
                <ChannelType value="ChannelType::RSSL_SOCKET"/>
                <Host value="***.***.***.***"/>
                <Port value="14002"/>
            </Channel>
            <Channel>
                <Name value="ALC_Secondary"/>
                <ChannelType value="ChannelType::RSSL_SOCKET"/>
                <Host value="***.***.***.***"/>
                <Port value="14002"/>
            </Channel>
            <Channel>
                <Name value="IPSec_Japan_1"/>
                <ChannelType value="ChannelType::RSSL_SOCKET"/>
                <Host value="***.***.***.***"/>
                <Port value="14002"/>
            </Channel>
            <Channel>
                <Name value="IPSec_Japan_2"/>
                <ChannelType value="ChannelType::RSSL_SOCKET"/>
                <Host value="***.***.***.***"/>
                <Port value="14002"/>
            </Channel>
        </ChannelList>
    </ChannelGroup>

    <DictionaryGroup>
        <DictionaryList>
            <Dictionary>
                <Name value="Dictionary_1"/>
                <DictionaryType value="DictionaryType::FileDictionary"/>
                <RdmFieldDictionaryFileName value="./etc/RDMFieldDictionary"/>
                <EnumTypeDefFileName value="./etc/enumtype.def"/>
            </Dictionary>
        </DictionaryList>
    </DictionaryGroup>

    <!--    <DictionaryGroup>-->
    <!--        <DictionaryList>-->
    <!--            <Dictionary>-->
    <!--                <Name value="Dictionary_1"/>-->
    <!--                <DictionaryType value="DictionaryType::RDM_FIELD_DICTIONARY"/>-->
    <!--                &lt;!&ndash; No local file paths, EMA will fetch from provider &ndash;&gt;-->
    <!--            </Dictionary>-->
    <!--        </DictionaryList>-->
    <!--    </DictionaryGroup>-->
</EmaConfig>

I'm seeing an entry for hEDD in the directory list result so my guess is that this isn't the problem.

Here are the trace logs just after the subscription request am I doing something stupid?

<!-- Outgoing Reactor message -->
<!-- java.nio.channels.SocketChannel[connected local=/10.95.210.54:60847 remote=/xxx.xxx.xxx.xxx:14002] -->
<!-- Tue Sep 16 18:47:44 AEST 2025 -->
 
<!-- rwfMajorVer="14" rwfMinorVer="1" -->
<REQUEST domainType="MARKET_PRICE" streamId="3" containerType="NO_DATA" flags="0x46 (HAS_PRIORITY|STREAMING|HAS_QOS)" Qos: Realtime/TickByTick/Static - timeInfo: 0 - rateInfo: 0 priorityClass="1" priorityCount="1" dataSize="0">
<key flags="0x03 (HAS_SERVICE_ID|HAS_NAME)" serviceId="425" name="AUCASH=RBAA"/>
<dataBody>
</dataBody>
</REQUEST>
 
18:47:44.599 [main] INFO  com.yieldbroker.ema.EmaClient - Subscribed to RIC: AUCASH=RBAA on service: hEDD
18:47:44.611 [pool-3-thread-1] WARN  com.refinitiv.ema.access.OmmConsumerImpl - loggerMsg
    ClientName: ChannelCallbackClient
    Severity: Warning
    Text:    Received ChannelDownReconnecting event on channel ALC_Primary
	RsslReactor @3bffddff
	RsslChannel @7609ccd2
	Error Id 0
	Internal sysError 0
	Error Location null
	Error text Connection reset
loggerMsgEnd
 
 
18:47:44.611 [pool-3-thread-1] DEBUG com.yieldbroker.ema.EmaClient - ALL Msg - Class: StatusMsgImpl, Domain: 1, Msg: StatusMsg
    streamId="1"
    domain="Login Domain"
    state="Open / Suspect / None / 'channel down'"
    name="xxxxxxxxxxxxxxxxxx"
    nameType="1"
StatusMsgEnd