Lesson 6: Subscribing for Data Changes

Content:

• Step 1: Implementing subscribe

Step 1: Implementing subscribe

Add the following code to header,

/* OPC UA service calls */ 
void subscribe();

and source file, rsp.:

void subscribe()
{
    printf("\n\n****************************************************************\n");
    printf("** Try to create a subscribe\n");
    if ( g_pUaSession == NULL )
    {
        printf("** Error: Server not connected\n");
        printf("****************************************************************\n");
        return;
    }

    UaStatus             status;
    UaSubscription* pUaSubscription         = NULL;
    SubscriptionSettings subscriptionSettings;
    subscriptionSettings.publishingInterval = 500;
    ServiceSettings      serviceSettings;

    /********************************************************************* 
     Create a Subscription 
    **********************************************************************/
    status = g_pUaSession->createSubscription(
        serviceSettings,        // Use default settings
        g_pCallback,            // Callback object
        0,                      // We have only one subscription, handle is not needed
        subscriptionSettings,   // general settings
        OpcUa_True,             // Publishing enabled
        &pUaSubscription);      // Returned Subscription instance
    /*********************************************************************/
    if ( status.isBad() )
    {
        printf("** Error: UaSession::createdSubscription failed [ret=%s] *********\n", status.toString().toUtf8());
    }
    else
    {
        printf("****************************************************************\n");
        printf("** Try to create monitored items\n");

        OpcUa_UInt32                  i;
        OpcUa_UInt32                  count;
        UaMonitoredItemCreateRequests monitoredItemCreateRequests;
        UaMonitoredItemCreateResults  monitoredItemCreateResults;

        // Initialize IN parameter monitoredItemCreateRequests 
        count = g_VariableNodeIds.length();
        monitoredItemCreateRequests.create(count + 1); // We create also an event monitored item
        for ( i=0; i<count; i++ )
        {
            g_VariableNodeIds[i].copyTo(&monitoredItemCreateRequests[i].ItemToMonitor.NodeId);
            monitoredItemCreateRequests[i].ItemToMonitor.AttributeId = OpcUa_Attributes_Value;
            monitoredItemCreateRequests[i].MonitoringMode = OpcUa_MonitoringMode_Reporting;
            monitoredItemCreateRequests[i].RequestedParameters.ClientHandle = i+1;
            monitoredItemCreateRequests[i].RequestedParameters.SamplingInterval = 1000;
            monitoredItemCreateRequests[i].RequestedParameters.QueueSize = 1;
        }

        // ------------------------------------------------------
        // Set Event item
        monitoredItemCreateRequests[count].ItemToMonitor.NodeId.Identifier.Numeric = OpcUaId_Server;
        monitoredItemCreateRequests[count].ItemToMonitor.AttributeId = OpcUa_Attributes_EventNotifier;
        monitoredItemCreateRequests[count].MonitoringMode = OpcUa_MonitoringMode_Reporting;
        monitoredItemCreateRequests[count].RequestedParameters.ClientHandle = count+1;
        monitoredItemCreateRequests[count].RequestedParameters.SamplingInterval = 1000;
        monitoredItemCreateRequests[count].RequestedParameters.QueueSize = 0;
        // Create extension object in monitoredItemCreateRequests[count].RequestedParameters.Filter
        OpcUa_EventFilter* pEventFilter = (OpcUa_EventFilter*)OpcUa_Null;
        OpcUa_EncodeableObject_CreateExtension( &OpcUa_EventFilter_EncodeableType, 
                                                &monitoredItemCreateRequests[count].RequestedParameters.Filter, 
                                                (OpcUa_Void**)&pEventFilter);
        OpcUa_EventFilter_Initialize(pEventFilter);
        UaSimpleAttributeOperands selectClauses;
        UaQualifiedNameArray      browsePath;
        UaQualifiedName           propName;
        // Build select statement with two elements
        selectClauses.create(3);
        // Select Event Field Message
        selectClauses[0].AttributeId = OpcUa_Attributes_Value;
        selectClauses[0].TypeDefinitionId.Identifier.Numeric = OpcUaId_BaseEventType;
        browsePath.create(1);
        propName.setQualifiedName( "Message", 0);
        propName.copyTo(&browsePath[0]);
        selectClauses[0].NoOfBrowsePath = browsePath.length();
        selectClauses[0].BrowsePath = browsePath.detach();
        // Select Event Field SourceName
        selectClauses[1].AttributeId = OpcUa_Attributes_Value;
        selectClauses[1].TypeDefinitionId.Identifier.Numeric = OpcUaId_BaseEventType;
        browsePath.create(1);
        propName.setQualifiedName( "SourceName", 0);
        propName.copyTo(&browsePath[0]);
        selectClauses[1].NoOfBrowsePath = browsePath.length();
        selectClauses[1].BrowsePath = browsePath.detach();
        // Select Event Field Temperature
        selectClauses[2].AttributeId = OpcUa_Attributes_Value;
        selectClauses[2].TypeDefinitionId.NamespaceIndex = 2;
        selectClauses[2].TypeDefinitionId.Identifier.Numeric = 4000; // ControlerEventType
        browsePath.create(1);
        propName.setQualifiedName( "Temperature", 2);
        propName.copyTo(&browsePath[0]);
        selectClauses[2].NoOfBrowsePath = browsePath.length();
        selectClauses[2].BrowsePath = browsePath.detach();
        // Detach to filter
        pEventFilter->NoOfSelectClauses = selectClauses.length();
        pEventFilter->SelectClauses = selectClauses.detach();
        // ------------------------------------------------------

        /********************************************************************* 
         Call createMonitoredItems service 
        **********************************************************************/
        status = pUaSubscription->createMonitoredItems(
            serviceSettings,               // Use default settings
            OpcUa_TimestampsToReturn_Both, // Select time stamps to return
            monitoredItemCreateRequests,   // monitored items to create
            monitoredItemCreateResults);   // Returned monitored items create result
        /*********************************************************************/
        if ( status.isBad() )
        {
            printf("** Error: UaSession::createMonitoredItems failed [ret=%s] *********\n", status.toString().toUtf8());
            printf("****************************************************************\n");
            return;
        }
        else
        {
            printf("** UaSession::createMonitoredItems result **********************\n");
            for ( i=0; i<count; i++ )
            {
                UaNodeId node(monitoredItemCreateRequests[i].ItemToMonitor.NodeId);
                if ( OpcUa_IsGood(monitoredItemCreateResults[i].StatusCode) )
                {
                    printf("** Variable %s MonitoredItemId = %d\n", node.toString().toUtf8(), monitoredItemCreateResults[i].MonitoredItemId);
                }
                else
                {
                    printf("** Variable %s failed!\n", node.toString().toUtf8());
                }
            }
            if ( OpcUa_IsGood(monitoredItemCreateResults[count].StatusCode) )
            {
                printf("** Event MonitoredItemId = %d\n", monitoredItemCreateResults[count].MonitoredItemId);
            }
            else
            {
                printf("** Event MonitoredItem for Server Object failed!\n");
            }
            printf("****************************************************************\n");
        }
    }

    printf("\n*******************************************************\n");
    printf("*******************************************************\n");
    printf("**        Press x to stop subscription          *******\n");
    printf("*******************************************************\n");
    printf("*******************************************************\n");

    int action;
    /******************************************************************************/
    /* Wait for user command to terminate the client thread. */
    while(!WaitForKeypress(action))
    {
       Sleep(100);
    }
    /******************************************************************************/

    /********************************************************************* 
     Delete Subscription 
    **********************************************************************/
    status = g_pUaSession->deleteSubscription(
        serviceSettings,    // Use default settings
        &pUaSubscription);  // Subcription
    /*********************************************************************/
    if ( status.isBad() )
    {
        printf("** UaSession::deleteSubscription failed! **********************\n");
        printf("****************************************************************\n");
    }
    else
    {
        pUaSubscription = NULL;
        printf("** UaSession::deleteSubscription succeeded!\n");
        printf("****************************************************************\n");
    }
}

subscribe() uses UaClientSdk::UaSubscription besides UaClientSdk::UaSession. UaSubscription provides all subscription related OPC UA Services. It features a real callback for Data Changes and Events. Publish is handled internally by this class.

subscribe() calls in the following order

• UaSession::createSubscription() in order to create a subscription for the session,
• UaSubscription::createMonitoredItems() in order to create monitored items,
• UaSession::deleteSubscription() in order to delete the subscription being created from the session.

UaSession::createSubscription() takes the following parameters:

• serviceSettings: the general Service settings like timeout,
• callRequest: the Object and Method to call and input arguments ( if necessary ),
• callResult: the output arguments and input argument results.

UaSubscription::createMonitoredItems() takes the following parameters:

• serviceSettings: the general Service settings like timeout,
• timeStamps: a parameter which allows the client to define which timestamps (source and server time stamp, rsp.) the server should return with the value,
• monitoredItemCreateRequests: list of monitored items to create,
• monitoredItemCreateResults: list create results.

UaSession::deleteSubscription() takes the following parameters:

• serviceSettings: the general Service settings like timeout,
• &pUaSubscription: the subscription object pointer. The inner pointer to the subscription object will be set to NULL when deleted.

Complete main()

Add

/********************************************************************* 
 Subscribing for data changes 
**********************************************************************/
subscribe();
/*********************************************************************/

to main().