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);
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.
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?
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>
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>
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>
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?
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.
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);