using Confluent.Kafka;

using System;

using System.Collections.Generic;

using log4net;

public class Program

{

    public static void Main()

    {

        Console.WriteLine("Hello World");

    }

    #region Fields & Properties

​

        private readonly static ILog log = null;

​

        public IConsumer<Ignore, string> Consumer { 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}, listening on topics: {1}", this.KafkaSettings.Servers, this.KafkaSettings.ConsumerTopics == this.KafkaSettings.AllTopics ? "*Any*" : this.KafkaSettings.ConsumerTopics); } }

​

        #endregion Fields & Properties

​

        #region Methods

​

        /// <summary>

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

        /// </summary>

        public void Connect()

        {   try

            {

                this.Consumer = new ConsumerBuilder<Ignore, string>(this.BuildConsumerConfig())

                    .Build();

​

                if (this.Consumer != 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 consumer and thus closes the connection

        /// </summary>

        public void Disconnect()

        {

            try

            {

                this.Consumer.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 consumer

        /// </summary>

        public void Dispose()

        {

            this.Disconnect();

        }

​

        /// <summary>

        /// Recieve message through the consumer

        /// </summary>

        /// <returns></returns>

        public XCMessage Receive()

        {

            if (this.Consumer == null)

            {

                this.Connect();

            }

​

            return this.ReceiveInternal(this.Consumer);

        }

​

        /// <summary>

        /// Recieve a batch of messages through the consumer

        /// </summary>

        /// <param name="maxBatchSize"></param>

        /// <returns></returns>

        public XCMessageBatch ReceiveBatch(int? maxBatchSize)

        {

            if (this.Consumer == null)

            {

                this.Connect();

            }

​

            return this.ReceiveBatchInternal(this.Consumer, maxBatchSize);

        }

​

        /// <summary>

        /// Internal method to get a message through the consumer

        /// </summary>

        /// <param name="consumer"></param>

        /// <returns></returns>

        private XCMessage ReceiveInternal(IConsumer<Ignore, string> consumer)

        {

            XCMessage xcMessage = null;

​

            try

            {

                List<string> topics = new List<string>() { this.KafkaSettings.ConsumerTopics };

​

                consumer.Subscribe(topics);

​

                // Poll for messages

                ConsumeResult<Ignore, string> result;

​

                result = consumer.Consume(this.KafkaSettings.ConsumerMessageTimeoutTimeSpan);

​

                if (result != null)

                {

                    xcMessage = new XCMessage() { Text = result.Message.Value };

                    consumer.Commit(result);

                }

                return xcMessage;

            }

            catch(Exception ex)

            {

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

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

            }

        }

​

        /// <summary>

        /// Internal method to get a batch of messgaes through the consumer

        /// </summary>

        /// <param name="consumer">Kafka consumer</param>

        /// <param name="maxMessages">Max number of messages you want to recieve.</param>

        /// <returns>XCMessageBatch</returns>

        private XCMessageBatch ReceiveBatchInternal(IConsumer<Ignore, string> consumer, int? maxMessages)

        {

            List<XCMessage> xcMessageBatch = new List<XCMessage>();

            maxMessages = maxMessages == null ? 1 : maxMessages;

​

            try

            {

                consumer.Subscribe(new List<string>() { this.KafkaSettings.ConsumerTopics });

​

                // Poll for messages while we still have some to consume

                ConsumeResult<Ignore, string> result;

​

                while ((xcMessageBatch.Count < maxMessages))

                {

                    result = consumer.Consume(this.KafkaSettings.ConsumerMessageTimeoutTimeSpan);

​

                    if (result != null)

                    {

                        xcMessageBatch.Add(new ScotiaXCMessage(result));

                        consumer.Commit(result); // This isn't that efficent.  If we can handle duplicate messages in the event of a failure further up the stack then we set an auto commit interval instead

                    }

                    else

                    {

                        break;

                    }

                }

​

                return (XCMessageBatch)xcMessageBatch;

            }

            catch (Exception ex)

            {

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

                throw new Exception("Error when receiving 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

}