<commit_before>﻿using System;
using System.Collections.Generic;
using System.Text;
using System.Threading;
using System.Threading.Tasks;
using NsqSharp.Core;
using NsqSharp.Utils;
using NsqSharp.Utils.Channels;
using NsqSharp.Utils.Loggers;

namespace NsqSharp
{
    // https://github.com/bitly/go-nsq/blob/master/producer.go

    /// <summary>
    /// IConn interface
    /// </summary>
    internal interface IConn
    {
        /// <summary>
        /// SetLogger assigns the logger to use as well as a level.
        ///
        /// The format parameter is expected to be a printf compatible string with
        /// a single {0} argument.  This is useful if you want to provide additional
        /// context to the log messages that the connection will print, the default
        /// is '({0})'.
        /// </summary>
        void SetLogger(ILogger l, string format);

        /// <summary>
        /// Connect dials and bootstraps the nsqd connection
        /// (including IDENTIFY) and returns the IdentifyResponse
        /// </summary>
        IdentifyResponse Connect();

        /// <summary>
        /// Close idempotently initiates connection close
        /// </summary>
        void Close();

        /// <summary>
        /// WriteCommand is a thread safe method to write a Command
        /// to this connection, and flush.
        /// </summary>
        void WriteCommand(Command command);
    }

    /// <summary>
    /// <para>Producer is a high-level type to publish to NSQ.</para>
    ///
    /// <para>A Producer instance is 1:1 with a destination nsqd
    /// and will lazily connect to that instance (and re-connect)
    /// when Publish commands are executed.</para>
    /// <seealso cref="Publish(string, string)"/>
    /// <seealso cref="Publish(string, byte[])"/>
    /// <seealso cref="Stop"/>
    /// </summary>
    public sealed partial class Producer
    {
        private static long _instCount;

        internal long _id;
        private readonly string _addr;
        private IConn _conn;
        private readonly Config _config;

        private readonly ILogger _logger;

        private readonly Chan<byte[]> _responseChan;
        private readonly Chan<byte[]> _errorChan;
        private Chan<int> _closeChan;

        private readonly Chan<ProducerResponse> _transactionChan;
        private readonly Queue<ProducerResponse> _transactions = new Queue<ProducerResponse>();
        private int _state;

        private int _concurrentProducers;
        private int _stopFlag;
        private readonly Chan<int> _exitChan;
        private readonly WaitGroup _wg = new WaitGroup();
        private readonly object _guard = new object();

        private readonly Func<Producer, IConn> _connFactory;
    }

    /// <summary>
    /// ProducerResponse is returned by the async publish methods
    /// to retrieve metadata about the command after the
    /// response is received.
    /// </summary>
    public class ProducerResponse
    {
        internal Command _cmd;
        internal Chan<ProducerResponse> _doneChan;

        /// <summary>
        /// the error (or nil) of the publish command
        /// </summary>
        public Exception Error { get; set; }

        /// <summary>
        /// the slice of variadic arguments passed to PublishAsync or MultiPublishAsync
        /// </summary>
        public object[] Args { get; set; }

        internal void finish()
        {
            if (_doneChan != null)
                _doneChan.Send(this);
        }
    }

    public sealed partial class Producer : IConnDelegate
    {
        /// <summary>
        /// Initializes a new instance of the Producer class.
        /// </summary>
        /// <param name="nsqdAddress">The nsqd address.</param>
        public Producer(string nsqdAddress)
            : this(nsqdAddress, new Config())
        {
        }

        /// <summary>
        /// Initializes a new instance of the Producer class.
        /// </summary>
        /// <param name="nsqdAddress">The nsqd address.</param>
        /// <param name="config">The config. After Config is passed in the values are
        /// no longer mutable (they are copied).</param>
        public Producer(string nsqdAddress, Config config)
            : this(nsqdAddress, new ConsoleLogger(LogLevel.Info), config, null)
        {
        }

        /// <summary>
        /// Initializes a new instance of the Producer class.
        /// </summary>
        /// <param name="nsqdAddress">The nsqd address.</param>
        /// <param name="logger">The logger.</param>
        public Producer(string nsqdAddress, ILogger logger)
            : this(nsqdAddress, logger, new Config(), null)
        {
        }

        /// <summary>
        /// Initializes a new instance of the Producer class.
        /// </summary>
        /// <param name="nsqdAddress">The nsqd address.</param>
        /// <param name="logger">The logger.</param>
        /// <param name="config">The config. Values are copied, changing the properties on <paramref name="config"/>
        /// after the constructor is called will have no effect on the <see cref="Producer"/>.</param>
        public Producer(string nsqdAddress, ILogger logger, Config config)
            : this(nsqdAddress, logger, config, null)
        {
        }

        private Producer(string addr, ILogger logger, Config config, Func<Producer, IConn> connFactory)
        {
            if (string.IsNullOrEmpty(addr))
                throw new ArgumentNullException("addr");
            if (logger == null)
                throw new ArgumentNullException("logger");
            if (config == null)
                throw new ArgumentNullException("config");

            _id = Interlocked.Increment(ref _instCount);

            config.Validate();

            _addr = addr;
            _config = config.Clone();

            _logger = logger;

            _transactionChan = new Chan<ProducerResponse>();
            _exitChan = new Chan<int>();
            _responseChan = new Chan<byte[]>();
            _errorChan = new Chan<byte[]>();
<commit_msg>Producer: add doc about default logger in ctor<commit_after>﻿using System;
using System.Collections.Generic;
using System.Text;
using System.Threading;
using System.Threading.Tasks;
using NsqSharp.Core;
using NsqSharp.Utils;
using NsqSharp.Utils.Channels;
using NsqSharp.Utils.Loggers;

namespace NsqSharp
{
    // https://github.com/bitly/go-nsq/blob/master/producer.go

    /// <summary>
    /// IConn interface
    /// </summary>
    internal interface IConn
    {
        /// <summary>
        /// SetLogger assigns the logger to use as well as a level.
        ///
        /// The format parameter is expected to be a printf compatible string with
        /// a single {0} argument.  This is useful if you want to provide additional
        /// context to the log messages that the connection will print, the default
        /// is '({0})'.
        /// </summary>
        void SetLogger(ILogger l, string format);

        /// <summary>
        /// Connect dials and bootstraps the nsqd connection
        /// (including IDENTIFY) and returns the IdentifyResponse
        /// </summary>
        IdentifyResponse Connect();

        /// <summary>
        /// Close idempotently initiates connection close
        /// </summary>
        void Close();

        /// <summary>
        /// WriteCommand is a thread safe method to write a Command
        /// to this connection, and flush.
        /// </summary>
        void WriteCommand(Command command);
    }

    /// <summary>
    /// <para>Producer is a high-level type to publish to NSQ.</para>
    ///
    /// <para>A Producer instance is 1:1 with a destination nsqd
    /// and will lazily connect to that instance (and re-connect)
    /// when Publish commands are executed.</para>
    /// <seealso cref="Publish(string, string)"/>
    /// <seealso cref="Publish(string, byte[])"/>
    /// <seealso cref="Stop"/>
    /// </summary>
    public sealed partial class Producer
    {
        private static long _instCount;

        internal long _id;
        private readonly string _addr;
        private IConn _conn;
        private readonly Config _config;

        private readonly ILogger _logger;

        private readonly Chan<byte[]> _responseChan;
        private readonly Chan<byte[]> _errorChan;
        private Chan<int> _closeChan;

        private readonly Chan<ProducerResponse> _transactionChan;
        private readonly Queue<ProducerResponse> _transactions = new Queue<ProducerResponse>();
        private int _state;

        private int _concurrentProducers;
        private int _stopFlag;
        private readonly Chan<int> _exitChan;
        private readonly WaitGroup _wg = new WaitGroup();
        private readonly object _guard = new object();

        private readonly Func<Producer, IConn> _connFactory;
    }

    /// <summary>
    /// ProducerResponse is returned by the async publish methods
    /// to retrieve metadata about the command after the
    /// response is received.
    /// </summary>
    public class ProducerResponse
    {
        internal Command _cmd;
        internal Chan<ProducerResponse> _doneChan;

        /// <summary>
        /// the error (or nil) of the publish command
        /// </summary>
        public Exception Error { get; set; }

        /// <summary>
        /// the slice of variadic arguments passed to PublishAsync or MultiPublishAsync
        /// </summary>
        public object[] Args { get; set; }

        internal void finish()
        {
            if (_doneChan != null)
                _doneChan.Send(this);
        }
    }

    public sealed partial class Producer : IConnDelegate
    {
        /// <summary>
        /// Initializes a new instance of the Producer class.
        /// </summary>
        /// <param name="nsqdAddress">The nsqd address.</param>
        public Producer(string nsqdAddress)
            : this(nsqdAddress, new Config())
        {
        }

        /// <summary>
        /// Initializes a new instance of the Producer class.
        /// </summary>
        /// <param name="nsqdAddress">The nsqd address.</param>
        /// <param name="config">The config. After Config is passed in the values are
        /// no longer mutable (they are copied).</param>
        public Producer(string nsqdAddress, Config config)
            : this(nsqdAddress, new ConsoleLogger(LogLevel.Info), config, null)
        {
        }

        /// <summary>
        /// Initializes a new instance of the Producer class.
        /// </summary>
        /// <param name="nsqdAddress">The nsqd address.</param>
        /// <param name="logger">
        /// The logger. Default = <see cref="ConsoleLogger"/>(<see cref="E:LogLevel.Info"/>).
        /// </param>
        public Producer(string nsqdAddress, ILogger logger)
            : this(nsqdAddress, logger, new Config(), null)
        {
        }

        /// <summary>
        /// Initializes a new instance of the Producer class.
        /// </summary>
        /// <param name="nsqdAddress">The nsqd address.</param>
        /// <param name="logger">
        /// The logger. Default = <see cref="ConsoleLogger"/>(<see cref="E:LogLevel.Info"/>).
        /// </param>
        /// <param name="config">The config. Values are copied, changing the properties on <paramref name="config"/>
        /// after the constructor is called will have no effect on the <see cref="Producer"/>.</param>
        public Producer(string nsqdAddress, ILogger logger, Config config)
            : this(nsqdAddress, logger, config, null)
        {
        }

        private Producer(string addr, ILogger logger, Config config, Func<Producer, IConn> connFactory)
        {
            if (string.IsNullOrEmpty(addr))
                throw new ArgumentNullException("addr");
            if (logger == null)
                throw new ArgumentNullException("logger");
            if (config == null)
                throw new ArgumentNullException("config");

            _id = Interlocked.Increment(ref _instCount);

            config.Validate();

            _addr = addr;
            _config = config.Clone();

            _logger = logger;

            _transactionChan = new Chan<ProducerResponse>();
            _exitChan = new Chan<int>();
            _responseChan = new Chan<byte[]>();
            _errorChan = new Chan<byte[]>();

{'fraction_non_alphanumeric': 0.1138913552344822, 'fraction_numerical': 0.0006457341189765114, 'mean_word_length': 1.9869816779170684, 'pattern_counts': {'":': 0, '<': 137, '<?xml version=': 0, '>': 137, 'https://': 2, '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': '4374500', 'n_tokens_mistral': 3287, 'n_tokens_neox': 3184, 'n_words': 1051}