Skip to content
22 changes: 10 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
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
23 changes: 20 additions & 3 deletions src/Opc.Ua.PubSub/Connections/PubSubConnection.cs
Original file line number Diff line number Diff line change
Expand Up @@ -2454,6 +2454,23 @@ private ValueTask SendNetworkMessageAsync(
return SendNetworkMessageAsync(networkMessage, topic, writerGroup: null, cancellationToken);
}

/// <summary>
/// Sends a DataSetMetaData announcement through the connection's send
/// pipeline, so a connection configured for message security signs and
/// encrypts the announcement like its DataSetMessages and never sends it
/// in the clear.
Comment thread
marcschier marked this conversation as resolved.
Outdated
/// </summary>
/// <param name="announcement">UADP discovery response or JSON metadata message.</param>
/// <param name="topic">Broker topic, or <see langword="null"/> for datagram transports.</param>
/// <param name="cancellationToken">Cancellation token.</param>
internal ValueTask SendMetaDataAnnouncementAsync(
PubSubNetworkMessage announcement,
string? topic,
CancellationToken cancellationToken)
{
return SendNetworkMessageAsync(announcement, topic, writerGroup: null, cancellationToken);
}

private async ValueTask SendNetworkMessageAsync(
PubSubNetworkMessage networkMessage,
string? topic,
Expand Down Expand Up @@ -2483,9 +2500,9 @@ private async ValueTask SendNetworkMessageAsync(

ReadOnlyMemory<byte> payload;
if (m_securityWrapper is not null &&
networkMessage is UadpNetworkMessage uadp)
networkMessage is UadpNetworkMessage or UadpDiscoveryResponseMessage)
Comment thread
marcschier marked this conversation as resolved.
Outdated
{
payload = await EncodeAndWrapUadpAsync(uadp, context, cancellationToken)
payload = await EncodeAndWrapUadpAsync(networkMessage, context, cancellationToken)
Comment thread
marcschier marked this conversation as resolved.
Outdated
.ConfigureAwait(false);
Comment thread
marcschier marked this conversation as resolved.
}
else if (RequiresInboundSecurity ||
Expand Down Expand Up @@ -2685,7 +2702,7 @@ await transport.SendAsync(frame, topic, cancellationToken)
}

private async ValueTask<ReadOnlyMemory<byte>> EncodeAndWrapUadpAsync(
UadpNetworkMessage message,
PubSubNetworkMessage message,
PubSubNetworkMessageContext context,
CancellationToken cancellationToken)
{
Expand Down
Loading
Loading