Download tutorial source code |
Click here to download |
| Last update | Dec 2020 |
| Compilers | Visual Studio 2015 |
| Prerequisites | EMA NI Provider - Publishing updates Declare the NI_PUB and TEST_NI_PUB services in your LSEG Real-Time Distribution System (see Before you start). |
Tutorial purpose
In this tutorial you will learn how to send status messages to indicate a change in the state of a data item published by your provider.
To this aim we will go through the following sections:
- Introduction
- Status messages
- Use case examples
- The NiProvider class
- The main workflow
- Build and run the application
Introduction
For each data item published by your provider, you can send status messages as part of the data item streams. Unlike refresh and update messages, status messages do not publish a payload of data but rather information about the state of the item. This can be useful to inform your consumer applications that an item stream is closed or that the published data is temporarily not reliable (for example, because of a failover procedure).
In the next sections, you will learn how to build status messages and how to publish them for your data items.
Status messages
A Status message conveys information about the state of an item stream and the related item data. This information is contained in an OmmState object transported by the message. The OmmState object contains 3 information items:
- The state of the item stream, via the StreamState enumeration (e.g. OpenEnum, ClosedEnum, etc.).
- The state of the data item, via the DataState enumeration (OkEnum, SuspectEnum, etc.) and via the StatusCode enumeration (e.g. FailoverStartedEnum, ItemRequestTimeoutEnum, etc.).
- An explanatory text of the status.
Note: Please refer to the Status sections defined for the relevant domain messages within the EMA C++ RDM Usage Guide. for more details about the OmmState class and its related enumerations.
Status messages are built using the default constructor of the StatusMsg class. The OmmState details are set via the state() method of the same class. Here is an example:
StatusMsg statusMsg;
statusMsg.state(
OmmState::OpenEnum, // Stream state
OmmState::OkEnum, // Data state
OmmState::NoneEnum, // Data status code
"All is good"); // Explanatory text
Use case examples
Below are two examples of disaster recovery scenarios and the status messages a provider could possibly send to notify its consumer applications.
The context
A provider application publishes data items on a LSEG Real-Time Distribution System for a single service. The application is also connected to several upstream systems that provide the source information for the published data items. Each upstream system provides data for a subset of items.
At the beginning of the scenarios below, all the item streams are open. The initial refresh message and some updates have been published for all of them.
- The connection with one of the upstream systems breaks. Because of this failure, the data items that were fed by this system may not be up-to-date.
- In order to notify its consumer applications, the provider builds the following status message and publishes it on the stream of the concerned items.
StatusMsg statusMsg;
statusMsg.state(
OmmState::OpenEnum, // Stream state
OmmState::SuspectEnum, // Data state
OmmState::FailoverStartedEnum, // Data status
"Recovering the upstream connection - The current data may not be reliable."); // Explanatory text
provider.submit(statusMsg, itemHandle);
This message contains:
- The item stream state that is still open (OpenEnum)
- The data state that is suspect (SuspectEnum) and the reason why (FailoverStartedEnum)
- The explanatory text.
3. The provider successfully recovers the connection and starts receiving up-to-date data again.
4. It builds and sends the following status message to notify the concerned consumers that everything is fine now.
StatusMsg statusMsg;
statusMsg.state(
OmmState::OpenEnum, // Stream state
OmmState::OkEnum, // Data state
OmmState::NoneEnum, // Data status
"All is good"); // Explanatory text
provider.submit(statusMsg, itemHandle);
This message contains:
- The item stream state that is still open (OpenEnum)
- The data state that is ok (OkEnum). As it is ok, no status code is required (NoneEnum)
- An explanatory text.
The previous scenario works fine. However, in the context of disaster recovery, the best practice is to publish an unsolicited refresh message with an Ok state in place of the Ok status message sent in scenario 1. This allows refreshing the ADH cache with the latest values, ensuring it is in sync with your provider application. With this best practice in mind, we changed the scenario by publishing a refresh message and an associated Ok state to indicate that the system is fully recovered:
- The connection with one of the upstream systems breaks. Because of this failure, the data items that were fed by this system may not be up-to-date.
- In order to notify its consumer applications, the provider builds the following status message and publishes it on the stream of the concerned items.
StatusMsg statusMsg;
statusMsg.state(
OmmState::OpenEnum, // Stream state
OmmState::SuspectEnum, // Data state
OmmState::FailoverStartedEnum, // Data status
"Recovering the upstream connection - The current data may not be reliable."); // Explanatory text
provider.submit(statusMsg, itemHandle);
This message contains:
- The item stream state is still open (OpenEnum)
- The data state that is suspect (SuspectEnum) and the reason why (FailoverStartedEnum)
- The explanatory text.
3. The provider successfully recovers the connection and starts receiving up-to-date data again.
4. It builds and sends an unsolicited refresh message that contains an Ok state and the complete data item. This message will refresh the ADH cache, send the latest data to the concerned consumers and notify them that everything is fine now.
provider->submit(
RefreshMsg()
.serviceName(serviceName)
.name(name)
.domainType(MMT_MARKET_PRICE)
.state(
OmmState::OpenEnum, // Stream state
OmmState::OkEnum, // Data state
OmmState::NoneEnum, // Data status
"UnSolicited Refresh Completed") // Explanatory text
.payload(
...
)
.complete(),
itemHandle);
Note: By default, EMA builds unsolicited refresh messages. You can build solicited messages by calling the solicited() method on the RefreshMsg object. This is however useless for Non-Interactive Providers as they must never publish this kind of message.
This best practice scenario is implemented by the main workflow below.
The NiProvider class
In order to be able to simulate such a use case (with data items that go Suspect and then Ok again), we added two methods to the NiProvider.
The sendSuspectStatusFor() method sends a status message that notifies consumer applications of suspect data. This method is built on the same model as the refresh() and the updated() methods. First of all, it checks if the provider is connected and if the item name is valid. Then, it builds the message and sends it to the concerned stream using the appropriate item handle.
void NiProvider::sendSuspectStatusFor(const std::string & itemName)
{
if (!isConnected())
{
cerr << " ERROR: Can't send a status for " << itemName << ". The provider is not connected." << endl;
return;
}
if (itemName != _theOnlyOneItem.getName())
{
cerr << " ERROR: Can't send a status for " << itemName << ". Unknown item name." << endl;
return;
}
Item & item = _theOnlyOneItem;
cout << " Sending a Stale status for " << item.getName() << endl;
_provider->submit(
StatusMsg()
.state(
OmmState::OpenEnum, // Stream state
OmmState::SuspectEnum, // Data state
OmmState::ErrorEnum, // Data status
"Houston, we have a problem!"), // Explanatory text
item.getHandle());
}
The sendOkStatusFor() method is built in a very similar way, except that it sends an ok status message instead of a suspect status message.
Note: This method is not used by the main workflow below but is provided in case you would like to implement scenario 1.
void NiProvider::sendOkStatusFor(const std::string & itemName)
{
if (!isConnected())
{
cerr << " ERROR: Can't send a status for " << itemName << ". The provider is not connected." << endl;
return;
}
if (itemName != _theOnlyOneItem.getName())
{
cerr << " ERROR: Can't send a status for " << itemName << ". Unknown item name." << endl;
return;
}
Item & item = _theOnlyOneItem;
cout << " Sending an Ok status for " << item.getName() << endl;
_provider->submit(
StatusMsg()
.state(
OmmState::OpenEnum, // Stream state
OmmState::OkEnum, // Data state
OmmState::NoneEnum, // Data status
"All is good" ), // Explanatory text
item.getHandle());
}
The main workflow
We changed the main workflow to simulate the situation when the item's data becomes suspect because of scenario 2 described above. To this aim, we use the NiProvider methods to send refresh messages, update messages, and statuses in a loop that simulates several disaster recoveries. For each iteration of the loop we do the following:
- We send an unsolicited refresh with its associated Ok state and wait 1second.
- Then we send an update and wait for 1 second again.
- Then we send a suspect status to notify consumers of the disaster.
- We wait for 3 seconds to simulate the recovery.
- After 3 seconds we go back to the beginning of the loop and send a refresh message with its Ok state again. This message refreshes the ADH cache, sends the latest data to consumers, and notifies them that everything has been recovered.
Here is the corresponding source code:
int main(int argc, char* argv[])
{
.
.
.
NiProvider provider;
provider.setServiceName("TEST_NI_PUB");
provider.connectAs("YOUR_PROVIDER_USER_NAME");
waitFor(5);
for (int i = 0; i < 30; ++i)
{
provider.refresh("SHARE-0");
waitFor(1);
provider.update("SHARE-0");
waitFor(1);
provider.sendSuspectStatusFor("SHARE-0");
waitFor(3);
}
provider.disconnect();
.
.
.
}
Build and run the application
Build the application and start it. Please refer to the Build and Run section within the first tutorial of this series (A barebones EMA NIP application shell) for detailed instructions.
This is what you should get:
- A console application should open and display something like:
-------------------------------------------------------------------------------
| Non Interactive Provider EMA Tutorial |
| |
| Tutorial 8 - Publishing statuses |
-------------------------------------------------------------------------------
Provider created
Connecting Provider to ADH 10.2.43.49:14003 as nip-user
Waiting for 5 seconds...
Provider is connected. OmmState:Open / Ok / None / 'Refresh Completed'
Refreshing SHARE-0
Waiting for 1 seconds...
Updating SHARE-0
Waiting for 1 seconds...
Sending a Stale status for SHARE-0
Waiting for 3 seconds...
.
.
.
Refreshing SHARE-0
Waiting for 1 seconds...
Updating SHARE-0
Waiting for 1 seconds...
Sending a Stale status for SHARE-0
Waiting for 3 seconds...
Disconnecting...
Provider is disconnected. OmmState:Open / Suspect / None / 'channel down'
Exiting the application
Press any key to continue . . .
2. Open a consuming application and subscribe to the SHARE-0 market price item of the TEST_NI_PUB service. After a short while, you should receive values for the 5 fields (DSPLY_NAME/3, OPEN_PRC/19, HST_CLOSE/21, BID/22 and ASK/25) published by the Ni Provider.
As an example, this is a screenshot of the Eikon Quote object that we used to subscribe to TEST_NI_PUB/SHARE-0:
3. After 1 second you should receive an update message for the BID and ASK fields.
In the Eikon Quote object the BID and ASK fields are displayed in yellow for a short while:
4. One second after you should receive a suspect status message.
In the Eikon Quote object the fields background turns red because of the suspect state of the item data:
5. After 3 seconds the loop goes back to step 3 and you should receive a refresh message with an Ok state again.
Thanks to this Ok state, the fields background of the Eikon Quote object turns black again:
6. After 30 iterations, the TEST_NI_PUB service goes down and the application exits.
7. Press a key to close the console when you are prompted to.
Troubleshooting
Q: When I build the tutorial project, I get errors like:
error MSB3073: The command "copy "\Ema\Libs\WIN_64_VS140\Debug_MDd\Shared\*.*" .\Debug_WIN_64_VS140_Shared\
A: The ElektronSDKInstallPath environment variable is not set or set to the wrong path. See Setup the development environment.
Q: The application is stuck after the "Connecting Provider to ADH…" message is displayed.
After a while the application displays an error like:
Exception Type='OmmInvalidUsageException', Text='login failed (timed out after waiting 45000 milliseconds) for 10.2.43.149:14003)'
A: Verify that the ADH of your LSEG Real-Time Distribution System infrastructure is up and that you properly set the host parameter in the EmaConfig.xml file.
You can also use the telnet command tool to verify that your NIP application machine can connect to the ADH (telnet <ADH host> <port>). If the telnet succeeds but you still can’t connect, verify that you don’t have any firewall blocking the messages sent/received by the application.
Ultimately, ask your administrator to help you to investigate with monitoring tools like adhmon.
Q: The “Waiting for 5 seconds...” message and the “Provider is connected” message get mixed up when displayed in the console.
A: This is perfectly normal and actually caused by the EMA background thread that prints the “Provider is connected” message at the same time the main application thread prints the “Waiting for 30 seconds...” message. This can be fixed either by choosing a mono threaded application model (see the Message Processing - dispatch section of the Requesting MarketPrice data EMA Consumer tutorial) or by printing these messages atomically (use critical sections or a single cout << call to print the whole message).