En mi aplicación .Net Core, hay un método en una biblioteca de terceros que escribe en una interfaz System.IO.Stream (toma la interfaz de transmisión como argumento y escribe en ella), pero quiero que esos datos vayan a mi fuente de datos que espera datos como una IAsyncEnumerable<bytes> . Me puse a escribir el código para implementar la interfaz Stream , de modo que cuando se llama Write() escribe en IAsyncEnumerable<bytes> , luego pensé 'esto debe haberse hecho antes', parece que sería de uso general.
Entonces, ¿hay una implementación estándar de esto en una biblioteca de terceros, o algún "buen truco" que me falta?
Aquí hay una implementación de Stream personalizada, diseñada para escenarios asíncronos de productor-consumidor. Es una secuencia de solo escritura, y su lectura (consumo) solo es posible a través del método especial GetConsumingEnumerable .
public class ProducerConsumerStream : Stream { private readonly Channel<byte> _channel; public ProducerConsumerStream(bool singleReader = true, bool singleWriter = true) { _channel = Channel.CreateUnbounded<byte>(new UnboundedChannelOptions() { SingleReader = singleReader, SingleWriter = singleWriter }); } public override bool CanRead { get { return false; } } public override bool CanSeek { get { return false; } } public override bool CanWrite { get { return true; } } public override long Length { get { throw new NotSupportedException(); } } public override void Flush() { } public override long Position { get { throw new NotSupportedException(); } set { throw new NotSupportedException(); } } public override long Seek(long offset, SeekOrigin origin) => throw new NotSupportedException(); public override void SetLength(long value) => throw new NotSupportedException(); public override int Read(byte[] buffer, int offset, int count) => throw new NotSupportedException(); public override void Write(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 (offset + count > buffer.Length) throw new ArgumentOutOfRangeException(nameof(count)); for (int i = offset; i < offset + count; i++) _channel.Writer.TryWrite(buffer[i]); } public override void WriteByte(byte value) { _channel.Writer.TryWrite(value); } public override void Close() { base.Close(); _channel.Writer.Complete(); } public IAsyncEnumerable<byte> GetConsumingEnumerable( CancellationToken cancellationToken = default) { return _channel.Reader.ReadAllAsync(cancellationToken); } } Esta implementación se basa en Channel<byte> . Si no está familiarizado con los canales, hay un tutorial aquí .