<commit_before>﻿using System;
using System.IO;
using System.Threading;
using System.Threading.Tasks;
using Cadena._Internals;
using JetBrains.Annotations;

namespace Cadena.Api.Streams
{
    /// <summary>
    /// Core of engine of receiving general streams.
    /// </summary>
    internal static class StreamWinder
    {
        public static async Task Run([NotNull] Stream stream, [NotNull] Action<string> parser,
            TimeSpan readTimeout, CancellationToken cancellationToken)
        {
            if (stream == null) throw new ArgumentNullException(nameof(stream));
            if (parser == null) throw new ArgumentNullException(nameof(parser));

            using (var reader = new CancellableStreamReader(stream))
            {
                while (true)
                {
                    cancellationToken.ThrowIfCancellationRequested();

                    /* 
                     * NOTE: performance information
                     * Creating CancellationTokenSource each time is faster than using Task.Delay.
                     * Simpler way always defeats complex and confusing one.
                     */
                    // create timeout cancellation token source
                    using (var tokenSource = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken))
                    {
                        tokenSource.CancelAfter(readTimeout);
                        // execute reading next line
                        var line = (await reader.ReadLineAsync(tokenSource.Token).ConfigureAwait(false));
                        if (line == null)
                        {
                            System.Diagnostics.Debug.WriteLine("#USERSTREAM# CONNECTION CLOSED.");
                            break;
                        }

                        // skip empty response
                        if (String.IsNullOrWhiteSpace(line)) continue;
                        // call parser with read line
#pragma warning disable  4014
                        Task.Run(() => parser(line), cancellationToken);
#pragma warning restore  4014
                    }
                }
            }
        }
    }
}<commit_msg>＼(#`・△・)/ < Improve behavior around CancellationTokenSource in StreamWinder.<commit_after>﻿using System;
using System.IO;
using System.Threading;
using System.Threading.Tasks;
using Cadena._Internals;
using JetBrains.Annotations;

namespace Cadena.Api.Streams
{
    /// <summary>
    /// Core of engine of receiving general streams.
    /// </summary>
    internal static class StreamWinder
    {
        public static async Task Run([NotNull] Stream stream, [NotNull] Action<string> parser,
            TimeSpan readTimeout, CancellationToken cancellationToken)
        {
            if (stream == null) throw new ArgumentNullException(nameof(stream));
            if (parser == null) throw new ArgumentNullException(nameof(parser));

            // create timeout cancellation token source
            using (var tokenSource = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken))
            using (var reader = new CancellableStreamReader(stream))
            {
                var localToken = tokenSource.Token;
                while (true)
                {
                    localToken.ThrowIfCancellationRequested();

                    // set read timeout(add epsilon time to timeout (100 msec))
                    tokenSource.CancelAfter(readTimeout + TimeSpan.FromTicks(10000 * 100));

                    // execute reading next line
                    var line = (await reader.ReadLineAsync(localToken).ConfigureAwait(false));

                    // disable timer
                    tokenSource.CancelAfter(Timeout.InfiniteTimeSpan);

                    if (line == null)
                    {
                        System.Diagnostics.Debug.WriteLine("#USERSTREAM# CONNECTION CLOSED.");
                        break;
                    }

                    // skip empty response
                    if (String.IsNullOrWhiteSpace(line)) continue;

                    // call parser with read line
#pragma warning disable 4014
                    Task.Run(() => parser(line), cancellationToken);
#pragma warning restore  4014
                }
            }
        }
    }
}
{'fraction_non_alphanumeric': 0.07966457023060797, 'fraction_numerical': 0.006289308176100629, 'mean_word_length': 1.6262996941896024, 'pattern_counts': {'":': 0, '<': 10, '<?xml version=': 0, '>': 11, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '3932383', 'n_tokens_mistral': 992, 'n_tokens_neox': 938, 'n_words': 259}