/* ========================================================================
* Copyright (c) 2005-2026 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.Threading;
using System.Threading.Tasks;
using Microsoft.Extensions.Time.Testing;
using NUnit.Framework;
using Opc.Ua.PubSub.DataSets;
using Opc.Ua.PubSub.Encoding;
using Opc.Ua.PubSub.Groups;
using Opc.Ua.PubSub.MetaData;
using UadpDataSetMessageV2 = Opc.Ua.PubSub.Encoding.Uadp.UadpDataSetMessage;
namespace Opc.Ua.PubSub.Tests.Groups
{
///
/// Validates the event-mode publisher:
/// drains pending events from the
/// sampler, projects them via
/// s and emits one
/// per event with
/// .
///
[TestFixture]
[TestSpec("5.3.3", Summary = "EventDataSetWriter event-message build")]
[TestSpec("6.2.4")]
public class EventDataSetWriterTests
{
[Test]
[TestSpec("5.3.3")]
public async Task BuildEventMessagesAsync_EmitsOneMessagePerEventAsync()
{
var clock = new FakeTimeProvider();
var sampler = new StubSampler();
sampler.Enqueue([new Variant("A1"), new Variant(1.0)]);
sampler.Enqueue([new Variant("A2"), new Variant(2.0)]);
EventDataSetWriter writer = BuildWriter(sampler, clock);
ArrayOf messages =
await writer.BuildEventMessagesAsync().ConfigureAwait(false);
Assert.That(messages, Has.Count.EqualTo(2));
Assert.That(messages[0].MessageType,
Is.EqualTo(PubSubDataSetMessageType.Event));
Assert.That(((DataSetField[]?)messages[0].Fields) ?? [], Has.Length.EqualTo(2));
Assert.That(messages[0].Fields[0].Value, Is.EqualTo(new Variant("A1")));
Assert.That(messages[1].Fields[1].Value, Is.EqualTo(new Variant(2.0)));
Assert.That(messages[0].SequenceNumber, Is.LessThan(messages[1].SequenceNumber));
}
[Test]
[TestSpec("5.3.3")]
public async Task BuildEventMessagesAsync_NoEvents_ReturnsEmptyAsync()
{
var sampler = new StubSampler();
EventDataSetWriter writer = BuildWriter(sampler, new FakeTimeProvider());
ArrayOf messages =
await writer.BuildEventMessagesAsync().ConfigureAwait(false);
Assert.That(messages, Is.Empty);
}
[Test]
[TestSpec("6.2.4")]
public async Task BuildEventMessagesAsync_HonoursFieldContentMaskAsync()
{
var sampler = new StubSampler();
sampler.Enqueue([new Variant(1.0), new Variant(2.0)]);
EventDataSetWriter writer = BuildWriter(
sampler,
new FakeTimeProvider(),
contentMask: (uint)DataSetFieldContentMask.StatusCode);
ArrayOf messages =
await writer.BuildEventMessagesAsync().ConfigureAwait(false);
Assert.That(messages, Has.Count.EqualTo(1));
var dsm = (UadpDataSetMessageV2)messages[0];
Assert.That(dsm.FieldContentMask & DataSetFieldContentMask.StatusCode,
Is.EqualTo(DataSetFieldContentMask.StatusCode));
}
[Test]
[TestSpec("6.2.4")]
public async Task EventPublishedDataSet_AlignsFieldsToMetaDataAsync()
{
var sampler = new StubSampler();
sampler.Enqueue([new Variant("event"), new Variant(99)]);
EventPublishedDataSet pds = BuildPublishedDataSet(sampler);
ArrayOf> rows =
await pds.SampleEventsAsync().ConfigureAwait(false);
Assert.That(((ArrayOf[]?)rows) ?? [], Has.Length.EqualTo(1));
Assert.That(rows[0][0].Name, Is.EqualTo("Message"));
Assert.That(rows[0][1].Name, Is.EqualTo("Severity"));
Assert.That(rows[0][0].Value, Is.EqualTo(new Variant("event")));
Assert.That(rows[0][1].Value, Is.EqualTo(new Variant(99)));
}
[Test]
[TestSpec("6.2.4")]
public async Task EventPublishedDataSet_SamplesOneOccurrencePerSnapshotAsync()
{
//
// An event dataset has no current state to sample, so the general
// IPublishedDataSet contract drains one occurrence per call and
// declares it as an event. Without the declaration a writer group
// would treat the occurrence as a state snapshot and derive delta
// frames by comparing it against an unrelated earlier occurrence.
//
var sampler = new StubSampler();
sampler.Enqueue([new Variant("first"), new Variant(1)]);
sampler.Enqueue([new Variant("second"), new Variant(2)]);
EventPublishedDataSet pds = BuildPublishedDataSet(sampler);
PublishedDataSetSnapshot first = await pds.SampleAsync().ConfigureAwait(false);
PublishedDataSetSnapshot second = await pds.SampleAsync().ConfigureAwait(false);
PublishedDataSetSnapshot exhausted = await pds.SampleAsync().ConfigureAwait(false);
Assert.Multiple(() =>
{
Assert.That(first.MessageType, Is.EqualTo(PubSubDataSetMessageType.Event));
Assert.That(first.Fields[0].Value, Is.EqualTo(new Variant("first")));
Assert.That(second.MessageType, Is.EqualTo(PubSubDataSetMessageType.Event));
Assert.That(second.Fields[0].Value, Is.EqualTo(new Variant("second")));
Assert.That(exhausted.MessageType, Is.Null);
Assert.That(exhausted.Fields.Count, Is.Zero);
});
}
[Test]
public async Task EventPublishedDataSetExposesClassIdAndFixedMetadataAsync()
{
var sampler = new StubSampler();
sampler.Enqueue([new Variant("event"), new Variant(1)]);
EventPublishedDataSet dataSet = BuildPublishedDataSet(sampler);
var classId = new Uuid(Guid.NewGuid());
dataSet.MetaData.DataSetClassId = classId;
bool metadataChanged = false;
EventHandler handler =
(_, _) => metadataChanged = true;
dataSet.MetaDataChanged += handler;
PublishedDataSetSnapshot snapshot =
await dataSet.SampleAsync().ConfigureAwait(false);
dataSet.MetaDataChanged -= handler;
Assert.Multiple(() =>
{
Assert.That(
((IPublishedDataSet)dataSet).DataSetClassId,
Is.EqualTo(classId));
Assert.That(snapshot.MessageType, Is.EqualTo(PubSubDataSetMessageType.Event));
Assert.That(metadataChanged, Is.False);
});
}
[Test]
public async Task EventDataSetWriterExposesContractAndBuildsJsonMessageAsync()
{
var sampler = new StubSampler();
sampler.Enqueue([new Variant("event"), new Variant(1)]);
EventPublishedDataSet dataSet = BuildPublishedDataSet(sampler);
var configuration = new DataSetWriterDataType
{
Name = "json-event-writer",
DataSetWriterId = 8
};
var writer = new EventDataSetWriter(
configuration,
dataSet,
new FakeTimeProvider(),
Profiles.PubSubMqttJsonTransport);
ArrayOf messages =
await writer.BuildEventMessagesAsync().ConfigureAwait(false);
Assert.Multiple(() =>
{
Assert.That(
((IDataSetWriter)writer).PublishedDataSet,
Is.SameAs(dataSet));
Assert.That(writer.KeyFrameCount, Is.Zero);
Assert.That(writer.State.ComponentName, Is.EqualTo(configuration.Name));
Assert.That(messages, Has.Count.EqualTo(1));
Assert.That(
messages[0],
Is.TypeOf());
});
}
private static EventDataSetWriter BuildWriter(
IEventSampler sampler,
TimeProvider clock,
uint contentMask = 0)
{
EventPublishedDataSet pds = BuildPublishedDataSet(sampler);
var writerCfg = new DataSetWriterDataType
{
Name = "evt-writer",
DataSetWriterId = 7,
DataSetFieldContentMask = contentMask
};
return new EventDataSetWriter(writerCfg, pds, clock);
}
private static EventPublishedDataSet BuildPublishedDataSet(IEventSampler sampler)
{
var pubEvents = new PublishedEventsDataType
{
EventNotifier = new NodeId("notifier", 1),
SelectedFields =
[
new SimpleAttributeOperand { TypeDefinitionId = new NodeId("Base", 1) },
new SimpleAttributeOperand { TypeDefinitionId = new NodeId("Base", 1) }
]
};
var pdsCfg = new PublishedDataSetDataType
{
Name = "events-pds",
DataSetMetaData = new DataSetMetaDataType
{
Fields =
[
new FieldMetaData { Name = "Message" },
new FieldMetaData { Name = "Severity" }
]
},
DataSetSource = new ExtensionObject(pubEvents)
};
return new EventPublishedDataSet(pdsCfg, sampler);
}
private sealed class StubSampler : IEventSampler
{
private readonly List> m_pending = [];
public string Name => "stub";
public void Enqueue(IReadOnlyList row)
{
m_pending.Add(row);
}
public ValueTask>> SampleEventsAsync(
ArrayOf selectedFields,
ContentFilter? filter,
CancellationToken cancellationToken = default)
{
IReadOnlyList> copy = [.. m_pending];
m_pending.Clear();
return new ValueTask>>(copy);
}
}
}
}