using System;

using System.Collections.Generic;

using Confluent.Kafka;

using log4net;

public class Program

{

    public static void Main()

    {

        Console.WriteLine("Hello World");

    }

    /// <summary>

    /// Kafka producer class for handling production of messages from a Kafka server

    /// </summary>

    [XmlType("kafkaProducer")]

    public class KafkaProducer : KafkaBase, IMessageSender, IXmlSerializable, IDisposable

    {

        #region Fields & Properties

​

        private static ILog log = ScotiaLogManagerWrapper.GetLogger(System.Reflection.MethodBase.GetCurrentMethod().DeclaringType);

​

        // The key is used to determine the partition that messages are logged to.  A key of Null will result in a random partition being chosen for us

        public IProducer<Null, string> Producer { get; private set; }

​

        /// <summary>

        /// Producer configuation collection

        /// </summary>

        public Dictionary<string, string> ProducerConfig { get; private set; }

​

        public string ShortDescription { get { return String.Format("Connecting to servers: {0}", this.KafkaSettings.Servers); } }

​

        public string LongDescription { get { return String.Format("Connecting to servers: {0} on topics: {1}", this.KafkaSettings.Servers, this.KafkaSettings.ProducerTopic); } }

​

        #endregion Fields & Properties

​

        #region Methods

​

        /// <summary>

        /// Method used to create a new PRODUCER with the apprpriate config settings.  Kafka handles the actual connecton within the Kafka PRODUCER class

        /// </summary>

        public void Connect()

        {

            try

            {

                this.Producer = new ProducerBuilder<Null, string>(this.BuildProducerConfig())

                    .Build();

​

                if (this.Producer != null)

                    this.Connected = true;

            }

            catch (Exception ex)

            {

                log.Error("Error when connecting: " + ex.Message, ex);

                throw new Exception("Error when connecting: " + ex.Message, ex);

            }

        }

​

        /// <summary>

        /// Disposes of producer and thus closes the connection

        /// </summary>

        public void Disconnect()

        {

            try

            {

                Producer.Flush(this.KafkaSettings.FlushTimeoutTimeSpan);

                Producer.Dispose();

            }

            catch (Exception ex)

            {

                log.Error("Error when disconnecting: " + ex.Message, ex);

                throw new Exception("Error when disconnecting: " + ex.Message, ex);

            }

        }

​

        /// <summary>

        /// Dispose method to dispose of producer

        /// </summary>

        public void Dispose()

        {

            this.Disconnect();

        }

​

        /// <summary>

        /// Send message through the producer

        /// </summary>

        /// <returns></returns>

        public XCMessage Send(XCMessage message)

        {

            if (this.Producer == null)

            {

                this.Connect();

            }

​

            return this.SendInternal(message, this.Producer);

        }

​

        /// <summary>

        /// Internal method to send a message through the producer

        /// </summary>

        /// <param name="message">XCMessage</param>

        /// <param name="producer">Kafka Producer</param>

        /// <returns>XCMessage</returns>

        private XCMessage SendInternal(XCMessage message, IProducer<Null, string> producer)

        {

            try

            {

                Message<Null, string> kafkaMessage = new Message<Null, string>() { Value = message.Text };

​

                DeliveryResult<Null, string> result = producer.ProduceAsync(this.KafkaSettings.ProducerTopic, kafkaMessage).GetAwaiter().GetResult();

​

                return new ScotiaXCMessage(result);

            }

            catch (Exception ex)

            {

                log.Error("Error when sending a message: " + ex.Message, ex);

                throw new Exception("Error when sending a message: " + ex.Message, ex);

            }

        }

​

        #endregion Methods

​

        #region Serialization\Deserialization

​

        /// <summary>

        /// Serialize a context

        /// </summary>

        /// <param name="context">XmlSerializationContext</param>

        public override void XmlSerialize(XmlSerializationContext context)

        {

            base.XmlSerialize(context);

        }

​

        /// <summary>

        /// De-Serialize a context

        /// </summary>

        /// <param name="context">XmlSerializationContext</param>

        public override void XmlDeserialize(XmlSerializationContext context)

        {

            base.XmlDeserialize(context);

        }

​

        #endregion Serialization\Deserialization

}