<commit_before>﻿using System;
using System.Collections.Generic;
using System.Collections.ObjectModel;
using System.Collections.Specialized;
using System.Diagnostics;
using System.Linq;
using System.Threading;
using System.Threading.Tasks;
using System.Threading.Tasks.Dataflow;
using Lime.Protocol.Listeners;
using Lime.Protocol.Network;

namespace Lime.Protocol.Client
{
    /// <summary>
    /// Implements a client channel that hold multiple connections with the server and distribute the load between these channels.
    /// </summary>
    /// <seealso cref="Lime.Protocol.Client.IOnDemandClientChannel" />
    /// <seealso cref="System.IDisposable" />
    public sealed class MultiplexerClientChannel : IOnDemandClientChannel, IDisposable
    {
        private readonly IOnDemandClientChannel[] _channels;
        private readonly IChannelListener[] _listeners;
        private readonly SemaphoreSlim _semaphore;
        private readonly BufferBlock<Envelope> _outputBufferBlock;
        private readonly ActionBlock<Envelope>[] _outputActionBlocks;
        private readonly BufferBlock<Message> _inputMessageBufferBlock;
        private readonly BufferBlock<Notification> _inputNotificationBufferBlock;
        private readonly BufferBlock<Command> _inputCommandBufferBlock;

        /// <summary>
        /// Initializes a new instance of the <see cref="MultiplexerClientChannel"/> class.
        /// </summary>
        /// <param name="builder">The channel builder.</param>
        /// <param name="count">The number of channels to create.</param>
        /// <param name="inputBoundedCapacity">The input buffer bounded capacity.</param>
        /// <param name="outputBoundedCapacity">The output buffer bounded capacity.</param>
        /// <exception cref="System.ArgumentNullException"></exception>
        /// <exception cref="System.ArgumentOutOfRangeException"></exception>
        public MultiplexerClientChannel(
            EstablishedClientChannelBuilder builder, 
            int count = 5,
            int inputBoundedCapacity = 1,
            int outputBoundedCapacity = 1)
        {
            if (builder == null) throw new ArgumentNullException(nameof(builder));
            if (count <= 0) throw new ArgumentOutOfRangeException(nameof(count));

            // Create observable collections to allow synchronization
            var channelCreatedHandlers = new ObservableCollection<Func<ChannelInformation, Task>>();
            var channelDiscardedHandlers = new ObservableCollection<Func<ChannelInformation, Task>>();
            var channelCreationFailedHandlers = new ObservableCollection<Func<FailedChannelInformation, Task<bool>>>();
            var channelOperationFailedHandlers = new ObservableCollection<Func<FailedChannelInformation, Task<bool>>>();
            ChannelCreatedHandlers = channelCreatedHandlers;
            ChannelDiscardedHandlers = channelDiscardedHandlers;
            ChannelCreationFailedHandlers = channelCreationFailedHandlers;
            ChannelOperationFailedHandlers = channelOperationFailedHandlers;

            var inputOptions = new DataflowBlockOptions()
            {
                BoundedCapacity = inputBoundedCapacity
            };

            // Global input buffers
            _inputMessageBufferBlock = new BufferBlock<Message>(inputOptions);
            _inputNotificationBufferBlock = new BufferBlock<Notification>(inputOptions);
            _inputCommandBufferBlock = new BufferBlock<Command>(inputOptions);

            // The global output buffer
            _outputBufferBlock = new BufferBlock<Envelope>(new DataflowBlockOptions()
            {
                BoundedCapacity = outputBoundedCapacity
            });
            // An output action block per channel
            _outputActionBlocks = new ActionBlock<Envelope>[count];

            _channels = new IOnDemandClientChannel[count];
            _listeners = new IChannelListener[count];

            for (var i = 0; i < _channels.Length; i++)
            {
                // Add an instance suffix to the builder
                var currentBuilder = builder
                    .ShallowCopy()
                    .WithInstance($"{builder.Instance}-{i}");
                var channel = new OnDemandClientChannel(currentBuilder);
                
                // Synchronize the handlers
                AttachCollection(channelCreatedHandlers, channel.ChannelCreatedHandlers);
                AttachCollection(channelDiscardedHandlers, channel.ChannelDiscardedHandlers);
                AttachCollection(channelCreationFailedHandlers, channel.ChannelCreationFailedHandlers);
                AttachCollection(channelOperationFailedHandlers, channel.ChannelOperationFailedHandlers);

                // Setup the listener for the channel
                _listeners[i] = new DataflowChannelListener(
                    _inputMessageBufferBlock,<commit_msg>Renamed bounded capacity variable on multiplexer channel<commit_after>﻿using System;
using System.Collections.Generic;
using System.Collections.ObjectModel;
using System.Collections.Specialized;
using System.Diagnostics;
using System.Linq;
using System.Threading;
using System.Threading.Tasks;
using System.Threading.Tasks.Dataflow;
using Lime.Protocol.Listeners;
using Lime.Protocol.Network;

namespace Lime.Protocol.Client
{
    /// <summary>
    /// Implements a client channel that hold multiple connections with the server and distribute the load between these channels.
    /// </summary>
    /// <seealso cref="Lime.Protocol.Client.IOnDemandClientChannel" />
    /// <seealso cref="System.IDisposable" />
    public sealed class MultiplexerClientChannel : IOnDemandClientChannel, IDisposable
    {
        private readonly IOnDemandClientChannel[] _channels;
        private readonly IChannelListener[] _listeners;
        private readonly SemaphoreSlim _semaphore;
        private readonly BufferBlock<Envelope> _outputBufferBlock;
        private readonly ActionBlock<Envelope>[] _outputActionBlocks;
        private readonly BufferBlock<Message> _inputMessageBufferBlock;
        private readonly BufferBlock<Notification> _inputNotificationBufferBlock;
        private readonly BufferBlock<Command> _inputCommandBufferBlock;

        /// <summary>
        /// Initializes a new instance of the <see cref="MultiplexerClientChannel"/> class.
        /// </summary>
        /// <param name="builder">The channel builder.</param>
        /// <param name="count">The number of channels to create.</param>
        /// <param name="inputBufferSize">The input buffer bounded capacity.</param>
        /// <param name="outputBufferSize">The output buffer bounded capacity.</param>
        /// <exception cref="System.ArgumentNullException"></exception>
        /// <exception cref="System.ArgumentOutOfRangeException"></exception>
        public MultiplexerClientChannel(
            EstablishedClientChannelBuilder builder, 
            int count = 5,
            int inputBufferSize = 1,
            int outputBufferSize = 1)
        {
            if (builder == null) throw new ArgumentNullException(nameof(builder));
            if (count <= 0) throw new ArgumentOutOfRangeException(nameof(count));

            // Create observable collections to allow synchronization
            var channelCreatedHandlers = new ObservableCollection<Func<ChannelInformation, Task>>();
            var channelDiscardedHandlers = new ObservableCollection<Func<ChannelInformation, Task>>();
            var channelCreationFailedHandlers = new ObservableCollection<Func<FailedChannelInformation, Task<bool>>>();
            var channelOperationFailedHandlers = new ObservableCollection<Func<FailedChannelInformation, Task<bool>>>();
            ChannelCreatedHandlers = channelCreatedHandlers;
            ChannelDiscardedHandlers = channelDiscardedHandlers;
            ChannelCreationFailedHandlers = channelCreationFailedHandlers;
            ChannelOperationFailedHandlers = channelOperationFailedHandlers;

            var inputOptions = new DataflowBlockOptions()
            {
                BoundedCapacity = inputBufferSize
            };

            // Global input buffers
            _inputMessageBufferBlock = new BufferBlock<Message>(inputOptions);
            _inputNotificationBufferBlock = new BufferBlock<Notification>(inputOptions);
            _inputCommandBufferBlock = new BufferBlock<Command>(inputOptions);

            // The global output buffer
            _outputBufferBlock = new BufferBlock<Envelope>(new DataflowBlockOptions()
            {
                BoundedCapacity = outputBufferSize
            });
            // An output action block per channel
            _outputActionBlocks = new ActionBlock<Envelope>[count];

            _channels = new IOnDemandClientChannel[count];
            _listeners = new IChannelListener[count];

            for (var i = 0; i < _channels.Length; i++)
            {
                // Add an instance suffix to the builder
                var currentBuilder = builder
                    .ShallowCopy()
                    .WithInstance($"{builder.Instance}-{i}");
                var channel = new OnDemandClientChannel(currentBuilder);
                
                // Synchronize the handlers
                AttachCollection(channelCreatedHandlers, channel.ChannelCreatedHandlers);
                AttachCollection(channelDiscardedHandlers, channel.ChannelDiscardedHandlers);
                AttachCollection(channelCreationFailedHandlers, channel.ChannelCreationFailedHandlers);
                AttachCollection(channelOperationFailedHandlers, channel.ChannelOperationFailedHandlers);

                // Setup the listener for the channel
                _listeners[i] = new DataflowChannelListener(
                    _inputMessageBufferBlock,
{'fraction_non_alphanumeric': 0.0789821882951654, 'fraction_numerical': 0.0010178117048346056, 'mean_word_length': 2.827814569536424, 'pattern_counts': {'":': 0, '<': 85, '<?xml version=': 0, '>': 81, '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': '2136720', 'n_tokens_mistral': 2245, 'n_tokens_neox': 2164, 'n_words': 545}