<commit_before>
//
// Juhta.NET, Copyright (c) 2017 <NAME>
//
// This source code may be used, modified and distributed under the terms of
// the MIT license. Please refer to the LICENSE.txt file for details.
//

using System;
using System.Threading;

namespace Juhta.Net.Common
{
    /// <summary>
    /// Defines an abstract base class for consumer threads. The class provides a thread-safe method for putting
    /// objects into consumption. Incoming objects will be consumed asynchronously with a single worker thread.
    /// </summary>
    /// <typeparam name="T">Specifies the type of objects to consume.</typeparam>
    public abstract class ConsumerThread<T>
    {
        #region Public Constructors

        /// <summary>
        /// Initializes a new instance.
        /// </summary>
        public ConsumerThread()
        {
            m_objectQueue = new ObjectQueue<T>();
        }

        #endregion

        #region Public Methods

        /// <summary>
        /// Consumes a specified object asynchronously.
        /// </summary>
        /// <param name="object">Specifies an object to consume.</param>
        /// <remarks>
        /// <para>This method returns immediately after the specified object has been put into an internal queue to
        /// wait for asynchronous consumption.</para>
        /// <para>Multiple threads can access this method concurrently.</para>
        /// </remarks>
        public void Consume(T @object)
        {
            if (m_workerThread == null)
                throw new InvalidOperationException(LibraryMessages.Error027.GetMessage());

            else if (!m_workerThread.IsAlive)
                throw new InvalidOperationException(LibraryMessages.Error028.GetMessage());

            m_objectQueue.PutObject(@object);
        }

        /// <summary>
        /// Starts this ConsumerThread instance.
        /// </summary>
        public void Start()
        {
            if (m_workerThread != null)
                throw new InvalidOperationException(CommonMessages.Error006.FormatMessage("Start", this.GetType()));

            m_workerThread = new Thread(new ThreadStart(WorkerThreadMain));

            m_workerThread.Start();
        }

        /// <summary>
        /// Stops this ConsumerThread instance.
        /// </summary>
        public void Stop()
        {
            if (m_workerThread == null)
                throw new InvalidOperationException(CommonMessages.Error006.FormatMessage("Stop", this.GetType()));

            else if (m_workerThread.ThreadState == ThreadState.Stopped)
                return;

            m_objectQueue.Close();

            m_workerThread.Join();
        }

        #endregion

        #region Protected Methods

        /// <summary>
        /// Consumes a specified object.
        /// </summary>
        /// <param name="object">Specifies an object to consume.</param>
        protected abstract void ConsumeObject(T @object);

        /// <summary>
        /// Performs necessary actions when an error occurs at the time an object is consumed.
        /// </summary><commit_msg>Operator nameof applied in the ConsumerThread class<commit_after>
//
// Juhta.NET, Copyright (c) 2017 <NAME>
//
// This source code may be used, modified and distributed under the terms of
// the MIT license. Please refer to the LICENSE.txt file for details.
//

using System;
using System.Threading;

namespace Juhta.Net.Common
{
    /// <summary>
    /// Defines an abstract base class for consumer threads. The class provides a thread-safe method for putting
    /// objects into consumption. Incoming objects will be consumed asynchronously with a single worker thread.
    /// </summary>
    /// <typeparam name="T">Specifies the type of objects to consume.</typeparam>
    public abstract class ConsumerThread<T>
    {
        #region Public Constructors

        /// <summary>
        /// Initializes a new instance.
        /// </summary>
        public ConsumerThread()
        {
            m_objectQueue = new ObjectQueue<T>();
        }

        #endregion

        #region Public Methods

        /// <summary>
        /// Consumes a specified object asynchronously.
        /// </summary>
        /// <param name="object">Specifies an object to consume.</param>
        /// <remarks>
        /// <para>This method returns immediately after the specified object has been put into an internal queue to
        /// wait for asynchronous consumption.</para>
        /// <para>Multiple threads can access this method concurrently.</para>
        /// </remarks>
        public void Consume(T @object)
        {
            if (m_workerThread == null)
                throw new InvalidOperationException(LibraryMessages.Error027.GetMessage());

            else if (!m_workerThread.IsAlive)
                throw new InvalidOperationException(LibraryMessages.Error028.GetMessage());

            m_objectQueue.PutObject(@object);
        }

        /// <summary>
        /// Starts this ConsumerThread instance.
        /// </summary>
        public void Start()
        {
            if (m_workerThread != null)
                throw new InvalidOperationException(CommonMessages.Error006.FormatMessage(nameof(Start), this.GetType()));

            m_workerThread = new Thread(new ThreadStart(WorkerThreadMain));

            m_workerThread.Start();
        }

        /// <summary>
        /// Stops this ConsumerThread instance.
        /// </summary>
        public void Stop()
        {
            if (m_workerThread == null)
                throw new InvalidOperationException(CommonMessages.Error006.FormatMessage(nameof(Stop), this.GetType()));

            else if (m_workerThread.ThreadState == ThreadState.Stopped)
                return;

            m_objectQueue.Close();

            m_workerThread.Join();
        }

        #endregion

        #region Protected Methods

        /// <summary>
        /// Consumes a specified object.
        /// </summary>
        /// <param name="object">Specifies an object to consume.</param>
        protected abstract void ConsumeObject(T @object);

        /// <summary>
        /// Performs necessary actions when an error occurs at the time an object is consumed.
        /// </summary>
{'fraction_non_alphanumeric': 0.10585225455708347, 'fraction_numerical': 0.005116725295810681, 'mean_word_length': 1.9799904716531682, 'pattern_counts': {'":': 0, '<': 61, '<?xml version=': 0, '>': 61, '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': '3297572', 'n_tokens_mistral': 1565, 'n_tokens_neox': 1477, 'n_words': 481}