src/Npgsql/Internal/Converters/StreamConverter.cs - test-logs - dev-npgsql/npgsql

src/Npgsql/Internal/Converters/StreamConverter.cs
73 строки3 KB
Nino Floris
Fix some more state cleanup cases
12 май 2026, 16:58
12 май 2026, 16:58467aca3
using System;
using System.IO;
using System.Threading;
using System.Threading.Tasks;

namespace Npgsql.Internal.Converters;

sealed class StreamConverter(bool supportsTextFormat) : PgStreamingConverter<Stream>
{
    public override bool CanConvert(DataFormat format, out BufferRequirements bufferRequirements)
    {
        bufferRequirements = BufferRequirements.Streaming;
        return supportsTextFormat
            ? format is DataFormat.Text or DataFormat.Binary
            : format is DataFormat.Binary;
    }

    public override Stream Read(PgReader reader)
        => reader.GetStream();

    public override ValueTask<Stream> ReadAsync(PgReader reader, CancellationToken cancellationToken = default)
        => new(reader.GetStream());

    protected override Size BindValue(in BindContext context, Stream value, ref object? writeState)
    {
        if (value.CanSeek)
            return checked((int)(value.Length - value.Position));

        var memoryStream = new MemoryStream();
        writeState = memoryStream;
        value.CopyTo(memoryStream);
        return checked((int)memoryStream.Length);
    }

    public override void Write(PgWriter writer, Stream value)
    {
        if (writer.Current.WriteState is not null)
        {
            if (!((MemoryStream)writer.Current.WriteState!).TryGetBuffer(out var writeStateSegment))
                throw new InvalidOperationException();

            writer.WriteBytes(writeStateSegment.AsSpan());
            return;
        }

        // Non-derived MemoryStream fast path
        if (value is MemoryStream memoryStream && memoryStream.TryGetBuffer(out var segment))
            writer.WriteBytes(segment.AsSpan((int)value.Position));
        else
            value.CopyTo(writer.GetStream());
    }

    public override ValueTask WriteAsync(PgWriter writer, Stream value, CancellationToken cancellationToken = default)
    {
        if (writer.Current.WriteState is not null)
        {
            if (!((MemoryStream)writer.Current.WriteState!).TryGetBuffer(out var writeStateSegment))
                throw new InvalidOperationException();

            return writer.WriteBytesAsync(writeStateSegment.AsMemory(), cancellationToken);
        }

        // Non-derived MemoryStream fast path
        if (value is MemoryStream memoryStream && memoryStream.TryGetBuffer(out var segment))
        {
            return writer.WriteBytesAsync(segment.AsMemory((int)value.Position), cancellationToken);
        }
        else
        {
            return new ValueTask(value.CopyToAsync(writer.GetStream(), cancellationToken));
        }
    }
}