/* ======================================================================== * Copyright (c) 2005-2025 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.Buffers; using System.IO; using System.Threading; using System.Threading.Tasks; namespace Opc.Ua.Client.FileSystem { /// /// A -derived wrapper around an OPC UA /// FileType handle. Async members hit the wire via the /// supplied proxy; sync members /// forward to their async counterparts via /// GetAwaiter().GetResult(). /// /// /// /// serialises concurrent calls from /// multiple threads on the same instance via an internal /// ; the resulting behaviour matches /// 's "not thread-safe but doesn't corrupt /// state" contract. /// /// /// is tracked locally because OPC UA does not /// expose a "current length" accessor on the file handle. It is /// initialised at construction (typically from the file's /// Size property) and bumped whenever a write extends past /// the current length. Callers that mutate the underlying file /// through other handles should call /// on the owning info object /// before relying on . /// /// /// is tracked locally and pushed to the /// server lazily — only when the requested cursor diverges from /// the last successfully transmitted position. /// /// /// Sync members (Read, Write, Seek, /// Flush, Dispose) forward to the async overrides /// via GetAwaiter().GetResult(). Calling them on a /// single-threaded synchronization context (e.g. WPF UI thread) /// can deadlock — prefer the async overrides. /// /// public sealed class UaFileStream : Stream #if !(NETSTANDARD2_1_OR_GREATER || NET) , IAsyncDisposable #endif { internal UaFileStream( FileTypeClient proxy, uint handle, UaFileMode mode, long initialLength, long initialPosition, int chunkSize) { m_proxy = proxy ?? throw new ArgumentNullException(nameof(proxy)); Handle = handle; Mode = mode; m_length = initialLength < 0 ? 0 : initialLength; m_position = initialPosition < 0 ? 0 : initialPosition; m_serverPosition = m_position; m_chunkSize = chunkSize <= 0 ? DefaultChunkSize : chunkSize; } /// public override bool CanRead => !m_disposed && (Mode & UaFileMode.Read) != 0; /// public override bool CanWrite => !m_disposed && (Mode & UaFileMode.Write) != 0; /// public override bool CanSeek => !m_disposed; /// public override bool CanTimeout => false; /// public override long Length { get { ThrowIfDisposed(); return Volatile.Read(ref m_length); } } /// public override long Position { get { ThrowIfDisposed(); return Volatile.Read(ref m_position); } set => Seek(value, SeekOrigin.Begin); } /// /// The OPC UA FileType handle wrapped by this stream. /// public uint Handle { get; } /// /// The mode this stream was opened with. /// public UaFileMode Mode { get; } /// public override Task ReadAsync( byte[] buffer, int offset, int count, CancellationToken cancellationToken) { ValidateBuffer(buffer, offset, count); return ReadCoreAsync(buffer, offset, count, cancellationToken); } /// public override Task WriteAsync( byte[] buffer, int offset, int count, CancellationToken cancellationToken) { ValidateBuffer(buffer, offset, count); return WriteCoreAsync(buffer, offset, count, cancellationToken); } #if NETSTANDARD2_1_OR_GREATER || NET /// public override async ValueTask ReadAsync( Memory buffer, CancellationToken cancellationToken = default) { if (buffer.IsEmpty) { ThrowIfDisposed(); return 0; } return await ReadIntoSpanAsync(buffer, cancellationToken).ConfigureAwait(false); } /// public override async ValueTask WriteAsync( ReadOnlyMemory buffer, CancellationToken cancellationToken = default) { if (buffer.IsEmpty) { ThrowIfDisposed(); return; } await WriteFromSpanAsync(buffer, cancellationToken).ConfigureAwait(false); } /// public override async ValueTask DisposeAsync() { await DisposeAsyncCore().ConfigureAwait(false); // base.DisposeAsync() calls Dispose(false) internally on the // standard Stream implementation; calling it keeps CA2215 // happy while keeping our (already idempotent) Dispose(bool) // override in the chain. await base.DisposeAsync().ConfigureAwait(false); GC.SuppressFinalize(this); } #else /// /// Asynchronously closes the underlying server file handle. /// Implements the recommended DisposeAsync pattern from /// the .NET docs. /// public async ValueTask DisposeAsync() { await DisposeAsyncCore().ConfigureAwait(false); Dispose(disposing: false); GC.SuppressFinalize(this); } #endif /// public override Task FlushAsync(CancellationToken cancellationToken) { ThrowIfDisposed(); // OPC UA file handles do not expose an explicit flush; writes // are applied immediately on the server. return Task.CompletedTask; } /// public override int Read(byte[] buffer, int offset, int count) { ValidateBuffer(buffer, offset, count); return ReadCoreAsync(buffer, offset, count, CancellationToken.None) .GetAwaiter() .GetResult(); } /// public override void Write(byte[] buffer, int offset, int count) { ValidateBuffer(buffer, offset, count); WriteCoreAsync(buffer, offset, count, CancellationToken.None) .GetAwaiter() .GetResult(); } /// public override long Seek(long offset, SeekOrigin origin) { ThrowIfDisposed(); long target = origin switch { SeekOrigin.Begin => offset, SeekOrigin.Current => Volatile.Read(ref m_position) + offset, SeekOrigin.End => Volatile.Read(ref m_length) + offset, _ => throw new ArgumentOutOfRangeException(nameof(origin)) }; if (target < 0) { throw new IOException("Cannot seek before the start of the stream."); } Volatile.Write(ref m_position, target); return target; } /// public override void SetLength(long value) { // OPC UA's FileType offers no truncation primitive; callers // that need truncation must reopen the file with // EraseExisting. throw new NotSupportedException( "OPC UA FileType does not support truncation; reopen the file with UaFileMode.EraseExisting."); } /// public override void Flush() { // No-op (see FlushAsync). } /// protected override void Dispose(bool disposing) { if (m_disposed) { base.Dispose(disposing); return; } if (disposing) { // Synchronous fallback for callers that use 'using' // instead of 'await using'. Mirrors the recommended // dispose pattern; the async path (DisposeAsync) is // preferred to avoid the GetAwaiter().GetResult() // dance. try { DisposeAsyncCore().AsTask().GetAwaiter().GetResult(); } catch { // Best-effort: async path surfaces exceptions. } } base.Dispose(disposing); } /// /// Performs the asynchronous cleanup of the server-side file /// handle. Idempotent. /// /// /// Implements the recommended async dispose core pattern from /// . /// The class is sealed, so this helper is private /// rather than the protected virtual shape recommended /// for unsealed types. /// private async ValueTask DisposeAsyncCore() { if (m_disposed) { return; } await m_lock.WaitAsync().ConfigureAwait(false); try { if (m_disposed) { return; } m_disposed = true; try { await m_proxy.CloseAsync(Handle, CancellationToken.None) .ConfigureAwait(false); } catch (ServiceResultException) { // Server may have already torn down the handle on // session close — best-effort close. } } finally { m_lock.Release(); m_lock.Dispose(); } } /// /// Marks this stream disposed without sending a Close to the /// server. Used by when the /// owner has already closed (or committed) the handle through a /// different path and wants to prevent further writes through /// this stream wrapper. /// internal void MarkDisposedWithoutClosing() { if (m_disposed) { return; } m_lock.Wait(); try { m_disposed = true; } finally { m_lock.Release(); } } private async Task ReadCoreAsync( byte[] buffer, int offset, int count, CancellationToken ct) { ThrowIfDisposed(); if (!CanRead) { throw new NotSupportedException("Stream not opened for reading."); } if (count == 0) { return 0; } await m_lock.WaitAsync(ct).ConfigureAwait(false); try { ThrowIfDisposed(); await SyncServerPositionAsync(ct).ConfigureAwait(false); int total = 0; while (total < count) { int chunkLen = Math.Min(m_chunkSize, count - total); ByteString data = await m_proxy.ReadAsync( Handle, chunkLen, ct).ConfigureAwait(false); // Empty payload = EOF: short-circuit before touching // the caller's buffer (or allocating anything). if (data.IsEmpty) { break; } int read = data.Length; data.Span.CopyTo(buffer.AsSpan(offset + total, read)); total += read; m_position += read; m_serverPosition = m_position; if (m_position > m_length) { m_length = m_position; } if (read < chunkLen) { // Short read = EOF on most servers. break; } } return total; } finally { m_lock.Release(); } } private async Task WriteCoreAsync( byte[] buffer, int offset, int count, CancellationToken ct) { ThrowIfDisposed(); if (!CanWrite) { throw new NotSupportedException("Stream not opened for writing."); } if (count == 0) { return; } await m_lock.WaitAsync(ct).ConfigureAwait(false); try { ThrowIfDisposed(); await SyncServerPositionAsync(ct).ConfigureAwait(false); int written = 0; while (written < count) { int chunkLen = Math.Min(m_chunkSize, count - written); // Wrap the caller's slice directly — the encoder // copies it onto the wire during WriteAsync, and we // await before reusing the underlying buffer. var chunk = new ByteString( buffer.AsMemory(offset + written, chunkLen)); await m_proxy.WriteAsync( Handle, chunk, ct).ConfigureAwait(false); written += chunkLen; m_position += chunkLen; m_serverPosition = m_position; if (m_position > m_length) { m_length = m_position; } } } finally { m_lock.Release(); } } #if NETSTANDARD2_1_OR_GREATER || NET private async ValueTask ReadIntoSpanAsync( Memory buffer, CancellationToken ct) { byte[] rented = ArrayPool.Shared.Rent(buffer.Length); try { int read = await ReadCoreAsync(rented, 0, buffer.Length, ct) .ConfigureAwait(false); if (read > 0) { rented.AsSpan(0, read).CopyTo(buffer.Span); } return read; } finally { ArrayPool.Shared.Return(rented); } } private async ValueTask WriteFromSpanAsync( ReadOnlyMemory buffer, CancellationToken ct) { byte[] rented = ArrayPool.Shared.Rent(buffer.Length); try { buffer.Span.CopyTo(rented); await WriteCoreAsync(rented, 0, buffer.Length, ct) .ConfigureAwait(false); } finally { ArrayPool.Shared.Return(rented); } } #endif private async Task SyncServerPositionAsync(CancellationToken ct) { if (m_position == m_serverPosition) { return; } await m_proxy.SetPositionAsync( Handle, (ulong)m_position, ct).ConfigureAwait(false); m_serverPosition = m_position; } private void ThrowIfDisposed() { if (m_disposed) { throw new ObjectDisposedException(nameof(UaFileStream)); } } private static void ValidateBuffer(byte[] buffer, int offset, int count) { if (buffer == null) { throw new ArgumentNullException(nameof(buffer)); } if (offset < 0) { throw new ArgumentOutOfRangeException(nameof(offset)); } if (count < 0) { throw new ArgumentOutOfRangeException(nameof(count)); } if (buffer.Length - offset < count) { throw new ArgumentException( "Offset + count exceeds buffer length.", nameof(count)); } } private const int DefaultChunkSize = 64 * 1024; private readonly FileTypeClient m_proxy; private readonly int m_chunkSize; private readonly SemaphoreSlim m_lock = new(1, 1); private long m_length; private long m_position; private long m_serverPosition; private bool m_disposed; } }