xIdentityService/Extensions/StreamExtensions.cs

using System.Collections.Generic;
using System.IO;
using System.Linq;
using System.Threading;
using xCommons.Extensions;

namespace xIdentityService.Extensions
{
    public static partial class StreamExtensions
    {
        /// <summary>
        /// Converts an Stream to Async Enumerable ...
        /// </summary>
        /// <param name="stream"></param>
        /// <param name="bufferSize"></param>
        /// <param name="cancellationToken"></param>
        /// <returns></returns>
        public static async IAsyncEnumerable<byte[]> ToAsyncEnumerable(
            this Stream stream,
            int bufferSize = 8192,
            [System.Runtime.CompilerServices.EnumeratorCancellation]
            CancellationToken cancellationToken = default
        )
        {
            //
            var buffer = new byte[bufferSize];
            while (true)
            {
                //
                int read = await stream
                    .ReadAsync(
                        buffer,
                        0,
                        buffer.Length,
                        cancellationToken
                );

                //
                if (read == 0)
                {
                    yield break;
                }

                //
                yield return buffer.Take(read).ToArray();
            }
        }

        /// <summary>
        /// Converts an Stream to Async Enumerable ...
        /// </summary>
        /// <param name="stream"></param>
        /// <param name="bufferSize"></param>
        /// <param name="cancellationToken"></param>
        /// <returns></returns>
        public static async IAsyncEnumerable<T> ToAsyncEnumerable<T>(
            this Stream stream,
            int bufferSize = 8192,
            [System.Runtime.CompilerServices.EnumeratorCancellation]
            CancellationToken cancellationToken = default
        )
        {
            //
            var buffer = new byte[bufferSize];
            while (true)
            {
                //
                int read = await stream
                    .ReadAsync(
                        buffer,
                        0,
                        buffer.Length,
                        cancellationToken
                );
                if (read == 0)
                {
                    yield break;
                }

                //
                T chunk = default(T);
                var readedBuffer = buffer.Take(read).ToArray();
                var str = readedBuffer.FromBytes();
                if (str.IsNullOrEmpty())
                {
                    yield return chunk;
                }
                try
                {
                    chunk = str.FromJSON<T>();
                }
                catch
                { }

                //
                yield return chunk;
            }
        }
    }
}