M2: implement SendEventAsync — event-send rides WCF AddS2, not the storage pipe
Roadmap Milestone 2 (event sending). Capture disproved the assumption that event delivery uses the non-WCF storage-engine pipe (which would block it like revision writes): a native AddStreamedValue(HistorianEvent) leaves over WCF as AddS2 (IHistoryServiceContract2.AddStreamValues2). CM_EVENT is a built-in registered tag, so the 129 TagNotFoundInCache gate that blocks AddS2 for user tags does not apply. - R2.1: NativeTraceHarness "event-send" scenario + Capture-EventSend.ps1; two captures diffed to separate constant framing from value-dependent fields. - R2.2: HistorianEventWriteProtocol serializes the AddS2 pBuf (storage sample buffer wrapping the event VTQ) — golden-byte tested. Decoded "OS" sig + length fields + CM_EVENT tag id + EventTime/ReceivedTime FILETIMEs + Opc 192 + 0x118D descriptor + event Id + Namespace + EventType + version 5 + typed property bag. - R2.3/R2.4: HistorianWcfEventOrchestrator.SendEventAsync (Open2 event-mode 0x501 -> reuse CM_EVENT RTag2/EnsT2 -> AddStreamValues2) + HistorianClient.SendEventAsync. - R2.5: gated live test; server accepts the AddS2 (success, empty error buffer). Server requires delivered byte[].Length == declared packet length (uint32@0x04); the native relies on the MDAS encoder adding a pad byte, so the SDK emits an explicit trailing 0x00 (else AddS2 rejects with "CValuStream buffer size vs packet length mismatch"). Original events only (RevisionVersion=0) with string properties; other property types + revision/update/delete throw ProtocolEvidenceMissingException. Caveat (documented): accepted events are not persisted on the local dev box; the native client behaves identically (event ingestion pipeline inactive) — not an SDK gap. 212 unit tests pass; 16/16 event tests pass live. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01B6mcaT2PjRFKcogzp9UkfC
This commit is contained in:
Joseph Doherty
@@ -93,6 +93,21 @@ public sealed class HistorianClient : IAsyncDisposable
|
||||
return _protocol.ReadEventsAsync(startUtc, endUtc, cancellationToken);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Sends a single <see cref="HistorianEvent"/> to the Historian's built-in CM_EVENT tag
|
||||
/// over the WCF event pipeline (Open2 event mode → CM_EVENT registration → AddS2). The
|
||||
/// event is appended to the historian's event history and is readable back via
|
||||
/// <see cref="ReadEventsAsync"/> / the <c>v_AlarmEventHistory2</c> view. Only original
|
||||
/// events (<see cref="HistorianEvent.RevisionVersion"/> = 0) with string-valued properties
|
||||
/// are supported; other property value types and revision/update/delete events throw
|
||||
/// <see cref="ProtocolEvidenceMissingException"/> until their wire encoding is captured.
|
||||
/// </summary>
|
||||
public Task<bool> SendEventAsync(HistorianEvent historianEvent, CancellationToken cancellationToken = default)
|
||||
{
|
||||
ArgumentNullException.ThrowIfNull(historianEvent);
|
||||
return new HistorianWcfEventOrchestrator(_options).SendEventAsync(historianEvent, cancellationToken);
|
||||
}
|
||||
|
||||
public IAsyncEnumerable<string> BrowseTagNamesAsync(string filter = "*", CancellationToken cancellationToken = default)
|
||||
{
|
||||
ArgumentException.ThrowIfNullOrWhiteSpace(filter);
|
||||
|
||||
@@ -0,0 +1,195 @@
|
||||
using System.Buffers.Binary;
|
||||
using System.Text;
|
||||
using AVEVA.Historian.Client.Models;
|
||||
|
||||
namespace AVEVA.Historian.Client.Wcf;
|
||||
|
||||
/// <remarks>
|
||||
/// Serializer for the <c>AddS2</c> (<c>IHistoryServiceContract2.AddStreamValues2</c>)
|
||||
/// <c>pBuf</c> that carries a single <see cref="HistorianEvent"/> to the built-in
|
||||
/// <c>CM_EVENT</c> tag. This is the inverse of <see cref="HistorianEventRowProtocol"/> and the
|
||||
/// managed equivalent of native <c>HistorianEvent.PackToVtq</c> →
|
||||
/// <c>CCommonArchestraEventValue::PackToVtq</c>.
|
||||
///
|
||||
/// Wire shape decoded byte-for-byte from two captured native event sends
|
||||
/// (instrument-wcf-writemessage; <c>User.Write</c> and <c>Alarm.Set</c>, diffed to separate
|
||||
/// constant framing from value-dependent fields):
|
||||
///
|
||||
/// <code>
|
||||
/// pBuf (storage sample buffer):
|
||||
/// 0x00 UInt16 0x534F // "OS" signature
|
||||
/// 0x02 UInt16 sampleCount = 1
|
||||
/// 0x04 UInt32 valueBlob.Length + 11
|
||||
/// 0x08 UInt16 valueBlob.Length + 1
|
||||
/// 0x0A valueBlob:
|
||||
/// +0x00 GUID CM_EVENT tag id (353b8145-5df0-4d46-a253-871aef49b321)
|
||||
/// +0x10 Int64 EventTime FILETIME (UTC, ms-truncated — the VTQ timestamp)
|
||||
/// +0x18 UInt16 OpcQuality = 192
|
||||
/// +0x1A UInt16 192
|
||||
/// +0x1C UInt16 0x118D // opaque CCommonArchestraEventValue descriptor (constant)
|
||||
/// +0x1E GUID event Id
|
||||
/// +0x2E Int64 ReceivedTime FILETIME (UTC, ms — unique/monotonic on the native path)
|
||||
/// +0x36 compact-ASCII string // Namespace
|
||||
/// +.... compact-ASCII string // EventType (e.g. "Alarm.Set")
|
||||
/// +.... UInt16 eventStructVersion = 5
|
||||
/// +.... UInt16 propertyCount
|
||||
/// +.... propertyCount × Property {
|
||||
/// compact-ASCII string // property name
|
||||
/// UInt8 typeMarker, UInt8 length, UInt8 status(=0), length×value bytes
|
||||
/// 0x43 → UTF-16 string: UInt16 charCount + charCount×UInt16
|
||||
/// }
|
||||
/// </code>
|
||||
///
|
||||
/// Compact ASCII string: <c>0x09 LEN 0x00 LEN×ASCII bytes</c> (same as
|
||||
/// <see cref="HistorianEventRowProtocol"/> and CTagMetadata strings).
|
||||
///
|
||||
/// Only string-valued properties are emitted here — the only property value type observed on
|
||||
/// the event-send wire. Non-string property values throw
|
||||
/// <see cref="ProtocolEvidenceMissingException"/> until captured (the read parser decodes more
|
||||
/// types, but the write framing for them is unverified). Likewise revision sends
|
||||
/// (Update/Delete/RevisionVersion ≠ 0) are not yet captured and are rejected by the caller.
|
||||
/// </remarks>
|
||||
internal static class HistorianEventWriteProtocol
|
||||
{
|
||||
public const ushort BufferSignature = 0x534F; // "OS"
|
||||
public const ushort EventStructVersion = 5;
|
||||
public const ushort OpcQualityGood = 192;
|
||||
private const ushort EventValueDescriptor = 0x118D;
|
||||
|
||||
/// <summary>Built-in CM_EVENT tag id every streamed event targets.</summary>
|
||||
public static readonly Guid CmEventTagId = new("353b8145-5df0-4d46-a253-871aef49b321");
|
||||
|
||||
private const byte CompactStringMarker = 0x09;
|
||||
private const byte ValueTypeUtf16String = 0x43;
|
||||
|
||||
/// <summary>
|
||||
/// Builds the <c>AddS2</c> <c>pBuf</c> for a single event. <paramref name="receivedTimeUtc"/>
|
||||
/// is the storage/received timestamp the native path generates uniquely; the orchestrator
|
||||
/// passes a current time.
|
||||
/// </summary>
|
||||
public static byte[] SerializeAddStreamValuesBuffer(HistorianEvent evt, DateTime receivedTimeUtc)
|
||||
{
|
||||
ArgumentNullException.ThrowIfNull(evt);
|
||||
byte[] valueBlob = SerializeEventValueBlob(evt, receivedTimeUtc);
|
||||
|
||||
// The server's CValuStream check requires the delivered byte[] length to equal the
|
||||
// declared packet length (UInt32 @0x04 = valueBlob.Length + 11). The native client's
|
||||
// content is only valueBlob.Length + 10 bytes and relies on the MDAS encoder appending
|
||||
// one trailing byte on the wire (its value is non-deterministic padding — 0x9F/0x00 in
|
||||
// captures — and is beyond the parsed event content). Over the SDK's WCF byte[] path no
|
||||
// such byte is added, so we emit it explicitly to satisfy the length check.
|
||||
byte[] buffer = new byte[8 + 2 + valueBlob.Length + 1];
|
||||
BinaryPrimitives.WriteUInt16LittleEndian(buffer.AsSpan(0, 2), BufferSignature);
|
||||
BinaryPrimitives.WriteUInt16LittleEndian(buffer.AsSpan(2, 2), 1); // sampleCount
|
||||
BinaryPrimitives.WriteUInt32LittleEndian(buffer.AsSpan(4, 4), checked((uint)(valueBlob.Length + 11)));
|
||||
BinaryPrimitives.WriteUInt16LittleEndian(buffer.AsSpan(8, 2), checked((ushort)(valueBlob.Length + 1)));
|
||||
valueBlob.CopyTo(buffer.AsSpan(10));
|
||||
// buffer[^1] left as 0x00 trailing pad.
|
||||
return buffer;
|
||||
}
|
||||
|
||||
private static byte[] SerializeEventValueBlob(HistorianEvent evt, DateTime receivedTimeUtc)
|
||||
{
|
||||
using MemoryStream stream = new();
|
||||
using BinaryWriter writer = new(stream, Encoding.Unicode, leaveOpen: true);
|
||||
|
||||
writer.Write(CmEventTagId.ToByteArray());
|
||||
// EventTime is the VTQ value timestamp; the native path floors it to whole
|
||||
// milliseconds (the event store's resolution).
|
||||
writer.Write(ToMillisecondFileTime(evt.EventTimeUtc));
|
||||
writer.Write(OpcQualityGood);
|
||||
writer.Write(OpcQualityGood);
|
||||
writer.Write(EventValueDescriptor);
|
||||
writer.Write(evt.Id.ToByteArray());
|
||||
// ReceivedTime keeps full 100ns FILETIME precision (a unique/monotonic time on the
|
||||
// native path), so it is NOT truncated to milliseconds.
|
||||
writer.Write(ToFileTime(receivedTimeUtc));
|
||||
WriteCompactAsciiString(writer, evt.Namespace ?? string.Empty);
|
||||
WriteCompactAsciiString(writer, evt.Type ?? string.Empty);
|
||||
writer.Write(EventStructVersion);
|
||||
|
||||
IReadOnlyList<KeyValuePair<string, object?>> properties = OrderedProperties(evt.Properties);
|
||||
writer.Write(checked((ushort)properties.Count));
|
||||
foreach (KeyValuePair<string, object?> property in properties)
|
||||
{
|
||||
WriteCompactAsciiString(writer, property.Key);
|
||||
WritePropertyValue(writer, property.Key, property.Value);
|
||||
}
|
||||
|
||||
writer.Flush();
|
||||
return stream.ToArray();
|
||||
}
|
||||
|
||||
private static IReadOnlyList<KeyValuePair<string, object?>> OrderedProperties(IReadOnlyDictionary<string, object?>? properties)
|
||||
{
|
||||
if (properties is null || properties.Count == 0)
|
||||
{
|
||||
return [];
|
||||
}
|
||||
|
||||
// The native client stores properties in a SortedList keyed by name; mirror that
|
||||
// deterministic ordering so the serialized bytes are stable.
|
||||
List<KeyValuePair<string, object?>> ordered = [.. properties];
|
||||
ordered.Sort(static (a, b) => string.CompareOrdinal(a.Key, b.Key));
|
||||
return ordered;
|
||||
}
|
||||
|
||||
private static void WritePropertyValue(BinaryWriter writer, string name, object? value)
|
||||
{
|
||||
if (value is not string s)
|
||||
{
|
||||
throw new ProtocolEvidenceMissingException(
|
||||
$"Event property '{name}' has value type '{value?.GetType().Name ?? "null"}'. Only string-valued " +
|
||||
"event properties have a captured write encoding; capture the typed-value framing before sending others.");
|
||||
}
|
||||
|
||||
byte[] chars = Encoding.Unicode.GetBytes(s);
|
||||
int valueByteLength = 2 + chars.Length; // UInt16 charCount + UTF-16 chars
|
||||
if (valueByteLength > byte.MaxValue)
|
||||
{
|
||||
throw new ProtocolEvidenceMissingException(
|
||||
$"Event property '{name}' string value is too long ({s.Length} chars); only a single-byte value " +
|
||||
"length has been captured on the event-send wire.");
|
||||
}
|
||||
|
||||
writer.Write(ValueTypeUtf16String);
|
||||
writer.Write((byte)valueByteLength);
|
||||
writer.Write((byte)0); // status
|
||||
writer.Write(checked((ushort)s.Length));
|
||||
writer.Write(chars);
|
||||
}
|
||||
|
||||
/// <summary>Compact ASCII string: <c>0x09 LEN 0x00 LEN×ASCII bytes</c>.</summary>
|
||||
private static void WriteCompactAsciiString(BinaryWriter writer, string value)
|
||||
{
|
||||
byte[] ascii = Encoding.ASCII.GetBytes(value);
|
||||
if (ascii.Length > byte.MaxValue)
|
||||
{
|
||||
throw new ProtocolEvidenceMissingException(
|
||||
$"String '{value}' exceeds the single-byte length captured for event compact strings.");
|
||||
}
|
||||
|
||||
writer.Write(CompactStringMarker);
|
||||
writer.Write((byte)ascii.Length);
|
||||
writer.Write((byte)0);
|
||||
writer.Write(ascii);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// FILETIME truncated to whole milliseconds — the native event path stores millisecond
|
||||
/// resolution and floors sub-millisecond ticks (observed in the capture).
|
||||
/// </summary>
|
||||
private static long ToMillisecondFileTime(DateTime value)
|
||||
{
|
||||
DateTime utc = value.Kind == DateTimeKind.Utc ? value : value.ToUniversalTime();
|
||||
long ms = utc.Ticks / TimeSpan.TicksPerMillisecond;
|
||||
DateTime truncated = new(ms * TimeSpan.TicksPerMillisecond, DateTimeKind.Utc);
|
||||
return truncated.ToFileTimeUtc();
|
||||
}
|
||||
|
||||
private static long ToFileTime(DateTime value)
|
||||
{
|
||||
DateTime utc = value.Kind == DateTimeKind.Utc ? value : value.ToUniversalTime();
|
||||
return utc.ToFileTimeUtc();
|
||||
}
|
||||
}
|
||||
@@ -87,6 +87,62 @@ internal sealed class HistorianWcfEventOrchestrator
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>Diagnostic: type+code description of the most recent AddS2 (event send) error buffer.</summary>
|
||||
public string LastSendErrorDescription { get; private set; } = string.Empty;
|
||||
|
||||
/// <summary>
|
||||
/// Sends a single <see cref="HistorianEvent"/> to the built-in CM_EVENT tag via the
|
||||
/// captured WCF event-send chain: Open2 (event mode 0x501) → CM_EVENT registration
|
||||
/// (RTag2 + EnsT2) → AddS2 (<c>AddStreamValues2</c>) carrying the serialized event VTQ.
|
||||
/// Returns the server's AddS2 result.
|
||||
/// </summary>
|
||||
public async Task<bool> SendEventAsync(HistorianEvent evt, CancellationToken cancellationToken)
|
||||
{
|
||||
ArgumentNullException.ThrowIfNull(evt);
|
||||
if (!_options.IntegratedSecurity && string.IsNullOrEmpty(_options.UserName))
|
||||
{
|
||||
throw new ProtocolEvidenceMissingException(
|
||||
"Managed event send currently requires IntegratedSecurity or an explicit UserName + Password.");
|
||||
}
|
||||
|
||||
if (evt.RevisionVersion != 0)
|
||||
{
|
||||
throw new ProtocolEvidenceMissingException(
|
||||
"Only original events (RevisionVersion = 0) have a captured send encoding; " +
|
||||
"revision/update/delete event sends are not yet supported.");
|
||||
}
|
||||
|
||||
cancellationToken.ThrowIfCancellationRequested();
|
||||
return await Task.Run(() => RunSendEventChain(evt, cancellationToken), cancellationToken).ConfigureAwait(false);
|
||||
}
|
||||
|
||||
private bool RunSendEventChain(HistorianEvent evt, CancellationToken cancellationToken)
|
||||
{
|
||||
Guid contextKey = Guid.NewGuid();
|
||||
var (histBinding, histEndpoint, retrBinding, retrEndpoint) = HistorianWcfBindingFactory.CreateBindingPair(_options);
|
||||
Binding auxBinding = HistorianWcfBindingFactory.CreateAuxiliaryBinding(_options);
|
||||
EndpointAddress statusEndpoint = HistorianWcfBindingFactory.CreateAuxiliaryEndpointAddress(_options, HistorianWcfServiceNames.Status);
|
||||
EndpointAddress transactionEndpoint = HistorianWcfBindingFactory.CreateAuxiliaryEndpointAddress(_options, HistorianWcfServiceNames.Transaction);
|
||||
|
||||
bool sent = false;
|
||||
HistorianWcfAuthChainHelper.OpenAuthenticatedConnection(
|
||||
_options, histBinding, histEndpoint, contextKey, cancellationToken,
|
||||
connectionMode: HistorianWcfAuthChainHelper.NativeIntegratedEventConnectionMode,
|
||||
additionalSetup: (historyChannel, context) =>
|
||||
{
|
||||
// Register CM_EVENT for this session (RTag2 + EnsT2) exactly as the native
|
||||
// event-send chain does, then stream the event value via AddS2 on the same
|
||||
// /Hist channel + storage-session handle.
|
||||
AddCmEventTagViaAddT(historyChannel, context, auxBinding, statusEndpoint, transactionEndpoint, retrBinding, retrEndpoint);
|
||||
|
||||
byte[] pBuf = HistorianEventWriteProtocol.SerializeAddStreamValuesBuffer(evt, DateTime.UtcNow);
|
||||
string handle = context.StorageSessionId.ToString("D").ToUpperInvariant();
|
||||
sent = historyChannel.AddStreamValues2(handle, pBuf, out byte[] errorBuffer);
|
||||
LastSendErrorDescription = DescribeNativeError(errorBuffer ?? []);
|
||||
});
|
||||
return sent;
|
||||
}
|
||||
|
||||
private List<HistorianEvent> RunEventChain(DateTime startUtc, DateTime endUtc, CancellationToken cancellationToken)
|
||||
{
|
||||
Guid contextKey = Guid.NewGuid();
|
||||
|
||||
Reference in New Issue
Block a user