ua-stress-test/src/ua-stress-test/main.cpp

// This is a minimal example that starts a number of workerThreads and terminates them afer a key press

// C includes
#include <stdio.h>
#include <libconfig.h>
#include <signal.h>
#include <open62541/client_config_default.h>
#include <open62541/client_highlevel.h>

// C++ includes
#include <iostream>
#include <thread>

// custom includes
#include "helper_functions.h"

// OPC UA general includes
#include <open62541/server.h>
#include "helper_functions.h"
#include <list>

int cfg_NumberOfConnections = 10;
int cfg_OpcUaReadsPerConnectionPerS = 1;
bool cfg_EnableConsoleOutput;
char *cfg_OpcUa_ServerAddress;

using namespace std;

bool stop = false; // when set to true, the client threads terminate

void workerThread()
{
    static int totalThreads = 0;
    int threadId = totalThreads++;
    cout << "Thread " << threadId << " starting" << endl;

    // open OPC UA connection to server
    UA_Client *client = UA_Client_new();
    UA_ClientConfig_setDefault(UA_Client_getConfig(client));
    UA_StatusCode retval = UA_Client_connect(client, cfg_OpcUa_ServerAddress); // Should be something like "opc.tcp://localhost:4840"
    if(retval != UA_STATUSCODE_GOOD) {
        UA_Client_delete(client);
        cerr << "Could not open OPC UA connection to " << cfg_OpcUa_ServerAddress << endl;
        return;
    }
    
    // TODO: maybe wait for a specific point in time to allow all threads to start before sending requests

    const UA_NodeId nodeId = UA_NODEID_NUMERIC(0, UA_NS0ID_SERVER_SERVERSTATUS_CURRENTTIME);
    chrono::system_clock::time_point startTime = chrono::system_clock::now();
    unsigned long intervalUs = 1.0 / ((double)cfg_OpcUaReadsPerConnectionPerS) * 1000000.0;
    UA_Variant value;
    UA_DateTime raw_date; 
    UA_DateTimeStruct dts; 
    for(int i = 0; !stop; i++) // loop repeats every intervalUs microseconds
    {
        std::this_thread::sleep_until(startTime + chrono::microseconds(i * intervalUs));
        retval = UA_Client_readValueAttribute(client, nodeId, &value);
        if (cfg_EnableConsoleOutput) {
            raw_date = *(UA_DateTime *) value.data;
            dts = UA_DateTime_toStruct(raw_date);
            cout << "Thread " << threadId << ": " << dts.day << "-" << dts.month << "-" << dts.year << "-" << dts.hour << "-" << dts.min << "-" << dts.sec << endl;
        }
        if (retval != UA_STATUSCODE_GOOD) {
            UA_Client_delete(client);
            cerr << "Could not read node NS=" << nodeId.namespaceIndex << ",i=" << nodeId.identifier.numeric << " from OPC UA server" << endl;
            return;
        }
    }

    cout << "Thread " << threadId << " terminating" << endl;
}

int main(int argc, char **argv) {
    printf("This is the ua-stress-test\n");

    // Check if a config file is provided via input arguments or use config.cfg
    string configFilename = "config.cfg";
    if (argc < 2) {
        cout << "No configuration file specified, using " << configFilename << endl;
    }
    else {
        configFilename = argv[1]; // use the input argument as confiFilename
    }
    // Check if the configuration file exists
    if(-1 == access(configFilename.c_str(), 0)) {
        cerr << "Could not find configuration file %s" << endl;
        return EXIT_FAILURE;
    }
    // Load the configuration file
    config_t cfg;
    config_setting_t *setting;
    config_init(&cfg);
    if(! config_read_file(&cfg, configFilename.c_str())){
        cerr << "Error in the config file: " << config_error_file(&cfg) << ":" << config_error_line(&cfg) << " - " << config_error_text(&cfg) << endl;
        config_destroy(&cfg);
        return EXIT_FAILURE;
    }
    // Read relevant values from the configuration file
    bool configOk = true;
    configOk &= getIntFromConfig(&cfg, "NumberOfConnections", &cfg_NumberOfConnections);
    configOk &= getIntFromConfig(&cfg, "OpcUaReadsPerConnectionPerS", &cfg_OpcUaReadsPerConnectionPerS);
    configOk &= getBoolFromConfig(&cfg, "EnableConsoleOutput", &cfg_EnableConsoleOutput);
    configOk &= getStringFromConfig(&cfg, "OpcUa.ServerAddress", &cfg_OpcUa_ServerAddress);
    if(!configOk){
        config_destroy(&cfg);
        return EXIT_FAILURE;
    }
    
    // start a thread for each connection
    list<thread> threadList;
    for (int i = 0; i < cfg_NumberOfConnections; i++) {  // loop repeats every intervalUs microseconds
        threadList.push_back(thread(workerThread));
    }

    cout << "Press ENTER to stop the stress test" << endl;
    getchar();
    cout << "Stopping..." << endl;

    // wait for threads to terminate
    stop = true;
    list<thread>::iterator it;
    for (it = threadList.begin(); it != threadList.end(); ++it){
        it->join();
    }

    return EXIT_SUCCESS;
}