Skip to content
27 changes: 15 additions & 12 deletions docs/PubSub.md
Original file line number Diff line number Diff line change
Expand Up @@ -95,17 +95,13 @@ DataSet #1 received (3 fields): BoolToggle=False, Int32=1, DateTime=09/28/2026 1
DataSet #2 received (3 fields): BoolToggle=True, Int32=2, DateTime=09/28/2026 12:18:51
```

The `udp-uadp` profile signs and encrypts the UADP messages with demo keys
that are built into the sample. A production deployment obtains its keys from
a [Security Key Service](#security-key-service-sks). Two warnings are expected:

- `DataSetReader Reader 1 faulted on MessageReceiveTimeout (>00:00:05)`
appears when no matching message arrives for five seconds, for example
before the publisher starts. The reader resumes when messages arrive.
- `Dropping unsecured inbound frame ... requiring SignAndEncrypt` appears once.
The publisher announces its DataSet metadata without message security, and
the secured subscriber discards the announcement. It decodes the DataSets
with the metadata configured on its DataSetReader instead.
The `udp-uadp` profile signs and encrypts the UADP messages, including the
DataSet metadata announcement, with demo keys that are built into the sample.
A production deployment obtains its keys from a
[Security Key Service](#security-key-service-sks).
`DataSetReader Reader 1 faulted on MessageReceiveTimeout (>00:00:05)` can
appear when no matching message arrives for five seconds, for example before
the publisher starts. The reader resumes when messages arrive.

The sample's default endpoint, `opc.udp://239.0.0.1:4840`, is a multicast
address. On one computer, the subscriber can miss multicast messages: the UDP
Expand Down Expand Up @@ -289,7 +285,9 @@ The publisher-side `MetaDataPublisher` ([§6.2.2.5](https://reference.opcfoundat
emits a retained `JsonMetaDataMessage` / `UadpDiscoveryResponseMessage`
on the well-known `ua-metadata` topic at startup and after each
configuration version bump; subscribers cache it before the first
KeyFrame arrives.
KeyFrame arrives. A connection configured for message security signs and
encrypts the UADP announcement with the same keys as its DataSetMessages,
so only subscribers that hold the keys can read the DataSet layout.
Comment thread
marcschier marked this conversation as resolved.
Outdated

A subscriber also registers the `DataSetMetaData` configured on each
DataSetReader whose filter names an exact PublisherId, WriterGroupId, and
Expand Down Expand Up @@ -1224,6 +1222,11 @@ subscribers that need to find publishers and bind to metadata at runtime.
duplicate probes are suppressed, and identical responses are throttled.
- MQTT publishes retained discovery messages on the standard status, connection,
application, endpoint, and metadata topics.
- A UADP connection configured for message security signs and encrypts its
discovery requests, responses, and announcements with the keys of its
SecurityGroup, so only applications that hold those keys can discover each
other. The MQTT last-will status message is the exception: the broker
publishes it later, so it is sent unsecured.

```csharp
PubSubDiscoveryResult result = await application.RequestDiscoveryAsync(
Expand Down
107 changes: 22 additions & 85 deletions src/Opc.Ua.PubSub/Application/MetaDataPublisher.cs
Original file line number Diff line number Diff line change
Expand Up @@ -75,6 +75,12 @@ namespace Opc.Ua.PubSub.Application
/// <see cref="UadpDiscoveryType.DataSetMetaData"/>.
/// </para>
/// <para>
/// Announcements go through the connection's send pipeline, so a
/// connection configured for message security signs and encrypts
/// them with the same keys as its DataSetMessages, and refuses to
/// send them when it cannot secure them.
Comment thread
marcschier marked this conversation as resolved.
Outdated
/// </para>
/// <para>
/// Lifetime is owned by <see cref="PubSubApplication"/>: started
/// after <c>EnableConnectionsAsync</c> returns, disposed before
/// the connections are torn down.
Expand All @@ -84,9 +90,6 @@ internal sealed class MetaDataPublisher : IAsyncDisposable
{
private readonly PubSubApplication m_application;
private readonly IDataSetMetaDataRegistry m_registry;
private readonly IReadOnlyDictionary<string, INetworkMessageEncoder> m_encoders;
private readonly IPubSubDiagnostics m_diagnostics;
private readonly ITelemetryContext m_telemetry;
private readonly BackgroundTaskScope m_backgroundWork;
private readonly TimeProvider m_timeProvider;
private readonly ILogger<MetaDataPublisher> m_logger;
Expand All @@ -106,17 +109,11 @@ internal sealed class MetaDataPublisher : IAsyncDisposable
/// the matching transport per writer group.
/// </param>
/// <param name="metaDataRegistry">Shared metadata registry.</param>
/// <param name="encoders">
/// Network-message encoders keyed by transport profile URI.
/// </param>
/// <param name="diagnostics">Diagnostics sink.</param>
/// <param name="telemetry">Telemetry context.</param>
/// <param name="timeProvider">Clock used to stamp MessageIds.</param>
public MetaDataPublisher(
PubSubApplication application,
IDataSetMetaDataRegistry metaDataRegistry,
IReadOnlyDictionary<string, INetworkMessageEncoder> encoders,
IPubSubDiagnostics diagnostics,
ITelemetryContext telemetry,
TimeProvider timeProvider)
{
Expand All @@ -128,14 +125,6 @@ public MetaDataPublisher(
{
throw new ArgumentNullException(nameof(metaDataRegistry));
}
if (encoders is null)
{
throw new ArgumentNullException(nameof(encoders));
}
if (diagnostics is null)
{
throw new ArgumentNullException(nameof(diagnostics));
}
if (telemetry is null)
{
throw new ArgumentNullException(nameof(telemetry));
Expand All @@ -146,9 +135,6 @@ public MetaDataPublisher(
}
m_application = application;
m_registry = metaDataRegistry;
m_encoders = encoders;
m_diagnostics = diagnostics;
m_telemetry = telemetry;
m_backgroundWork = new BackgroundTaskScope(nameof(MetaDataPublisher), telemetry);
m_timeProvider = timeProvider;
m_logger = telemetry.CreateLogger<MetaDataPublisher>();
Expand Down Expand Up @@ -497,22 +483,16 @@ private async ValueTask PublishMetaDataAsync(
{
return;
}
string profile = connection.TransportProfileUri;
string family = TransportProfileFamily(profile);
Uuid classId = metaData.DataSetClassId == Guid.Empty
? Uuid.Empty
: new Uuid(metaData.DataSetClassId);
ReadOnlyMemory<byte> payload;
string? topic;
if (string.Equals(family, "Json", StringComparison.Ordinal))
PubSubNetworkMessage message;
if (string.Equals(
TransportProfileFamily(connection.TransportProfileUri),
"Json",
StringComparison.Ordinal))
{
if (!TryResolveEncoder(profile, family, out INetworkMessageEncoder? encoder) ||
encoder is null)
{
m_logger.NoJsonEncoderRegistered(profile);
return;
}
var message = new JsonMetaDataMessage
message = new JsonMetaDataMessage
{
MessageId = NewMessageId(),
PublisherId = connection.PublisherId,
Expand All @@ -530,22 +510,10 @@ private async ValueTask PublishMetaDataAsync(
MetaData = metaData,
MetaDataPayload = metaData
};
var context = new PubSubNetworkMessageContext(
ServiceMessageContext.CreateEmpty(m_telemetry),
m_registry,
m_diagnostics,
m_timeProvider);
payload = await encoder.EncodeAsync(message, context, cancellationToken)
.ConfigureAwait(false);
topic = ResolveMetaDataTopic(
transport,
connection.PublisherId,
writerGroup.WriterGroupId,
writer.DataSetWriterId);
}
else
{
var message = new UadpDiscoveryResponseMessage
message = new UadpDiscoveryResponseMessage
{
PublisherId = connection.PublisherId,
WriterGroupId = writerGroup.WriterGroupId,
Expand All @@ -556,44 +524,17 @@ private async ValueTask PublishMetaDataAsync(
SequenceNumber = NewSequenceNumber(),
StatusCode = StatusCodes.Good
};
var context = new PubSubNetworkMessageContext(
ServiceMessageContext.CreateEmpty(m_telemetry),
m_registry,
m_diagnostics,
m_timeProvider);
payload = UadpDiscoveryCoder.Encode(message, context);
topic = ResolveMetaDataTopic(
transport,
connection.PublisherId,
writerGroup.WriterGroupId,
writer.DataSetWriterId);
}
string? topic = ResolveMetaDataTopic(
transport,
connection.PublisherId,
writerGroup.WriterGroupId,
writer.DataSetWriterId);

await transport.SendAsync(payload, topic, cancellationToken).ConfigureAwait(false);
}

private bool TryResolveEncoder(
string profile,
string family,
out INetworkMessageEncoder? encoder)
{
if (m_encoders.TryGetValue(profile, out encoder))
{
return true;
}
foreach (KeyValuePair<string, INetworkMessageEncoder> entry in m_encoders)
{
if (string.Equals(
TransportProfileFamily(entry.Key),
family,
StringComparison.Ordinal))
{
encoder = entry.Value;
return true;
}
}
encoder = null;
return false;
// The connection applies its message security, so a secured
// connection never announces the DataSet layout in the clear.
await connection.SendMetaDataAnnouncementAsync(message, topic, cancellationToken)
.ConfigureAwait(false);
}

private static string? ResolveMetaDataTopic(
Expand Down Expand Up @@ -681,10 +622,6 @@ public static partial void FailedToPublishMetadataChange(
Exception exception,
ushort writer,
ushort group);

[LoggerMessage(EventId = PubSubEventIds.MetaDataPublisher + 2, Level = LogLevel.Debug,
Message = "No JSON encoder registered for {Profile}; metadata publish skipped.")]
public static partial void NoJsonEncoderRegistered(this ILogger logger, string? profile);
}

}
2 changes: 0 additions & 2 deletions src/Opc.Ua.PubSub/Application/PubSubApplication.cs
Original file line number Diff line number Diff line change
Expand Up @@ -662,8 +662,6 @@ public async ValueTask StartAsync(CancellationToken cancellationToken = default)
var metaDataPublisher = new MetaDataPublisher(
this,
MetaDataRegistry,
m_encoderMap,
m_aggregatingDiagnostics,
m_telemetry,
m_timeProvider);
try
Expand Down
Loading
Loading