Interrupted streaming service from ERT in cloud with MRN
I am trying to get real time news data using ERT in cloud and the MRN service. I am using the template code shown here https://github.com/Refinitiv-API-Samples/Example.WebSocketAPI.Python.TestClient in pywstestclient.py, but I added changes so that in the process_message function I added the processMRNUpdate function from https://github.com/Refinitiv-API-Samples/Example.WebSocketAPI.Python.MRN/blob/master/mrn_console_app.py, which is unique to MRN and process the data coming through. I also added code to put the streamed data into a mongo DB. My problem is that after a random amount of time, I would get one of the several errors:
[ E r r n o 1 0 4 ] C o n n e c t i o n r e s e t b y p e e r
W e b S o c k e t C l o s e d
or
C o n n e c t i o n i s a l r e a d y c l o s e d .
W e b S o c k e t C l o s e d
or
Traceback (most recent call last):
File "MRNclient.py", line 348, in <module>
market_data.reissue_token(ws_app,sts_token)
File "/mnt/batch/tasks/shared/LS_root/mounts/clusters/bokun/code/Users/Bokun/Refinitive_real_time_data/market_data.py", line 246, in reissue_token
send_login_request(ws, True)
File "/mnt/batch/tasks/shared/LS_root/mounts/clusters/bokun/code/Users/Bokun/Refinitive_real_time_data/market_data.py", line 277, in send_login_request
ws.send(json.dumps(login_json))
File "/anaconda/envs/azureml_py36/lib/python3.6/site-packages/websocket/_app.py", line 154, in send
if not self.sock or self.sock.send(data, opcode) == 0:
File "/anaconda/envs/azureml_py36/lib/python3.6/site-packages/websocket/_core.py", line 253, in send
return self.send_frame(frame)
File "/anaconda/envs/azureml_py36/lib/python3.6/site-packages/websocket/_core.py", line 279, in send_frame
l = self._send(data)
File "/anaconda/envs/azureml_py36/lib/python3.6/site-packages/websocket/_core.py", line 449, in _send
return send(self.sock, data)
File "/anaconda/envs/azureml_py36/lib/python3.6/site-packages/websocket/_socket.py", line 157, in send
return _send()
File "/anaconda/envs/azureml_py36/lib/python3.6/site-packages/websocket/_socket.py", line 139, in _send
return sock.send(data)
File "/anaconda/envs/azureml_py36/lib/python3.6/ssl.py", line 944, in send
return self._sslobj.write(data)
File "/anaconda/envs/azureml_py36/lib/python3.6/ssl.py", line 642, in write
return self._sslobj.write(data)
BrokenPipeError: [Errno 32] Broken pipe
And the stream would stop when the error occur. I understand that this problem would not occur if I do not process the MRN messages and put it on mongoDB but I am wondering why is it causing my streaming to stop? Has anyone else ran into similar problems when trying to process the streamed data (in my case doing the MRN processing and put in to mongoDB)? Also it is weird that one of the 3 errors would occur randomly.
Thank you!
Hi @bokun-a.wei,
Thanks for confirming.
You are correct that you are not the first one trying to use MRN over WebSockets. There are examples that retrieve MRN data from ERT in Cloud:
MRN example viewer using JavaScript
MRN example using RDP Library for .NET
And a number of other examples getting data via deployed ADS that are available within Refinitiv API Samples
When using MRN vs the raw code (which I assume subscribes to Market Price data) should be quite similar. That is, they both subscribe to data and capture updates within a callback. If you find that each individual example (WebSocket ERT in Cloud example and MRN example) works as expected independently, then there is likely an issue with how you merged the code. I would personally start simple - replace the subscription from MarketPrice data to MRN news and just echo the fact that you received data - no processing. This should work perfectly given the raw WebSocket ERT in Cloud code is likely doing something similar.
@Wasin Waeosri, you may be able to provide some troubleshooting steps regarding how to merge MRN code within your example.
Hi @bokun-a.wei,
Can you confirm how you are connecting into your streaming service? Via Elektron? Refinitiv Data Platform (ERT in Cloud)? Eikon? This is important to understand. You categorized your question under Eikon Data APIs but at this moment, MRN is not available via Eikon.
I'll assume you want to get MRN data from the Refinitiv Data Platform (RDP) given I can see from your stack trace you are trying to refresh tokens.
Given your application is rich with a combination of 2 examples (you linked above) plus the integration of Mongo DB, it would be difficult to narrow down at first glance. One area I would first look into is whether the access token expired, thus causing the WebSocket channel to close. I do see above that the code does refresh the access token so here are some suggestions.
Can you clarify random? Every few minutes? Is it consistent?
I would first run the basic streaming example (without MRN subscriptions/processing and Mongo DB) and observe the behavior. If you find the example runs perfectly for a long time, then the algorithm to get streaming data from RDP and refreshing access tokens is operating as expected. I would then include the code to process/manage MRN streaming data and continue from there.
Hi! Apologise for not making the post clear. I am using Refinitiv Data Platform (ERT in Cloud) as shown in the examples here (https://github.com/Refinitiv-API-Samples/Example.WebSocketAPI.Python.TestClient ). With some changes regarding process MRN messages.
The access token are being refreshed as seen from the log. And by random I meant sometimes it would run for a few min and stops itself, and sometimes it would run for half an hour to an hour before it stops itself. I have tried running the raw code without me processing and it works fine...but the problem occur when I added the MRN processing code to the pure ERT in cloud as shown in github. This is why I am posting on this forum to see if anyone has tried what I have done, which is using the ERT in cloud Websocket API code to get MRN message and process it. I assume I am not the first one trying to do this.
Hello @bokun-a.wei,
In addition to the suggestions by @nick.zincone.1 and @Umer Nalla,
Depending on your use case, your organization may be interested in the news already stored, please see full details on News Archive, you will need to login with your customer account credentials, if yes, please contact your Refinitiv account manager to discuss.
You may be interested in NewsAnalytics Database Store Example (MRN) example, it it uses EMA Java and stores MRN into MySQL db, demonstrates analogous concepts, so you may find useful hints.
Hi @bokun-a.wei
Based on the information provided, both myself and colleague think this could be a case of Slow Consumer scenario.
i.e. the writes to the Mongo database are taking too long, your python script is not reading the data quickly enough from the server and eventually, the buffer (that the server reserves for your connection) overflows and the server disconnects your application.
With our other (Java/C++) APIs we usually recommend farming the database writes (or any other slower activity) to a separate thread.
Whilst not being a Python expert, I understand that Python is not truly multi-threaded– so not sure just how much that will help in this situation.