<commit_before>﻿using System;
using System.IO;
using System.Net;
using System.Net.Http;
using System.Text;
using System.Threading;
using System.Threading.Tasks;
using Tweetinvi.Core.Enum;
using Tweetinvi.Core.Events;
using Tweetinvi.Core.Events.EventArguments;
using Tweetinvi.Core.Exceptions;
using Tweetinvi.Core.Extensions;
using Tweetinvi.Core.Helpers;
using Tweetinvi.Core.Injectinvi;
using Tweetinvi.Core.Interfaces.Exceptions;
using Tweetinvi.Core.Interfaces.Models;
using HttpMethod = System.Net.Http.HttpMethod;

namespace Tweetinvi.Streams
{
    public interface IStreamTask
    {
        event EventHandler StreamStarted;
        event EventHandler<GenericEventArgs<StreamState>> StreamStateChanged;

        StreamState StreamState { get; }
        Exception LastException { get; }

        void Start();
        void Resume();
        void Pause();
        void Stop();
    }

    public class StreamTask : IStreamTask
    {
        public event EventHandler StreamStarted;
        public event EventHandler<GenericEventArgs<StreamState>> StreamStateChanged;

        // https://dev.twitter.com/streaming/overview/connecting#stalls
        private const int STREAM_DISCONNECTED_DELAY = 90000;
        private const int STREAM_RESUME_DELAY = 1000;

        private readonly Func<string, bool> _processObject;
        private readonly Func<ITwitterQuery> _generateTwitterQuery;
        private readonly IExceptionHandler _exceptionHandler;
        private readonly ITweetinviEvents _tweetinviEvents;
        private readonly IFactory<ITwitterTimeoutException> _twitterTimeoutExceptionFactory;
        private readonly IHttpClientWebHelper _httpClientWebHelper;

        private bool _isNew;
        private Exception _lastException;

        private ITwitterQuery _twitterQuery;
        private StreamReader _currentStreamReader;
        private HttpClient _currentHttpClient;

        public StreamTask(
            Func<string, bool> processObject,
            Func<ITwitterQuery> generateTwitterQuery,
            IExceptionHandler exceptionHandler,
            ITweetinviEvents tweetinviEvents,
            IFactory<ITwitterTimeoutException> twitterTimeoutExceptionFactory,
            IHttpClientWebHelper httpClientWebHelper)
        {
            _processObject = processObject;
            _generateTwitterQuery = generateTwitterQuery;
            _exceptionHandler = exceptionHandler;
            _tweetinviEvents = tweetinviEvents;
            _twitterTimeoutExceptionFactory = twitterTimeoutExceptionFactory;
            _httpClientWebHelper = httpClientWebHelper;
            _isNew = true;
        }

        public StreamState StreamState { get; private set; }
        public Exception LastException { get { return _lastException; } }

        public void Start()
        {
            if (StreamState == StreamState.Stop && !_isNew)
            {
                return;
            }

            this.Raise(StreamStarted);
            SetStreamState(StreamState.Resume);

            _twitterQuery = _generateTwitterQuery();

            if (_twitterQuery.TwitterCredentials == null)
            {
                throw new TwitterNullCredentialsException();
            }

            if (!_twitterQuery.TwitterCredentials.AreSetupForUserAuthentication())
            {
                throw new TwitterInvalidCredentialsException(_twitterQuery.TwitterCredentials);
            }

            _currentHttpClient = GetHttpClient(_twitterQuery);
            _currentStreamReader = GetStreamReader(_currentHttpClient, _twitterQuery).Result;

            int numberOfRepeatedFailures = 0;

            while (StreamState != StreamState.Stop)
            {
                if (StreamState == StreamState.Pause)
                {
                    using (EventWaitHandle tmpEvent = new ManualResetEvent(false))
                    {
                        tmpEvent.WaitOne(TimeSpan.FromMilliseconds(STREAM_RESUME_DELAY));
                    }

                    continue;
                }

                try
                {
                    var json = GetJsonResponseFromReader(_currentStreamReader, _twitterQuery);

                    var isJsonResponseValid = json.IsMatchingJsonFormat();
                    if (!isJsonResponseValid)
                    {
                        if (json != null)
                        {
                            throw new WebException(json);
                        }

                        if (TryHandleInvalidResponse(numberOfRepeatedFailures))
                        {
                            ++numberOfRepeatedFailures;
                            continue;
                        }

                        throw new WebException("Stream cannot be read.");
                    }

                    numberOfRepeatedFailures = 0;

                    if (json == string.Empty)
                    {
                        continue;
                    }

                    if (StreamState == StreamState.Resume && !_processObject(json))
                    {
                        SetStreamState(StreamState.Stop);
                        break;
                    }
                }
                catch (Exception ex)
                {
                    if (!ShouldContinueAfterHandlingException(ex))
                    {
                        SetStreamState(StreamState.Stop);
                        break;
                    }
                }
            }

            if (_currentStreamReader != null)
            {
                _currentStreamReader.Dispose();
            }

            if (_currentHttpClient != null)
            {
                _currentHttpClient.Dispose();<commit_msg>Tweetinvi 0.9.10.2 - Fixed critical bug affecting all streams<commit_after>﻿using System;
using System.IO;
using System.Net;
using System.Net.Http;
using System.Text;
using System.Threading;
using System.Threading.Tasks;
using Tweetinvi.Core.Enum;
using Tweetinvi.Core.Events;
using Tweetinvi.Core.Events.EventArguments;
using Tweetinvi.Core.Exceptions;
using Tweetinvi.Core.Extensions;
using Tweetinvi.Core.Helpers;
using Tweetinvi.Core.Injectinvi;
using Tweetinvi.Core.Interfaces.Exceptions;
using Tweetinvi.Core.Interfaces.Models;
using HttpMethod = System.Net.Http.HttpMethod;

namespace Tweetinvi.Streams
{
    public interface IStreamTask
    {
        event EventHandler StreamStarted;
        event EventHandler<GenericEventArgs<StreamState>> StreamStateChanged;

        StreamState StreamState { get; }
        Exception LastException { get; }

        void Start();
        void Resume();
        void Pause();
        void Stop();
    }

    public class StreamTask : IStreamTask
    {
        public event EventHandler StreamStarted;
        public event EventHandler<GenericEventArgs<StreamState>> StreamStateChanged;

        // https://dev.twitter.com/streaming/overview/connecting#stalls
        private const int STREAM_DISCONNECTED_DELAY = 90000;
        private const int STREAM_RESUME_DELAY = 1000;

        private readonly Func<string, bool> _processObject;
        private readonly Func<ITwitterQuery> _generateTwitterQuery;
        private readonly IExceptionHandler _exceptionHandler;
        private readonly ITweetinviEvents _tweetinviEvents;
        private readonly IFactory<ITwitterTimeoutException> _twitterTimeoutExceptionFactory;
        private readonly IHttpClientWebHelper _httpClientWebHelper;

        private bool _isNew;
        private Exception _lastException;

        private ITwitterQuery _twitterQuery;
        private StreamReader _currentStreamReader;
        private HttpClient _currentHttpClient;

        public StreamTask(
            Func<string, bool> processObject,
            Func<ITwitterQuery> generateTwitterQuery,
            IExceptionHandler exceptionHandler,
            ITweetinviEvents tweetinviEvents,
            IFactory<ITwitterTimeoutException> twitterTimeoutExceptionFactory,
            IHttpClientWebHelper httpClientWebHelper)
        {
            _processObject = processObject;
            _generateTwitterQuery = generateTwitterQuery;
            _exceptionHandler = exceptionHandler;
            _tweetinviEvents = tweetinviEvents;
            _twitterTimeoutExceptionFactory = twitterTimeoutExceptionFactory;
            _httpClientWebHelper = httpClientWebHelper;
            _isNew = true;
        }

        public StreamState StreamState { get; private set; }
        public Exception LastException { get { return _lastException; } }

        public void Start()
        {
            if (StreamState == StreamState.Stop && !_isNew)
            {
                return;
            }

            this.Raise(StreamStarted);
            SetStreamState(StreamState.Resume);

            _twitterQuery = _generateTwitterQuery();

            if (_twitterQuery.TwitterCredentials == null)
            {
                throw new TwitterNullCredentialsException();
            }

            if (!_twitterQuery.TwitterCredentials.AreSetupForUserAuthentication())
            {
                throw new TwitterInvalidCredentialsException(_twitterQuery.TwitterCredentials);
            }

            _currentHttpClient = GetHttpClient(_twitterQuery);
            _currentStreamReader = GetStreamReader(_currentHttpClient, _twitterQuery).Result;

            int numberOfRepeatedFailures = 0;

            while (StreamState != StreamState.Stop)
            {
                if (StreamState == StreamState.Pause)
                {
                    using (EventWaitHandle tmpEvent = new ManualResetEvent(false))
                    {
                        tmpEvent.WaitOne(TimeSpan.FromMilliseconds(STREAM_RESUME_DELAY));
                    }

                    continue;
                }

                try
                {
                    var json = GetJsonResponseFromReader(_currentStreamReader, _twitterQuery);

                    var isJsonResponseValid = json.IsMatchingJsonFormat();
                    if (!isJsonResponseValid)
                    {
                        if (json == string.Empty)
                        {
                            continue;
                        }

                        if (json != null)
                        {
                            throw new WebException(json);
                        }

                        if (TryHandleInvalidResponse(numberOfRepeatedFailures))
                        {
                            ++numberOfRepeatedFailures;
                            continue;
                        }

                        throw new WebException("Stream cannot be read.");
                    }

                    numberOfRepeatedFailures = 0;

                    if (StreamState == StreamState.Resume && !_processObject(json))
                    {
                        SetStreamState(StreamState.Stop);
                        break;
                    }
                }
                catch (Exception ex)
                {
                    if (!ShouldContinueAfterHandlingException(ex))
                    {
                        SetStreamState(StreamState.Stop);
                        break;
                    }
                }
            }

            if (_currentStreamReader != null)
            {
                _currentStreamReader.Dispose();
            }

            if (_currentHttpClient != null)
            {
                _currentHttpClient.Dispose();
{'fraction_non_alphanumeric': 0.06233675779209472, 'fraction_numerical': 0.002350687793835974, 'mean_word_length': 1.6794961511546536, 'pattern_counts': {'":': 0, '<': 23, '<?xml version=': 0, '>': 23, 'https://': 2, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 1, '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': '2738762', 'n_tokens_mistral': 2755, 'n_tokens_neox': 2572, 'n_words': 530}