/* ========================================================================
* 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.Net;
using System.Threading;
using System.Threading.Tasks;
using Microsoft.Extensions.Logging;
using Microsoft.Extensions.Logging.Abstractions;
using Opc.Ua.Bindings;
namespace Opc.Ua.Bindings.Pcap.Bindings
{
///
/// decorator that forwards every call to
/// an inner socket and, when the
/// has an active observer, taps
/// a copy of the bytes that move across the wire.
///
///
///
/// Hot path on send / receive:
///
/// -
/// One volatile read of
/// .
///
/// -
/// If null - one predicted branch + forward to the inner socket.
///
/// -
/// If non-null - one view over the
/// buffer and a single virtual call into the observer (which is
/// expected to enqueue and return without I/O).
///
///
///
///
/// Channel id is read from the wrapped
/// (which is always a ) at the moment
/// of the call so renewals and reattachments are handled
/// transparently.
///
///
public sealed class CapturingMessageSocket : IMessageSocket
{
private readonly IMessageSocket m_inner;
private readonly IChannelCaptureRegistry m_registry;
private readonly ILogger m_logger;
private CapturingSinkWrapper m_sinkWrapper;
///
/// Constructs a new capturing socket wrapper.
///
/// The wrapped socket; ownership transfers.
/// The capture registry whose observer is
/// consulted on the send / receive hot path.
/// The sink supplied by the channel.
/// Must be a so the channel id
/// can be projected to the observer.
/// Optional logger factory.
/// Any required argument is null.
public CapturingMessageSocket(
IMessageSocket inner,
IChannelCaptureRegistry registry,
IMessageSink originalSink,
ILoggerFactory? loggerFactory = null)
{
ArgumentNullException.ThrowIfNull(inner);
ArgumentNullException.ThrowIfNull(registry);
ArgumentNullException.ThrowIfNull(originalSink);
m_inner = inner;
m_registry = registry;
m_logger = (loggerFactory ?? NullLoggerFactory.Instance)
.CreateLogger();
m_sinkWrapper = new CapturingSinkWrapper(originalSink, this);
m_inner.ChangeSink(m_sinkWrapper);
}
///
public int Handle => m_inner.Handle;
///
public EndPoint? LocalEndpoint => m_inner.LocalEndpoint;
///
public EndPoint? RemoteEndpoint => m_inner.RemoteEndpoint;
///
public TransportChannelFeatures MessageSocketFeatures
=> m_inner.MessageSocketFeatures;
///
public Task ConnectAsync(Uri endpointUrl, CancellationToken ct = default)
=> m_inner.ConnectAsync(endpointUrl, ct);
///
public void Close() => m_inner.Close();
///
public void ReadNextMessage() => m_inner.ReadNextMessage();
///
public void ChangeSink(IMessageSink sink)
{
ArgumentNullException.ThrowIfNull(sink);
m_sinkWrapper = new CapturingSinkWrapper(sink, this);
m_inner.ChangeSink(m_sinkWrapper);
}
///
public bool Send(IMessageSocketAsyncEventArgs args)
{
ArgumentNullException.ThrowIfNull(args);
IFrameCaptureSink? observer = m_registry.CurrentObserver;
if (observer is not null)
{
TapOutbound(observer, args);
}
return m_inner.Send(args);
}
///
public IMessageSocketAsyncEventArgs MessageSocketEventArgs()
=> m_inner.MessageSocketEventArgs();
///
public void Dispose() => m_inner.Dispose();
///
/// Called by the wrapped sink on the receive path. Reads the
/// active observer (single volatile load) and forwards a span
/// over the chunk bytes.
///
internal void TapInbound(uint channelId, ArraySegment message)
{
IFrameCaptureSink? observer = m_registry.CurrentObserver;
if (observer is null || message.Array is null || message.Count == 0)
{
return;
}
try
{
observer.OnFrameReceived(
channelId,
new ReadOnlySpan(message.Array, message.Offset, message.Count));
}
catch (Exception ex)
{
m_logger.LogWarning(
ex,
"CapturingMessageSocket: observer.OnFrameReceived threw on channel {ChannelId}.",
channelId);
}
}
private void TapOutbound(IFrameCaptureSink observer, IMessageSocketAsyncEventArgs args)
{
uint channelId = m_sinkWrapper.ChannelId;
try
{
BufferCollection? list = args.BufferList;
if (list is { Count: > 0 })
{
for (int i = 0; i < list.Count; i++)
{
ArraySegment segment = list[i];
if (segment.Array is null || segment.Count == 0)
{
continue;
}
observer.OnFrameSent(
channelId,
new ReadOnlySpan(segment.Array, segment.Offset, segment.Count));
}
return;
}
byte[]? buffer = args.Buffer;
if (buffer is null || args.Count == 0)
{
return;
}
observer.OnFrameSent(
channelId,
new ReadOnlySpan(buffer, args.Offset, args.Count));
}
catch (Exception ex)
{
m_logger.LogWarning(
ex,
"CapturingMessageSocket: observer.OnFrameSent threw on channel {ChannelId}.",
m_sinkWrapper.ChannelId);
}
}
///
/// Sink wrapper that taps received chunks before forwarding to the
/// real channel sink. Constructed once per
/// call so the cast to happens
/// only once.
///
private sealed class CapturingSinkWrapper : IMessageSink
{
private readonly IMessageSink m_inner;
private readonly CapturingMessageSocket m_owner;
private readonly UaSCUaBinaryChannel? m_channel;
public CapturingSinkWrapper(IMessageSink inner, CapturingMessageSocket owner)
{
m_inner = inner;
m_owner = owner;
m_channel = inner as UaSCUaBinaryChannel;
}
public uint ChannelId => m_channel?.Id ?? 0u;
public bool ChannelFull => m_inner.ChannelFull;
public void OnMessageReceived(IMessageSocket source, ArraySegment message)
{
m_owner.TapInbound(ChannelId, message);
m_inner.OnMessageReceived(source, message);
}
public void OnReceiveError(IMessageSocket source, ServiceResult result)
{
m_inner.OnReceiveError(source, result);
}
}
}
}