/* ======================================================================== * 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 Opc.Ua; using Opc.Ua.PubSub.Configuration; using Opc.Ua.PubSub.Eth; using Opc.Ua.PubSub.Kafka; namespace Quickstarts.ConsoleReferencePubSubClient { /// /// Builds minimal Part 14 /// payloads for the demo wire profiles using the fluent /// . Each payload wires one /// PubSubConnection > ReaderGroup > DataSetReader filtered on /// PublisherId / WriterGroupId / DataSetWriterId. /// public static class SubscriberConfigurationBuilder { public const string ReaderName = "Reader 1"; public const string DataSetName = "Simple"; public const string DefaultUdpEndpoint = "opc.udp://239.0.0.1:4840"; public const string DefaultEthEndpoint = "opc.eth://01-00-5E-7F-00-01"; public const string DefaultMqttEndpoint = "mqtt://localhost:1883"; public const string DefaultKafkaEndpoint = "kafka://localhost:9092"; private const string MqttQueueName = "Quickstarts/Reference/Simple"; private const string KafkaTopicName = "Quickstarts.Reference.Simple"; public static string DefaultEndpointFor(SubscriberProfile profile) { return profile switch { SubscriberProfile.UdpUadp => DefaultUdpEndpoint, SubscriberProfile.EthUadp => DefaultEthEndpoint, SubscriberProfile.KafkaUadp or SubscriberProfile.KafkaJson => DefaultKafkaEndpoint, SubscriberProfile.MqttUadp or SubscriberProfile.MqttJson => DefaultMqttEndpoint, _ => throw new ArgumentOutOfRangeException(nameof(profile)) }; } public static PubSubConfigurationDataType Build( SubscriberProfile profile, string endpoint, ushort publisherIdFilter, ushort writerGroupIdFilter, ushort dataSetWriterIdFilter) { // UDP and Ethernet are datagram transports (no broker queue); // broker profiles use broker transport settings instead. bool udp = profile is SubscriberProfile.UdpUadp or SubscriberProfile.EthUadp; // UADP message security (SignAndEncrypt) is wired for the UADP // profiles via the shared StaticSecurityKeyProvider. The JSON // profile has no UADP security wrapper, so it stays unsecured. bool secured = profile is not (SubscriberProfile.MqttJson or SubscriberProfile.KafkaJson); string brokerQueueName = profile is SubscriberProfile.KafkaUadp or SubscriberProfile.KafkaJson ? KafkaTopicName : MqttQueueName; string transportProfileUri = profile switch { SubscriberProfile.UdpUadp => Profiles.PubSubUdpUadpTransport, SubscriberProfile.EthUadp => EthProfiles.PubSubEthUadpTransport, SubscriberProfile.MqttUadp => Profiles.PubSubMqttUadpTransport, SubscriberProfile.MqttJson => Profiles.PubSubMqttJsonTransport, SubscriberProfile.KafkaUadp => KafkaProfiles.PubSubKafkaUadpTransport, SubscriberProfile.KafkaJson => KafkaProfiles.PubSubKafkaJsonTransport, _ => throw new ArgumentOutOfRangeException(nameof(profile)) }; return PubSubConfigurationBuilder.Create() .AddConnection("Subscriber Connection", connection => connection .WithPublisherId(new Variant(publisherIdFilter)) .WithTransportProfile(transportProfileUri) .WithAddress(endpoint) .AddReaderGroup("ReaderGroup 1", group => { group.WithMaxNetworkMessageSize(1500); if (secured) { group.WithSecurity( MessageSecurityMode.SignAndEncrypt, SampleSecurity.SecurityGroupId, SampleSecurity.SecurityKeyServiceUrl); } group.AddDataSetReader(ReaderName, reader => { reader .WithFilter( new Variant(publisherIdFilter), writerGroupIdFilter, dataSetWriterIdFilter) .WithFieldContentMask(DataSetFieldContentMask.RawData) .WithMessageReceiveTimeout(5000) .WithMessageSettings(ReaderMessageSettings(profile)) .WithMirrorSubscribedDataSet(ReaderName) .WithDataSetMetaData(DataSetName, metaData => metaData .WithoutFieldIds() .AddField("BoolToggle", (byte)DataTypes.Boolean, DataTypeIds.Boolean) .AddField("Int32", (byte)DataTypes.Int32, DataTypeIds.Int32) .AddField("DateTime", (byte)DataTypes.DateTime, DataTypeIds.DateTime)); if (!udp) { reader.WithTransportSettings( new BrokerDataSetReaderTransportDataType { QueueName = brokerQueueName, RequestedDeliveryGuarantee = BrokerTransportQualityOfService.BestEffort }); } }); })) .Build(); } private static IEncodeable ReaderMessageSettings(SubscriberProfile profile) { if (profile is SubscriberProfile.MqttJson or SubscriberProfile.KafkaJson) { return new JsonDataSetReaderMessageDataType { NetworkMessageContentMask = (uint)( JsonNetworkMessageContentMask.NetworkMessageHeader | JsonNetworkMessageContentMask.DataSetMessageHeader | JsonNetworkMessageContentMask.PublisherId), DataSetMessageContentMask = (uint)( JsonDataSetMessageContentMask.DataSetWriterId | JsonDataSetMessageContentMask.SequenceNumber | JsonDataSetMessageContentMask.Status | JsonDataSetMessageContentMask.Timestamp) }; } return new UadpDataSetReaderMessageDataType { NetworkMessageContentMask = (uint)( UadpNetworkMessageContentMask.PublisherId | UadpNetworkMessageContentMask.GroupHeader | UadpNetworkMessageContentMask.WriterGroupId | UadpNetworkMessageContentMask.PayloadHeader | UadpNetworkMessageContentMask.NetworkMessageNumber | UadpNetworkMessageContentMask.SequenceNumber), DataSetMessageContentMask = (uint)( UadpDataSetMessageContentMask.Status | UadpDataSetMessageContentMask.SequenceNumber) }; } } }