Listening for connection errors for RTO

simon.dando · June 2022

We are connected to RTO 24/7 and we sometimes see connection errors during your maintenance window. When implementing a solution to reconnect, we didn't see that we could listen for connection errors and hence reconnect some time after seeing the event. Please advise if we can do this or pls raise an enhancement request. Currently if we see a connection error when requesting data we can only reconnect then (and probably will require retries) hence our clients won't receive market data in a timely fashion.

This is the usual connection error that we see:

2022-05-29 05:05:42,629 [pool-5-thread-1] ERROR c.r.ema.access.OmmConsumerImpl - loggerMsg

ClientName: LoginCallbackClient

Severity: Error

Text: RDMLogin stream was closed with status message

username <removed>

usernameType 1

State: Closed/Suspect/Not entitled - text: "Force Logout from DACS."

loggerMsgEnd

Jirapongse · June 2022

@simon.dando

I tested the following code in ex113.

consumer  = EmaFactory.createOmmConsumer(config.consumerName(connectWebSocket ? "Consumer_5" : "Consumer_4"));

LoginReq loginReq = EmaFactory.Domain.createLoginReq();
consumer.registerClient(loginReq.message(), appClient);

It works properly.

The example can get the login refresh and status messages properly.

    public void onRefreshMsg(RefreshMsg refreshMsg, OmmConsumerEvent event)
    {
        if(refreshMsg.domainType() == EmaRdm.MMT_LOGIN)
        {
            System.out.println("#### Login Refresh #####");
        }
        System.out.println(refreshMsg);
    }


    public void onUpdateMsg(UpdateMsg updateMsg, OmmConsumerEvent event)
    {
        if(updateMsg.domainType() == EmaRdm.MMT_LOGIN)
        {
            System.out.println("#### Login Update #####");
        }
        System.out.println(updateMsg);
    }


    public void onStatusMsg(StatusMsg statusMsg, OmmConsumerEvent event)
    {
        if(statusMsg.domainType() == EmaRdm.MMT_LOGIN)
        {
            System.out.println("#### Login Status #####");
        }
        System.out.println(statusMsg);
    }

The output is:

#### Login Refresh #####
RefreshMsg
    streamId="1"
    domain="Login Domain"
    solicited
    RefreshComplete
    state="Open / Ok / None / 'Login accepted by host WIN-2NOVD7CLMV1'"
    itemGroup="00 00"
    Attrib dataType="ElementList"
        ElementList
            ElementEntry name="ApplicationId" dataType="Ascii" value="256"
            ElementEntry name="ApplicationName" dataType="Ascii" value="rsslProvider"
            ElementEntry name="Position" dataType="Ascii" value="192.168.0.184/WIN-2NOVD7CLMV1"
            ElementEntry name="ProvidePermissionProfile" dataType="UInt" value="1"
            ElementEntry name="ProvidePermissionExpressions" dataType="UInt" value="1"
            ElementEntry name="SingleOpen" dataType="UInt" value="0"
            ElementEntry name="AllowSuspectData" dataType="UInt" value="1"
            ElementEntry name="SupportPauseResume" dataType="UInt" value="0"
            ElementEntry name="SupportOptimizedPauseResume" dataType="UInt" value="0"
            ElementEntry name="SupportOMMPost" dataType="UInt" value="1"
            ElementEntry name="SupportViewRequests" dataType="UInt" value="0"
            ElementEntry name="SupportBatchRequests" dataType="UInt" value="5"
            ElementEntry name="SupportStandby" dataType="UInt" value="0"
        ElementListEnd
    AttribEnd
RefreshMsgEnd
...
...
#### Login Status #####
StatusMsg
    streamId="1"
    domain="Login Domain"
    state="Closed / Suspect / Not entitled / 'Simulated Force Logout from DACS.

Jirapongse · June 2022

@simon.dando

You can refer to the answers on this thread.

In summary, you can register to get the login closed event, and then create a new OMMConsumer to re-open the login stream.

For an enhancement request, you can raise this issue on GitHub.

simon.dando · June 2022

Thanks, that should pretty much solve our problem since the only connection error we have seen so far is when we have been forcibly logged out. Our existing code to create a new consumer upon error should cater for all other scenarios. I'm a bit surprised you can't listen for errors in general but if one of your HFT clients hasn't mentioned it then it probably isn't an issue.

simon.dando · June 2022

Jirapongse. I've looked at the answer mentioned in the thread above and it seems that the latest library no longer supports the specific Login request messages. Would this be the equivalent in the latest ema lib? (Since it may take up to 3 weeks to see the problem it would be good to be certain sooner rather than later). Thanks.

ommConsumer.registerClient(EmaFactory.createReqMsg().domainType(LOGIN).name(username), logoutListener);

simon.dando · June 2022

Jirapongse. I've looked at the answer mentioned in the thread above and it seems that the latest library no longer supports the specific Login request messages. Would this be the equivalent in the latest ema lib? (Since it may take up to 3 weeks to see the problem it would be good to be certain sooner rather than later). Thanks.


  ommConsumer.registerClient(EmaFactory.createReqMsg().domainType(LOGIN).name(username), logoutListener);

simon.dando · June 2022

Hi @Jirapongse - we were able to test this over the weekend however we ran into some problems when reinitialising the OmmConsumerImpl which left us in a state where every request for market data had a response that the service was not available (until we rebooted the service).

We saw the force logout:

2022-06-19 04:58:37,594 [pool-6-thread-1] INFO  c.c.m.r.RefinitivLogoutOmmConsumerClient - Status msg received: StatusMsg
    streamId="1"
    domain="Login Domain"
    state="Closed / Suspect / Not entitled / 'Force Logout from DACS.'"
    name="AQIC5wM2LY4SfczT8oWhac76IOgLWUWiZ45vTbWWiHjVit8%3D%40AAJTSQACMzAAAlNLABM1NDYzODQ0Mjk5MzM4NDgwMDgwAAJTMQACMjQ%3D%23"
    nameType="1"
StatusMsgEnd

And tried uninitialising and saw

2022-06-19 04:58:40,594 [pool-6-thread-1] ERROR c.r.ema.access.OmmConsumerImpl - loggerMsg
    ClientName: Consumer_4_1
    Severity: Error
    Text:    Failed to uninitialize OmmBaseImpl (_executor.awaitTermination() timed out).
loggerMsgEnd

Created a new consumer and logged in successfully

2022-06-19 04:58:52,204 [pool-11-thread-1] INFO  c.c.m.r.RefinitivLogoutOmmConsumerClient - Login Refresh msg received: RefreshMsg
    streamId="1"
    domain="Login Domain"
    solicited
    RefreshComplete
    state="Open / Ok / None / 'Login accepted by host ads-fanout-sm-az2-apse1-prd.'"

but then saw

2022-06-19 04:58:52,203 [pool-6-thread-1] ERROR c.r.ema.access.OmmConsumerImpl - loggerMsg
    ClientName: Consumer_4_1
    Severity: Error
    Text:    Failed to close reactor channel (rsslReactorChannel).' RsslChannel='0Error Id -1Internal sysError 0Error Location ReactorChannel.closeError Text Reactor is shutdown, close aborted.'. 
loggerMsgEnd

and then were always told:

2022-06-19 10:01:09,209 [pool-11-thread-1] INFO  c.c.m.r.RefinitivOmmConsumerClient - Status message received: StatusMsg
    streamId="5"
    domain="MarketPrice Domain"
    state="Open / Suspect / None / 'Service not available'"
    name="SPK.NZ"
    serviceName="ELEKTRON_DD"
StatusMsgEnd

This was presumably caused by some sort of upgrade during the maintenance window. I'm unclear as to how we can reinitialise our connection asap once we detect the force logout.

Jirapongse · June 2022

@simon.dando

From the log, the new OmmConsumer can connect to the server but the service (ELEKTRON_DD) is not available.

You may verify if the application uses the latest version of RTSDK (Real-Time-SDK-2.0.5.L1). If not, you may consider upgrading the application to use the latest version of RTSDK. You can download it from here.

You can register the source directory request to receive the source directory response messages. Therefore, we can verify the state of the service.

The code looks like this:

consumer.registerClient(EmaFactory.createReqMsg().domainType(EmaRdm.MMT_DIRECTORY), appClient);

The source response looks like this:

#### Source Refresh #####
RefreshMsg
    streamId="5"
    domain="Directory Domain"
    solicited
    RefreshComplete
    state="Open / Ok / None / ''"
    itemGroup="00 00"
    filter="0"
    Payload dataType="Map"
        Map
            MapEntry action="Add" key dataType="UInt" value="1" dataType="FilterList"
                FilterList
                    FilterEntry action="Set" filterId="1 dataType="ElementList"
                        ElementList
                            ElementEntry name="Name" dataType="Ascii" value="ELEKTRON_DD"
                            ElementEntry name="Vendor" dataType="Ascii" value="Refinitiv"
                            ElementEntry name="IsSource" dataType="UInt" value="1"
                            ElementEntry name="Capabilities" dataType="OmmArray"

With this message, we may be able to verify the state of the service when the new consumer can connect to the server.

simon.dando · June 2022

@Jirapongse the problem is that we are told that the service is unavailable until we rebooted. That would imply that we need to keep recreating the OmmConsumerImpl until the service is ok or am i missing something? Not sure about our rtsdk lib but com.refinitiv.ema/eta are on 3.6.3.1.

Jirapongse · June 2022

@simon.dando

I would like to check if the application received any source directory refresh or update messages from the server.

The best way to verify it is to enable XML tracing (XmlTraceToStdout) in the Consumer configuration but the log will be very large.

        <Consumer>         
            ...
            <XmlTraceToStdout value="1"/>
        </Consumer>

Otherwise, the application can register the source directory request to receive the source directory response messages.

Real-Time-SDK-2.0.5.L1 contains EMA/ETA 3.6.5.L1. The new Real-Time-SDK-2.0.6.L1containing EMA/ETA 3.6.6.L1 has been released. You can download it from here.

Listening for connection errors for RTO

Best Answer

Answers

Categories