<commit_before>﻿using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Linq;
using System.Threading;
using System.Threading.Tasks;
using System.Threading.Tasks.Dataflow;

namespace Lime.Protocol.Network.Modules
{
    /// <summary>
    /// Defines a module that resend messages that doesn't have <see cref="Event.Received"/> receipts from the destination.
    /// </summary>
    public sealed class ResendMessagesChannelModule : IChannelModule<Message>, IChannelModule<Notification>
    {        
        private readonly int _resendMessageTryCount;
        private readonly TimeSpan _resendMessageInterval;
        private readonly bool _filterByDestination;
        private readonly ConcurrentDictionary<MessageIdDestination, SentMessage> _sentMessageDictionary;
        private readonly object _syncRoot = new object();        
        private readonly BufferBlock<SentMessage> _inputBlock;
        private readonly TransformBlock<SentMessage, SentMessage> _waitForRetryBlock;
        private readonly ActionBlock<SentMessage> _resendBlock;

        private IChannel _channel;
        private bool _unbindWhenClosed;
        private IDisposable _link;

        public ResendMessagesChannelModule(int resendMessageTryCount, TimeSpan resendMessageInterval, bool filterByDestination = false)
        {            
            if (resendMessageTryCount <= 0) throw new ArgumentOutOfRangeException(nameof(resendMessageTryCount));            
            _resendMessageTryCount = resendMessageTryCount;
            _resendMessageInterval = resendMessageInterval;
            _filterByDestination = filterByDestination;

            _sentMessageDictionary = new ConcurrentDictionary<MessageIdDestination, SentMessage>();            
            _inputBlock = new BufferBlock<SentMessage>();
            _waitForRetryBlock =  new TransformBlock<SentMessage, SentMessage>(
                m => WaitForRetryAsync(m),
                new ExecutionDataflowBlockOptions() { BoundedCapacity = DataflowBlockOptions.Unbounded});
            _resendBlock = new ActionBlock<SentMessage>(
                ResendMessageAsync, 
                new ExecutionDataflowBlockOptions() { MaxDegreeOfParallelism = 1 });
            _inputBlock.LinkTo(_waitForRetryBlock);
        }

        private async Task<SentMessage> WaitForRetryAsync(SentMessage sentMessage)
        {                                        
            try
            {
                await Task.Delay(_resendMessageInterval, sentMessage.CancellationToken);
                return sentMessage;
            }            
            catch (OperationCanceledException) when (sentMessage.CancellationToken.IsCancellationRequested)
            {
                return null;
            }                       
        }

        private async Task ResendMessageAsync(SentMessage sentMessage)
        {
            if (sentMessage == null || sentMessage.CancellationToken.IsCancellationRequested) return;
            
            try
            {
                await _channel.SendMessageAsync(sentMessage.Message);
            }
            catch
            {
                Unbind();
            }            
        }

        public void OnStateChanged(SessionState state)
        {            
            if (state > SessionState.Established &&
                IsBound &&
                _unbindWhenClosed)
            {
                Unbind();
            }            
        }

        Task<Notification> IChannelModule<Notification>.OnReceivingAsync(Notification envelope, CancellationToken cancellationToken)
        {            
            if (envelope.Event == Event.Received || envelope.Event == Event.Failed)
            {                
                SentMessage sentMessage;
                if (_sentMessageDictionary.TryRemove(Create(envelope), out sentMessage))
                {
                    sentMessage.Dispose();
                }
            }

            return envelope.AsCompletedTask();
        }

        Task<Notification> IChannelModule<Notification>.OnSendingAsync(Notification envelope, CancellationToken cancellationToken)
        {
            return envelope.AsCompletedTask();
        }

        Task<Message> IChannelModule<Message>.OnReceivingAsync(Message envelope, CancellationToken cancellationToken)
        {
            return envelope.AsCompletedTask();
        }

        async Task<Message> IChannelModule<Message>.OnSendingAsync(Message envelope, CancellationToken cancellationToken)
        {
            if (envelope.Id != Guid.Empty)
            {
                var key = Create(envelope);
                SentMessage sentMessage;
                if (_sentMessageDictionary.TryGetValue(key, out sentMessage))
                {
                    sentMessage.IncrementResentCount();
                }
                else
                {
                    sentMessage = new SentMessage(envelope);
                    _sentMessageDictionary.TryAdd(Create(envelope), sentMessage);
                }

                if (sentMessage.ResentCount <= _resendMessageTryCount)
                {
                    await _inputBlock.SendAsync(sentMessage, cancellationToken);
                }
                else
                {
                    _sentMessageDictionary.TryRemove(key, out sentMessage);
                }                                
            }

            return envelope;
        }

        public bool IsBound => _channel != null;
<commit_msg>Reuse of created key to add the sentmessage dictionary<commit_after>﻿using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Linq;
using System.Threading;
using System.Threading.Tasks;
using System.Threading.Tasks.Dataflow;

namespace Lime.Protocol.Network.Modules
{
    /// <summary>
    /// Defines a module that resend messages that doesn't have <see cref="Event.Received"/> receipts from the destination.
    /// </summary>
    public sealed class ResendMessagesChannelModule : IChannelModule<Message>, IChannelModule<Notification>
    {        
        private readonly int _resendMessageTryCount;
        private readonly TimeSpan _resendMessageInterval;
        private readonly bool _filterByDestination;
        private readonly ConcurrentDictionary<MessageIdDestination, SentMessage> _sentMessageDictionary;
        private readonly object _syncRoot = new object();        
        private readonly BufferBlock<SentMessage> _inputBlock;
        private readonly TransformBlock<SentMessage, SentMessage> _waitForRetryBlock;
        private readonly ActionBlock<SentMessage> _resendBlock;

        private IChannel _channel;
        private bool _unbindWhenClosed;
        private IDisposable _link;

        public ResendMessagesChannelModule(int resendMessageTryCount, TimeSpan resendMessageInterval, bool filterByDestination = false)
        {            
            if (resendMessageTryCount <= 0) throw new ArgumentOutOfRangeException(nameof(resendMessageTryCount));            
            _resendMessageTryCount = resendMessageTryCount;
            _resendMessageInterval = resendMessageInterval;
            _filterByDestination = filterByDestination;

            _sentMessageDictionary = new ConcurrentDictionary<MessageIdDestination, SentMessage>();            
            _inputBlock = new BufferBlock<SentMessage>();
            _waitForRetryBlock =  new TransformBlock<SentMessage, SentMessage>(
                m => WaitForRetryAsync(m),
                new ExecutionDataflowBlockOptions() { BoundedCapacity = DataflowBlockOptions.Unbounded});
            _resendBlock = new ActionBlock<SentMessage>(
                ResendMessageAsync, 
                new ExecutionDataflowBlockOptions() { MaxDegreeOfParallelism = 1 });
            _inputBlock.LinkTo(_waitForRetryBlock);
        }

        private async Task<SentMessage> WaitForRetryAsync(SentMessage sentMessage)
        {                                        
            try
            {
                await Task.Delay(_resendMessageInterval, sentMessage.CancellationToken);
                return sentMessage;
            }            
            catch (OperationCanceledException) when (sentMessage.CancellationToken.IsCancellationRequested)
            {
                return null;
            }                       
        }

        private async Task ResendMessageAsync(SentMessage sentMessage)
        {
            if (sentMessage == null || sentMessage.CancellationToken.IsCancellationRequested) return;
            
            try
            {
                await _channel.SendMessageAsync(sentMessage.Message);
            }
            catch
            {
                Unbind();
            }            
        }

        public void OnStateChanged(SessionState state)
        {            
            if (state > SessionState.Established &&
                IsBound &&
                _unbindWhenClosed)
            {
                Unbind();
            }            
        }

        Task<Notification> IChannelModule<Notification>.OnReceivingAsync(Notification envelope, CancellationToken cancellationToken)
        {            
            if (envelope.Event == Event.Received || envelope.Event == Event.Failed)
            {                
                SentMessage sentMessage;
                if (_sentMessageDictionary.TryRemove(Create(envelope), out sentMessage))
                {
                    sentMessage.Dispose();
                }
            }

            return envelope.AsCompletedTask();
        }

        Task<Notification> IChannelModule<Notification>.OnSendingAsync(Notification envelope, CancellationToken cancellationToken)
        {
            return envelope.AsCompletedTask();
        }

        Task<Message> IChannelModule<Message>.OnReceivingAsync(Message envelope, CancellationToken cancellationToken)
        {
            return envelope.AsCompletedTask();
        }

        async Task<Message> IChannelModule<Message>.OnSendingAsync(Message envelope, CancellationToken cancellationToken)
        {
            if (envelope.Id != Guid.Empty)
            {
                var key = Create(envelope);
                SentMessage sentMessage;
                if (_sentMessageDictionary.TryGetValue(key, out sentMessage))
                {
                    sentMessage.IncrementResentCount();
                }
                else
                {
                    sentMessage = new SentMessage(envelope);
                    _sentMessageDictionary.TryAdd(key, sentMessage);
                }

                if (sentMessage.ResentCount <= _resendMessageTryCount)
                {
                    await _inputBlock.SendAsync(sentMessage, cancellationToken);
                }
                else
                {
                    _sentMessageDictionary.TryRemove(key, out sentMessage);
                }                                
            }

            return envelope;
        }

        public bool IsBound => _channel != null;

{'fraction_non_alphanumeric': 0.0645045045045045, 'fraction_numerical': 0.00036036036036036037, 'mean_word_length': 1.821092757306226, 'pattern_counts': {'":': 0, '<': 51, '<?xml version=': 0, '>': 53, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 8, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '3871859', 'n_tokens_mistral': 2528, 'n_tokens_neox': 2445, 'n_words': 502}