Files
UA-.NETStandard/samples/Reference/ConsoleReferenceClient/HistorianClientSample.cs
T
Marc Schier 7411654c8d Complete OPC UA Part 11 Historical Access (#4390)
# Description

Completes the OPC UA Part 11 Historical Access work requested by #4387
and
the historian framework integration requested by #4400.

The conformance/evidence catalog covers all 37 released UACore 1.05
Historical Access profiles: 15 Server facets and 22 Client facets,
together
with the standard aggregate functions and their conformance units.
Server
facets are advertised only when the resolved providers expose their
required
interfaces and capabilities.

## Server and provider support

- Completes raw, modified, at-time, processed, annotation, structured,
and event history.
- Dispatches best-effort, bulk, and transactional updates with aligned
statuses, old values, diagnostics, and rollback state.
- Supports mixed history-update detail types and batched
structured/annotation operations.
- Uses composite identities and exclusive cursors for same-timestamp
structured, modified, and event values.
- Captures live data and events, installs historical configuration
objects, and emits typed audit payloads.
- Derives aggregate monitoring filters from per-node historian
capabilities and rejects unsupported processed aggregates before
provider dispatch.
- Supports tri-state `Historizing`: set, clear, or preserve
provider-owned state.

## Client support

- Exposes modified values together with `ModificationInfo`.
- Adds event-history reads and event insert/replace/update/delete
operations.
- Adds structured and annotation batching with strict response
validation.
- Shares continuation handling and release-on-abandon across read
families.
- Accepts inherited standard event fields rooted at event subtypes.

## Review hardening

- Registers DI-selected historians before node-manager reconciliation
and capability publication, while preserving provider ownership.
- Uses committed Raft read barriers for linearizable `TryGetAsync` and
`ScanAsync`.
- Keeps claimed continuation state immutable until a fresh successor is
saved, restoring the original after transient failures.
- Uses exact-incarnation compare-and-delete so delayed cleanup cannot
remove a restored same-ID continuation.
- Cleans up indeterminate initial and replacement saves, and propagates
shutdown cancellation through cleanup resolution.
- Preserves legacy continuation envelopes/cursors and distinct
annotation parent/property identities.
- Deletes every structured entry at a requested timestamp and uses the
complete modified-history ordering tuple.
- Shares native/fallback at-time and AnnotationCount calculations,
including reverse endpoints and cross-framework fractional intervals.
- Primes modified aggregate items before buffered live delivery and
deduplicates history/live overlap using provider-specific identities.
- Keeps committed Modify results successful while reporting priming
failures
  through the monitored item's notification queue.
- Keeps protected-notification state live-only for queues larger than
one.
Queue size one retains the newest notification without pinning errors or
transferring priority during resizing. Durable restore retains ordinary
definitions, last values/errors, and raw queues without recreating
priority.
- Isolates aggregate calculation, modification preparation, and
historical/live
handoff in `AggregationFilterHandler`, retaining existing
provider/factory
wiring and supported past-start behavior. The larger-queue protection
policy
  is unchanged pending the open review discussion.
- Uses one `IMonitoringFilter` owner for protocol filters or the
aggregation
handler, eliminating parallel effective-filter storage and retaining the
current definition during preparation. The original client filter stays
distinct.
- Integrates current master, retaining custom monitored-item creation
decisions and per-node history callbacks.

## Redundant-server alignment and samples

- Adds a protected shared historian with immutable segments, manifest
CAS publication, writer fencing, and restart recovery.
- Persists portable continuation state before exposing it to clients and
transfers ownership during mirrored-session takeover.
- Rejects eventual-consistency and active/active multi-writer historian
configurations rather than silently degrading.
- ReferenceServer and ConsoleReferenceClient demonstrate discovery,
paging, raw/modified/at-time/processed reads, annotation/structured
CRUD, event capture, and event CRUD.
- RedundantServer and RedundantClient demonstrate active/passive
history, continuation recovery after replica termination, and
post-promotion writes.
- Updates Historical Access, High Availability, profile, migration, and
sample documentation.

## Compatibility

- New 2.0 collection boundaries use `ArrayOf<T>` and public byte
payloads use `ByteString`.
- Removes the legacy public virtual
`SamplingGroupManager.CreateMonitoredItem` /
`ModifyMonitoredItem` entry points and protected creation factory, as
agreed
  in review. Migration guidance points to the monitored-item manager and
  construction-decision APIs.
- Retains established persisted formats and current write layouts.
Retired interim
notification-state formats (sample v2 / shared v4) are rejected, as
agreed in review.
- Keeps the historian integration compatible with .NET Framework and
NativeAOT.

## Validation

### Latest CI source repair

`73c619435` forwards the existing cancellation token at both
predefined-node
creation calls in the PR-added History event fixture. The merged
lifecycle
overload exposed these two CA2016 errors, which blocked History jobs,
the
solution build matrix, and CodeQL on GitHub and Azure. The exact
compiler
failure was reproduced locally; the repaired fixture passes 4/4
integration
cases on net10.0 and 4/4 on net48 with clean scoped diagnostics. No
assertions,
tests, coverage thresholds, or CI configuration were weakened.

The repaired History jobs on Linux/macOS and all four Azure solution
builds
now pass. The Windows all-TFM GitHub job instead lost its hosted
runner's
connection; no build log was recoverable, and GitHub recorded a
runner-loss
annotation rather than a compiler/test diagnostic. One unchanged retry
was
requested. Remaining checks are monitored with
`gh pr checks --watch --fail-fast --interval 30`; CI is not yet reported
as green.

### Latest base integration

`bd52e5fea` merges upstream `master` at `da7412c42`. The lifecycle
conflict
retains upstream's shared cancellation-aware registration helper and the
historian behavior. The hosted-historian fixture forwards its existing
token
to the new lifecycle overload. GDS project renames and all three AOT
test
hosts are retained, with their project references resolving after the
merge.

| Integrated scope | net10.0 | net48 |
| --- | --- | --- |
| Server lifecycle, fluent authoring/import, historian, monitored-item
and hosting regressions | 732 passed | 732 passed |
| NodeState lifecycle regressions | 216 passed | 216 passed |
| Automatically merged Raft client fixtures | 28 passed | 28 passed |

No selected tests failed or were skipped. The manually resolved files
pass
scoped formatting/analyzer checks. One CA1861 warning remains in the
incoming
`NodeSetImportIntegrationTests.cs`; that file is byte-identical to
upstream
master and was not rewritten or suppressed during integration.

### Latest feedback follow-up

`102954eaa` replaces split filter references with the shared
`IMonitoringFilter`
contract implemented by protocol filters and `AggregationFilterHandler`.
Targeted server regressions pass **103/103 on net10.0 and 103/103 on
net48**;
scoped formatting/analyzers are clean, and the updated XML-documentation
inventory has zero gaps across 188 non-generated PR C# files.

`93c5b5db8` renames `AggregationFilterHandler`, makes it the owner of
the
server-revised aggregate filter while preserving the original client
filter,
and removes the retired interim notification-state readers. Server
regressions
pass **101/101** and shared-store regressions **64/64**, on both net10.0
and
net48, with no failures or skips. Runtime-file formatting/analyzers are
clean.

`5ccf45630` completes the public/internal XML-documentation sweep. The
Roslyn
inventory finds zero undocumented eligible declarations across all 186
non-generated PR C# files; the 110-file documentation slice changes only
XML
comments and spacing. `f17765475` applies the three requested Markdown
edits.

### Earlier feedback follow-ups

`a53950737` removes single-slot notification pinning and extracts
aggregate
implementation from MonitoredItem. Targeted sampling, aggregate
modification,
history/live handoff, queue/lifecycle, restore, and serialization cases
pass
**101/101 on net10.0 and 101/101 on net48**, without failures or skips.
Scoped formatting/analyzers are clean; the focused review found no
significant
issues. The final formatted monitored-item fixtures also pass 45/45 on
net48.

The preceding `b21555568` removes the legacy sampling API bridge and
durable
notification-priority state. Its targeted Server cases passed 97/97 and
shared
subscription-store cases passed 64/64 on each framework. The broader
evidence
below remains pinned to its original tested commit, not attributed to
these
follow-ups.

### Earlier integrated-implementation validation

Validated implementation commit: **`3efae597f`**. The following commit,
`f6dbefb1c`, only adds two explanatory comment lines for the final
standards
review; it changes no executable code.

| Scope | net10.0 | net48 |
| --- | --- | --- |
| Server | 5,187 passed; 56 skipped | Affected surface: 2,160 passed; 1
reproduced baseline failure; 56 skipped |
| Client | 2,202 passed; 2 skipped | 2,204 passed; 5 skipped |
| History integration | 519 passed; 21 skipped | 519 passed; 21 skipped
|
| Redundancy server | 641 passed | 641 passed |
| Redundancy client | 124 passed | 124 passed |
| Core shared-store regression suite | 23 passed | 23 passed |

All six affected samples passed non-incremental Release/net10.0 builds
with
zero warnings and errors: ConsoleReferenceClient,
ConsoleReferenceServer,
RedundantServer, RedundantClient, MinimalBoilerServer, and
PumpDeviceIntegrationServer.

The published self-contained win-x64 NativeAOT executable is 116,924,928
bytes and has no CLR header or managed runtime dependency. Direct
execution
passed **HistoryAotTests 3/3** and **RaftAotTests 2/2**.

Final standards and specification reviews found no confirmed outstanding
high/medium functional gaps or hard standards violations. The remaining
constructor-purpose clarification is addressed in `f6dbefb1c`.

### Reproduced baseline exceptions

The net48 hosted-startup timeout also reproduces in a pristine worktree
at
the previous PR head, `24cf1416d`, in

`RegisterPostStartRegistriesWiresHistorianAndAliasStoresOnPlainServerAsync`,
with the same subsequent certificate-leak assertion. The corresponding
renamed case fails before startup in the reviewed branch as well. This
is
reported separately from the net48 affected-surface results, not counted
as
a passing full Server suite. The net48 matrix selected the historian,
aggregate, monitored-item, fluent, and continuation surfaces and
excluded
the Hosting namespace.

`DurableDataValueQueueVerifyReferenceBatchingAsync` failed at item 1000
in
the net48 affected-surface run and passed its single isolated rerun. The
same
failure at the same assertion reproduces on pristine `849e16acc`; the
test,
sample queue, queue factory, and batch persistor are unchanged. The
original
failure remains in the table rather than being erased by the rerun.

### CI source repairs

- `4be7eb8f7` keeps diagnostics/configuration managers out of
distributed
address-space replication. Those managers inherit the source interface,
but their live server state and redundancy metadata must remain
replica-local.
  The ownership regression passes on net10.0 and net48, and all nine
  redundancy sample CI tests pass locally.
- `d17f467b3` completes the sample startup repair: core managers are
also
replica-local; the startup banner no longer advertises readiness before
initialization; the producer starts after distributed services; and
native
Raft startup no longer forces disruptive extra elections. The full
sample
suite passed 9/9, with initialized `RedundancySupport=Hot`, and focused
  Raft/address-space regressions passed 8/8 and 3/3.
- `81b0b72be` moves the unchanged distributed-historian NativeAOT
scenario
into a smaller, always-run companion executable to bound Apple's linker
input. All 137 AOT tests remain present; the moved test and helpers are
  unchanged. GitHub and Azure publish/run both native executables, with
separate result directories where they share a job. The companion passed
actual Windows NativeAOT execution. Linux and Intel macOS CI passed both
  executables, but this split alone did not fix ARM64 macOS linking.
The remaining historian graph is rooted by ReferenceServer startup, not
just the relocated test. No test exclusion or speculative linker setting
  was added to conceal it.
- `8a5842913` isolates the independent MCP test graph into an always-run
native host. The existing MCP source and assembly leak hooks are linked
  unchanged, and evaluated compiler inputs prove exactly one owner for
  every test source: 133 main + 3 MCP + 1 historian = 137 declarations.
All three hosts run on every AOT platform. The main Windows native
object
is 9.35% smaller; both changed publishes had zero diagnostics. MCP
passed
  3/3; the identical main executable passed 133/133 on one unfiltered
followup after two initial endpoint-discovery connection failures, which
  remain recorded in the local evidence.
- The previous Intel macOS job successfully published and executed its
native
tests, then failed on artifact-upload DNS (`ENOTFOUND`). No upload gate,
  shared environment setting, or cache configuration was disabled.

The first repair round also encountered the unrelated
`MetricsAreEmittedForChannelLifetimeAsync` timing failure on macOS Core.
Neither that test nor channel-manager code is changed by this PR, and
its
assertions remain intact. It passed unchanged on the final repair head.

On `d17f467b3`, the Linux and macOS redundancy sample jobs pass; the
remaining
failure was ARM64 Apple's `too many large addends` assertion. The
verified
upstream relocation-anchor fix (dotnet/runtime#124721, backport
dotnet/runtime#132171) is not present in the shipping ILCompiler
10.0.11.
An experimental multi-module probe failed and was rejected, not added to
CI.
On `8a5842913`,

[`aot-macos-latest`](https://github.com/OPCFoundation/UA-.NETStandard/actions/runs/34102158680/job/101678934057)
successfully publishes and executes all three native hosts on ARM64:
136 main runtime cases, 1 historian case, and 3 MCP cases, with no
skips.
The previous Apple linker blocker is resolved. Linux NativeAOT also
passes.
No assertion, test, shared environment setting, cache, or required gate
was
weakened.

The same CI run reports a separate macOS Sessions teardown failure: all
662 executed tests pass, but the global leak assertion detects one
undisposed certificate. The MCP partition changes no runtime, Sessions,
or
test-fixture code. A single unchanged diagnostic rerun is queued once
GitHub
permits rerunning that job; the failure is not suppressed or counted as
green.
Other CI checks are still pending.

## Related issues

- Fixes #4387
- Fixes #4400

## Checklist

- [ ] I have signed the CLA and read the contributing guide.
- [x] I have added regression tests and directly related documentation.
- [x] I have retained working reference and redundant-server sample
wiring.
- [x] Final validation and review evidence is complete, including the
baseline exceptions above.
- [ ] All required CI checks pass; the fail-fast rerun for `73c619435`
is pending.

---------

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
Copilot-Session: 9dbd9527-d818-47a0-984c-fbb1472d3d6b
2026-09-08 11:32:23 +02:00

842 lines
34 KiB
C#

/* ========================================================================
* 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 Opc.Ua;
using Opc.Ua.Client;
using Opc.Ua.Client.Historian;
namespace Quickstarts.ConsoleReferenceClient
{
/// <summary>
/// End-to-end Historical Access workflow for the ReferenceServer.
/// </summary>
public static class HistorianClientSample
{
/// <summary>
/// Discovers historian capabilities and exercises data, annotations,
/// aggregates, and event history through <see cref="HistoryClient"/>.
/// </summary>
/// <exception cref="ServiceResultException"></exception>
public static async Task RunAsync(
ISession session,
CancellationToken cancellationToken = default)
{
int namespaceIndex = session.NamespaceUris.GetIndex(ReferenceServerNamespaceUri);
if (namespaceIndex < 0)
{
throw new ServiceResultException(
StatusCodes.BadNodeIdUnknown,
$"The server does not expose {ReferenceServerNamespaceUri}.");
}
var dataNodeId = new NodeId(DataNodeName, (ushort)namespaceIndex);
var structuredNodeId = new NodeId(
StructuredNodeName,
(ushort)namespaceIndex);
var eventNotifierId = new NodeId(EventNotifierName, (ushort)namespaceIndex);
HistoryClient client = session.Historian(new HistoryClientOptions
{
MaxPagesPerRead = 10_000,
MaxReadDuration = TimeSpan.FromMinutes(1)
});
HistoryServerCapabilitiesInfo capabilities =
await client.GetServerCapabilitiesAsync(cancellationToken).ConfigureAwait(false);
HistoricalConformanceInfo conformance =
await client.GetConformanceInfoAsync(
cancellationToken).ConfigureAwait(false);
HistoricalDataConfigurationInfo configuration =
await client.GetConfigurationAsync(
dataNodeId,
cancellationToken).ConfigureAwait(false);
HistoricalDataConfigurationInfo structuredConfiguration =
await client.GetConfigurationAsync(
structuredNodeId,
cancellationToken).ConfigureAwait(false);
HistoricalEventConfigurationInfo eventConfiguration =
await client.GetEventConfigurationAsync(
eventNotifierId,
cancellationToken).ConfigureAwait(false);
ValidateDiscovery(
capabilities,
conformance,
configuration,
structuredConfiguration,
eventConfiguration);
Console.WriteLine(
"Historian capabilities: data={0}, events={1}, data writes={2}, event writes={3}.",
capabilities.AccessHistoryData,
capabilities.AccessHistoryEvents,
capabilities.InsertData && capabilities.ReplaceData && capabilities.UpdateData,
capabilities.InsertEvent &&
capabilities.ReplaceEvent &&
capabilities.UpdateEvent &&
capabilities.DeleteEvent);
Console.WriteLine(
"Historical claims: server profiles={0}, conformance units={1}.",
conformance.ServerProfiles.Count,
conformance.ConformanceUnits.Count);
Console.WriteLine(
"Data configuration: installed={0}, server timestamps={1}, archive start={2:O}.",
configuration.HasConfiguration,
configuration.ServerTimestampSupported,
configuration.StartOfArchive);
Console.WriteLine(
"Event configuration: installed={0}, event types={1}, archive start={2:O}.",
eventConfiguration.HasConfiguration,
eventConfiguration.EventTypes.Count,
eventConfiguration.StartOfArchive);
DateTime endTime = DateTime.UtcNow;
DateTime startTime = endTime.AddHours(-4);
int rawCount = await CountAsync(
client.ReadRawAsync(
dataNodeId,
startTime,
endTime,
maxValuesPerNode: 50,
returnBounds: true,
timestampsToReturn: TimestampsToReturn.Both,
cancellationToken: cancellationToken),
cancellationToken).ConfigureAwait(false);
int atTimeCount = await CountAsync(
client.ReadAtTimeAsync(
dataNodeId,
[startTime.AddMinutes(30), startTime.AddHours(2)],
useSimpleBounds: false,
TimestampsToReturn.Both,
cancellationToken: cancellationToken),
cancellationToken).ConfigureAwait(false);
Console.WriteLine("Raw history values={0}; at-time values={1}.", rawCount, atTimeCount);
await ReadAggregateFamiliesAsync(
client,
dataNodeId,
startTime,
endTime,
cancellationToken).ConfigureAwait(false);
await DemonstrateDataUpdatesAsync(
client,
dataNodeId,
cancellationToken).ConfigureAwait(false);
await DemonstrateStructuredDataAsync(
client,
structuredNodeId,
cancellationToken).ConfigureAwait(false);
await DemonstrateEventsAsync(
client,
eventNotifierId,
cancellationToken).ConfigureAwait(false);
await DemonstrateAnnotationsAsync(
client,
dataNodeId,
cancellationToken).ConfigureAwait(false);
}
private static async ValueTask ReadAggregateFamiliesAsync(
HistoryClient client,
NodeId nodeId,
DateTime startTime,
DateTime endTime,
CancellationToken cancellationToken)
{
(string Name, NodeId Id)[] aggregateFamilies =
[
("interpolative", ObjectIds.AggregateFunction_Interpolative),
("average/total", ObjectIds.AggregateFunction_Average),
("minimum/maximum", ObjectIds.AggregateFunction_Minimum),
("count/state", ObjectIds.AggregateFunction_Count),
("start/end/delta", ObjectIds.AggregateFunction_Start),
("quality", ObjectIds.AggregateFunction_DurationGood),
("statistical", ObjectIds.AggregateFunction_StandardDeviationPopulation)
];
foreach ((string name, NodeId aggregateId) in aggregateFamilies)
{
int count = await CountAsync(
client.ReadProcessedAsync(
nodeId,
aggregateId,
startTime,
endTime,
processingInterval: TimeSpan.FromMinutes(30).TotalMilliseconds,
cancellationToken: cancellationToken),
cancellationToken).ConfigureAwait(false);
Console.WriteLine("Aggregate family {0}: {1} processed values.", name, count);
}
}
private static async ValueTask DemonstrateDataUpdatesAsync(
HistoryClient client,
NodeId nodeId,
CancellationToken cancellationToken)
{
DateTime timestamp = DateTime.UtcNow.AddYears(-20);
DataValue insert = CreateHistoricalValue(4387, timestamp);
AssertGood(
await client.InsertAsync(
nodeId,
[insert],
cancellationToken).ConfigureAwait(false),
"insert raw value");
DataValue replacement = CreateHistoricalValue(4388, timestamp);
AssertGood(
await client.ReplaceAsync(
nodeId,
[replacement],
cancellationToken).ConfigureAwait(false),
"replace raw value");
DataValue update = CreateHistoricalValue(4389, timestamp);
AssertGood(
await client.UpdateAsync(
nodeId,
[update],
cancellationToken).ConfigureAwait(false),
"update raw value");
int modifiedCount = await CountAsync(
client.ReadModifiedAsync(
nodeId,
timestamp.AddMilliseconds(-1),
timestamp.AddMilliseconds(1),
maxValuesPerNode: 1,
cancellationToken: cancellationToken),
cancellationToken).ConfigureAwait(false);
Console.WriteLine("Modified-history entries={0}.", modifiedCount);
AssertGood(
await client.DeleteAtTimeAsync(
nodeId,
[timestamp],
cancellationToken).ConfigureAwait(false),
"delete raw value");
DateTime rangeTimestamp = timestamp.AddSeconds(10);
AssertGood(
await client.InsertAsync(
nodeId,
[CreateHistoricalValue(4390, rangeTimestamp)],
cancellationToken).ConfigureAwait(false),
"insert value for range delete");
AssertGood(
await client.DeleteRawAsync(
nodeId,
rangeTimestamp.AddMilliseconds(-1),
rangeTimestamp.AddMilliseconds(1),
cancellationToken: cancellationToken)
.ConfigureAwait(false),
"delete raw range");
}
private static async ValueTask DemonstrateAnnotationsAsync(
HistoryClient client,
NodeId nodeId,
CancellationToken cancellationToken)
{
DateTime annotationTime = DateTime.UtcNow.AddYears(-20).AddMinutes(1);
ArrayOf<Annotation> annotations =
[
new Annotation
{
AnnotationTime = annotationTime,
Message = "ReferenceClient historian sample",
UserName = "ReferenceClient"
},
new Annotation
{
AnnotationTime = annotationTime.AddSeconds(1),
Message = "ReferenceClient historian sample batch",
UserName = "ReferenceClient"
}
];
AssertGood(
await client.WriteAnnotationsAsync(
nodeId,
annotations,
cancellationToken: cancellationToken).ConfigureAwait(false),
"insert annotation");
ArrayOf<Annotation> replacements =
[
new Annotation
{
AnnotationTime = annotations[0].AnnotationTime,
Message = "ReferenceClient replaced annotation",
UserName = "ReferenceClient"
},
new Annotation
{
AnnotationTime = annotations[1].AnnotationTime,
Message = "ReferenceClient replaced annotation batch",
UserName = "ReferenceClient"
}
];
AssertGood(
await client.WriteAnnotationsAsync(
nodeId,
replacements,
PerformUpdateType.Replace,
cancellationToken).ConfigureAwait(false),
"replace annotations");
var updated = new Annotation
{
AnnotationTime = replacements[1].AnnotationTime,
Message = "ReferenceClient updated annotation",
UserName = "ReferenceClient"
};
AssertGood(
await client.WriteAnnotationsAsync(
nodeId,
[updated],
PerformUpdateType.Update,
cancellationToken).ConfigureAwait(false),
"update annotation");
int annotationCount = await CountAsync(
client.ReadAnnotationsAsync(
nodeId,
annotationTime.AddMilliseconds(-1),
updated.AnnotationTime.ToDateTime().AddMilliseconds(1),
cancellationToken: cancellationToken),
cancellationToken).ConfigureAwait(false);
Console.WriteLine("Annotations read={0}.", annotationCount);
AssertGood(
await client.WriteAnnotationsAsync(
nodeId,
[replacements[0], updated],
PerformUpdateType.Remove,
cancellationToken).ConfigureAwait(false),
"remove annotation");
}
private static async ValueTask DemonstrateStructuredDataAsync(
HistoryClient client,
NodeId nodeId,
CancellationToken cancellationToken)
{
DateTime timestamp =
DateTime.UtcNow.AddYears(-20).AddMinutes(3);
DataValue pressure = CreateStructuredValue(
nodeId.NamespaceIndex,
"Pressure",
timestamp,
42.5);
DataValue temperature = CreateStructuredValue(
nodeId.NamespaceIndex,
"Temperature",
timestamp,
21.25);
AssertGood(
await client.UpdateStructureDataAsync(
nodeId,
PerformUpdateType.Insert,
[pressure, temperature],
cancellationToken).ConfigureAwait(false),
"insert structured values");
// IDE0001 is suppressed: "KeyValuePair" is ambiguous between Opc.Ua.KeyValuePair
// and System.Collections.Generic.KeyValuePair in this file (CS0104), so the
// qualification cannot be simplified away.
#pragma warning disable IDE0001
var readBack = new List<Opc.Ua.KeyValuePair>();
#pragma warning restore IDE0001
await foreach (DataValue value in client.ReadRawAsync(
nodeId,
timestamp.AddMilliseconds(-1),
timestamp.AddMilliseconds(1),
maxValuesPerNode: 1,
nodeOptions: new HistoryReadNodeOptions
{
DataEncoding = new QualifiedName(
BrowseNames.DefaultBinary)
},
cancellationToken: cancellationToken).ConfigureAwait(false))
{
readBack.Add(ReadStructuredValue(value));
}
if (readBack.Count != 2)
{
throw new ServiceResultException(
StatusCodes.BadNoData,
"The structured history read did not return both same-time keys.");
}
DataValue replacedPressure = CreateStructuredValue(
nodeId.NamespaceIndex,
"Pressure",
timestamp,
43.0);
AssertGood(
await client.UpdateStructureDataAsync(
nodeId,
PerformUpdateType.Replace,
[replacedPressure],
cancellationToken).ConfigureAwait(false),
"replace structured value");
DataValue updatedTemperature = CreateStructuredValue(
nodeId.NamespaceIndex,
"Temperature",
timestamp,
22.0);
AssertGood(
await client.UpdateStructureDataAsync(
nodeId,
PerformUpdateType.Update,
[updatedTemperature],
cancellationToken).ConfigureAwait(false),
"update structured value");
int modifiedCount = await CountAsync(
client.ReadModifiedAsync(
nodeId,
timestamp.AddMilliseconds(-1),
timestamp.AddMilliseconds(1),
maxValuesPerNode: 1,
cancellationToken: cancellationToken),
cancellationToken).ConfigureAwait(false);
int atTimeCount = await CountAsync(
client.ReadAtTimeAsync(
nodeId,
[timestamp],
nodeOptions: new HistoryReadNodeOptions
{
DataEncoding = new QualifiedName(
BrowseNames.DefaultBinary)
},
cancellationToken: cancellationToken),
cancellationToken).ConfigureAwait(false);
if (modifiedCount < 2 || atTimeCount != 1)
{
throw new ServiceResultException(
StatusCodes.BadNoData,
"Structured modified/at-time history was incomplete.");
}
AssertGood(
await client.UpdateStructureDataAsync(
nodeId,
PerformUpdateType.Remove,
[replacedPressure, updatedTemperature],
cancellationToken).ConfigureAwait(false),
"remove structured values");
Console.WriteLine(
"Structured history CRUD completed; same-time keys={0}, modified={1}.",
readBack.Count,
modifiedCount);
}
private static async ValueTask DemonstrateEventsAsync(
HistoryClient client,
NodeId notifierId,
CancellationToken cancellationToken)
{
EventFilter filter = CreateEventFilter();
await DemonstrateCapturedServerEventAsync(
client,
notifierId,
filter,
cancellationToken).ConfigureAwait(false);
DateTime eventTime = DateTime.UtcNow.AddYears(-20).AddMinutes(2);
HistoryEventFieldList inserted = CreateEvent(
ByteString.Empty,
notifierId,
eventTime,
"ReferenceClient inserted historical event");
AssertGood(
await client.InsertEventsAsync(
notifierId,
filter,
[inserted],
cancellationToken).ConfigureAwait(false),
"insert event");
ByteString generatedEventId = ByteString.Empty;
await foreach (HistoryEventFieldList historicalEvent in client.ReadEventsAsync(
notifierId,
eventTime.AddMilliseconds(-1),
eventTime.AddMilliseconds(1),
filter,
maxValuesPerNode: 1,
cancellationToken: cancellationToken).ConfigureAwait(false))
{
if (historicalEvent.EventFields.Count > 0 &&
historicalEvent.EventFields[0].TryGetValue(out ByteString eventId))
{
generatedEventId = eventId;
break;
}
}
if (generatedEventId.IsEmpty)
{
throw new ServiceResultException(
StatusCodes.BadNoData,
"The server did not return the generated EventId.");
}
AssertGood(
await client.ReplaceEventsAsync(
notifierId,
filter,
[
CreateEvent(
generatedEventId,
notifierId,
eventTime,
"ReferenceClient replaced historical event")
],
cancellationToken).ConfigureAwait(false),
"replace event");
AssertGood(
await client.UpdateEventsAsync(
notifierId,
filter,
[
CreateEvent(
generatedEventId,
notifierId,
eventTime,
"ReferenceClient updated historical event")
],
cancellationToken).ConfigureAwait(false),
"update event");
AssertGood(
await client.DeleteEventsAsync(
notifierId,
[generatedEventId],
cancellationToken).ConfigureAwait(false),
"delete event");
Console.WriteLine("Historical event CRUD completed; generated EventId={0}.", generatedEventId);
}
private static async ValueTask DemonstrateCapturedServerEventAsync(
HistoryClient client,
NodeId notifierId,
EventFilter filter,
CancellationToken cancellationToken)
{
DateTime startTime = DateTime.UtcNow.AddSeconds(-1);
var triggerNodeId = new NodeId(
EventTriggerNodeName,
notifierId.NamespaceIndex);
WriteResponse response = await client.Session.WriteAsync(
null,
[
new WriteValue
{
NodeId = triggerNodeId,
AttributeId = Attributes.Value,
Value = new DataValue(Variant.From(Environment.TickCount))
}
],
cancellationToken).ConfigureAwait(false);
if (response.Results.Count != 1 ||
StatusCode.IsBad(response.Results[0]))
{
StatusCode status = response.Results.Count == 1
? response.Results[0]
: StatusCodes.BadUnexpectedError;
throw new ServiceResultException(
status,
"Writing the ReferenceServer event trigger failed.");
}
bool found = false;
for (int attempt = 0; attempt < 20 && !found; attempt++)
{
await foreach (HistoryEventFieldList historicalEvent in
client.ReadEventsAsync(
notifierId,
startTime,
DateTime.UtcNow.AddSeconds(1),
filter,
cancellationToken: cancellationToken)
.ConfigureAwait(false))
{
if (historicalEvent.EventFields.Count > 4 &&
historicalEvent.EventFields[4].TryGetValue(
out LocalizedText message) &&
message.Text?.Contains(
"Trigger event",
StringComparison.Ordinal) == true)
{
found = true;
break;
}
}
if (!found)
{
await Task.Delay(
TimeSpan.FromMilliseconds(50),
cancellationToken).ConfigureAwait(false);
}
}
if (!found)
{
throw new ServiceResultException(
StatusCodes.BadNoData,
"The server-reported event was not captured in event history.");
}
Console.WriteLine("Server-reported event was captured automatically.");
}
private static EventFilter CreateEventFilter()
{
var filter = new EventFilter();
filter.AddSelectClause(
ObjectTypeIds.BaseEventType,
BrowseNames.EventId,
Attributes.Value);
filter.AddSelectClause(
ObjectTypeIds.BaseEventType,
BrowseNames.EventType,
Attributes.Value);
filter.AddSelectClause(
ObjectTypeIds.BaseEventType,
BrowseNames.SourceNode,
Attributes.Value);
filter.AddSelectClause(
ObjectTypeIds.BaseEventType,
BrowseNames.Time,
Attributes.Value);
filter.AddSelectClause(
ObjectTypeIds.BaseEventType,
BrowseNames.Message,
Attributes.Value);
return filter;
}
private static void ValidateDiscovery(
HistoryServerCapabilitiesInfo capabilities,
HistoricalConformanceInfo conformance,
HistoricalDataConfigurationInfo dataConfiguration,
HistoricalDataConfigurationInfo structuredConfiguration,
HistoricalEventConfigurationInfo eventConfiguration)
{
if (!capabilities.AccessHistoryData ||
!capabilities.AccessHistoryEvents ||
!capabilities.InsertData ||
!capabilities.ReplaceData ||
!capabilities.UpdateData ||
!capabilities.DeleteRaw ||
!capabilities.DeleteAtTime ||
!capabilities.InsertAnnotation ||
!capabilities.InsertEvent ||
!capabilities.ReplaceEvent ||
!capabilities.UpdateEvent ||
!capabilities.DeleteEvent ||
!capabilities.ServerTimestampSupported)
{
throw new ServiceResultException(
StatusCodes.BadConfigurationError,
"The ReferenceServer does not expose the complete Historical Access capability set.");
}
if (!dataConfiguration.HasConfiguration ||
!structuredConfiguration.HasConfiguration ||
!eventConfiguration.HasConfiguration)
{
throw new ServiceResultException(
StatusCodes.BadConfigurationError,
"The ReferenceServer is missing a historical configuration object.");
}
ArrayOf<HistoricalAccessProfileDescriptor> serverProfiles =
HistoricalAccessProfileCatalog.GetProfiles(
HistoricalAccessProfileSide.Server);
if (conformance.ServerProfiles.Count !=
serverProfiles.Count)
{
throw new ServiceResultException(
StatusCodes.BadConfigurationError,
"The ReferenceServer does not advertise all 15 Historical Access Server facets.");
}
for (int i = 0; i < serverProfiles.Count; i++)
{
HistoricalAccessProfileDescriptor profile =
serverProfiles[i];
if (!conformance.ServerProfiles.Contains(
profile.ProfileUri))
{
throw new ServiceResultException(
StatusCodes.BadConfigurationError,
$"The ReferenceServer is missing profile {profile.ProfileUri}.");
}
foreach (string unitName in
profile.MandatoryConformanceUnits)
{
if (!conformance.ConformanceUnits.Contains(
new QualifiedName(unitName)))
{
throw new ServiceResultException(
StatusCodes.BadConfigurationError,
$"The ReferenceServer is missing conformance unit {unitName}.");
}
}
}
foreach (HistoricalAggregateFunctionDescriptor aggregate in
HistoricalAggregateFunctionCatalog.AllFunctions)
{
if (!conformance.ConformanceUnits.Contains(
new QualifiedName(
aggregate.ServerConformanceUnit)))
{
throw new ServiceResultException(
StatusCodes.BadConfigurationError,
$"The ReferenceServer is missing aggregate unit {aggregate.ServerConformanceUnit}.");
}
}
}
private static HistoryEventFieldList CreateEvent(
ByteString eventId,
NodeId sourceNode,
DateTime eventTime,
string message)
{
return new HistoryEventFieldList
{
EventFields =
[
new Variant(eventId),
new Variant(ObjectTypeIds.BaseEventType),
new Variant(sourceNode),
new Variant((DateTimeUtc)eventTime),
new Variant(new LocalizedText(message))
]
};
}
private static DataValue CreateHistoricalValue(int value, DateTime timestamp)
{
return new DataValue(
Variant.From(value),
StatusCodes.Good,
sourceTimestamp: timestamp,
serverTimestamp: timestamp);
}
private static DataValue CreateStructuredValue(
ushort namespaceIndex,
string key,
DateTime timestamp,
double value)
{
// IDE0001 is suppressed: "KeyValuePair" is ambiguous between Opc.Ua.KeyValuePair
// and System.Collections.Generic.KeyValuePair in this file (CS0104), so the
// qualification cannot be simplified away.
#pragma warning disable IDE0001
var pair = new Opc.Ua.KeyValuePair
{
Key = new QualifiedName(key, namespaceIndex),
Value = Variant.From(value)
};
#pragma warning restore IDE0001
return new DataValue(
new Variant(new ExtensionObject(pair)),
StatusCodes.Good,
timestamp,
timestamp);
}
// IDE0001 is suppressed: "KeyValuePair" is ambiguous between Opc.Ua.KeyValuePair
// and System.Collections.Generic.KeyValuePair in this file (CS0104), so the
// qualification cannot be simplified away.
#pragma warning disable IDE0001
private static Opc.Ua.KeyValuePair ReadStructuredValue(
DataValue value)
{
if (value.WrappedValue.TryGetValue(
out ExtensionObject extension) &&
extension.TryGetValue(
out Opc.Ua.KeyValuePair? pair) &&
pair != null)
{
return pair;
}
#pragma warning restore IDE0001
throw new ServiceResultException(
StatusCodes.BadDecodingError,
"Structured history did not contain a KeyValuePair.");
}
private static async ValueTask<int> CountAsync<T>(
IAsyncEnumerable<T> values,
CancellationToken cancellationToken)
{
int count = 0;
await foreach (T _ in values.WithCancellation(cancellationToken).ConfigureAwait(false))
{
count++;
}
return count;
}
private static void AssertGood(
ArrayOf<StatusCode> statuses,
string operation)
{
for (int i = 0; i < statuses.Count; i++)
{
if (StatusCode.IsBad(statuses[i]))
{
throw new ServiceResultException(
statuses[i],
$"Historical Access sample failed to {operation}.");
}
}
}
private static void AssertGood(
StatusCode status,
string operation)
{
if (StatusCode.IsBad(status))
{
throw new ServiceResultException(
status,
$"Historical Access sample failed to {operation}.");
}
}
private const string ReferenceServerNamespaceUri =
"http://opcfoundation.org/Quickstarts/ReferenceServer";
private const string DataNodeName = "Scalar_Static_Int32";
private const string StructuredNodeName =
"Historical_KeyValuePairs";
private const string EventNotifierName = "CTT";
private const string EventTriggerNodeName = "NodeIds_Events_TriggerNode01";
}
}