EMA Java - Is it possible to open a stream in a paused state?

Is there a way to open a stream and make it not publish data until another request is made? The use case is that a warm server is on standby, it needs to make subscriptions in advance and pause the streams, so that when it takes over, it can resume/unpause and start processing OMM data as quickly as possible.

I've tried the EMA Java code below, the pause boolean has no effect. If I instead send a pause request immediately after the initial request, there's a chance that some data may get through during that period which I want to avoid.

EmaFactory.createReqMsg().serviceName("theService").name("theName").pause(true);

Find more posts tagged with

java

elektron

request

elektron-sdk

refinitiv-realtime

rrt

elektron-message-api

ema-api

Accepted answers

veerapath.rungruengrayubkul

Hi @ahmire.kailer,

The "SupportPauseResume" = "0" means that your connected server doesn't support/enable pause and resume feature. You could contact server administrator to enable the feature to make pause work.

All comments

pimchaya.wongrukun01

Hello @ahmire.kailer

I tried example350__MarketPrice__PauseAll application shipped with EMA package. Sometime I got some data messages after the pause request was sent out as shown below:

As far as I understand this happened because the data messages were sent out from the server before the server proceeded the pause request. After that I did not receive any data messages. In my opinion, it was not 100% that you will not get any data message after the pause request is sent out.

ahmire.kailer

Hi @Pimchaya.Wongrukun

Thank you for your answer.

I checked example350__MarketPrice__PauseAll, that example sends pause request after the initial request. I assume there isn't a way to make a paused subscription? Subscriptions/streams could only be paused after opening the stream which means some data may get through?

veerapath.rungruengrayubkul

Hi @ahmire.kailer,

I have tried the following code to send the initial request with pause flag on the EMA Java 1.5 version. Application will receive only Refresh message, and then continue receiving updates after resume all. My XML tracing file is attached. console.txt

    ReqMsg reqMsg = EmaFactory.createReqMsg(
    long loginHandle = consumer.registerClient(reqMsg.domainType( EmaRdm.MMT_LOGIN ), appClient);

    appClient.setOmmConsumer(consumer, loginHandle);
    
    //request item with pause
    consumer.registerClient(reqMsg.clear().serviceName("API_ELEKTRON_EDGE_TOKYO").name("EUR=").pause(true), appClient, 1);
    
    Thread.sleep(10000);
    //Resume all
    consumer.reissue(reqMsg.clear().domainType( EmaRdm.MMT_LOGIN ).initialImage(false).interestAfterRefresh(true).name("user"), loginHandle);
    
    Thread.sleep(60000);            // API calls onRefreshMsg(), onUpdateMsg() and onStatusMsg()

If the code doesn't work in your environment, please enable the XML message tracing and share the console output. The tracing can be enabled by add the XmlTraceToStdout parameter in your EMAConfig.xml file.

<Consumer>
    <!-- Name is mandatory                                                                    -->
    <Name value="Consumer_1"
    <!-- Channel is optional: defaulted to "RSSL_SOCKET + localhost + 14002"                    -->
    <!-- Channel or ChannelSet may be specified                                                    -->
    <Channel value="Channel_1"/>

    <!-- Dictionary is optional: defaulted to "ChannelDictionary"                                -->
    <Dictionary value="Dictionary_2"/>
    <XmlTraceToStdout value="1"/>
</Consumer>

ahmire.kailer

Hi @veerapath.rungruengrayubkul

In my environment, .pause(true) in the initial ReqMsg has no effect. both RefreshMsg and UpdateMsg keep coming through just like normal. I was using EMA Java 1.4, then I changed to EMA Java 1.5, it made no difference. Redacted XML output when using EMA Java 1.5 is included below.

<!-- rwfMajorVer="14" rwfMinorVer="1" -->
<REQUEST domainType="MARKET_PRICE" streamId="6" containerType="NO_DATA" flags="0x246 (HAS_PRIORITY|STREAMING|HAS_QOS|PAUSE)" Qos: Realtime/TickByTick/Static - timeInfo: 0 - rateInfo: 0 priorityClass="1" priorityCount="1" dataSize="0">
    <key flags="0x03 (HAS_SERVICE_ID|HAS_NAME)" serviceId="000" name="REDACTED"/>
    <dataBody>
    </dataBody>
</REQUEST>

<!-- rwfMajorVer="14" rwfMinorVer="1" -->
<REFRESH domainType="MARKET_PRICE" streamId="6" containerType="FIELD_LIST" flags="0x1EA (HAS_PERM_DATA|HAS_MSG_KEY|SOLICITED|REFRESH_COMPLETE|HAS_QOS|CLEAR_CACHE)" groupId="20" permData="0000 0000 0000" Qos: Realtime/TickByTick/Static - timeInfo: 0 - rateInfo: 0 State: Open/Ok/None - text: "" dataSize="3365">
    <key flags="0x07 (HAS_SERVICE_ID|HAS_NAME|HAS_NAME_TYPE)" serviceId="000" name="REDACTED" nameType="1"/>
    <dataBody>
        <fieldList flags="0x08 (HAS_STANDARD_DATA)">
            <fieldEntry fieldId="1" data="0000"/>
        </fieldList>
    </dataBody>
</REFRESH>

<!-- rwfMajorVer="14" rwfMinorVer="1" -->
<UPDATE domainType="MARKET_PRICE" streamId="6" containerType="FIELD_LIST" flags="0x00" updateType="0" dataSize="175">
    <dataBody>
        <fieldList flags="0x08 (HAS_STANDARD_DATA)">
            <fieldEntry fieldId="1" data="1111"/>
        </fieldList>
    </dataBody>
</UPDATE>

ahmire.kailer

XML output for EUR= is included below

Request code:

EmaFactory.createReqMsg().serviceName("serviceName").name("EUR=").pause(true);

<!-- rwfMajorVer="14" rwfMinorVer="1" -->
<REQUEST domainType="MARKET_PRICE" streamId="5" containerType="NO_DATA" flags="0x246 (HAS_PRIORITY|STREAMING|HAS_QOS|PAUSE)" Qos: Realtime/TickByTick/Static - timeInfo: 0 - rateInfo: 0 priorityClass="1" priorityCount="1" dataSize="0">
    <key flags="0x03 (HAS_SERVICE_ID|HAS_NAME)" serviceId="000" name="EUR="/>
    <dataBody>
    </dataBody>
</REQUEST>

<!-- rwfMajorVer="14" rwfMinorVer="1" -->
<REFRESH domainType="MARKET_PRICE" streamId="5" containerType="FIELD_LIST" flags="0x1FA (HAS_PERM_DATA|HAS_MSG_KEY|HAS_SEQ_NUM|SOLICITED|REFRESH_COMPLETE|HAS_QOS|CLEAR_CACHE)" groupId="35" seqNum="61408" permData="0301 0152 6C" Qos: Realtime/TickByTick/Static - timeInfo: 0 - rateInfo: 0 State: Open/Ok/None - text: "All is well" dataSize="2072">
    <key flags="0x07 (HAS_SERVICE_ID|HAS_NAME|HAS_NAME_TYPE)" serviceId="000" name="EUR=" nameType="1"/>
    <dataBody>
        <fieldList flags="0x09 (HAS_FIELD_LIST_INFO|HAS_STANDARD_DATA)" fieldListNum="99" dictionaryId="1">
            <fieldEntry fieldId="1" data="0000"/>
        </fieldList>
    </dataBody>
</REFRESH>

<!-- rwfMajorVer="14" rwfMinorVer="1" -->
<UPDATE domainType="MARKET_PRICE" streamId="5" containerType="FIELD_LIST" flags="0x90 (HAS_SEQ_NUM|DO_NOT_CONFLATE)" updateType="0" seqNum="61438" dataSize="233">
    <dataBody>
        <fieldList flags="0x08 (HAS_STANDARD_DATA)">
            <fieldEntry fieldId="1" data="0000"/>
        </fieldList>
    </dataBody>
</UPDATE>

ahmire.kailer

I've just found this entry in one of the responses.

<elementEntry name="SupportPauseResume" dataType="UINT" data="0"/>

I assume I need to turn it on on the server side to make pause work?

I also tried to request for a blank view to suppress the data by adding .payload(EmaFactory.createElementList()) to the initial request, but it had no effect. Apart from using pause, is there another way to make a placeholder subscription and make it not send data (or send completely blank data) until another reissue request is made?

veerapath.rungruengrayubkul

Hi @ahmire.kailer,

The "SupportPauseResume" = "0" means that your connected server doesn't support/enable pause and resume feature. You could contact server administrator to enable the feature to make pause work.

ahmire.kailer

Hi @veerapath.rungruengrayubkul

How do I resume after sending initial request with pause? I tried below and only got one RefreshMsg, there're no UpdateMsg after that. If I send the initial request without pause, then pause, then invoke the function below, it works fine, but that's not what I want, I need to include pause in the initial request.

consumer.reissue(reqMsg.clear(), instrumentHandle);

EXPLORE OUR SITES