mirror of
https://github.com/OPCFoundation/UA-.NETStandard.git
synced 2026-10-28 22:55:49 +08:00
Added KeyFrameCount and DeltaFrame tests (#2579)
Added tests which include PubSub KeyFrameCount and DeltaFrames.
This commit is contained in:
Suciu Mircea Adrian
File diff suppressed because it is too large
Load Diff
@@ -0,0 +1,296 @@
|
||||
/* ========================================================================
|
||||
* Copyright (c) 2005-2021 The OPC Foundation, Inc. All rights reserved.
|
||||
*
|
||||
* OPC Foundation MIT License 1.00
|
||||
*
|
||||
* Permission is hereby granted, free of charge, to any person
|
||||
* obtaining a copy of this software and associated documentation
|
||||
* files (the "Software"), to deal in the Software without
|
||||
* restriction, including without limitation the rights to use,
|
||||
* copy, modify, merge, publish, distribute, sublicense, and/or sell
|
||||
* copies of the Software, and to permit persons to whom the
|
||||
* Software is furnished to do so, subject to the following
|
||||
* conditions:
|
||||
*
|
||||
* The above copyright notice and this permission notice shall be
|
||||
* included in all copies or substantial portions of the Software.
|
||||
* THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,
|
||||
* EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES
|
||||
* OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND
|
||||
* NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT
|
||||
* HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY,
|
||||
* WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING
|
||||
* FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR
|
||||
* OTHER DEALINGS IN THE SOFTWARE.
|
||||
*
|
||||
* The complete license agreement can be found here:
|
||||
* http://opcfoundation.org/License/MIT/1.00/
|
||||
* ======================================================================*/
|
||||
|
||||
using System;
|
||||
using System.Collections.Generic;
|
||||
using System.Linq;
|
||||
using NUnit.Framework;
|
||||
using Opc.Ua.PubSub.Encoding;
|
||||
using Opc.Ua.PubSub.Tests.Encoding;
|
||||
using Opc.Ua.PubSub.PublishedData;
|
||||
using System.Reflection;
|
||||
|
||||
namespace Opc.Ua.PubSub.Tests.PublishedData
|
||||
{
|
||||
public class WriterGroupPublishedStateTests
|
||||
{
|
||||
|
||||
/// <summary>
|
||||
/// PubSub message type mapping
|
||||
/// </summary>
|
||||
public enum PubSubMessageType
|
||||
{
|
||||
Uadp,
|
||||
Json
|
||||
}
|
||||
|
||||
private const UInt16 NamespaceIndexAllTypes = 3;
|
||||
|
||||
private const string UdpUrlFormat = "{0}://{1}:4840";
|
||||
private const string MqttUrlFormat = "{0}://{1}:1883";
|
||||
//private const string MqttsUrlFormat = "{0}://{1}:8883";
|
||||
|
||||
[Test(Description = "Publish Uadp | Json DataSetMessages with KeyFrameCount and delta frames")]
|
||||
public void PublishDataSetMessages(
|
||||
[Values(PubSubMessageType.Uadp, PubSubMessageType.Json)]
|
||||
PubSubMessageType pubSubMessageType,
|
||||
[Values(1, 2, 3, 4)] Int32 keyFrameCount)
|
||||
{
|
||||
//Arrange
|
||||
object publisherId = 1;
|
||||
UInt16 writerGroupId = 1;
|
||||
|
||||
string addressUrl = "http://localhost:1883";
|
||||
|
||||
DataSetFieldContentMask dataSetFieldContentMask = DataSetFieldContentMask.None;
|
||||
|
||||
DataSetMetaDataType[] dataSetMetaDataArray = new DataSetMetaDataType[]
|
||||
{
|
||||
MessagesHelper.CreateDataSetMetaData1("DataSet1"),
|
||||
MessagesHelper.CreateDataSetMetaData2("DataSet2"),
|
||||
MessagesHelper.CreateDataSetMetaData2("DataSet3"),
|
||||
};
|
||||
|
||||
PubSubConfigurationDataType publisherConfiguration = null;
|
||||
|
||||
if (pubSubMessageType == PubSubMessageType.Uadp)
|
||||
{
|
||||
UadpNetworkMessageContentMask uadpNetworkMessageContentMask = UadpNetworkMessageContentMask.PublisherId
|
||||
| UadpNetworkMessageContentMask.WriterGroupId
|
||||
| UadpNetworkMessageContentMask.PayloadHeader;
|
||||
UadpDataSetMessageContentMask uadpDataSetMessageContentMask = UadpDataSetMessageContentMask.None;
|
||||
|
||||
publisherConfiguration = MessagesHelper.CreatePublisherConfiguration(
|
||||
Profiles.PubSubMqttUadpTransport,
|
||||
addressUrl, publisherId: publisherId, writerGroupId: writerGroupId,
|
||||
uadpNetworkMessageContentMask: uadpNetworkMessageContentMask,
|
||||
uadpDataSetMessageContentMask: uadpDataSetMessageContentMask,
|
||||
dataSetFieldContentMask: dataSetFieldContentMask,
|
||||
dataSetMetaDataArray: dataSetMetaDataArray, nameSpaceIndexForData: NamespaceIndexAllTypes,
|
||||
keyFrameCount: Convert.ToUInt32(keyFrameCount));
|
||||
}
|
||||
|
||||
if (pubSubMessageType == PubSubMessageType.Json)
|
||||
{
|
||||
JsonNetworkMessageContentMask jsonNetworkMessageContentMask = JsonNetworkMessageContentMask.NetworkMessageHeader
|
||||
| JsonNetworkMessageContentMask.PublisherId
|
||||
| JsonNetworkMessageContentMask.DataSetMessageHeader;
|
||||
JsonDataSetMessageContentMask jsonDataSetMessageContentMask = JsonDataSetMessageContentMask.DataSetWriterId;
|
||||
|
||||
publisherConfiguration = MessagesHelper.CreatePublisherConfiguration(
|
||||
Profiles.PubSubMqttJsonTransport,
|
||||
addressUrl, publisherId: publisherId, writerGroupId: writerGroupId,
|
||||
jsonNetworkMessageContentMask: jsonNetworkMessageContentMask,
|
||||
jsonDataSetMessageContentMask: jsonDataSetMessageContentMask,
|
||||
dataSetFieldContentMask: dataSetFieldContentMask,
|
||||
dataSetMetaDataArray: dataSetMetaDataArray, nameSpaceIndexForData: NamespaceIndexAllTypes,
|
||||
keyFrameCount: Convert.ToUInt32(keyFrameCount));
|
||||
}
|
||||
|
||||
Assert.IsNotNull(publisherConfiguration, "publisherConfiguration should not be null");
|
||||
|
||||
// Create publisher application for multiple datasets
|
||||
UaPubSubApplication publisherApplication = UaPubSubApplication.Create(publisherConfiguration);
|
||||
MessagesHelper.LoadData(publisherApplication, NamespaceIndexAllTypes);
|
||||
|
||||
IUaPubSubConnection publisherConnection = publisherApplication.PubSubConnections.First();
|
||||
Assert.IsNotNull(publisherConnection, "Publisher first connection should not be null");
|
||||
|
||||
Assert.IsNotNull(publisherConfiguration.Connections.First(), "publisherConfiguration first connection should not be null");
|
||||
Assert.IsNotNull(publisherConfiguration.Connections.First().WriterGroups.First(), "publisherConfiguration first writer group of first connection should not be null");
|
||||
|
||||
WriterGroupPublishState writerGroupPublishState = new WriterGroupPublishState();
|
||||
var networkMessages = publisherConnection.CreateNetworkMessages(publisherConfiguration.Connections.First().WriterGroups.First(), writerGroupPublishState);
|
||||
Assert.IsNotNull(networkMessages, "connection.CreateNetworkMessages shall not return null");
|
||||
Assert.GreaterOrEqual(networkMessages.Count, 1, "connection.CreateNetworkMessages shall have at least one network message");
|
||||
|
||||
object uaNetworkMessagesList = null;
|
||||
List<UaNetworkMessage> uaNetworkMessages = null;
|
||||
if (pubSubMessageType == PubSubMessageType.Uadp)
|
||||
{
|
||||
uaNetworkMessagesList = MessagesHelper.GetUaDataNetworkMessages(networkMessages.Cast<UadpNetworkMessage>().ToList());
|
||||
Assert.IsNotNull(uaNetworkMessagesList, "uaNetworkMessagesList should not be null");
|
||||
uaNetworkMessages = ((IEnumerable<UaNetworkMessage>)uaNetworkMessagesList).Cast<UaNetworkMessage>().ToList();
|
||||
}
|
||||
if (pubSubMessageType == PubSubMessageType.Json)
|
||||
{
|
||||
uaNetworkMessagesList = MessagesHelper.GetUaDataNetworkMessages(networkMessages.Cast<JsonNetworkMessage>().ToList());
|
||||
uaNetworkMessages = ((IEnumerable<UaNetworkMessage>)uaNetworkMessagesList).Cast<UaNetworkMessage>().ToList();
|
||||
}
|
||||
Assert.IsNotNull(uaNetworkMessages, "uaNetworkMessages should not be null. Data entry is missing from configuration!?");
|
||||
|
||||
// get datastore data
|
||||
Dictionary<NodeId, DataValue> dataStoreData = new Dictionary<NodeId, DataValue>();
|
||||
foreach (UaNetworkMessage uaDataNetworkMessage in uaNetworkMessages)
|
||||
{
|
||||
Dictionary<NodeId, DataValue> dataSetsData = MessagesHelper.GetDataStoreData(publisherApplication, uaDataNetworkMessage, NamespaceIndexAllTypes);
|
||||
foreach (NodeId nodeId in dataSetsData.Keys)
|
||||
{
|
||||
if (!dataStoreData.ContainsKey(nodeId))
|
||||
{
|
||||
dataStoreData.Add(nodeId, dataSetsData[nodeId]);
|
||||
}
|
||||
}
|
||||
}
|
||||
Assert.IsNotEmpty(dataStoreData, "datastore entries should be greater than 0");
|
||||
|
||||
// check if received data is valid
|
||||
foreach (UaNetworkMessage uaDataNetworkMessage in uaNetworkMessages)
|
||||
{
|
||||
ValidateDataSetMessageData(uaDataNetworkMessage, dataStoreData);
|
||||
}
|
||||
|
||||
for (int keyCount = 0; keyCount < keyFrameCount - 1; keyCount++)
|
||||
{
|
||||
// change the values and get one more time the dataset(s) data
|
||||
MessagesHelper.UpdateSnapshotData(publisherApplication, NamespaceIndexAllTypes);
|
||||
networkMessages = publisherConnection.CreateNetworkMessages(publisherConfiguration.Connections.First().WriterGroups.First(), writerGroupPublishState);
|
||||
Assert.IsNotNull(networkMessages, "connection.CreateNetworkMessages shall not be null");
|
||||
Assert.GreaterOrEqual(networkMessages.Count, 1, "connection.CreateNetworkMessages should have at least one network message");
|
||||
|
||||
if (pubSubMessageType == PubSubMessageType.Uadp)
|
||||
{
|
||||
uaNetworkMessagesList = MessagesHelper.GetUaDataNetworkMessages(networkMessages.Cast<UadpNetworkMessage>().ToList());
|
||||
Assert.IsNotNull(uaNetworkMessagesList, "uaNetworkMessagesList shall not be null");
|
||||
uaNetworkMessages = ((IEnumerable<UaNetworkMessage>)uaNetworkMessagesList).Cast<UaNetworkMessage>().ToList();
|
||||
}
|
||||
if (pubSubMessageType == PubSubMessageType.Json)
|
||||
{
|
||||
uaNetworkMessagesList = MessagesHelper.GetUaDataNetworkMessages(networkMessages.Cast<JsonNetworkMessage>().ToList());
|
||||
uaNetworkMessages = ((IEnumerable<UaNetworkMessage>)uaNetworkMessagesList).Cast<UaNetworkMessage>().ToList();
|
||||
}
|
||||
Assert.IsNotNull(uaNetworkMessages, "uaNetworkMessages should not be null. Data entry is missing from configuration!?");
|
||||
|
||||
// check if delta received data is valid
|
||||
Dictionary<NodeId, DataValue> snapshotData = MessagesHelper.GetSnapshotData(publisherApplication, NamespaceIndexAllTypes);
|
||||
foreach (UaNetworkMessage uaDataNetworkMessage in uaNetworkMessages)
|
||||
{
|
||||
ValidateDataSetMessageData(uaDataNetworkMessage, keyFrameCount == 1 ? dataStoreData : snapshotData, keyFrameCount, writerGroupPublishState);
|
||||
}
|
||||
}
|
||||
|
||||
// check one more time if delta received data is valid
|
||||
Dictionary<NodeId, DataValue> snapshotDataCopy = MessagesHelper.GetSnapshotData(publisherApplication, NamespaceIndexAllTypes);
|
||||
foreach (UaNetworkMessage uaDataNetworkMessage in uaNetworkMessages)
|
||||
{
|
||||
ValidateDataSetMessageData(uaDataNetworkMessage, keyFrameCount == 1 ? dataStoreData : snapshotDataCopy, keyFrameCount, writerGroupPublishState);
|
||||
}
|
||||
}
|
||||
|
||||
#region Private methods
|
||||
|
||||
/// <summary>
|
||||
/// Validate dataset message data
|
||||
/// </summary>
|
||||
/// <param name="uaDataNetworkMessage"></param>
|
||||
/// <param name="keyFrameCount"></param>
|
||||
private void ValidateDataSetMessageData(UaNetworkMessage uaDataNetworkMessage, Dictionary<NodeId, DataValue> dataStoreData,
|
||||
Int32 keyFrameCount = 1, WriterGroupPublishState writerGroupPublishState = null)
|
||||
{
|
||||
IEnumerable<object> writerGroupDataSetStates = null;
|
||||
if (writerGroupPublishState != null)
|
||||
{
|
||||
var dataSetStates = writerGroupPublishState.GetType()
|
||||
.GetField("m_dataSetStates", BindingFlags.Instance | BindingFlags.NonPublic)
|
||||
.GetValue(writerGroupPublishState);
|
||||
|
||||
var dataSetStatesValues = dataSetStates.GetType()
|
||||
.GetProperty("Values", BindingFlags.Instance | BindingFlags.Public)
|
||||
.GetValue(dataSetStates);
|
||||
|
||||
writerGroupDataSetStates = ((IEnumerable<object>)dataSetStatesValues).Cast<object>();
|
||||
}
|
||||
|
||||
foreach (UaDataSetMessage datasetMessage in uaDataNetworkMessage.DataSetMessages)
|
||||
{
|
||||
if (datasetMessage.DataSet.IsDeltaFrame)
|
||||
{
|
||||
Assert.Greater(keyFrameCount, 1, "keyFrameCount > 1 if dataset is delta!");
|
||||
Assert.IsNotNull(writerGroupPublishState, "WriterGroupPublishState should not be null");
|
||||
Assert.IsNotNull(writerGroupDataSetStates, "writerGroupDataSetStates that contains last saved detaset should not be null");
|
||||
|
||||
DataSet lastDataSetFound = null;
|
||||
foreach (object dataSetState in writerGroupDataSetStates)
|
||||
{
|
||||
var writerGroupLastDataSet = dataSetState.GetType()
|
||||
.GetField("LastDataSet", BindingFlags.Instance | BindingFlags.Public)
|
||||
.GetValue(dataSetState);
|
||||
if (writerGroupLastDataSet != null)
|
||||
{
|
||||
string dataSetName = writerGroupLastDataSet.GetType()
|
||||
.GetProperty("Name", BindingFlags.Instance | BindingFlags.Public)
|
||||
.GetValue(writerGroupLastDataSet) as string;
|
||||
if (!string.IsNullOrEmpty(dataSetName) &&
|
||||
datasetMessage.DataSet.Name == dataSetName)
|
||||
{
|
||||
lastDataSetFound = writerGroupLastDataSet as DataSet;
|
||||
}
|
||||
}
|
||||
}
|
||||
Assert.IsNotNull(lastDataSetFound, "lastDataSetFound dataset should not be null");
|
||||
|
||||
int fieldIndex = 0;
|
||||
foreach (Field field in datasetMessage.DataSet.Fields)
|
||||
{
|
||||
// ghost field should still be hold it in the state.LastDataSet
|
||||
Field lastDataSetField = lastDataSetFound.Fields[fieldIndex++];
|
||||
Assert.IsNotNull(lastDataSetField, "lastDataSetField should not be null even if the partial field is missing due to delta");
|
||||
// for delta frames dataset might contains partial filled data
|
||||
if (field == null)
|
||||
{
|
||||
continue;
|
||||
}
|
||||
NodeId targetNodeId = new NodeId(field.FieldMetaData.Name, NamespaceIndexAllTypes);
|
||||
Assert.IsTrue(dataStoreData.ContainsKey(targetNodeId), "field name: '{0}' should be exists in partial received dataset", field.FieldMetaData.Name);
|
||||
Assert.IsNotNull(dataStoreData[targetNodeId], "field: '{0}' should not be null", field.FieldMetaData.Name);
|
||||
Assert.AreEqual(field.Value.Value, dataStoreData[targetNodeId].Value, "field: '{0}' value: {1} should be equal to datastore value: {2}",
|
||||
field.FieldMetaData.Name, field.Value, dataStoreData[targetNodeId].Value);
|
||||
Assert.AreEqual(lastDataSetField.Value.Value, dataStoreData[targetNodeId].Value, "lastDataSetField: '{0}' value: {1} should be equal to datastore value: {2}",
|
||||
lastDataSetField.FieldMetaData.Name, lastDataSetField.Value, dataStoreData[targetNodeId].Value);
|
||||
}
|
||||
}
|
||||
else
|
||||
{
|
||||
Assert.AreEqual(keyFrameCount, 1, "keyFrameCount = 1 if dataset is not delta!");
|
||||
foreach (Field field in datasetMessage.DataSet.Fields)
|
||||
{
|
||||
Assert.IsNotNull(field, "field {0}: should not be null if dataset is not delta!", field.FieldMetaData.Name);
|
||||
NodeId targetNodeId = new NodeId(field.FieldMetaData.Name, NamespaceIndexAllTypes);
|
||||
Assert.IsTrue(dataStoreData.ContainsKey(targetNodeId), "field name: {0} should be exists in partial received dataset", field.FieldMetaData.Name);
|
||||
Assert.IsNotNull(dataStoreData[targetNodeId], "field {0}: should not be null", field.FieldMetaData.Name);
|
||||
Assert.AreEqual(field.Value.Value, dataStoreData[targetNodeId].Value, "field: '{0}' value: {1} should be equal to datastore value: {2}",
|
||||
field.FieldMetaData.Name, field.Value, dataStoreData[targetNodeId].Value);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
#endregion Private methods
|
||||
}
|
||||
}
|
||||
@@ -37,6 +37,7 @@ using Opc.Ua.PubSub.Encoding;
|
||||
using Opc.Ua.PubSub.Transport;
|
||||
using Opc.Ua.PubSub.Tests.Encoding;
|
||||
using System.Collections.Generic;
|
||||
using Opc.Ua.PubSub.PublishedData;
|
||||
|
||||
namespace Opc.Ua.PubSub.Tests.Transport
|
||||
{
|
||||
@@ -46,10 +47,19 @@ namespace Opc.Ua.PubSub.Tests.Transport
|
||||
private const UInt16 kNamespaceIndexAllTypes = 3;
|
||||
|
||||
private ManualResetEvent m_uaDataShutdownEvent;
|
||||
private ManualResetEvent m_uaDeltaDataShutdownEvent;
|
||||
private ManualResetEvent m_uaMetaDataShutdownEvent;
|
||||
private ManualResetEvent m_uaConfigurationUpdateEvent;
|
||||
private bool m_isDeltaFrame;
|
||||
private Dictionary<NodeId, DataValue> m_snapshotData;
|
||||
private const int kEstimatedPublishingTime = 60000;
|
||||
|
||||
private const string DefaultBrokerProcessName = "mosquitto";
|
||||
|
||||
private const string UdpUrlFormat = "{0}://{1}:4840";
|
||||
private const string MqttUrlFormat = "{0}://{1}:1883";
|
||||
private const string MqttsUrlFormat = "{0}://{1}:8883";
|
||||
|
||||
[OneTimeSetUp()]
|
||||
public void MyTestInitialize()
|
||||
{
|
||||
@@ -63,12 +73,12 @@ namespace Opc.Ua.PubSub.Tests.Transport
|
||||
public void ValidateMqttLocalPubSubConnectionWithUadp(
|
||||
[Values((byte)1, (UInt16)1, (UInt32)1, (UInt64)1, "abc")] object publisherId)
|
||||
{
|
||||
RestartMosquitto("mosquitto");
|
||||
RestartMosquitto();
|
||||
|
||||
//Arrange
|
||||
UInt16 writerGroupId = 1;
|
||||
|
||||
string mqttLocalBrokerUrl = "mqtt://localhost:1883";
|
||||
string mqttLocalBrokerUrl = string.Format(MqttUrlFormat, Utils.UriSchemeMqtt, "localhost");
|
||||
|
||||
ITransportProtocolConfiguration mqttConfiguration = new MqttClientProtocolConfiguration(version: EnumMqttProtocolVersion.V500);
|
||||
|
||||
@@ -149,6 +159,7 @@ namespace Opc.Ua.PubSub.Tests.Transport
|
||||
// it will signal if the uadp message was received from local ip
|
||||
m_uaDataShutdownEvent = new ManualResetEvent(false);
|
||||
|
||||
m_isDeltaFrame = false;
|
||||
subscriberApplication.DataReceived += UaPubSubApplication_DataReceived;
|
||||
subscriberConnection.Start();
|
||||
|
||||
@@ -159,6 +170,142 @@ namespace Opc.Ua.PubSub.Tests.Transport
|
||||
{
|
||||
Assert.Fail("The UADP message was not received");
|
||||
}
|
||||
|
||||
subscriberConnection.Stop();
|
||||
publisherConnection.Stop();
|
||||
|
||||
}
|
||||
|
||||
[Test(Description = "Validate mqtt local pub/sub connection with uadp data.")]
|
||||
#if !CUSTOM_TESTS
|
||||
[Ignore("A mosquitto tool should be installed local in order to run correctly.")]
|
||||
#endif
|
||||
public void ValidateMqttLocalPubSubConnectionWithDeltaUadp(
|
||||
[Values((byte)1, (UInt16)1, (UInt32)1, (UInt64)1, "abc")] object publisherId,
|
||||
[Values(1, 2, 3, 4)] Int32 keyFrameCount)
|
||||
{
|
||||
RestartMosquitto();
|
||||
|
||||
//Arrange
|
||||
UInt16 writerGroupId = 1;
|
||||
|
||||
string mqttLocalBrokerUrl = string.Format(MqttUrlFormat, Utils.UriSchemeMqtt, "localhost");
|
||||
|
||||
ITransportProtocolConfiguration mqttConfiguration = new MqttClientProtocolConfiguration(version: EnumMqttProtocolVersion.V500);
|
||||
|
||||
UadpNetworkMessageContentMask uadpNetworkMessageContentMask = UadpNetworkMessageContentMask.PublisherId
|
||||
| UadpNetworkMessageContentMask.WriterGroupId
|
||||
| UadpNetworkMessageContentMask.PayloadHeader;
|
||||
UadpDataSetMessageContentMask uadpDataSetMessageContentMask = UadpDataSetMessageContentMask.None;
|
||||
|
||||
DataSetFieldContentMask dataSetFieldContentMask = DataSetFieldContentMask.None;
|
||||
|
||||
DataSetMetaDataType[] dataSetMetaDataArray = new DataSetMetaDataType[]
|
||||
{
|
||||
MessagesHelper.CreateDataSetMetaData1("DataSet1"),
|
||||
MessagesHelper.CreateDataSetMetaData2("DataSet2"),
|
||||
};
|
||||
|
||||
PubSubConfigurationDataType publisherConfiguration = MessagesHelper.CreatePublisherConfiguration(
|
||||
Profiles.PubSubMqttUadpTransport,
|
||||
mqttLocalBrokerUrl, publisherId: publisherId, writerGroupId: writerGroupId,
|
||||
uadpNetworkMessageContentMask: uadpNetworkMessageContentMask,
|
||||
uadpDataSetMessageContentMask: uadpDataSetMessageContentMask,
|
||||
dataSetFieldContentMask: dataSetFieldContentMask,
|
||||
dataSetMetaDataArray: dataSetMetaDataArray, nameSpaceIndexForData: kNamespaceIndexAllTypes,
|
||||
keyFrameCount: Convert.ToUInt32(keyFrameCount));
|
||||
Assert.IsNotNull(publisherConfiguration, "publisherConfiguration should not be null");
|
||||
|
||||
// Configure the mqtt publisher configuration with the MQTTbroker
|
||||
PubSubConnectionDataType mqttPublisherConnection = MessagesHelper.GetConnection(publisherConfiguration, publisherId);
|
||||
Assert.IsNotNull(mqttPublisherConnection, "The MQTT publisher connection is invalid.");
|
||||
mqttPublisherConnection.ConnectionProperties = mqttConfiguration.ConnectionProperties;
|
||||
Assert.IsNotNull(mqttPublisherConnection.ConnectionProperties, "The MQTT publisher connection properties are not valid.");
|
||||
|
||||
// Create publisher application for multiple datasets
|
||||
UaPubSubApplication publisherApplication = UaPubSubApplication.Create(publisherConfiguration);
|
||||
MessagesHelper.LoadData(publisherApplication, kNamespaceIndexAllTypes);
|
||||
|
||||
IUaPubSubConnection publisherConnection = publisherApplication.PubSubConnections.First();
|
||||
Assert.IsNotNull(publisherConnection, "Publisher first connection should not be null");
|
||||
|
||||
Assert.IsNotNull(publisherConfiguration.Connections.First(), "publisherConfiguration first connection should not be null");
|
||||
Assert.IsNotNull(publisherConfiguration.Connections.First().WriterGroups.First(), "publisherConfiguration first writer group of first connection should not be null");
|
||||
|
||||
var networkMessages = publisherConnection.CreateNetworkMessages(publisherConfiguration.Connections.First().WriterGroups.First(), new WriterGroupPublishState());
|
||||
Assert.IsNotNull(networkMessages, "connection.CreateNetworkMessages shall not return null");
|
||||
Assert.GreaterOrEqual(networkMessages.Count, 1, "connection.CreateNetworkMessages shall have at least one network message");
|
||||
|
||||
UadpNetworkMessage uaNetworkMessage = networkMessages[0] as UadpNetworkMessage;
|
||||
Assert.IsNotNull(uaNetworkMessage, "networkMessageEncode should not be null");
|
||||
|
||||
bool hasDataSetWriterId = (uadpNetworkMessageContentMask & UadpNetworkMessageContentMask.PayloadHeader) != 0;
|
||||
|
||||
PubSubConfigurationDataType subscriberConfiguration = MessagesHelper.CreateSubscriberConfiguration(
|
||||
Profiles.PubSubMqttUadpTransport,
|
||||
mqttLocalBrokerUrl, publisherId: publisherId, writerGroupId: writerGroupId, setDataSetWriterId: hasDataSetWriterId,
|
||||
uadpNetworkMessageContentMask: uadpNetworkMessageContentMask,
|
||||
uadpDataSetMessageContentMask: uadpDataSetMessageContentMask,
|
||||
dataSetFieldContentMask: dataSetFieldContentMask,
|
||||
dataSetMetaDataArray: dataSetMetaDataArray,
|
||||
nameSpaceIndexForData: kNamespaceIndexAllTypes, keyFrameCount: Convert.ToUInt32(keyFrameCount));
|
||||
Assert.IsNotNull(subscriberConfiguration, "subscriberConfiguration should not be null");
|
||||
|
||||
// Create subscriber application for multiple datasets
|
||||
UaPubSubApplication subscriberApplication = UaPubSubApplication.Create(subscriberConfiguration);
|
||||
Assert.IsNotNull(subscriberApplication, "subscriberApplication should not be null");
|
||||
Assert.IsNotNull(subscriberApplication.PubSubConnections.First(), "subscriberConfiguration first connection should not be null");
|
||||
|
||||
// Configure the mqtt subscriber configuration with the MQTTbroker
|
||||
PubSubConnectionDataType mqttSubcriberConnection = MessagesHelper.GetConnection(subscriberConfiguration, publisherId);
|
||||
Assert.IsNotNull(mqttSubcriberConnection, "The MQTT subscriber connection is invalid.");
|
||||
mqttSubcriberConnection.ConnectionProperties = mqttConfiguration.ConnectionProperties;
|
||||
Assert.IsNotNull(mqttSubcriberConnection.ConnectionProperties, "The MQTT subscriber connection properties are not valid.");
|
||||
|
||||
var dataSetReaders = subscriberApplication.PubSubConnections.First().GetOperationalDataSetReaders();
|
||||
Assert.IsNotNull(dataSetReaders, "dataSetReaders should not be null");
|
||||
IUaPubSubConnection subscriberConnection = subscriberApplication.PubSubConnections.First();
|
||||
Assert.IsNotNull(subscriberConnection, "Subscriber first connection should not be null");
|
||||
|
||||
//Act
|
||||
// it will signal if the uadp message was received from local ip
|
||||
m_uaDataShutdownEvent = new ManualResetEvent(false);
|
||||
// it will signal if the mqtt with delta frame message was received from local ip
|
||||
m_uaDeltaDataShutdownEvent = new ManualResetEvent(false);
|
||||
|
||||
m_isDeltaFrame = keyFrameCount > 1;
|
||||
subscriberApplication.DataReceived += UaPubSubApplication_DataReceived;
|
||||
subscriberConnection.Start();
|
||||
|
||||
publisherConnection.Start();
|
||||
|
||||
//Assert
|
||||
m_snapshotData = MessagesHelper.GetSnapshotData(publisherApplication, kNamespaceIndexAllTypes);
|
||||
if (!m_uaDataShutdownEvent.WaitOne(kEstimatedPublishingTime))
|
||||
{
|
||||
Assert.Fail("The UADP message was not received");
|
||||
}
|
||||
if (keyFrameCount > 1)
|
||||
{
|
||||
MessagesHelper.UpdateSnapshotData(publisherApplication, kNamespaceIndexAllTypes);
|
||||
if (!m_uaDeltaDataShutdownEvent.WaitOne(kEstimatedPublishingTime))
|
||||
{
|
||||
Assert.Fail("The UADP delta message was not received");
|
||||
}
|
||||
}
|
||||
if (keyFrameCount > 2)
|
||||
{
|
||||
for (int keyCount = 0; keyCount < keyFrameCount - 1; keyCount++)
|
||||
{
|
||||
m_uaDeltaDataShutdownEvent.Reset();
|
||||
m_snapshotData = MessagesHelper.GetSnapshotData(publisherApplication, kNamespaceIndexAllTypes);
|
||||
MessagesHelper.UpdateSnapshotData(publisherApplication, kNamespaceIndexAllTypes);
|
||||
if (!m_uaDeltaDataShutdownEvent.WaitOne(kEstimatedPublishingTime))
|
||||
{
|
||||
Assert.Fail("The UADP delta message was not received");
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
subscriberConnection.Stop();
|
||||
publisherConnection.Stop();
|
||||
@@ -173,12 +320,12 @@ namespace Opc.Ua.PubSub.Tests.Transport
|
||||
[Values((byte)1, (UInt16)1, (UInt32)1, (UInt64)1, "abc")] object publisherId,
|
||||
[Values(0, 10000)] double metaDataUpdateTime)
|
||||
{
|
||||
RestartMosquitto("mosquitto");
|
||||
RestartMosquitto();
|
||||
|
||||
//Arrange
|
||||
UInt16 writerGroupId = 1;
|
||||
|
||||
string mqttLocalBrokerUrl = "mqtt://localhost:1883";
|
||||
string mqttLocalBrokerUrl = string.Format(MqttUrlFormat, Utils.UriSchemeMqtt, "localhost");
|
||||
|
||||
ITransportProtocolConfiguration mqttConfiguration = new MqttClientProtocolConfiguration(version: EnumMqttProtocolVersion.V500);
|
||||
|
||||
@@ -204,7 +351,7 @@ namespace Opc.Ua.PubSub.Tests.Transport
|
||||
jsonDataSetMessageContentMask: jsonDataSetMessageContentMask,
|
||||
dataSetFieldContentMask: dataSetFieldContentMask,
|
||||
dataSetMetaDataArray: dataSetMetaDataArray, nameSpaceIndexForData: kNamespaceIndexAllTypes,
|
||||
metaDataUpdateTime);
|
||||
metaDataUpdateTime: metaDataUpdateTime);
|
||||
Assert.IsNotNull(publisherConfiguration, "publisherConfiguration should not be null");
|
||||
|
||||
// Configure the mqtt publisher configuration with the MQTTbroker
|
||||
@@ -269,6 +416,7 @@ namespace Opc.Ua.PubSub.Tests.Transport
|
||||
// it will signal if the changed configuration message was received on local ip
|
||||
m_uaConfigurationUpdateEvent = new ManualResetEvent(false);
|
||||
|
||||
m_isDeltaFrame = false;
|
||||
subscriberApplication.DataReceived += UaPubSubApplication_DataReceived;
|
||||
subscriberApplication.MetaDataReceived += UaPubSubApplication_MetaDataReceived;
|
||||
subscriberApplication.ConfigurationUpdating += UaPubSubApplication_ConfigurationUpdating;
|
||||
@@ -290,6 +438,146 @@ namespace Opc.Ua.PubSub.Tests.Transport
|
||||
publisherConnection.Stop();
|
||||
}
|
||||
|
||||
[Test(Description = "Validate mqtt local pub/sub connection with json data.")]
|
||||
#if !CUSTOM_TESTS
|
||||
[Ignore("A mosquitto tool should be installed local in order to run correctly.")]
|
||||
#endif
|
||||
public void ValidateMqttLocalPubSubConnectionWithDeltaJson(
|
||||
[Values((byte)1, (UInt16)1, (UInt32)1, (UInt64)1, "abc")] object publisherId,
|
||||
[Values(2, 3, 4)] Int32 keyFrameCount)
|
||||
{
|
||||
RestartMosquitto();
|
||||
|
||||
//Arrange
|
||||
UInt16 writerGroupId = 1;
|
||||
|
||||
string mqttLocalBrokerUrl = string.Format(MqttUrlFormat, Utils.UriSchemeMqtt, "localhost");
|
||||
|
||||
ITransportProtocolConfiguration mqttConfiguration = new MqttClientProtocolConfiguration(version: EnumMqttProtocolVersion.V500);
|
||||
|
||||
JsonNetworkMessageContentMask jsonNetworkMessageContentMask = JsonNetworkMessageContentMask.NetworkMessageHeader
|
||||
| JsonNetworkMessageContentMask.PublisherId
|
||||
| JsonNetworkMessageContentMask.DataSetMessageHeader;
|
||||
JsonDataSetMessageContentMask jsonDataSetMessageContentMask = JsonDataSetMessageContentMask.None;
|
||||
|
||||
DataSetFieldContentMask dataSetFieldContentMask = DataSetFieldContentMask.None;
|
||||
|
||||
DataSetMetaDataType[] dataSetMetaDataArray = new DataSetMetaDataType[]
|
||||
{
|
||||
MessagesHelper.CreateDataSetMetaData1("DataSet1"),
|
||||
MessagesHelper.CreateDataSetMetaData2("DataSet2"),
|
||||
};
|
||||
|
||||
PubSubConfigurationDataType publisherConfiguration = MessagesHelper.CreatePublisherConfiguration(
|
||||
Profiles.PubSubMqttJsonTransport,
|
||||
mqttLocalBrokerUrl, publisherId: publisherId, writerGroupId: writerGroupId,
|
||||
jsonNetworkMessageContentMask: jsonNetworkMessageContentMask,
|
||||
jsonDataSetMessageContentMask: jsonDataSetMessageContentMask,
|
||||
dataSetFieldContentMask: dataSetFieldContentMask,
|
||||
dataSetMetaDataArray: dataSetMetaDataArray, nameSpaceIndexForData: kNamespaceIndexAllTypes,
|
||||
metaDataUpdateTime: 1000, keyFrameCount: Convert.ToUInt32(keyFrameCount));
|
||||
Assert.IsNotNull(publisherConfiguration, "publisherConfiguration should not be null");
|
||||
|
||||
// Configure the mqtt publisher configuration with the MQTTbroker
|
||||
PubSubConnectionDataType mqttPublisherConnection = MessagesHelper.GetConnection(publisherConfiguration, publisherId);
|
||||
Assert.IsNotNull(mqttPublisherConnection, "The MQTT publisher connection is invalid.");
|
||||
mqttPublisherConnection.ConnectionProperties = mqttConfiguration.ConnectionProperties;
|
||||
Assert.IsNotNull(mqttPublisherConnection.ConnectionProperties, "The MQTT publisher connection properties are not valid.");
|
||||
|
||||
// Create publisher application for multiple datasets
|
||||
UaPubSubApplication publisherApplication = UaPubSubApplication.Create(publisherConfiguration);
|
||||
MessagesHelper.LoadData(publisherApplication, kNamespaceIndexAllTypes);
|
||||
|
||||
IUaPubSubConnection publisherConnection = publisherApplication.PubSubConnections.First();
|
||||
Assert.IsNotNull(publisherConnection, "Publisher first connection should not be null");
|
||||
|
||||
Assert.IsNotNull(publisherConfiguration.Connections.First(), "publisherConfiguration first connection should not be null");
|
||||
Assert.IsNotNull(publisherConfiguration.Connections.First().WriterGroups.First(), "publisherConfiguration first writer group of first connection should not be null");
|
||||
|
||||
var networkMessages = publisherConnection.CreateNetworkMessages(publisherConfiguration.Connections.First().WriterGroups.First(), new WriterGroupPublishState());
|
||||
Assert.IsNotNull(networkMessages, "connection.CreateNetworkMessages shall not return null");
|
||||
Assert.GreaterOrEqual(networkMessages.Count, 1, "connection.CreateNetworkMessages shall have at least one network message");
|
||||
|
||||
List<JsonNetworkMessage> uaNetworkMessages = MessagesHelper.GetJsonUaDataNetworkMessages(networkMessages.Cast<JsonNetworkMessage>().ToList());
|
||||
Assert.IsNotNull(uaNetworkMessages, "Json ua-data entries are missing from configuration!");
|
||||
|
||||
List<JsonNetworkMessage> uaMetaDataNetworkMessages = MessagesHelper.GetJsonUaMetaDataNetworkMessages(networkMessages.Cast<JsonNetworkMessage>().ToList());
|
||||
Assert.IsNotNull(uaMetaDataNetworkMessages, "Json ua-metadata entries are missing from configuration!");
|
||||
|
||||
bool hasDataSetWriterId = (jsonNetworkMessageContentMask & JsonNetworkMessageContentMask.DataSetMessageHeader) != 0
|
||||
&& (jsonDataSetMessageContentMask & JsonDataSetMessageContentMask.DataSetWriterId) != 0;
|
||||
|
||||
PubSubConfigurationDataType subscriberConfiguration = MessagesHelper.CreateSubscriberConfiguration(
|
||||
Profiles.PubSubMqttJsonTransport,
|
||||
mqttLocalBrokerUrl, publisherId: publisherId, writerGroupId: writerGroupId, setDataSetWriterId: hasDataSetWriterId,
|
||||
jsonNetworkMessageContentMask: jsonNetworkMessageContentMask,
|
||||
jsonDataSetMessageContentMask: jsonDataSetMessageContentMask,
|
||||
dataSetFieldContentMask: dataSetFieldContentMask,
|
||||
dataSetMetaDataArray: dataSetMetaDataArray,
|
||||
nameSpaceIndexForData: kNamespaceIndexAllTypes, keyFrameCount: Convert.ToUInt32(keyFrameCount));
|
||||
Assert.IsNotNull(subscriberConfiguration, "subscriberConfiguration should not be null");
|
||||
|
||||
// Create subscriber application for multiple datasets
|
||||
UaPubSubApplication subscriberApplication = UaPubSubApplication.Create(subscriberConfiguration);
|
||||
Assert.IsNotNull(subscriberApplication, "subscriberApplication should not be null");
|
||||
Assert.IsNotNull(subscriberApplication.PubSubConnections.First(), "subscriberConfiguration first connection should not be null");
|
||||
|
||||
// Configure the mqtt subscriber configuration with the MQTTbroker
|
||||
PubSubConnectionDataType mqttSubcriberConnection = MessagesHelper.GetConnection(subscriberConfiguration, publisherId);
|
||||
Assert.IsNotNull(mqttSubcriberConnection, "The MQTT subscriber connection is invalid.");
|
||||
mqttSubcriberConnection.ConnectionProperties = mqttConfiguration.ConnectionProperties;
|
||||
Assert.IsNotNull(mqttSubcriberConnection.ConnectionProperties, "The MQTT subscriber connection properties are not valid.");
|
||||
|
||||
var dataSetReaders = subscriberApplication.PubSubConnections.First().GetOperationalDataSetReaders();
|
||||
Assert.IsNotNull(dataSetReaders, "dataSetReaders should not be null");
|
||||
IUaPubSubConnection subscriberConnection = subscriberApplication.PubSubConnections.First();
|
||||
Assert.IsNotNull(subscriberConnection, "Subscriber first connection should not be null");
|
||||
|
||||
//Act
|
||||
// it will signal if the mqtt message was received from local ip
|
||||
m_uaDataShutdownEvent = new ManualResetEvent(false);
|
||||
// it will signal if the mqtt with delta frame message was received from local ip
|
||||
m_uaDeltaDataShutdownEvent = new ManualResetEvent(false);
|
||||
|
||||
m_isDeltaFrame = keyFrameCount > 1;
|
||||
subscriberApplication.DataReceived += UaPubSubApplication_DataReceived;
|
||||
subscriberConnection.Start();
|
||||
|
||||
publisherConnection.Start();
|
||||
|
||||
//Assert
|
||||
m_snapshotData = MessagesHelper.GetSnapshotData(publisherApplication, kNamespaceIndexAllTypes);
|
||||
Assert.IsNotNull(m_snapshotData, "snapshot data should not be null");
|
||||
if (!m_uaDataShutdownEvent.WaitOne(kEstimatedPublishingTime))
|
||||
{
|
||||
Assert.Fail("The JSON message was not received");
|
||||
}
|
||||
if (keyFrameCount > 1)
|
||||
{
|
||||
MessagesHelper.UpdateSnapshotData(publisherApplication, kNamespaceIndexAllTypes);
|
||||
if (!m_uaDeltaDataShutdownEvent.WaitOne(kEstimatedPublishingTime))
|
||||
{
|
||||
Assert.Fail("The JSON delta message was not received");
|
||||
}
|
||||
}
|
||||
if (keyFrameCount > 2)
|
||||
{
|
||||
for (int keyCount = 0; keyCount < keyFrameCount - 1; keyCount++)
|
||||
{
|
||||
m_uaDeltaDataShutdownEvent.Reset();
|
||||
m_snapshotData = MessagesHelper.GetSnapshotData(publisherApplication, kNamespaceIndexAllTypes);
|
||||
MessagesHelper.UpdateSnapshotData(publisherApplication, kNamespaceIndexAllTypes);
|
||||
if (!m_uaDeltaDataShutdownEvent.WaitOne(kEstimatedPublishingTime))
|
||||
{
|
||||
Assert.Fail("The JSON delta message was not received");
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
subscriberConnection.Stop();
|
||||
publisherConnection.Stop();
|
||||
}
|
||||
|
||||
#region Private methods
|
||||
|
||||
/// <summary>
|
||||
@@ -299,7 +587,37 @@ namespace Opc.Ua.PubSub.Tests.Transport
|
||||
/// <param name="e"></param>
|
||||
private void UaPubSubApplication_DataReceived(object sender, SubscribedDataEventArgs e)
|
||||
{
|
||||
m_uaDataShutdownEvent.Set();
|
||||
if (m_isDeltaFrame)
|
||||
{
|
||||
bool hasChanged = false;
|
||||
foreach (UaDataSetMessage dataSetMessage in e.NetworkMessage.DataSetMessages)
|
||||
{
|
||||
foreach(Field field in dataSetMessage.DataSet.Fields)
|
||||
{
|
||||
if (m_snapshotData.Keys.Contains(field.TargetNodeId))
|
||||
{
|
||||
// check if the published value changed
|
||||
DataValue snapshotValue = m_snapshotData[field.TargetNodeId];
|
||||
if(!field.Value.Equals(snapshotValue))
|
||||
{
|
||||
hasChanged = true;
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
if(!hasChanged)
|
||||
{
|
||||
m_uaDataShutdownEvent.Set();
|
||||
}
|
||||
else
|
||||
{
|
||||
m_uaDeltaDataShutdownEvent.Set();
|
||||
}
|
||||
}
|
||||
else
|
||||
{
|
||||
m_uaDataShutdownEvent.Set();
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
@@ -327,11 +645,11 @@ namespace Opc.Ua.PubSub.Tests.Transport
|
||||
/// </summary>
|
||||
/// <param name="processName"></param>
|
||||
/// <param name="arguments"></param>
|
||||
private void RestartMosquitto(string processName, string arguments = "")
|
||||
private void RestartMosquitto(string arguments = "")
|
||||
{
|
||||
try
|
||||
{
|
||||
Process[] processes = Process.GetProcessesByName(processName);
|
||||
Process[] processes = Process.GetProcessesByName(DefaultBrokerProcessName);
|
||||
if (processes.Length > 0)
|
||||
{
|
||||
Process mosquittoProcess = processes[0];
|
||||
@@ -340,10 +658,13 @@ namespace Opc.Ua.PubSub.Tests.Transport
|
||||
|
||||
using (Process process = new Process())
|
||||
{
|
||||
string programFilesPath = Environment.Is64BitOperatingSystem ?
|
||||
Environment.GetEnvironmentVariable("ProgramW6432") :
|
||||
Environment.GetFolderPath(Environment.SpecialFolder.ProgramFilesX86);
|
||||
|
||||
ProcessStartInfo startInfo =
|
||||
new ProcessStartInfo(
|
||||
Path.Combine(Environment.GetFolderPath(Environment.SpecialFolder.ProgramFiles),
|
||||
Path.Combine(processName, $"{processName}.exe")));
|
||||
Path.Combine(programFilesPath, Path.Combine(DefaultBrokerProcessName, $"{DefaultBrokerProcessName}.exe")));
|
||||
startInfo.WindowStyle = ProcessWindowStyle.Hidden;
|
||||
//startInfo.CreateNoWindow = true;
|
||||
//startInfo.RedirectStandardOutput = true;
|
||||
|
||||
Reference in New Issue
Block a user