<commit_before>﻿using System;
using System.Collections.Generic;
using System.Diagnostics;
using System.Linq;
using System.Net;
using System.Threading;
using System.Threading.Tasks;
using Amazon;
using Amazon.Runtime;
using Amazon.SQS;
using Amazon.SQS.Model;

namespace RockLib.Messaging.SQS
{
    /// <summary>
    /// An implementation of <see cref="IReceiver"/> that receives messages from SQS.
    /// </summary>
    public class SQSQueueReceiver : Receiver
    {
        private readonly IAmazonSQS _sqs;
        private readonly Lazy<Task> _receiveMessagesTask;

        private bool _stopped;

        private const int _defaultMaxMessages = 3;
        private const int _maxAcknowledgeAttempts = 3;
        private const int _maxReceiveAttempts = 3;

        /// <summary>
        /// Initializes a new instance of the <see cref="SQSQueueReceiver"/> class.
        /// Uses a default implementation of the <see cref="AmazonSQSClient"/> to
        /// communicate with SQS.
        /// </summary>
        /// <param name="name">The name of the receiver.</param>
        /// <param name="queueUrl">The url of the SQS queue.</param>
        /// <param name="region">The region of the SQS queue.</param>
        /// <param name="maxMessages">
        /// The maximum number of messages to return with each call to the SQS endpoint.
        /// Amazon SQS never returns more messages than this value (however, fewer messages
        /// might be returned). Valid values are 1 to 10.
        /// </param>
        /// <param name="autoAcknowledge">
        /// Whether messages will be automatically acknowledged after any event handlers execute.
        /// </param>
        public SQSQueueReceiver(string name,
            string queueUrl,
            string region = null,
            int maxMessages = _defaultMaxMessages,
            bool autoAcknowledge = true)
            : this(region == null ? new AmazonSQSClient() : new AmazonSQSClient(RegionEndpoint.GetBySystemName(region)), name, queueUrl, maxMessages, autoAcknowledge)
        {
        }

        /// <summary>
        /// Initializes a new instance of the <see cref="SQSQueueReceiver"/> class.
        /// </summary>
        /// <param name="sqs">An object that communicates with SQS.</param>
        /// <param name="name">The name of the receiver.</param>
        /// <param name="queueUrl">The url of the SQS queue.</param>
        /// <param name="maxMessages">
        /// The maximum number of messages to return with each call to the SQS endpoint.
        /// Amazon SQS never returns more messages than this value (however, fewer messages
        /// might be returned). Valid values are 1 to 10.
        /// </param>
        /// <param name="autoAcknowledge">
        /// Whether messages will be automatically acknowledged after any event handlers execute.
        /// </param>
        public SQSQueueReceiver(IAmazonSQS sqs,
            string name,
            string queueUrl,
            int maxMessages = _defaultMaxMessages,
            bool autoAcknowledge = true)
            : base(name)
        {
            if (maxMessages < 1 || maxMessages > 10)
                throw new ArgumentOutOfRangeException(nameof(maxMessages), "Value must be from 1 to 10, inclusive.");

            _sqs = sqs ?? throw new ArgumentNullException(nameof(sqs));
            QueueUrl = queueUrl ?? throw new ArgumentNullException(nameof(queueUrl));
            MaxMessages = maxMessages;
            AutoAcknwoledge = autoAcknowledge;

            _receiveMessagesTask = new Lazy<Task>(ReceiveMessages);
        }

        /// <summary>
        /// Gets the url of the SQS queue.
        /// </summary>
        public string QueueUrl { get; }

        /// <summary>
        /// Gets the maximum number of messages to return with each call to the SQS endpoint.
        /// Amazon SQS never returns more messages than this value (however, fewer messages
        /// might be returned). Valid values are 1 to 10.
        /// </summary>
        public int MaxMessages { get; }

        /// <summary>
        /// Gets a value indicating whether messages will be automatically acknowledged after
        /// any event handlers execute.
        /// </summary>
        public bool AutoAcknwoledge { get; }

        /// <summary>
        /// Starts the polling background thread that listens for messages.
        /// </summary>
        protected override void Start()
        {
            if (!_receiveMessagesTask.IsValueCreated && !_stopped)
            {
                var dummy = _receiveMessagesTask.Value;
            }
        }

        private async Task ReceiveMessages()
        {
            await Task.Yield();

            bool? connected = null;

            while (!_stopped)
            {
                var receiveMessageRequest = new ReceiveMessageRequest
                {
                    MaxNumberOfMessages = MaxMessages,
                    QueueUrl = QueueUrl,
                    MessageAttributeNames = new List<string> { "*" }
                };

                ReceiveMessageResponse response = null;
                Exception exception = null;

                for (int i = 0; i < _maxReceiveAttempts; i++)
                {
                    try
                    {
                        response = await _sqs.ReceiveMessageAsync(receiveMessageRequest).ConfigureAwait(false);
<commit_msg>Add WaitTimeSeconds setting to SQSQueueReceiver<commit_after>﻿using System;
using System.Collections.Generic;
using System.Diagnostics;
using System.Linq;
using System.Net;
using System.Threading;
using System.Threading.Tasks;
using Amazon;
using Amazon.Runtime;
using Amazon.SQS;
using Amazon.SQS.Model;

namespace RockLib.Messaging.SQS
{
    /// <summary>
    /// An implementation of <see cref="IReceiver"/> that receives messages from SQS.
    /// </summary>
    public class SQSQueueReceiver : Receiver
    {
        private readonly IAmazonSQS _sqs;
        private readonly Lazy<Task> _receiveMessagesTask;

        private bool _stopped;

        private const int _defaultMaxMessages = 3;
        private const int _maxAcknowledgeAttempts = 3;
        private const int _maxReceiveAttempts = 3;

        /// <summary>
        /// Initializes a new instance of the <see cref="SQSQueueReceiver"/> class.
        /// Uses a default implementation of the <see cref="AmazonSQSClient"/> to
        /// communicate with SQS.
        /// </summary>
        /// <param name="name">The name of the receiver.</param>
        /// <param name="queueUrl">The url of the SQS queue.</param>
        /// <param name="region">The region of the SQS queue.</param>
        /// <param name="maxMessages">
        /// The maximum number of messages to return with each call to the SQS endpoint.
        /// Amazon SQS never returns more messages than this value (however, fewer messages
        /// might be returned). Valid values are 1 to 10.
        /// </param>
        /// <param name="autoAcknowledge">
        /// Whether messages will be automatically acknowledged after the message handler executes.
        /// </param>
        /// <param name="waitTimeSeconds">
        /// The duration (in seconds) for which calls to ReceiveMessage wait for a message
        /// to arrive in the queue before returning. If a message is available, the call returns
        /// sooner than WaitTimeSeconds. If no messages are available and the wait time expires,
        /// the call returns successfully with an empty list of messages.
        /// </param>
        public SQSQueueReceiver(string name,
            string queueUrl,
            string region = null,
            int maxMessages = _defaultMaxMessages,
            bool autoAcknowledge = true,
            int waitTimeSeconds = 0)
            : this(region == null ? new AmazonSQSClient() : new AmazonSQSClient(RegionEndpoint.GetBySystemName(region)), name, queueUrl, maxMessages, autoAcknowledge, waitTimeSeconds)
        {
        }

        /// <summary>
        /// Initializes a new instance of the <see cref="SQSQueueReceiver"/> class.
        /// </summary>
        /// <param name="sqs">An object that communicates with SQS.</param>
        /// <param name="name">The name of the receiver.</param>
        /// <param name="queueUrl">The url of the SQS queue.</param>
        /// <param name="maxMessages">
        /// The maximum number of messages to return with each call to the SQS endpoint.
        /// Amazon SQS never returns more messages than this value (however, fewer messages
        /// might be returned). Valid values are 1 to 10.
        /// </param>
        /// <param name="autoAcknowledge">
        /// Whether messages will be automatically acknowledged after the message handler executes.
        /// </param>
        /// <param name="waitTimeSeconds">
        /// The duration (in seconds) for which calls to ReceiveMessage wait for a message
        /// to arrive in the queue before returning. If a message is available, the call returns
        /// sooner than WaitTimeSeconds. If no messages are available and the wait time expires,
        /// the call returns successfully with an empty list of messages.
        /// </param>
        public SQSQueueReceiver(IAmazonSQS sqs,
            string name,
            string queueUrl,
            int maxMessages = _defaultMaxMessages,
            bool autoAcknowledge = true,
            int waitTimeSeconds = 0)
            : base(name)
        {
            if (maxMessages < 1 || maxMessages > 10)
                throw new ArgumentOutOfRangeException(nameof(maxMessages), "Value must be from 1 to 10, inclusive.");
            if (waitTimeSeconds < 0)
                throw new ArgumentOutOfRangeException(nameof(waitTimeSeconds), "Value cannot be negative.");

            _sqs = sqs ?? throw new ArgumentNullException(nameof(sqs));
            QueueUrl = queueUrl ?? throw new ArgumentNullException(nameof(queueUrl));
            MaxMessages = maxMessages;
            AutoAcknwoledge = autoAcknowledge;
            WaitTimeSeconds = waitTimeSeconds;

            _receiveMessagesTask = new Lazy<Task>(ReceiveMessages);
        }

        /// <summary>
        /// Gets the url of the SQS queue.
        /// </summary>
        public string QueueUrl { get; }

        /// <summary>
        /// Gets the maximum number of messages to return with each call to the SQS endpoint.
        /// Amazon SQS never returns more messages than this value (however, fewer messages
        /// might be returned). Valid values are 1 to 10.
        /// </summary>
        public int MaxMessages { get; }

        /// <summary>
        /// Gets a value indicating whether messages will be automatically acknowledged after
        /// the message handler executes.
        /// </summary>
        public bool AutoAcknwoledge { get; }

        /// <summary>
        /// Gets the duration (in seconds) for which calls to ReceiveMessage wait for a message
        /// to arrive in the queue before returning. If a message is available, the call returns
        /// sooner than WaitTimeSeconds. If no messages are available and the wait time expires,
        /// the call returns successfully with an empty list of messages.
        /// </summary>
        public int WaitTimeSeconds { get; }

        /// <summary>
        /// Starts the polling background thread that listens for messages.
        /// </summary>
        protected override void Start()
        {
            if (!_receiveMessagesTask.IsValueCreated && !_stopped)
            {
                var dummy = _receiveMessagesTask.Value;
            }
        }

        private async Task ReceiveMessages()
        {
            await Task.Yield();

            bool? connected = null;

            while (!_stopped)
            {
                var receiveMessageRequest = new ReceiveMessageRequest
                {
                    MaxNumberOfMessages = MaxMessages,
                    QueueUrl = QueueUrl,
                    MessageAttributeNames = new List<string> { "*" },
                    WaitTimeSeconds = WaitTimeSeconds
                };

                ReceiveMessageResponse response = null;
                Exception exception = null;

                for (int i = 0; i < _maxReceiveAttempts; i++)
                {
                    try
                    {
                        response = await _sqs.ReceiveMessageAsync(receiveMessageRequest).ConfigureAwait(false);

{'fraction_non_alphanumeric': 0.09307550349033138, 'fraction_numerical': 0.0032897376233651607, 'mean_word_length': 1.925134944848627, 'pattern_counts': {'":': 0, '<': 96, '<?xml version=': 0, '>': 93, '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': '2637234', 'n_tokens_mistral': 3156, 'n_tokens_neox': 2951, 'n_words': 1094}