<commit_before>﻿// --------------------------------------------------------------------------------------------------------------------
// <copyright file="PublisherCallbackChannelImpl.cs" company="The original author or authors.">
//   Copyright 2002-2012 the original author or authors.
//   
//   Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with
//   the License. You may obtain a copy of the License at
//   
//   http://www.apache.org/licenses/LICENSE-2.0
//   
//   Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on
//   an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the
//   specific language governing permissions and limitations under the License.
// </copyright>
// --------------------------------------------------------------------------------------------------------------------

#region Using Directives
using System;
using System.Collections;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Linq;
using System.Runtime.CompilerServices;
using Common.Logging;
using RabbitMQ.Client;
using RabbitMQ.Client.Events;
using RabbitMQ.Client.Impl;
using Spring.Util;
#endregion

namespace Spring.Messaging.Amqp.Rabbit.Support
{
    /// <summary>
    /// Channel wrapper to allow a single listener able to handle confirms from multiple channels.
    /// </summary>
    /// <author><NAME></author>
    /// <author><NAME></author>
    public class PublisherCallbackChannelImpl : IPublisherCallbackChannel // , BasicAckEventHandler, BasicNackEventHandler
    {
        private readonly ILog logger = LogManager.GetCurrentClassLogger();

        private readonly IModel channelDelegate;

        private readonly ConcurrentDictionary<string, IPublisherCallbackChannelListener> listeners = new ConcurrentDictionary<string, IPublisherCallbackChannelListener>();

        private readonly SortedDictionary<IPublisherCallbackChannelListener, SortedDictionary<long, PendingConfirm>> pendingConfirms = new SortedDictionary<IPublisherCallbackChannelListener, SortedDictionary<long, PendingConfirm>>();

        private readonly ConcurrentDictionary<long, IPublisherCallbackChannelListener> listenerForSeq = new ConcurrentDictionary<long, IPublisherCallbackChannelListener>();

        /// <summary>Initializes a new instance of the <see cref="PublisherCallbackChannelImpl"/> class.</summary>
        /// <param name="channelDelegate">The channel delegate.</param>
        public PublisherCallbackChannelImpl(IModel channelDelegate) { this.channelDelegate = channelDelegate; }

        #region Delegate Methods

        /// <summary>The add shutdown listener.</summary>
        /// <param name="listener">The listener.</param>
        public void AddShutdownListener(ModelShutdownEventHandler listener) { this.channelDelegate.ModelShutdown += listener; }

        /// <summary>The remove shutdown listener.</summary>
        /// <param name="listener">The listener.</param>
        public void RemoveShutdownListener(ModelShutdownEventHandler listener) { this.channelDelegate.ModelShutdown -= listener; }

        /// <summary>Gets the close reason.</summary>
        public ShutdownEventArgs CloseReason { get { return this.channelDelegate.CloseReason; } }

        /// <summary>The notify listeners.</summary>
        public void NotifyListeners() { ((ModelBase)this.channelDelegate).m_session.Notify(); }

        /// <summary>Gets a value indicating whether is open.</summary>
        public bool IsOpen { get { return this.channelDelegate.IsOpen; } }

        /// <summary>Gets the channel number.</summary>
        public int ChannelNumber { get { return ((ModelBase)this.channelDelegate).m_session.ChannelNumber; } }

        /// <summary>Gets the connection.</summary>
        public IConnection Connection { get { return ((ModelBase)this.channelDelegate).m_session.Connection; } }

        /// <summary>The close.</summary>
        /// <param name="closeCode">The close code.</param>
        /// <param name="closeMessage">The close message.</param>
        public void Close(ushort closeCode, string closeMessage) { this.channelDelegate.Close(closeCode, closeMessage); }

        /// <summary>The channel flow.</summary>
        /// <param name="active">The active.</param>
        public void ChannelFlow(bool active) { this.channelDelegate.ChannelFlow(active); }

        // public ChannelFlowOk Flow() { return this.channelDelegate.ChannelFlow()); }

        /// <summary>The abort.</summary>
        public void Abort() { this.channelDelegate.Abort(); }

        /// <summary>The abort.</summary>
        /// <param name="closeCode">The close code.</param>
        /// <param name="closeMessage">The close message.</param>
        public void Abort(ushort closeCode, string closeMessage) { this.channelDelegate.Abort(closeCode, closeMessage); }

        private readonly List<FlowControlEventHandler> flowListeners = new List<FlowControlEventHandler>();

        /// <summary>The flow control.</summary>
        public event FlowControlEventHandler FlowControl;

        /// <summary>The add flow listener.</summary>
        /// <param name="listener">The listener.</param>
        public void AddFlowListener(FlowControlEventHandler listener)
        {
            this.flowListeners.Add(listener);
            this.channelDelegate.FlowControl += listener;
        }

        /// <summary>The remove flow listener.</summary>
        /// <param name="listener">The listener.</param>
        public void RemoveFlowListener(FlowControlEventHandler listener)
        {
            if (this.flowListeners.Contains(listener))
            {
                this.flowListeners.Remove(listener);
                this.channelDelegate.FlowControl -= listener;
            }
        }

        /// <summary>The clear flow listeners.</summary>
        public void ClearFlowListeners()
        {
            foreach (var listener in this.flowListeners)
            {
                this.channelDelegate.FlowControl -= listener;
            }
        }

        /// <summary>Gets or sets the default consumer.</summary>
        public IBasicConsumer DefaultConsumer { get { return this.channelDelegate.DefaultConsumer; } set { this.channelDelegate.DefaultConsumer = value; } }

        /// <summary>The basic qos.</summary>
        /// <param name="prefetchSize">The prefetch size.</param>
        /// <param name="prefetchCount">The prefetch count.</param>
        /// <param name="global">The global.</param>
        public void BasicQos(uint prefetchSize, ushort prefetchCount, bool global) { this.channelDelegate.BasicQos(prefetchSize, prefetchCount, global); }

        /// <summary>The basic qos.</summary>
        /// <param name="prefetchCount">The prefetch count.</param>
        public void BasicQos(int prefetchCount) { this.channelDelegate.BasicQos(0, (ushort)prefetchCount, false); }

        /// <summary>The basic publish.</summary>
        /// <param name="exchange">The exchange.</param>
        /// <param name="routingKey">The routing key.</param>
        /// <param name="props">The props.</param>
        /// <param name="body">The body.</param>
        public void BasicPublish(string exchange, string routingKey, IBasicProperties props, byte[] body) { this.channelDelegate.BasicPublish(exchange, routingKey, props, body); }

        /// <summary>The basic publish.</summary>
        /// <param name="exchange">The exchange.</param>
        /// <param name="routingKey">The routing key.</param>
        /// <param name="mandatory">The mandatory.</param>
        /// <param name="basicProperties">The basic properties.</param>
        /// <param name="body">The body.</param>
        /// <exception cref="NotImplementedException">This is not implemented.</exception>
        public void BasicPublish(string exchange, string routingKey, bool mandatory, IBasicProperties basicProperties, byte[] body) { this.channelDelegate.BasicPublish(exchange, routingKey, mandatory, basicProperties, body); }

        /// <summary>The basic publish.</summary>
        /// <param name="exchange">The exchange.</param>
        /// <param name="routingKey">The routing key.</param>
        /// <param name="mandatory">The mandatory.</param>
        /// <param name="immediate">The immediate.</param>
        /// <param name="props">The props.</param>
        /// <param name="body">The body.</param>
        public void BasicPublish(string exchange, string routingKey, bool mandatory, bool immediate, IBasicProperties props, byte[] body) { this.channelDelegate.BasicPublish(exchange, routingKey, mandatory, immediate, props, body); }

        /// <summary>The exchange declare.</summary>
        /// <param name="exchange">The exchange.</param>
        /// <param name="type">The type.</param>
        public void ExchangeDeclare(string exchange, string type) { this.channelDelegate.ExchangeDeclare(exchange, type); }

        /// <summary>The exchange declare.</summary>
        /// <param name="exchange">The exchange.</param>
        /// <param name="type">The type.</param>
        /// <param name="durable">The durable.</param>
        public void ExchangeDeclare(string exchange, string type, bool durable) { this.channelDelegate.ExchangeDeclare(exchange, type, durable); }

        /// <summary>The exchange declare.</summary>
        /// <param name="exchange">The exchange.</param>
        /// <param name="type">The type.</param>
        /// <param name="durable">The durable.</param>
        /// <param name="autoDelete">The auto delete.</param>
        /// <param name="arguments">The arguments.</param>
        public void ExchangeDeclare(string exchange, string type, bool durable, bool autoDelete, IDictionary arguments) { this.channelDelegate.ExchangeDeclare(exchange, type, durable, autoDelete, arguments); }

        /// <summary>The exchange declare.</summary><commit_msg>Removing invalid method<commit_after>﻿// --------------------------------------------------------------------------------------------------------------------
// <copyright file="PublisherCallbackChannelImpl.cs" company="The original author or authors.">
//   Copyright 2002-2012 the original author or authors.
//   
//   Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with
//   the License. You may obtain a copy of the License at
//   
//   http://www.apache.org/licenses/LICENSE-2.0
//   
//   Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on
//   an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the
//   specific language governing permissions and limitations under the License.
// </copyright>
// --------------------------------------------------------------------------------------------------------------------

#region Using Directives
using System;
using System.Collections;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Linq;
using System.Runtime.CompilerServices;
using Common.Logging;
using RabbitMQ.Client;
using RabbitMQ.Client.Events;
using RabbitMQ.Client.Impl;
using Spring.Util;
#endregion

namespace Spring.Messaging.Amqp.Rabbit.Support
{
    /// <summary>
    /// Channel wrapper to allow a single listener able to handle confirms from multiple channels.
    /// </summary>
    /// <author><NAME></author>
    /// <author><NAME></author>
    public class PublisherCallbackChannelImpl : IPublisherCallbackChannel // , BasicAckEventHandler, BasicNackEventHandler
    {
        private readonly ILog logger = LogManager.GetCurrentClassLogger();

        private readonly IModel channelDelegate;

        private readonly ConcurrentDictionary<string, IPublisherCallbackChannelListener> listeners = new ConcurrentDictionary<string, IPublisherCallbackChannelListener>();

        private readonly SortedDictionary<IPublisherCallbackChannelListener, SortedDictionary<long, PendingConfirm>> pendingConfirms = new SortedDictionary<IPublisherCallbackChannelListener, SortedDictionary<long, PendingConfirm>>();

        private readonly ConcurrentDictionary<long, IPublisherCallbackChannelListener> listenerForSeq = new ConcurrentDictionary<long, IPublisherCallbackChannelListener>();

        /// <summary>Initializes a new instance of the <see cref="PublisherCallbackChannelImpl"/> class.</summary>
        /// <param name="channelDelegate">The channel delegate.</param>
        public PublisherCallbackChannelImpl(IModel channelDelegate) { this.channelDelegate = channelDelegate; }

        #region Delegate Methods

        /// <summary>The add shutdown listener.</summary>
        /// <param name="listener">The listener.</param>
        public void AddShutdownListener(ModelShutdownEventHandler listener) { this.channelDelegate.ModelShutdown += listener; }

        /// <summary>The remove shutdown listener.</summary>
        /// <param name="listener">The listener.</param>
        public void RemoveShutdownListener(ModelShutdownEventHandler listener) { this.channelDelegate.ModelShutdown -= listener; }

        /// <summary>Gets the close reason.</summary>
        public ShutdownEventArgs CloseReason { get { return this.channelDelegate.CloseReason; } }

        /// <summary>The notify listeners.</summary>
        public void NotifyListeners() { ((ModelBase)this.channelDelegate).m_session.Notify(); }

        /// <summary>Gets a value indicating whether is open.</summary>
        public bool IsOpen { get { return this.channelDelegate.IsOpen; } }

        /// <summary>Gets the channel number.</summary>
        public int ChannelNumber { get { return ((ModelBase)this.channelDelegate).m_session.ChannelNumber; } }

        /// <summary>Gets the connection.</summary>
        public IConnection Connection { get { return ((ModelBase)this.channelDelegate).m_session.Connection; } }

        /// <summary>The close.</summary>
        /// <param name="closeCode">The close code.</param>
        /// <param name="closeMessage">The close message.</param>
        public void Close(ushort closeCode, string closeMessage) { this.channelDelegate.Close(closeCode, closeMessage); }

        /// <summary>The channel flow.</summary>
        /// <param name="active">The active.</param>
        public void ChannelFlow(bool active) { this.channelDelegate.ChannelFlow(active); }

        // public ChannelFlowOk Flow() { return this.channelDelegate.ChannelFlow()); }

        /// <summary>The abort.</summary>
        public void Abort() { this.channelDelegate.Abort(); }

        /// <summary>The abort.</summary>
        /// <param name="closeCode">The close code.</param>
        /// <param name="closeMessage">The close message.</param>
        public void Abort(ushort closeCode, string closeMessage) { this.channelDelegate.Abort(closeCode, closeMessage); }

        private readonly List<FlowControlEventHandler> flowListeners = new List<FlowControlEventHandler>();

        /// <summary>The flow control.</summary>
        public event FlowControlEventHandler FlowControl;

        /// <summary>The add flow listener.</summary>
        /// <param name="listener">The listener.</param>
        public void AddFlowListener(FlowControlEventHandler listener)
        {
            this.flowListeners.Add(listener);
            this.channelDelegate.FlowControl += listener;
        }

        /// <summary>The remove flow listener.</summary>
        /// <param name="listener">The listener.</param>
        public void RemoveFlowListener(FlowControlEventHandler listener)
        {
            if (this.flowListeners.Contains(listener))
            {
                this.flowListeners.Remove(listener);
                this.channelDelegate.FlowControl -= listener;
            }
        }

        /// <summary>The clear flow listeners.</summary>
        public void ClearFlowListeners()
        {
            foreach (var listener in this.flowListeners)
            {
                this.channelDelegate.FlowControl -= listener;
            }
        }

        /// <summary>Gets or sets the default consumer.</summary>
        public IBasicConsumer DefaultConsumer { get { return this.channelDelegate.DefaultConsumer; } set { this.channelDelegate.DefaultConsumer = value; } }

        /// <summary>The basic qos.</summary>
        /// <param name="prefetchSize">The prefetch size.</param>
        /// <param name="prefetchCount">The prefetch count.</param>
        /// <param name="global">The global.</param>
        public void BasicQos(uint prefetchSize, ushort prefetchCount, bool global) { this.channelDelegate.BasicQos(prefetchSize, prefetchCount, global); }

        /// <summary>The basic qos.</summary>
        /// <param name="prefetchCount">The prefetch count.</param>
        public void BasicQos(int prefetchCount) { this.channelDelegate.BasicQos(0, (ushort)prefetchCount, false); }

        /// <summary>The basic publish.</summary>
        /// <param name="exchange">The exchange.</param>
        /// <param name="routingKey">The routing key.</param>
        /// <param name="props">The props.</param>
        /// <param name="body">The body.</param>
        public void BasicPublish(string exchange, string routingKey, IBasicProperties props, byte[] body) { this.channelDelegate.BasicPublish(exchange, routingKey, props, body); }

        /// <summary>The basic publish.</summary>
        /// <param name="exchange">The exchange.</param>
        /// <param name="routingKey">The routing key.</param>
        /// <param name="mandatory">The mandatory.</param>
        /// <param name="immediate">The immediate.</param>
        /// <param name="props">The props.</param>
        /// <param name="body">The body.</param>
        public void BasicPublish(string exchange, string routingKey, bool mandatory, bool immediate, IBasicProperties props, byte[] body) { this.channelDelegate.BasicPublish(exchange, routingKey, mandatory, immediate, props, body); }

        /// <summary>The exchange declare.</summary>
        /// <param name="exchange">The exchange.</param>
        /// <param name="type">The type.</param>
        public void ExchangeDeclare(string exchange, string type) { this.channelDelegate.ExchangeDeclare(exchange, type); }

        /// <summary>The exchange declare.</summary>
        /// <param name="exchange">The exchange.</param>
        /// <param name="type">The type.</param>
        /// <param name="durable">The durable.</param>
        public void ExchangeDeclare(string exchange, string type, bool durable) { this.channelDelegate.ExchangeDeclare(exchange, type, durable); }

        /// <summary>The exchange declare.</summary>
        /// <param name="exchange">The exchange.</param>
        /// <param name="type">The type.</param>
        /// <param name="durable">The durable.</param>
        /// <param name="autoDelete">The auto delete.</param>
        /// <param name="arguments">The arguments.</param>
        public void ExchangeDeclare(string exchange, string type, bool durable, bool autoDelete, IDictionary arguments) { this.channelDelegate.ExchangeDeclare(exchange, type, durable, autoDelete, arguments); }

        /// <summary>The exchange declare.</summary>
{'fraction_non_alphanumeric': 0.1459665772642872, 'fraction_numerical': 0.0013410356921807305, 'mean_word_length': 2.8500794281175534, 'pattern_counts': {'":': 0, '<': 295, '<?xml version=': 0, '>': 295, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, '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': '6818027', 'n_tokens_mistral': 4704, 'n_tokens_neox': 4578, 'n_words': 1237}