From 63cbcabc963fe09ffad02a8f92b5d815da87ee06 Mon Sep 17 00:00:00 2001
From: Yury Pliner <yury.pliner@gmail.com>
Date: Wed, 23 Jul 2014 23:26:19 +0600
Subject: [PATCH 1/2] An attempt to #12

---
 src/TestHarness/Program.cs                    |  21 +--
 src/kafka-net-client/JsonConsumer.cs          |   6 +-
 src/kafka-net-client/JsonProducer.cs          |   5 +-
 src/kafka-net/BrokerRouter.cs                 |  35 ++--
 src/kafka-net/Configuration/Bus.cs            |  37 ++++
 src/kafka-net/Configuration/BusFactory.cs     |  16 ++
 .../Configuration/DefaultContainer.cs         |  37 ++++
 .../DefaultServiceRegistrator.cs              |  16 ++
 src/kafka-net/Configuration/IContainer.cs     |   6 +
 .../Configuration/IServiceProvider.cs         |   7 +
 .../Configuration/IServiceRegistrator.cs      |  10 ++
 .../ServiceImplementationNotFound.cs          |   8 +
 src/kafka-net/Consumer.cs                     |  28 +--
 src/kafka-net/ConsumerFactory.cs              |  21 +++
 .../Default/DefaultKafkaConnectionFactory.cs  |  15 +-
 src/kafka-net/Interfaces/IBrokerRouter.cs     |   2 -
 src/kafka-net/Interfaces/IBus.cs              |  14 ++
 src/kafka-net/Interfaces/IConsumer.cs         |  12 ++
 src/kafka-net/Interfaces/IConsumerFactory.cs  |   7 +
 .../Interfaces/IKafkaConnectionFactory.cs     |   8 +-
 src/kafka-net/Interfaces/IKafkaOptions.cs     |  12 ++
 .../Interfaces/IPartitionSelector.cs          |   4 +-
 src/kafka-net/Interfaces/IProducer.cs         |  12 ++
 src/kafka-net/KafkaConnection.cs              |   6 +-
 src/kafka-net/Model/ConsumerOptions.cs        |  12 +-
 src/kafka-net/Model/KafkaOptions.cs           |  40 +----
 src/kafka-net/Producer.cs                     |  10 +-
 src/kafka-net/kafka-net.csproj                |  18 +-
 src/kafka-tests/Fakes/BrokerRouterProxy.cs    |  13 +-
 .../Integration/GzipProducerConsumerTests.cs  |  17 +-
 .../KafkaConnectionIntegrationTests.cs        |   4 +-
 .../Integration/OffsetManagementTests.cs      |  18 +-
 .../ProducerConsumerIntegrationTests.cs       |  46 +++--
 src/kafka-tests/Unit/BrokerRouterTests.cs     |  21 +--
 src/kafka-tests/Unit/ConsumerTests.cs         |  41 ++---
 src/kafka-tests/Unit/KafkaConnectionTests.cs  |  12 +-
 src/kafka-tests/Unit/ProducerTests.cs         | 159 ++++++++++--------
 37 files changed, 492 insertions(+), 264 deletions(-)
 create mode 100644 src/kafka-net/Configuration/Bus.cs
 create mode 100644 src/kafka-net/Configuration/BusFactory.cs
 create mode 100644 src/kafka-net/Configuration/DefaultContainer.cs
 create mode 100644 src/kafka-net/Configuration/DefaultServiceRegistrator.cs
 create mode 100644 src/kafka-net/Configuration/IContainer.cs
 create mode 100644 src/kafka-net/Configuration/IServiceProvider.cs
 create mode 100644 src/kafka-net/Configuration/IServiceRegistrator.cs
 create mode 100644 src/kafka-net/Configuration/ServiceImplementationNotFound.cs
 create mode 100644 src/kafka-net/ConsumerFactory.cs
 create mode 100644 src/kafka-net/Interfaces/IBus.cs
 create mode 100644 src/kafka-net/Interfaces/IConsumer.cs
 create mode 100644 src/kafka-net/Interfaces/IConsumerFactory.cs
 create mode 100644 src/kafka-net/Interfaces/IKafkaOptions.cs
 create mode 100644 src/kafka-net/Interfaces/IProducer.cs

diff --git a/src/TestHarness/Program.cs b/src/TestHarness/Program.cs
index 2dd3e84f..440f6612 100644
--- a/src/TestHarness/Program.cs
+++ b/src/TestHarness/Program.cs
@@ -1,9 +1,8 @@
 using System;
 using System.Threading.Tasks;
-using KafkaNet;
+using KafkaNet.Configuration;
 using KafkaNet.Model;
 using KafkaNet.Protocol;
-using System.Collections.Generic;
 
 namespace TestHarness
 {
@@ -11,17 +10,15 @@ class Program
     {
         static void Main(string[] args)
         {
-            var options = new KafkaOptions(new Uri("http://CSDKAFKA01:9092"), new Uri("http://CSDKAFKA02:9092"))
+            var bus = BusFactory.Create(new KafkaOptions
                 {
-                    Log = new ConsoleLog()
-                };
-            var router = new BrokerRouter(options);
-            var client = new Producer(router);
+                    Hosts = new[] {new Uri("http://CSDKAFKA01:9092"), new Uri("http://CSDKAFKA02:9092")}
+                }, x => { });
+
 
             Task.Factory.StartNew(() =>
                 {
-                    var consumer = new Consumer(new ConsumerOptions("TestHarness", router));
-                    foreach (var data in consumer.Consume())
+                    foreach (var data in bus.Consume("TestHarness"))
                     {
                         Console.WriteLine("Response: P{0},O{1} : {2}", data.Meta.PartitionId, data.Meta.Offset, data.Value);
                     }
@@ -33,13 +30,11 @@ static void Main(string[] args)
             {
                 var message = Console.ReadLine();
                 if (message == "quit") break;
-                client.SendMessageAsync("TestHarness", new[] {new Message {Value = message}});
+                bus.SendMessageAsync("TestHarness", new[] {new Message {Value = message}});
             }
 
-            using (client)
-            using (router)
+            using (bus)
             {
-
             }
         }
     }
diff --git a/src/kafka-net-client/JsonConsumer.cs b/src/kafka-net-client/JsonConsumer.cs
index 07efb02d..ab329a51 100644
--- a/src/kafka-net-client/JsonConsumer.cs
+++ b/src/kafka-net-client/JsonConsumer.cs
@@ -9,13 +9,11 @@ namespace KafkaNet.Client
 {
     public class JsonConsumer<T> : IDisposable
     {
-        private readonly ConsumerOptions _options;
         private readonly Consumer _consumer;
 
-        public JsonConsumer(ConsumerOptions options)
+        public JsonConsumer(IBrokerRouter brokerRouter, IKafkaLog log, ConsumerOptions options)
         {
-            _options = options;
-            _consumer = new Consumer(options);
+            _consumer = new Consumer(brokerRouter, log, options);
         }
 
         public IEnumerable<Message<T>> Consume()
diff --git a/src/kafka-net-client/JsonProducer.cs b/src/kafka-net-client/JsonProducer.cs
index 666fec03..067da55a 100644
--- a/src/kafka-net-client/JsonProducer.cs
+++ b/src/kafka-net-client/JsonProducer.cs
@@ -2,6 +2,7 @@
 using System.Collections.Generic;
 using System.Linq;
 using System.Threading.Tasks;
+using KafkaNet.Configuration;
 using KafkaNet.Protocol;
 using Newtonsoft.Json;
 
@@ -11,9 +12,9 @@ public class JsonProducer : IDisposable
     {
         private readonly Producer _producer;
 
-        public JsonProducer(IBrokerRouter brokerRouter)
+        public JsonProducer(IBrokerRouter brokerRouter, IKafkaOptions options)
         {
-            _producer = new Producer(brokerRouter);
+            _producer = new Producer(brokerRouter, options);
         }
 
         public Task<List<ProduceResponse>> Publish<T>(string topic, IEnumerable<T> messages, Int16 acks = 1, int timeoutMS = 1000) where T : class 
diff --git a/src/kafka-net/BrokerRouter.cs b/src/kafka-net/BrokerRouter.cs
index 544b02b0..f2f2c2c7 100644
--- a/src/kafka-net/BrokerRouter.cs
+++ b/src/kafka-net/BrokerRouter.cs
@@ -2,7 +2,7 @@
 using System.Collections.Concurrent;
 using System.Collections.Generic;
 using System.Linq;
-using KafkaNet.Model;
+using KafkaNet.Configuration;
 using KafkaNet.Protocol;
 
 namespace KafkaNet
@@ -20,17 +20,23 @@ namespace KafkaNet
     public class BrokerRouter : IBrokerRouter
     {
         private readonly object _threadLock = new object();
-        private readonly KafkaOptions _kafkaOptions;
+        private readonly IKafkaOptions options;
+        private readonly IKafkaLog log;
+        private readonly IPartitionSelector partitionSelector;
+        private readonly IKafkaConnectionFactory connectionFactory;
         private readonly ConcurrentDictionary<int, IKafkaConnection> _brokerConnectionIndex = new ConcurrentDictionary<int, IKafkaConnection>();
         private readonly ConcurrentDictionary<string, Topic> _topicIndex = new ConcurrentDictionary<string, Topic>();
         private readonly List<IKafkaConnection> _defaultConnections = new List<IKafkaConnection>();
 
-        public BrokerRouter(KafkaOptions kafkaOptions)
+        public BrokerRouter(IKafkaOptions options, IKafkaLog log, IPartitionSelector partitionSelector, IKafkaConnectionFactory connectionFactory)
         {
-            _kafkaOptions = kafkaOptions;
+            this.options = options;
+            this.log = log;
+            this.partitionSelector = partitionSelector;
+            this.connectionFactory = connectionFactory;
             _defaultConnections
-                .AddRange(kafkaOptions.KafkaServerUri.Distinct()
-                .Select(uri => _kafkaOptions.KafkaConnectionFactory.Create(uri, _kafkaOptions.ResponseTimeoutMs, _kafkaOptions.Log)));
+                .AddRange(options.Hosts.Distinct()
+                .Select(uri => connectionFactory.Create(uri, options.Timeout)));
         }
 
         /// <summary>
@@ -76,7 +82,7 @@ public BrokerRoute SelectBrokerRoute(string topic, string key = null)
             if (cachedTopic == null)
                 throw new InvalidTopicMetadataException(string.Format("The Metadata is invalid as it returned no data for the given topic:{0}", topic));
 
-            var partition = _kafkaOptions.PartitionSelector.Select(cachedTopic, key);
+            var partition = partitionSelector.Select(cachedTopic, key);
 
             return GetCachedRoute(cachedTopic.Name, partition);
         }
@@ -119,7 +125,7 @@ public void RefreshTopicMetadata(params string[] topics)
         {
             lock (_threadLock)
             {
-                _kafkaOptions.Log.DebugFormat("BrokerRouter: Refreshing metadata for topics: {0}", string.Join(",", topics));
+                log.DebugFormat("BrokerRouter: Refreshing metadata for topics: {0}", string.Join(",", topics));
 
                 //use the initial default connections to retrieve metadata
                 if (_defaultConnections.Count > 0)
@@ -216,14 +222,14 @@ private void CycleConnectionsForTopicMetadata(IEnumerable<IKafkaConnection> conn
                 }
                 catch (Exception ex)
                 {
-                    _kafkaOptions.Log.WarnFormat("Failed to contact Kafka server={0}.  Trying next default server.  Exception={1}", conn.KafkaUri, ex);
+                    log.WarnFormat("Failed to contact Kafka server={0}.  Trying next default server.  Exception={1}", conn.KafkaUri, ex);
                 }
             }
 
             throw new ServerUnreachableException(
                     string.Format(
                         "Unable to query for metadata from any of the default Kafka servers.  At least one provided server must be available.  Server list: {0}",
-                        string.Join(", ", _kafkaOptions.KafkaServerUri.Select(x => x.ToString()))));
+                        string.Join(", ", options.Hosts.Select(x => x.ToString()))));
         }
 
         private void UpdateInternalMetadataCache(MetadataResponse metadata)
@@ -232,16 +238,13 @@ private void UpdateInternalMetadataCache(MetadataResponse metadata)
             {
                 var localBroker = broker;
                 _brokerConnectionIndex.AddOrUpdate(broker.BrokerId,
-                    i =>
-                    {
-                        return _kafkaOptions.KafkaConnectionFactory.Create(localBroker.Address, _kafkaOptions.ResponseTimeoutMs, _kafkaOptions.Log);
-                    },
+                    i => connectionFactory.Create(localBroker.Address, options.Timeout),
                     (i, connection) =>
                     {
                         //if a connection changes for a broker close old connection and create a new one
                         if (connection.KafkaUri == localBroker.Address) return connection;
-                        _kafkaOptions.Log.WarnFormat("Broker:{0} Uri changed from:{1} to {2}", localBroker.BrokerId, connection.KafkaUri, localBroker.Address);
-                        using (connection) { return _kafkaOptions.KafkaConnectionFactory.Create(localBroker.Address, _kafkaOptions.ResponseTimeoutMs, _kafkaOptions.Log); }
+                        log.WarnFormat("Broker:{0} Uri changed from:{1} to {2}", localBroker.BrokerId, connection.KafkaUri, localBroker.Address);
+                        using (connection) { return connectionFactory.Create(localBroker.Address, options.Timeout); }
                     });
             }
 
diff --git a/src/kafka-net/Configuration/Bus.cs b/src/kafka-net/Configuration/Bus.cs
new file mode 100644
index 00000000..4316ca74
--- /dev/null
+++ b/src/kafka-net/Configuration/Bus.cs
@@ -0,0 +1,37 @@
+using System;
+using System.Collections.Concurrent;
+using System.Collections.Generic;
+using System.Threading;
+using System.Threading.Tasks;
+using KafkaNet.Protocol;
+
+namespace KafkaNet.Configuration
+{
+    public class Bus : IBus
+    {
+        private readonly IConsumerFactory _consumerFactory;
+        private readonly IProducer _producer;
+        
+        public Bus(IProducer producer, IConsumerFactory consumerFactory)
+        {
+            _producer = producer;
+            _consumerFactory = consumerFactory;
+        }
+
+        public Task<List<ProduceResponse>> SendMessageAsync(string topic, IEnumerable<Message> messages, short acks = 1, int timeoutMS = 1000, MessageCodec codec = MessageCodec.CodecNone)
+        {
+            return _producer.SendMessageAsync(topic, messages, acks, timeoutMS, codec);
+        }
+
+        public IEnumerable<Message> Consume(string topic, CancellationToken? token = null)
+        {
+            var consumer = _consumerFactory.GetConsumer(topic);
+            return consumer.Consume(token);
+        }
+
+        public void Dispose()
+        {
+            _producer.Dispose();
+        }
+    }
+}
\ No newline at end of file
diff --git a/src/kafka-net/Configuration/BusFactory.cs b/src/kafka-net/Configuration/BusFactory.cs
new file mode 100644
index 00000000..be75b585
--- /dev/null
+++ b/src/kafka-net/Configuration/BusFactory.cs
@@ -0,0 +1,16 @@
+using System;
+
+namespace KafkaNet.Configuration
+{
+    public static class BusFactory
+    {
+        public static IBus Create(IKafkaOptions settings, Action<IServiceRegistrator> configure)
+        {
+            var container = new DefaultContainer();
+            DefaultServiceRegistrator.Register(container);
+            container.Register(settings);
+            configure(container);
+            return container.Resolve<IBus>();
+        }
+    }
+}
\ No newline at end of file
diff --git a/src/kafka-net/Configuration/DefaultContainer.cs b/src/kafka-net/Configuration/DefaultContainer.cs
new file mode 100644
index 00000000..52fa02ca
--- /dev/null
+++ b/src/kafka-net/Configuration/DefaultContainer.cs
@@ -0,0 +1,37 @@
+using System;
+using System.Collections.Generic;
+
+namespace KafkaNet.Configuration
+{
+    public class DefaultContainer : IContainer
+    {
+        private readonly Dictionary<Type, object> _implementations = new Dictionary<Type, object>();
+        private readonly Dictionary<Type, object> _factories = new Dictionary<Type, object>();
+
+        public T Resolve<T>() where T: class
+        {
+            object implementation;
+            if (_implementations.TryGetValue(typeof (T), out implementation))
+                return (T) implementation;
+            object factory;
+            if (_factories.TryGetValue(typeof(T), out factory))
+            {
+                var serviceFactory = (Func<IServiceProvider, T>)factory;
+                var serviceImplementation = serviceFactory(this);
+                _implementations.Add(typeof (T), serviceImplementation);
+                return serviceImplementation;
+            }
+            throw new ServiceImplementationNotFound();
+        }
+
+        public void Register<T>(T implementation) where T : class
+        {
+            _implementations.Add(typeof (T), implementation);
+        }
+
+        public void Register<T>(Func<IServiceProvider, T> factory) where T : class 
+        {
+            _factories.Add(typeof(T), factory);
+        }
+    }
+}
\ No newline at end of file
diff --git a/src/kafka-net/Configuration/DefaultServiceRegistrator.cs b/src/kafka-net/Configuration/DefaultServiceRegistrator.cs
new file mode 100644
index 00000000..99c45bb2
--- /dev/null
+++ b/src/kafka-net/Configuration/DefaultServiceRegistrator.cs
@@ -0,0 +1,16 @@
+namespace KafkaNet.Configuration
+{
+    public static class DefaultServiceRegistrator
+    {
+        public static void Register(IServiceRegistrator registrator)
+        {
+            registrator.Register<IKafkaLog>(new DefaultTraceLog());
+            registrator.Register<IPartitionSelector>(new DefaultPartitionSelector());
+            registrator.Register<IKafkaConnectionFactory>(x => new DefaultKafkaConnectionFactory(x.Resolve<IKafkaLog>()));
+            registrator.Register<IBrokerRouter>(x => new BrokerRouter(x.Resolve<IKafkaOptions>(), x.Resolve<IKafkaLog>(), x.Resolve<IPartitionSelector>(), x.Resolve<IKafkaConnectionFactory>()));
+            registrator.Register<IProducer>(x => new Producer(x.Resolve<IBrokerRouter>(), x.Resolve<IKafkaOptions>()));
+            registrator.Register<IConsumerFactory>(x => new ConsumerFactory(x.Resolve<IBrokerRouter>(), x.Resolve<IKafkaLog>()));
+            registrator.Register<IBus>(x => new Bus(x.Resolve<IProducer>(), x.Resolve<IConsumerFactory>()));
+        }
+    }
+}
\ No newline at end of file
diff --git a/src/kafka-net/Configuration/IContainer.cs b/src/kafka-net/Configuration/IContainer.cs
new file mode 100644
index 00000000..0e8c54f5
--- /dev/null
+++ b/src/kafka-net/Configuration/IContainer.cs
@@ -0,0 +1,6 @@
+namespace KafkaNet.Configuration
+{
+    public interface IContainer : IServiceProvider, IServiceRegistrator
+    {
+    }
+}
\ No newline at end of file
diff --git a/src/kafka-net/Configuration/IServiceProvider.cs b/src/kafka-net/Configuration/IServiceProvider.cs
new file mode 100644
index 00000000..c43e1a1e
--- /dev/null
+++ b/src/kafka-net/Configuration/IServiceProvider.cs
@@ -0,0 +1,7 @@
+namespace KafkaNet.Configuration
+{
+    public interface IServiceProvider
+    {
+        T Resolve<T>() where T : class;
+    }
+}
\ No newline at end of file
diff --git a/src/kafka-net/Configuration/IServiceRegistrator.cs b/src/kafka-net/Configuration/IServiceRegistrator.cs
new file mode 100644
index 00000000..ae6c2e24
--- /dev/null
+++ b/src/kafka-net/Configuration/IServiceRegistrator.cs
@@ -0,0 +1,10 @@
+using System;
+
+namespace KafkaNet.Configuration
+{
+    public interface IServiceRegistrator
+    {
+        void Register<T>(T implementation) where T : class;
+        void Register<T>(Func<IServiceProvider, T> factory) where T : class;
+    }
+}
\ No newline at end of file
diff --git a/src/kafka-net/Configuration/ServiceImplementationNotFound.cs b/src/kafka-net/Configuration/ServiceImplementationNotFound.cs
new file mode 100644
index 00000000..3fb24aeb
--- /dev/null
+++ b/src/kafka-net/Configuration/ServiceImplementationNotFound.cs
@@ -0,0 +1,8 @@
+using System;
+
+namespace KafkaNet.Configuration
+{
+    public class ServiceImplementationNotFound : Exception
+    {
+    }
+}
\ No newline at end of file
diff --git a/src/kafka-net/Consumer.cs b/src/kafka-net/Consumer.cs
index 19b69c46..aed38a76 100644
--- a/src/kafka-net/Consumer.cs
+++ b/src/kafka-net/Consumer.cs
@@ -16,8 +16,10 @@ namespace KafkaNet
     /// TODO: provide automatic offset saving when the feature is available in 0.8.2
     /// https://cwiki.apache.org/confluence/display/KAFKA/A+Guide+To+The+Kafka+Protocol#AGuideToTheKafkaProtocol-OffsetCommit/FetchAPI
     /// </summary>
-    public class Consumer : IMetadataQueries, IDisposable
+    public class Consumer : IMetadataQueries, IConsumer
     {
+        private readonly IBrokerRouter _brokerRouter;
+        private readonly IKafkaLog _log;
         private readonly ConsumerOptions _options;
         private readonly BlockingCollection<Message> _fetchResponseQueue;
         private readonly CancellationTokenSource _disposeToken = new CancellationTokenSource();
@@ -29,11 +31,13 @@ public class Consumer : IMetadataQueries, IDisposable
         private int _ensureOneThread;
         private Topic _topic;
 
-        public Consumer(ConsumerOptions options, params OffsetPosition[] positions)
+        public Consumer(IBrokerRouter brokerRouter, IKafkaLog log, ConsumerOptions options, params OffsetPosition[] positions)
         {
+            _brokerRouter = brokerRouter;
+            _log = log;
             _options = options;
             _fetchResponseQueue = new BlockingCollection<Message>(_options.ConsumerBufferSize);
-            _metadataQueries = new MetadataQueries(_options.Router);
+            _metadataQueries = new MetadataQueries(brokerRouter);
 
             //this timer will periodically look for new partitions and automatically add them to the consuming queue
             //using the same whitelist logic
@@ -56,7 +60,7 @@ public Consumer(ConsumerOptions options, params OffsetPosition[] positions)
         /// <returns>Blocking enumberable of messages from Kafka.</returns>
         public IEnumerable<Message> Consume(CancellationToken? cancellationToken = null)
         {
-            _options.Log.DebugFormat("Consumer: Beginning consumption of topic: {0}", _options.Topic);
+            _log.DebugFormat("Consumer: Beginning consumption of topic: {0}", _options.Topic);
             _topicPartitionQueryTimer.Begin();
 
             return _fetchResponseQueue.GetConsumingEnumerable(cancellationToken ?? new CancellationToken(false));
@@ -91,8 +95,8 @@ private void RefreshTopicPartitions()
             {
                 if (Interlocked.Increment(ref _ensureOneThread) == 1)
                 {
-                    _options.Log.DebugFormat("Consumer: Refreshing partitions for topic: {0}", _options.Topic);
-                    var topic = _options.Router.GetTopicMetadata(_options.Topic);
+                    _log.DebugFormat("Consumer: Refreshing partitions for topic: {0}", _options.Topic);
+                    var topic = _brokerRouter.GetTopicMetadata(_options.Topic);
                     if (topic.Count <= 0) throw new ApplicationException(string.Format("Unable to get metadata for topic:{0}.", _options.Topic));
                     _topic = topic.First();
 
@@ -111,7 +115,7 @@ private void RefreshTopicPartitions()
             }
             catch (Exception ex)
             {
-                _options.Log.ErrorFormat("Exception occured trying to setup consumer for topic:{0}.  Exception={1}", _options.Topic, ex);
+                _log.ErrorFormat("Exception occured trying to setup consumer for topic:{0}.  Exception={1}", _options.Topic, ex);
             }
             finally
             {
@@ -125,7 +129,7 @@ private Task ConsumeTopicPartitionAsync(string topic, int partitionId)
             {
                 try
                 {
-                    _options.Log.DebugFormat("Consumer: Creating polling task for topic: {0} on parition: {1}", topic, partitionId);
+                    _log.DebugFormat("Consumer: Creating polling task for topic: {0} on parition: {1}", topic, partitionId);
                     while (_disposeToken.IsCancellationRequested == false)
                     {
                         try
@@ -151,7 +155,7 @@ private Task ConsumeTopicPartitionAsync(string topic, int partitionId)
                                 };
 
                             //make request and post to queue
-                            var route = _options.Router.SelectBrokerRoute(topic, partitionId);
+                            var route = _brokerRouter.SelectBrokerRoute(topic, partitionId);
                             var responses = route.Connection.SendAsync(fetchRequest).Result;
 
                             if (responses.Count > 0)
@@ -179,13 +183,13 @@ private Task ConsumeTopicPartitionAsync(string topic, int partitionId)
                         }
                         catch (Exception ex)
                         {
-                        	_options.Log.ErrorFormat("Exception occured while polling topic:{0} partition:{1}.  Polling will continue.  Exception={2}", topic, partitionId, ex);
+                        	_log.ErrorFormat("Exception occured while polling topic:{0} partition:{1}.  Polling will continue.  Exception={2}", topic, partitionId, ex);
                         }
                 	}
                 }
                 finally
                 {
-                    _options.Log.DebugFormat("Consumer: Disabling polling task for topic: {0} on parition: {1}", topic, partitionId);
+                    _log.DebugFormat("Consumer: Disabling polling task for topic: {0} on parition: {1}", topic, partitionId);
                     Task tempTask;
                     _partitionPollingIndex.TryRemove(partitionId, out tempTask);
                 }
@@ -204,7 +208,7 @@ public Task<List<OffsetResponse>> GetTopicOffsetAsync(string topic, int maxOffse
 
         public void Dispose()
         {
-            _options.Log.DebugFormat("Consumer: Disposing...");
+            _log.DebugFormat("Consumer: Disposing...");
             _disposeToken.Cancel();
             using (_topicPartitionQueryTimer)
             using (_metadataQueries)
diff --git a/src/kafka-net/ConsumerFactory.cs b/src/kafka-net/ConsumerFactory.cs
new file mode 100644
index 00000000..b0abb009
--- /dev/null
+++ b/src/kafka-net/ConsumerFactory.cs
@@ -0,0 +1,21 @@
+using KafkaNet.Model;
+
+namespace KafkaNet
+{
+    public class ConsumerFactory : IConsumerFactory
+    {
+        private readonly IBrokerRouter _brokerRouter;
+        private readonly IKafkaLog _log;
+
+        public ConsumerFactory(IBrokerRouter brokerRouter, IKafkaLog log)
+        {
+            _brokerRouter = brokerRouter;
+            _log = log;
+        }
+
+        public IConsumer GetConsumer(string topic)
+        {
+            return new Consumer(_brokerRouter, _log, new ConsumerOptions(topic));
+        }
+    }
+}
\ No newline at end of file
diff --git a/src/kafka-net/Default/DefaultKafkaConnectionFactory.cs b/src/kafka-net/Default/DefaultKafkaConnectionFactory.cs
index c55bd162..9dd894df 100644
--- a/src/kafka-net/Default/DefaultKafkaConnectionFactory.cs
+++ b/src/kafka-net/Default/DefaultKafkaConnectionFactory.cs
@@ -1,16 +1,19 @@
 using System;
-using System.Collections.Generic;
-using System.Linq;
-using System.Text;
-using System.Threading.Tasks;
 
 namespace KafkaNet
 {
     public class DefaultKafkaConnectionFactory : IKafkaConnectionFactory
     {
-        public IKafkaConnection Create(Uri kafkaAddress, int responseTimeoutMs, IKafkaLog log)
+        private readonly IKafkaLog kafkaLog;
+        
+        public DefaultKafkaConnectionFactory(IKafkaLog kafkaLog)
         {
-            return new KafkaConnection(new KafkaTcpSocket(log, kafkaAddress), responseTimeoutMs, log);
+            this.kafkaLog = kafkaLog;
+        }
+
+        public IKafkaConnection Create(Uri kafkaAddress, TimeSpan timeout)
+        {
+            return new KafkaConnection(new KafkaTcpSocket(kafkaLog, kafkaAddress), (int)timeout.TotalMilliseconds, kafkaLog);
         }
     }
 }
diff --git a/src/kafka-net/Interfaces/IBrokerRouter.cs b/src/kafka-net/Interfaces/IBrokerRouter.cs
index 6225232d..3908963a 100644
--- a/src/kafka-net/Interfaces/IBrokerRouter.cs
+++ b/src/kafka-net/Interfaces/IBrokerRouter.cs
@@ -1,7 +1,5 @@
 using System;
 using System.Collections.Generic;
-using System.Threading.Tasks;
-using KafkaNet.Model;
 using KafkaNet.Protocol;
 
 namespace KafkaNet
diff --git a/src/kafka-net/Interfaces/IBus.cs b/src/kafka-net/Interfaces/IBus.cs
new file mode 100644
index 00000000..23b757ff
--- /dev/null
+++ b/src/kafka-net/Interfaces/IBus.cs
@@ -0,0 +1,14 @@
+using System;
+using System.Collections.Generic;
+using System.Threading;
+using System.Threading.Tasks;
+using KafkaNet.Protocol;
+
+namespace KafkaNet
+{
+    public interface IBus : IDisposable
+    {
+        Task<List<ProduceResponse>> SendMessageAsync(string topic, IEnumerable<Message> messages, Int16 acks = 1, int timeoutMS = 1000, MessageCodec codec = MessageCodec.CodecNone);
+        IEnumerable<Message> Consume(string topic, CancellationToken? token = null);
+    }
+}
\ No newline at end of file
diff --git a/src/kafka-net/Interfaces/IConsumer.cs b/src/kafka-net/Interfaces/IConsumer.cs
new file mode 100644
index 00000000..a7c6de22
--- /dev/null
+++ b/src/kafka-net/Interfaces/IConsumer.cs
@@ -0,0 +1,12 @@
+using System;
+using System.Collections.Generic;
+using System.Threading;
+using KafkaNet.Protocol;
+
+namespace KafkaNet
+{
+    public interface IConsumer : IDisposable
+    {
+        IEnumerable<Message> Consume(CancellationToken? cancellationToken = null);
+    }
+}
\ No newline at end of file
diff --git a/src/kafka-net/Interfaces/IConsumerFactory.cs b/src/kafka-net/Interfaces/IConsumerFactory.cs
new file mode 100644
index 00000000..0c64759a
--- /dev/null
+++ b/src/kafka-net/Interfaces/IConsumerFactory.cs
@@ -0,0 +1,7 @@
+namespace KafkaNet
+{
+    public interface IConsumerFactory
+    {
+        IConsumer GetConsumer(string topic);
+    }
+}
\ No newline at end of file
diff --git a/src/kafka-net/Interfaces/IKafkaConnectionFactory.cs b/src/kafka-net/Interfaces/IKafkaConnectionFactory.cs
index 47ab8a61..5fd04e5f 100644
--- a/src/kafka-net/Interfaces/IKafkaConnectionFactory.cs
+++ b/src/kafka-net/Interfaces/IKafkaConnectionFactory.cs
@@ -1,13 +1,9 @@
-using KafkaNet.Protocol;
-using System;
-using System.Collections.Generic;
-using System.Linq;
-using System.Text;
+using System;
 
 namespace KafkaNet
 {
     public interface IKafkaConnectionFactory
     {
-        IKafkaConnection Create(Uri kafkaAddress, int responseTimeoutMs, IKafkaLog log);
+        IKafkaConnection Create(Uri kafkaAddress, TimeSpan timeout);
     }
 }
diff --git a/src/kafka-net/Interfaces/IKafkaOptions.cs b/src/kafka-net/Interfaces/IKafkaOptions.cs
new file mode 100644
index 00000000..665ec37a
--- /dev/null
+++ b/src/kafka-net/Interfaces/IKafkaOptions.cs
@@ -0,0 +1,12 @@
+using System;
+using System.Collections.Generic;
+
+namespace KafkaNet
+{
+    public interface IKafkaOptions
+    {
+        IEnumerable<Uri> Hosts { get; }
+        TimeSpan Timeout { get; }
+        int QueueSize { get; }
+    }
+}
\ No newline at end of file
diff --git a/src/kafka-net/Interfaces/IPartitionSelector.cs b/src/kafka-net/Interfaces/IPartitionSelector.cs
index bb877a95..5582d6c9 100644
--- a/src/kafka-net/Interfaces/IPartitionSelector.cs
+++ b/src/kafka-net/Interfaces/IPartitionSelector.cs
@@ -1,6 +1,4 @@
-using System.Collections.Generic;
-using KafkaNet.Model;
-using KafkaNet.Protocol;
+using KafkaNet.Protocol;
 
 namespace KafkaNet
 {
diff --git a/src/kafka-net/Interfaces/IProducer.cs b/src/kafka-net/Interfaces/IProducer.cs
new file mode 100644
index 00000000..48ae64b8
--- /dev/null
+++ b/src/kafka-net/Interfaces/IProducer.cs
@@ -0,0 +1,12 @@
+using System;
+using System.Collections.Generic;
+using System.Threading.Tasks;
+using KafkaNet.Protocol;
+
+namespace KafkaNet
+{
+    public interface IProducer : IDisposable
+    {
+        Task<List<ProduceResponse>> SendMessageAsync(string topic, IEnumerable<Message> messages, Int16 acks = 1, int timeoutMS = 1000, MessageCodec codec = MessageCodec.CodecNone);
+    }
+}
\ No newline at end of file
diff --git a/src/kafka-net/KafkaConnection.cs b/src/kafka-net/KafkaConnection.cs
index 76cad887..c3aae670 100644
--- a/src/kafka-net/KafkaConnection.cs
+++ b/src/kafka-net/KafkaConnection.cs
@@ -20,8 +20,6 @@ namespace KafkaNet
     /// </summary>
     public class KafkaConnection : IKafkaConnection
     {
-        private const int DefaultResponseTimeoutMs = 30000;
-
         private readonly ConcurrentDictionary<int, AsyncRequestItem> _requestIndex = new ConcurrentDictionary<int, AsyncRequestItem>();
         private readonly IScheduledTimer _responseTimeoutTimer;
         private readonly int _responseTimeoutMS;
@@ -40,10 +38,10 @@ public class KafkaConnection : IKafkaConnection
         /// <param name="log">Logging interface used to record any log messages created by the connection.</param>
         /// <param name="client">The kafka socket initialized to the kafka server.</param>
         /// <param name="responseTimeoutMs">The amount of time to wait for a message response to be received after sending message to Kafka.</param>
-        public KafkaConnection(IKafkaTcpSocket client, int responseTimeoutMs = DefaultResponseTimeoutMs, IKafkaLog log = null)
+        public KafkaConnection(IKafkaTcpSocket client, int responseTimeoutMs, IKafkaLog log)
         {
             _client = client;
-            _log = log ?? new DefaultTraceLog();
+            _log = log;
             _responseTimeoutMS = responseTimeoutMs;
             _responseTimeoutTimer = new ScheduledTimer()
                 .Do(ResponseTimeoutCheck)
diff --git a/src/kafka-net/Model/ConsumerOptions.cs b/src/kafka-net/Model/ConsumerOptions.cs
index d097822f..7b415878 100644
--- a/src/kafka-net/Model/ConsumerOptions.cs
+++ b/src/kafka-net/Model/ConsumerOptions.cs
@@ -17,14 +17,6 @@ public class ConsumerOptions
         /// </summary>
         public List<int> PartitionWhitelist { get; set; }
         /// <summary>
-        /// Log object to record operational messages.
-        /// </summary>
-        public IKafkaLog Log { get; set; }
-        /// <summary>
-        /// The broker router used to provide connection to each partition server.
-        /// </summary>
-        public IBrokerRouter Router { get; set; }
-        /// <summary>
         /// The time in milliseconds between queries to look for any new partitions being created.
         /// </summary>
         public int TopicPartitionQueryTimeMs { get; set; }
@@ -37,12 +29,10 @@ public class ConsumerOptions
         /// </summary>
         public int BackoffInterval { get; set; }
 
-        public ConsumerOptions(string topic, IBrokerRouter router)
+        public ConsumerOptions(string topic)
         {
             Topic = topic;
-            Router = router;
             PartitionWhitelist = new List<int>();
-            Log = new DefaultTraceLog();
             TopicPartitionQueryTimeMs = (int)TimeSpan.FromMinutes(15).TotalMilliseconds;
             ConsumerBufferSize = DefaultMaxConsumerBufferSize;
             BackoffInterval = DefaultBackoffIntervalMS;
diff --git a/src/kafka-net/Model/KafkaOptions.cs b/src/kafka-net/Model/KafkaOptions.cs
index 7b5b178f..3201f09b 100644
--- a/src/kafka-net/Model/KafkaOptions.cs
+++ b/src/kafka-net/Model/KafkaOptions.cs
@@ -1,41 +1,19 @@
 using System;
 using System.Collections.Generic;
-using System.Linq;
+using KafkaNet.Configuration;
 
 namespace KafkaNet.Model
 {
-    public class KafkaOptions
+    public class KafkaOptions : IKafkaOptions
     {
-        private const int DefaultResponseTimeout = 30000;
-
-        /// <summary>
-        /// List of Uri connections to kafka servers.  The are used to query for metadata from Kafka.  More than one is recommended.
-        /// </summary>
-        public List<Uri> KafkaServerUri { get; set; }
-        /// <summary>
-        /// Provides a factory for creating new kafka connections.
-        /// </summary>
-        public IKafkaConnectionFactory KafkaConnectionFactory { get; set; }
-        /// <summary>
-        /// Selector function for routing messages to partitions. Default is key/hash and round robin.
-        /// </summary>
-        public IPartitionSelector PartitionSelector { get; set; }
-        /// <summary>
-        /// Timeout length in milliseconds waiting for a response from kafka.
-        /// </summary>
-        public int ResponseTimeoutMs { get; set; }
-        /// <summary>
-        /// Log object to record operational messages.
-        /// </summary>
-        public IKafkaLog Log { get; set; }
-
-        public KafkaOptions(params Uri[] kafkaServerUri)
+        public KafkaOptions()
         {
-            KafkaServerUri = kafkaServerUri.ToList();
-            KafkaConnectionFactory = new DefaultKafkaConnectionFactory();
-            PartitionSelector = new DefaultPartitionSelector();
-            Log = new DefaultTraceLog();
-            ResponseTimeoutMs = DefaultResponseTimeout;
+            Timeout = TimeSpan.FromSeconds(30);
+            QueueSize = int.MaxValue;
         }
+
+        public IEnumerable<Uri> Hosts { get; set; }
+        public TimeSpan Timeout { get; set; }
+        public int QueueSize { get; set; }
     }
 }
diff --git a/src/kafka-net/Producer.cs b/src/kafka-net/Producer.cs
index 3e0fabff..d3f15990 100644
--- a/src/kafka-net/Producer.cs
+++ b/src/kafka-net/Producer.cs
@@ -1,9 +1,8 @@
 using System;
-using System.Collections.Concurrent;
 using System.Collections.Generic;
 using System.Linq;
 using System.Threading.Tasks;
-using KafkaNet.Model;
+using KafkaNet.Configuration;
 using KafkaNet.Protocol;
 using System.Threading;
 
@@ -13,7 +12,7 @@ namespace KafkaNet
     /// <summary>
     /// Provides a simplified high level API for producing messages on a topic.
     /// </summary>
-    public class Producer : IMetadataQueries, IDisposable
+    public class Producer : IMetadataQueries, IProducer
     {
         private readonly IBrokerRouter _router;
         private readonly SemaphoreSlim _sendSemaphore;
@@ -30,6 +29,7 @@ public class Producer : IMetadataQueries, IDisposable
         /// </summary>
         /// <param name="brokerRouter">The router used to direct produced messages to the correct partition.</param>
         /// <param name="maximumAsyncQueue">The maximum async calls allowed before blocking new requests.  -1 indicates unlimited.</param>
+        /// <param name="options"></param>
         /// <remarks>
         /// The maximumAsyncQueue parameter provides a mechanism for blocking an async request return if the amount of requests queue is 
         /// over a certain limit.  This is usefull if a client is trying to push a large stream of documents through the producer and
@@ -39,11 +39,11 @@ public class Producer : IMetadataQueries, IDisposable
         /// messages sitting in the async queue then a message may spend its entire timeout cycle waiting in this queue and never getting
         /// attempted to send to Kafka before a timeout exception is thrown.
         /// </remarks>
-        public Producer(IBrokerRouter brokerRouter, int maximumAsyncQueue = -1)
+        public Producer(IBrokerRouter brokerRouter, IKafkaOptions options)
         {
             _router = brokerRouter;
             _metadataQueries = new MetadataQueries(_router);
-            _maximumAsyncQueue = maximumAsyncQueue == -1 ? int.MaxValue : maximumAsyncQueue;
+            _maximumAsyncQueue = options.QueueSize;
             _sendSemaphore = new SemaphoreSlim(_maximumAsyncQueue, _maximumAsyncQueue);
         }
 
diff --git a/src/kafka-net/kafka-net.csproj b/src/kafka-net/kafka-net.csproj
index a0a5dc6a..c66a01db 100644
--- a/src/kafka-net/kafka-net.csproj
+++ b/src/kafka-net/kafka-net.csproj
@@ -43,9 +43,23 @@
   </ItemGroup>
   <ItemGroup>
     <Compile Include="Common\ThreadWall.cs" />
+    <Compile Include="Configuration\Bus.cs" />
+    <Compile Include="Configuration\BusFactory.cs" />
+    <Compile Include="Configuration\DefaultContainer.cs" />
+    <Compile Include="Configuration\DefaultServiceRegistrator.cs" />
+    <Compile Include="Interfaces\IBus.cs" />
+    <Compile Include="Configuration\IContainer.cs" />
+    <Compile Include="ConsumerFactory.cs" />
+    <Compile Include="Interfaces\IConsumer.cs" />
+    <Compile Include="Interfaces\IConsumerFactory.cs" />
+    <Compile Include="Interfaces\IKafkaOptions.cs" />
+    <Compile Include="Configuration\IServiceProvider.cs" />
+    <Compile Include="Configuration\IServiceRegistrator.cs" />
+    <Compile Include="Configuration\ServiceImplementationNotFound.cs" />
     <Compile Include="Default\DefaultKafkaConnectionFactory.cs" />
     <Compile Include="Interfaces\IMetadataQueries.cs" />
     <Compile Include="Interfaces\IKafkaTcpSocket.cs" />
+    <Compile Include="Interfaces\IProducer.cs" />
     <Compile Include="Model\BrokerRoute.cs" />
     <Compile Include="Common\ScheduledTimer.cs" />
     <Compile Include="Consumer.cs" />
@@ -85,9 +99,7 @@
   <ItemGroup>
     <None Include="kafka-net.nuspec" />
   </ItemGroup>
-  <ItemGroup>
-    <Folder Include="lib\net45\" />
-  </ItemGroup>
+  <ItemGroup />
   <Import Project="$(MSBuildToolsPath)\Microsoft.CSharp.targets" />
   <!-- To modify your build process, add your task inside one of the targets below and uncomment it. 
        Other similar extension points exist, see Microsoft.Common.targets.
diff --git a/src/kafka-tests/Fakes/BrokerRouterProxy.cs b/src/kafka-tests/Fakes/BrokerRouterProxy.cs
index 025dcead..3f14968a 100644
--- a/src/kafka-tests/Fakes/BrokerRouterProxy.cs
+++ b/src/kafka-tests/Fakes/BrokerRouterProxy.cs
@@ -31,6 +31,7 @@ public class BrokerRouterProxy
         public Func<MetadataResponse> MetadataResponse = () => DefaultMetadataResponse();
 
         public IPartitionSelector PartitionSelector = new DefaultPartitionSelector();
+        public IKafkaLog KafkaLog = new DefaultTraceLog();
 
         public BrokerRouterProxy(MoqMockingKernel kernel)
         {
@@ -50,18 +51,16 @@ public BrokerRouterProxy(MoqMockingKernel kernel)
             _fakeConn1.FetchResponseFunction = () => { Thread.Sleep(500); return null; };
             
             _factoryMock = _kernel.GetMock<IKafkaConnectionFactory>();
-            _factoryMock.Setup(x => x.Create(It.Is<Uri>(uri => uri.Port == 1), It.IsAny<int>(), It.IsAny<IKafkaLog>())).Returns(() => _fakeConn0);
-            _factoryMock.Setup(x => x.Create(It.Is<Uri>(uri => uri.Port == 2), It.IsAny<int>(), It.IsAny<IKafkaLog>())).Returns(() => _fakeConn1);
+            _factoryMock.Setup(x => x.Create(It.Is<Uri>(uri => uri.Port == 1), It.IsAny<TimeSpan>())).Returns(() => _fakeConn0);
+            _factoryMock.Setup(x => x.Create(It.Is<Uri>(uri => uri.Port == 2), It.IsAny<TimeSpan>())).Returns(() => _fakeConn1);
         }
 
         public IBrokerRouter Create()
         {
-            return new BrokerRouter(new KafkaNet.Model.KafkaOptions
+            return new BrokerRouter(new KafkaOptions
             {
-                KafkaServerUri = new List<Uri> { new Uri("http://localhost:1"), new Uri("http://localhost:2") },
-                KafkaConnectionFactory = _factoryMock.Object,
-                PartitionSelector = PartitionSelector
-            });
+                Hosts = new List<Uri> { new Uri("http://localhost:1"), new Uri("http://localhost:2") },
+            }, KafkaLog, PartitionSelector, _factoryMock.Object);
         }
 
         public static MetadataResponse DefaultMetadataResponse()
diff --git a/src/kafka-tests/Integration/GzipProducerConsumerTests.cs b/src/kafka-tests/Integration/GzipProducerConsumerTests.cs
index a26b6b5c..6f5b2fa8 100644
--- a/src/kafka-tests/Integration/GzipProducerConsumerTests.cs
+++ b/src/kafka-tests/Integration/GzipProducerConsumerTests.cs
@@ -13,11 +13,12 @@ namespace kafka_tests.Integration
     [Category("Integration")]
     public class GzipProducerConsumerTests
     {
-        private readonly KafkaOptions _options = new KafkaOptions(IntegrationConfig.IntegrationUri);
+        private readonly KafkaOptions _options = new KafkaOptions { Hosts = new [] {IntegrationConfig.IntegrationUri} };
 
-       private KafkaConnection GetKafkaConnection()
+        private KafkaConnection GetKafkaConnection()
         {
-            return new KafkaConnection(new KafkaTcpSocket(new DefaultTraceLog(), _options.KafkaServerUri.First()), _options.ResponseTimeoutMs, _options.Log);
+            var log = new DefaultTraceLog();
+            return new KafkaConnection(new KafkaTcpSocket(log, _options.Hosts.First()), (int)_options.Timeout.TotalMilliseconds, log);
         }
 
         [Test]
@@ -29,7 +30,8 @@ public void EnsureGzipCompressedMessageCanSend()
                 conn.SendAsync(new MetadataRequest { Topics = new List<string>(new[] { IntegrationConfig.IntegrationCompressionTopic }) }).Wait(TimeSpan.FromSeconds(10));
             }
 
-            using (var router = new BrokerRouter(_options))
+            var log = new DefaultTraceLog();
+            using (var router = new BrokerRouter(_options, log, new DefaultPartitionSelector(), new DefaultKafkaConnectionFactory(log)))
             {
                 var conn = router.SelectBrokerRoute(IntegrationConfig.IntegrationCompressionTopic, 0);
 
@@ -62,12 +64,13 @@ public void EnsureGzipCompressedMessageCanSend()
         [Test]
         public void EnsureGzipCanDecompressMessageFromKafka()
         {
-            var router = new BrokerRouter(_options);
-            var producer = new Producer(router);
+            var log = new DefaultTraceLog();
+            var router = new BrokerRouter(_options, log, new DefaultPartitionSelector(), new DefaultKafkaConnectionFactory(log));
+            var producer = new Producer(router, _options);
 
             var offsets = producer.GetTopicOffsetAsync(IntegrationConfig.IntegrationCompressionTopic).Result;
 
-            var consumer = new Consumer(new ConsumerOptions(IntegrationConfig.IntegrationCompressionTopic, router),
+            var consumer = new Consumer(router, log, new ConsumerOptions(IntegrationConfig.IntegrationCompressionTopic),
                 offsets.Select(x => new OffsetPosition(x.PartitionId, 0)).ToArray());
 
             var results = consumer.Consume().Take(3).ToList();
diff --git a/src/kafka-tests/Integration/KafkaConnectionIntegrationTests.cs b/src/kafka-tests/Integration/KafkaConnectionIntegrationTests.cs
index a4166759..bff38080 100644
--- a/src/kafka-tests/Integration/KafkaConnectionIntegrationTests.cs
+++ b/src/kafka-tests/Integration/KafkaConnectionIntegrationTests.cs
@@ -21,9 +21,9 @@ public class KafkaConnectionIntegrationTests
         [SetUp]
         public void Setup()
         {
-            var options = new KafkaOptions(IntegrationConfig.IntegrationUri);
+            var options = new KafkaOptions { Hosts = new [] {IntegrationConfig.IntegrationUri}};
             
-            _conn = new KafkaConnection(new KafkaTcpSocket(new DefaultTraceLog(), options.KafkaServerUri.First()), options.ResponseTimeoutMs, options.Log);
+            _conn = new KafkaConnection(new KafkaTcpSocket(new DefaultTraceLog(), options.Hosts.First()), (int)options.Timeout.TotalMilliseconds, new DefaultTraceLog());
         }
         
         [Test]
diff --git a/src/kafka-tests/Integration/OffsetManagementTests.cs b/src/kafka-tests/Integration/OffsetManagementTests.cs
index 673cfda0..c7c81bb6 100644
--- a/src/kafka-tests/Integration/OffsetManagementTests.cs
+++ b/src/kafka-tests/Integration/OffsetManagementTests.cs
@@ -13,7 +13,7 @@ namespace kafka_tests.Integration
     [Category("Integration")]
     public class OffsetManagementTests
     {
-        private readonly KafkaOptions Options = new KafkaOptions(IntegrationConfig.IntegrationUri);
+        private readonly KafkaOptions Options = new KafkaOptions { Hosts = new [] {IntegrationConfig.IntegrationUri}};
 
         [SetUp]
         public void Setup()
@@ -28,7 +28,8 @@ public void OffsetFetchRequestOfNonExistingGroupShouldReturnNoError()
             //Note that if there is no offset associated with a topic-partition under that consumer group the broker does not set an error code 
             //(since it is not really an error), but returns empty metadata and sets the offset field to -1.
             const int partitionId = 0;
-            using (var router = new BrokerRouter(Options))
+            var log = new DefaultTraceLog();
+            using (var router = new BrokerRouter(Options, log, new DefaultPartitionSelector(), new DefaultKafkaConnectionFactory(log)))
             {
                 var request = CreateOffsetFetchRequest(Guid.NewGuid().ToString(), partitionId);
 
@@ -46,7 +47,8 @@ public void OffsetFetchRequestOfNonExistingGroupShouldReturnNoError()
         public void OffsetCommitShouldStoreAndReturnSuccess()
         {
             const int partitionId = 0;
-            using (var router = new BrokerRouter(Options))
+            var log = new DefaultTraceLog();
+            using (var router = new BrokerRouter(Options, log, new DefaultPartitionSelector(), new DefaultKafkaConnectionFactory(log)))
             {
                 var conn = router.SelectBrokerRoute(IntegrationConfig.IntegrationTopic, partitionId);
 
@@ -64,9 +66,9 @@ public void OffsetCommitShouldStoreOffsetValue()
             const int partitionId = 0;
             const long offset = 99;
 
-            using (var router = new BrokerRouter(Options))
+            var log = new DefaultTraceLog();
+            using (var router = new BrokerRouter(Options, log, new DefaultPartitionSelector(), new DefaultKafkaConnectionFactory(log)))
             {
-
                 var conn = router.SelectBrokerRoute(IntegrationConfig.IntegrationTopic, partitionId);
 
                 var commit = CreateOffsetCommitRequest(IntegrationConfig.IntegrationConsumer, partitionId, offset);
@@ -92,7 +94,8 @@ public void OffsetCommitShouldStoreMetadata()
             const long offset = 101;
             const string metadata = "metadata";
 
-            using (var router = new BrokerRouter(Options))
+            var log = new DefaultTraceLog();
+            using (var router = new BrokerRouter(Options, log, new DefaultPartitionSelector(), new DefaultKafkaConnectionFactory(log)))
             {
                 var conn = router.SelectBrokerRoute(IntegrationConfig.IntegrationTopic, partitionId);
 
@@ -116,7 +119,8 @@ public void OffsetCommitShouldStoreMetadata()
         [Ignore("Not supported currently in 8.1.1?")]
         public void ConsumerMetadataRequestShouldReturnWithoutError()
         {
-            using (var router = new BrokerRouter(Options))
+            var log = new DefaultTraceLog();
+            using (var router = new BrokerRouter(Options, log, new DefaultPartitionSelector(), new DefaultKafkaConnectionFactory(log)))
             {
                 var conn = router.SelectBrokerRoute(IntegrationConfig.IntegrationTopic);
 
diff --git a/src/kafka-tests/Integration/ProducerConsumerIntegrationTests.cs b/src/kafka-tests/Integration/ProducerConsumerIntegrationTests.cs
index 17f4b3bf..2e42b11b 100644
--- a/src/kafka-tests/Integration/ProducerConsumerIntegrationTests.cs
+++ b/src/kafka-tests/Integration/ProducerConsumerIntegrationTests.cs
@@ -20,8 +20,10 @@ public class ProducerConsumerTests
         [TestCase(1000, -1)]
         public void SendAsyncShouldHandleHighVolumeOfMessages(int amount, int maxAsync)
         {
-            using (var router = new BrokerRouter(new KafkaOptions(IntegrationConfig.IntegrationUri)))
-            using (var producer = new Producer(router, maxAsync))
+            var log = new DefaultTraceLog();
+            var options = new KafkaOptions {Hosts = new[] {IntegrationConfig.IntegrationUri}, QueueSize = maxAsync};
+            using (var router = new BrokerRouter(options, log, new DefaultPartitionSelector(), new DefaultKafkaConnectionFactory(log)))
+            using (var producer = new Producer(router, options))
             {
                 var tasks = new Task<List<ProduceResponse>>[amount];
 
@@ -43,13 +45,15 @@ public void ConsumerShouldConsumeInSameOrderAsProduced()
             var expected = new List<string> { "0", "1", "2", "3", "4", "5", "6", "7", "8", "9", "10", "11", "12", "13", "14", "15", "16", "17", "18", "19" };
             var testId = Guid.NewGuid().ToString();
 
-            using (var router = new BrokerRouter(new KafkaOptions(IntegrationConfig.IntegrationUri)))
-            using (var producer = new Producer(router))
+            var log = new DefaultTraceLog();
+            var options = new KafkaOptions {Hosts = new[] {IntegrationConfig.IntegrationUri}};
+            using (var router = new BrokerRouter(options, log, new DefaultPartitionSelector(), new DefaultKafkaConnectionFactory(log)))
+            using (var producer = new Producer(router, options))
             {
 
                 var offsets = producer.GetTopicOffsetAsync(IntegrationConfig.IntegrationTopic).Result;
 
-                using (var consumer = new Consumer(new ConsumerOptions(IntegrationConfig.IntegrationTopic, router),
+                using (var consumer = new Consumer(router, log, new ConsumerOptions(IntegrationConfig.IntegrationTopic),
                     offsets.Select(x => new OffsetPosition(x.PartitionId, x.Offsets.Max())).ToArray()))
                 {
 
@@ -76,13 +80,15 @@ public void ConsumerShouldBeAbleToSeekBackToEarlierOffset()
             var expected = new List<string> { "0", "1", "2", "3", "4", "5", "6", "7", "8", "9", "10", "11", "12", "13", "14", "15", "16", "17", "18", "19" };
             var testId = Guid.NewGuid().ToString();
 
-            using (var router = new BrokerRouter(new KafkaOptions(IntegrationConfig.IntegrationUri)))
-            using (var producer = new Producer(router))
+            var log = new DefaultTraceLog();
+            var options = new KafkaOptions {Hosts = new[] {IntegrationConfig.IntegrationUri}};
+            using (var router = new BrokerRouter(options, log, new DefaultPartitionSelector(), new DefaultKafkaConnectionFactory(log)))
+            using (var producer = new Producer(router, options))
             {
                 var offsets = producer.GetTopicOffsetAsync(IntegrationConfig.IntegrationTopic).Result
                     .Select(x => new OffsetPosition(x.PartitionId, x.Offsets.Max())).ToArray();
 
-                using (var consumer = new Consumer(new ConsumerOptions(IntegrationConfig.IntegrationTopic, router), offsets))
+                using (var consumer = new Consumer(router, log, new ConsumerOptions(IntegrationConfig.IntegrationTopic), offsets))
                 {
                     for (int i = 0; i < 20; i++)
                     {
@@ -116,14 +122,16 @@ public void ConsumerShouldBeAbleToSeekBackToEarlierOffset()
         [Test]
         public void ConsumerShouldBeAbleToGetCurrentOffsetInformation()
         {
-            using (var router = new BrokerRouter(new KafkaOptions(IntegrationConfig.IntegrationUri)))
-            using (var producer = new Producer(router))
+            var log = new DefaultTraceLog();
+            var options = new KafkaOptions {Hosts = new[] {IntegrationConfig.IntegrationUri}};
+            using (var router = new BrokerRouter(options, log, new DefaultPartitionSelector(), new DefaultKafkaConnectionFactory(log)))
+            using (var producer = new Producer(router, options))
             {
 
                 var startOffsets = producer.GetTopicOffsetAsync(IntegrationConfig.IntegrationTopic).Result
                     .Select(x => new OffsetPosition(x.PartitionId, x.Offsets.Max())).ToArray();
 
-                using (var consumer = new Consumer(new ConsumerOptions(IntegrationConfig.IntegrationTopic, router), startOffsets))
+                using (var consumer = new Consumer(router, log, new ConsumerOptions(IntegrationConfig.IntegrationTopic), startOffsets))
                 {
 
                     for (int i = 0; i < 20; i++)
@@ -151,13 +159,15 @@ public void ConsumerShouldNotLoseMessageWhenBlocked()
         {
             var testId = Guid.NewGuid().ToString();
 
-            using (var router = new BrokerRouter(new KafkaOptions(IntegrationConfig.IntegrationUri)))
-            using (var producer = new Producer(router))
+            var log = new DefaultTraceLog();
+            var options = new KafkaOptions {Hosts = new[] {IntegrationConfig.IntegrationUri}};
+            using (var router = new BrokerRouter(options, log, new DefaultPartitionSelector(), new DefaultKafkaConnectionFactory(log)))
+            using (var producer = new Producer(router, options))
             {
                 var offsets = producer.GetTopicOffsetAsync(IntegrationConfig.IntegrationTopic).Result;
 
                 //create consumer with buffer size of 1 (should block upstream)
-                using (var consumer = new Consumer(new ConsumerOptions(IntegrationConfig.IntegrationTopic, router) { ConsumerBufferSize = 1 },
+                using (var consumer = new Consumer(router, log, new ConsumerOptions(IntegrationConfig.IntegrationTopic) { ConsumerBufferSize = 1 },
                     offsets.Select(x => new OffsetPosition(x.PartitionId, x.Offsets.Max())).ToArray()))
                 {
 
@@ -180,15 +190,17 @@ public void ConsumerShouldNotLoseMessageWhenBlocked()
         [Test]
         public void ConsumerShouldMoveToNextAvailableOffsetWhenQueryingForNextMessage()
         {
-            using (var router = new BrokerRouter(new KafkaOptions(IntegrationConfig.IntegrationUri)))
-            using (var producer = new Producer(router))
+            var log = new DefaultTraceLog();
+            var options = new KafkaOptions {Hosts = new[] {IntegrationConfig.IntegrationUri}};
+            using (var router = new BrokerRouter(options, log, new DefaultPartitionSelector(), new DefaultKafkaConnectionFactory(log)))
+            using (var producer = new Producer(router, options))
             {
                 var offsets = producer.GetTopicOffsetAsync(IntegrationConfig.IntegrationTopic).Result;
                 Assert.That(offsets.Count, Is.EqualTo(2), "This test requires there to be exactly two paritions.");
                 Assert.That(offsets.Count(x => x.Offsets.Max(o => o) > 1000), Is.EqualTo(2), "Need more than 1000 messages in each topic for this test to work.");
 
                 //set offset 1000 messages back on one partition.  We should be able to get all 1000 messages over multiple calls.
-                using (var consumer = new Consumer(new ConsumerOptions(IntegrationConfig.IntegrationTopic, router),
+                using (var consumer = new Consumer(router, log, new ConsumerOptions(IntegrationConfig.IntegrationTopic),
                      offsets.Select(x => new OffsetPosition(x.PartitionId, x.Offsets.Max() - 1000)).ToArray()))
                 {
                     var data = new List<Message>();
diff --git a/src/kafka-tests/Unit/BrokerRouterTests.cs b/src/kafka-tests/Unit/BrokerRouterTests.cs
index 1b3a83b1..d0cd1c24 100644
--- a/src/kafka-tests/Unit/BrokerRouterTests.cs
+++ b/src/kafka-tests/Unit/BrokerRouterTests.cs
@@ -1,4 +1,5 @@
 using KafkaNet;
+using KafkaNet.Model;
 using KafkaNet.Protocol;
 using Moq;
 using Ninject.MockingKernel.Moq;
@@ -18,6 +19,7 @@ public class BrokerRouterTests
         private Mock<IKafkaConnection> _connMock1;
         private Mock<IKafkaConnectionFactory> _factoryMock;
         private Mock<IPartitionSelector> _partitionSelectorMock;
+        private Mock<IKafkaLog> _logMock;
 
         [SetUp]
         public void Setup()
@@ -26,19 +28,19 @@ public void Setup()
 
             //setup mock IKafkaConnection
             _partitionSelectorMock = _kernel.GetMock<IPartitionSelector>();
+            _logMock = _kernel.GetMock<IKafkaLog>();
             _connMock1 = _kernel.GetMock<IKafkaConnection>();
             _factoryMock = _kernel.GetMock<IKafkaConnectionFactory>();
-            _factoryMock.Setup(x => x.Create(It.Is<Uri>(uri => uri.Port == 1), It.IsAny<int>(), It.IsAny<IKafkaLog>())).Returns(() => _connMock1.Object);
+            _factoryMock.Setup(x => x.Create(It.Is<Uri>(uri => uri.Port == 1), It.IsAny<TimeSpan>())).Returns(() => _connMock1.Object);
         }
 
         [Test]
         public void BrokerRouterCanConstruct()
         {
-            var result = new BrokerRouter(new KafkaNet.Model.KafkaOptions
+            var result = new BrokerRouter(new KafkaOptions
             {
-                KafkaServerUri = new List<Uri> { new Uri("http://localhost:1") },
-                KafkaConnectionFactory = _factoryMock.Object
-            });
+                Hosts = new List<Uri> { new Uri("http://localhost:1") },
+            }, _logMock.Object, _partitionSelectorMock.Object, _factoryMock.Object);
 
             Assert.That(result, Is.Not.Null);
         }
@@ -46,17 +48,16 @@ public void BrokerRouterCanConstruct()
         [Test]
         public void BrokerRouterUsesFactoryToAddNewBrokers()
         {
-            var router = new BrokerRouter(new KafkaNet.Model.KafkaOptions
+            var router = new BrokerRouter(new KafkaOptions
             {
-                KafkaServerUri = new List<Uri> { new Uri("http://localhost:1") },
-                KafkaConnectionFactory = _factoryMock.Object
-            });
+                Hosts = new List<Uri> { new Uri("http://localhost:1") },
+            }, _logMock.Object, _partitionSelectorMock.Object, _factoryMock.Object);
 
             _connMock1.Setup(x => x.SendAsync(It.IsAny<IKafkaRequest<MetadataResponse>>()))
                       .Returns(() => Task.Factory.StartNew(() => new List<MetadataResponse> { CreateMetaResponse() }));
 
             var topics = router.GetTopicMetadata(TestTopic);
-            _factoryMock.Verify(x => x.Create(It.Is<Uri>(uri => uri.Port == 2), It.IsAny<int>(), It.IsAny<IKafkaLog>()), Times.Once());
+            _factoryMock.Verify(x => x.Create(It.Is<Uri>(uri => uri.Port == 2), It.IsAny<TimeSpan>()), Times.Once());
         }
 
         #region MetadataRequest Tests...
diff --git a/src/kafka-tests/Unit/ConsumerTests.cs b/src/kafka-tests/Unit/ConsumerTests.cs
index 9ee33fff..a40c2678 100644
--- a/src/kafka-tests/Unit/ConsumerTests.cs
+++ b/src/kafka-tests/Unit/ConsumerTests.cs
@@ -22,13 +22,14 @@ public class ConsumerTests
         public void CancellationShouldInterruptConsumption()
         {
             var routerProxy = new BrokerRouterProxy(new MoqMockingKernel());
-            routerProxy.BrokerConn0.FetchResponseFunction = () => { return new FetchResponse(); };
+            routerProxy.BrokerConn0.FetchResponseFunction = () => new FetchResponse();
 
             var router = routerProxy.Create();
 
-            var options = CreateOptions(router);
 
-            using (var consumer = new Consumer(options))
+            var options = CreateOptions();
+
+            using (var consumer = new Consumer(router, new DefaultTraceLog(), options))
             {
                 var tokenSrc = new CancellationTokenSource();
 
@@ -49,11 +50,11 @@ public void CancellationShouldInterruptConsumption()
         public void ConsumerWhitelistShouldOnlyConsumeSpecifiedPartition()
         {
             var routerProxy = new BrokerRouterProxy(new MoqMockingKernel());
-            routerProxy.BrokerConn0.FetchResponseFunction = () => { return new FetchResponse(); };
+            routerProxy.BrokerConn0.FetchResponseFunction = () => new FetchResponse();
             var router = routerProxy.Create();
-            var options = CreateOptions(router);
+            var options = CreateOptions();
             options.PartitionWhitelist = new List<int> { 0 };
-            using (var consumer = new Consumer(options))
+            using (var consumer = new Consumer(router, new DefaultTraceLog(), options))
             {
                 var test = consumer.Consume();
 
@@ -72,10 +73,10 @@ public void ConsumerWithEmptyWhitelistShouldConsumeAllPartition()
             var routerProxy = new BrokerRouterProxy(new MoqMockingKernel());
 
             var router = routerProxy.Create();
-            var options = CreateOptions(router);
+            var options = CreateOptions();
             options.PartitionWhitelist = new List<int>();
 
-            using (var consumer = new Consumer(options))
+            using (var consumer = new Consumer(router, new DefaultTraceLog(), options))
             {
                 var test = consumer.Consume();
 
@@ -93,11 +94,11 @@ public void ConsumerWithEmptyWhitelistShouldConsumeAllPartition()
         public void ConsumerShouldCreateTaskForEachBroker()
         {
             var routerProxy = new BrokerRouterProxy(new MoqMockingKernel());
-            routerProxy.BrokerConn0.FetchResponseFunction = () => { return new FetchResponse(); };
+            routerProxy.BrokerConn0.FetchResponseFunction = () => new FetchResponse();
             var router = routerProxy.Create();
-            var options = CreateOptions(router);
+            var options = CreateOptions();
             options.PartitionWhitelist = new List<int>();
-            using (var consumer = new Consumer(options))
+            using (var consumer = new Consumer(router, new DefaultTraceLog(), options))
             {
                 var test = consumer.Consume();
                 TaskTest.WaitFor(() => consumer.ConsumerTaskCount >= 2);
@@ -111,11 +112,11 @@ public void ConsumerShouldCreateTaskForEachBroker()
         public void ConsumerShouldReturnOffset()
         {
             var routerProxy = new BrokerRouterProxy(new MoqMockingKernel());
-            routerProxy.BrokerConn0.FetchResponseFunction = () => { return new FetchResponse(); };
+            routerProxy.BrokerConn0.FetchResponseFunction = () => new FetchResponse();
             var router = routerProxy.Create();
-            var options = CreateOptions(router);
+            var options = CreateOptions();
             options.PartitionWhitelist = new List<int>();
-            using (var consumer = new Consumer(options))
+            using (var consumer = new Consumer(router, new DefaultTraceLog(), options))
             {
                 var test = consumer.Consume();
                 TaskTest.WaitFor(() => consumer.ConsumerTaskCount >= 2);
@@ -128,7 +129,7 @@ public void ConsumerShouldReturnOffset()
         public void EnsureConsumerDisposesRouter()
         {
             var router = new MoqMockingKernel().GetMock<IBrokerRouter>();
-            var consumer = new Consumer(CreateOptions(router.Object));
+            var consumer = new Consumer(router.Object, new DefaultTraceLog(), CreateOptions());
             using (consumer) { }
             router.Verify(x => x.Dispose(), Times.Once());
         }
@@ -137,12 +138,12 @@ public void EnsureConsumerDisposesRouter()
         public void EnsureConsumerDisposesAllTasks()
         {
             var routerProxy = new BrokerRouterProxy(new MoqMockingKernel());
-            routerProxy.BrokerConn0.FetchResponseFunction = () => { return new FetchResponse(); };
+            routerProxy.BrokerConn0.FetchResponseFunction = () => new FetchResponse();
             var router = routerProxy.Create();
-            var options = CreateOptions(router);
+            var options = CreateOptions();
             options.PartitionWhitelist = new List<int>();
 
-            var consumer = new Consumer(options);
+            var consumer = new Consumer(router, new DefaultTraceLog(), options);
             using (consumer)
             {
                 var test = consumer.Consume();
@@ -153,9 +154,9 @@ public void EnsureConsumerDisposesAllTasks()
             Assert.That(consumer.ConsumerTaskCount, Is.EqualTo(0));
         }
 
-        private ConsumerOptions CreateOptions(IBrokerRouter router)
+        private ConsumerOptions CreateOptions()
         {
-            return new ConsumerOptions(BrokerRouterProxy.TestTopic, router);
+            return new ConsumerOptions(BrokerRouterProxy.TestTopic);
         }
     }
 }
diff --git a/src/kafka-tests/Unit/KafkaConnectionTests.cs b/src/kafka-tests/Unit/KafkaConnectionTests.cs
index 27c5817b..73c70a99 100644
--- a/src/kafka-tests/Unit/KafkaConnectionTests.cs
+++ b/src/kafka-tests/Unit/KafkaConnectionTests.cs
@@ -33,7 +33,7 @@ public void Setup()
         public void ShouldStartReadPollingOnConstruction()
         {
             using (var socket = new KafkaTcpSocket(_log, new Uri("http://localhost:8999")))
-            using (var conn = new KafkaConnection(socket, log: _log))
+            using (var conn = new KafkaConnection(socket, 10000, _log))
             {
                 TaskTest.WaitFor(() => conn.ReadPolling);
                 Assert.That(conn.ReadPolling, Is.True);
@@ -45,7 +45,7 @@ public void ShouldReportServerUriOnConstruction()
         {
             var expectedUrl = new Uri("http://localhost:8999");
             using (var socket = new KafkaTcpSocket(_log, expectedUrl))
-            using (var conn = new KafkaConnection(socket, log: _log))
+            using (var conn = new KafkaConnection(socket, 10000, _log))
             {
                 Assert.That(conn.KafkaUri, Is.EqualTo(expectedUrl));
             }
@@ -59,7 +59,7 @@ public void ShouldDisposeWithoutExceptionThrown()
             using (var server = new FakeTcpServer(8999))
             using (var socket = new KafkaTcpSocket(_log, new Uri("http://localhost:8999")))
             {
-                var conn = new KafkaConnection(socket, log: _log);
+                var conn = new KafkaConnection(socket, 10000, _log);
                 TaskTest.WaitFor(() => server.ConnectionEventcount > 0);
                 using (conn) { }
             }
@@ -69,7 +69,7 @@ public void ShouldDisposeWithoutExceptionThrown()
         public void ShouldDisposeWithoutExceptionEvenWhileCallingSendAsync()
         {
             using (var socket = new KafkaTcpSocket(_log, new Uri("http://localhost:8999")))
-            using (var conn = new KafkaConnection(socket, log: _log))
+            using (var conn = new KafkaConnection(socket, 10000, _log))
             {
                 var task = conn.SendAsync(new MetadataRequest());
                 task.Wait(TimeSpan.FromMilliseconds(1000));
@@ -86,7 +86,7 @@ public void ReadShouldLogDisconnectAndRecover()
 
             using (var server = new FakeTcpServer(8999))
             using (var socket = new KafkaTcpSocket(mockLog.Object, new Uri("http://localhost:8999")))
-            using (var conn = new KafkaConnection(socket, log: mockLog.Object))
+            using (var conn = new KafkaConnection(socket, 10000, mockLog.Object))
             {
                 TaskTest.WaitFor(() => server.ConnectionEventcount > 0);
                 Assert.That(server.ConnectionEventcount, Is.EqualTo(1));
@@ -112,7 +112,7 @@ public void ReadShouldIgnoreMessageWithUnknownCorrelationId()
 
             using (var server = new FakeTcpServer(8999))
             using (var socket = new KafkaTcpSocket(mockLog.Object, new Uri("http://localhost:8999")))
-            using (var conn = new KafkaConnection(socket, log: mockLog.Object))
+            using (var conn = new KafkaConnection(socket, 10000, mockLog.Object))
             {
                 //send correlation message
                 server.SendDataAsync(CreateCorrelationMessage(correlationId)).Wait(TimeSpan.FromSeconds(1));
diff --git a/src/kafka-tests/Unit/ProducerTests.cs b/src/kafka-tests/Unit/ProducerTests.cs
index 2b26283e..fda520ea 100644
--- a/src/kafka-tests/Unit/ProducerTests.cs
+++ b/src/kafka-tests/Unit/ProducerTests.cs
@@ -1,7 +1,6 @@
 using System;
 using System.Collections.Generic;
-using System.Linq;
-using System.Text;
+using System.Threading;
 using System.Threading.Tasks;
 using KafkaNet;
 using KafkaNet.Model;
@@ -9,7 +8,6 @@
 using Moq;
 using NUnit.Framework;
 using Ninject.MockingKernel.Moq;
-using System.Threading;
 using kafka_tests.Helpers;
 
 namespace kafka_tests.Unit
@@ -18,9 +16,6 @@ namespace kafka_tests.Unit
     [Category("Unit")]
     public class ProducerTests
     {
-        private MoqMockingKernel _kernel;
-        private BrokerRouterProxy _routerProxy;
-
         [SetUp]
         public void Setup()
         {
@@ -28,117 +23,141 @@ public void Setup()
             _routerProxy = new BrokerRouterProxy(_kernel);
         }
 
-        #region SendMessageAsync Tests...
+        private MoqMockingKernel _kernel;
+        private BrokerRouterProxy _routerProxy;
+
         [Test]
-        public void ProducerShouldGroupMessagesByBroker()
+        [Ignore("is there a way to communicate back which client failed and which succeeded.")]
+        public void ConnectionExceptionOnOneShouldCommunicateBackWhichMessagesFailed()
         {
-            var router = _routerProxy.Create();
-            using (var producer = new Producer(router))
+            //TODO is there a way to communicate back which client failed and which succeeded.
+            var routerProxy = new BrokerRouterProxy(_kernel);
+            routerProxy.BrokerConn1.ProduceResponseFunction = () => { throw new ApplicationException("some exception"); };
+
+            var router = routerProxy.Create();
+            using (var producer = new Producer(router, new KafkaOptions
+                {
+                    Hosts = new List<Uri> {new Uri("http://localhost:1"), new Uri("http://localhost:2")},
+                }))
             {
                 var messages = new List<Message>
-                {
-                    new Message{Value = "1"}, new Message{Value = "2"}
-                };
-
-                var response = producer.SendMessageAsync("UnitTest", messages).Result;
+                    {
+                        new Message {Value = "1"}, new Message {Value = "2"}
+                    };
 
-                Assert.That(response.Count, Is.EqualTo(2));
-                Assert.That(_routerProxy.BrokerConn0.ProduceRequestCallCount, Is.EqualTo(1));
-                Assert.That(_routerProxy.BrokerConn1.ProduceRequestCallCount, Is.EqualTo(1));
+                //this will produce an exception, but message 1 succeeded and message 2 did not.  
+                //should we return a ProduceResponse with an error and no error for the other messages?
+                //at this point though the client does not know which message is routed to which server.  
+                //the whole batch of messages would need to be returned.
+                var test = producer.SendMessageAsync("UnitTest", messages).Result;
             }
         }
 
         [Test]
-        public void ShouldSendAsyncToAllConnectionsEvenWhenExceptionOccursOnOne()
+        public void EnsureProducerDisposesRouter()
         {
-            var routerProxy = new BrokerRouterProxy(_kernel);
-            routerProxy.BrokerConn1.ProduceResponseFunction = () => { throw new ApplicationException("some exception"); };
+            var router = _kernel.GetMock<IBrokerRouter>();
+            var producer = new Producer(router.Object, new KafkaOptions
+                {
+                    Hosts = new List<Uri> {new Uri("http://localhost:1"), new Uri("http://localhost:2")},
+                });
+            using (producer)
+            {
+            }
+            router.Verify(x => x.Dispose(), Times.Once());
+        }
 
-            var router = routerProxy.Create();
-            using (var producer = new Producer(router))
+        [Test]
+        public void ProducerShouldGroupMessagesByBroker()
+        {
+            var router = _routerProxy.Create();
+            using (var producer = new Producer(router, new KafkaOptions
+                {
+                    Hosts = new List<Uri> {new Uri("http://localhost:1"), new Uri("http://localhost:2")},
+                }))
             {
                 var messages = new List<Message>
-                {
-                    new Message{Value = "1"}, new Message{Value = "2"}
-                };
+                    {
+                        new Message {Value = "1"}, new Message {Value = "2"}
+                    };
 
-                producer.SendMessageAsync("UnitTest", messages).ContinueWith(t =>
-                {
-                    Assert.That(t.IsFaulted, Is.True);
-                    Assert.That(t.Exception, Is.Not.Null);
-                    Assert.That(t.Exception.ToString(), Is.StringContaining("ApplicationException"));
-                    Assert.That(routerProxy.BrokerConn0.ProduceRequestCallCount, Is.EqualTo(1));
-                    Assert.That(routerProxy.BrokerConn1.ProduceRequestCallCount, Is.EqualTo(1));
-                }).Wait(TimeSpan.FromSeconds(10));
+                var response = producer.SendMessageAsync("UnitTest", messages).Result;
+
+                Assert.That(response.Count, Is.EqualTo(2));
+                Assert.That(_routerProxy.BrokerConn0.ProduceRequestCallCount, Is.EqualTo(1));
+                Assert.That(_routerProxy.BrokerConn1.ProduceRequestCallCount, Is.EqualTo(1));
             }
         }
 
         [Test]
         public void SendAsyncShouldBlockWhenMaximumAsyncQueueReached()
         {
-            int count = 0;
+            var count = 0;
             var semaphore = new SemaphoreSlim(0);
             var routerProxy = new BrokerRouterProxy(_kernel);
             //block the second call returning from send message async
-            routerProxy.BrokerConn0.ProduceResponseFunction = () => { semaphore.Wait(); return new ProduceResponse(); };
+            routerProxy.BrokerConn0.ProduceResponseFunction = () =>
+                {
+                    semaphore.Wait();
+                    return new ProduceResponse();
+                };
 
             var router = routerProxy.Create();
-            using (var producer = new Producer(router, 1))
-            {
-                var messages = new List<Message>
+            var options = new KafkaOptions
                 {
-                    new Message{Value = "1"}, new Message{Value = "2"}
+                    Hosts = new List<Uri> {new Uri("http://localhost:1"), new Uri("http://localhost:2")}, 
+                    QueueSize = 1
                 };
+            using (var producer = new Producer(router, options))
+            {
+                var messages = new List<Message>
+                    {
+                        new Message {Value = "1"}, new Message {Value = "2"}
+                    };
 
                 Task.Factory.StartNew(() =>
-                {
-                    producer.SendMessageAsync(BrokerRouterProxy.TestTopic, messages);
-                    count++;
-                    producer.SendMessageAsync(BrokerRouterProxy.TestTopic, messages);
-                    count++;
-                });
+                    {
+                        producer.SendMessageAsync(BrokerRouterProxy.TestTopic, messages);
+                        count++;
+                        producer.SendMessageAsync(BrokerRouterProxy.TestTopic, messages);
+                        count++;
+                    });
 
                 TaskTest.WaitFor(() => count > 0);
                 Assert.That(count, Is.EqualTo(1), "Only one SendMessageAsync should continue.");
                 Assert.That(producer.ActiveCount, Is.EqualTo(1), "One async call shoud be active.");
                 semaphore.Release();
                 TaskTest.WaitFor(() => count > 1);
-                Assert.That(count, Is.EqualTo(2), "The second SendMessageAsync should continue after semaphore is released.");
+                Assert.That(Thread.VolatileRead(ref count), Is.EqualTo(2), "The second SendMessageAsync should continue after semaphore is released.");
             }
         }
 
         [Test]
-        [Ignore("is there a way to communicate back which client failed and which succeeded.")]
-        public void ConnectionExceptionOnOneShouldCommunicateBackWhichMessagesFailed()
+        public void ShouldSendAsyncToAllConnectionsEvenWhenExceptionOccursOnOne()
         {
-            //TODO is there a way to communicate back which client failed and which succeeded.
             var routerProxy = new BrokerRouterProxy(_kernel);
             routerProxy.BrokerConn1.ProduceResponseFunction = () => { throw new ApplicationException("some exception"); };
 
             var router = routerProxy.Create();
-            using (var producer = new Producer(router))
+            using (var producer = new Producer(router, new KafkaOptions
+            {
+                Hosts = new List<Uri> { new Uri("http://localhost:1"), new Uri("http://localhost:2") },
+            }))
             {
                 var messages = new List<Message>
-                {
-                    new Message{Value = "1"}, new Message{Value = "2"}
-                };
+                    {
+                        new Message {Value = "1"}, new Message {Value = "2"}
+                    };
 
-                //this will produce an exception, but message 1 succeeded and message 2 did not.  
-                //should we return a ProduceResponse with an error and no error for the other messages?
-                //at this point though the client does not know which message is routed to which server.  
-                //the whole batch of messages would need to be returned.
-                var test = producer.SendMessageAsync("UnitTest", messages).Result;
+                producer.SendMessageAsync("UnitTest", messages).ContinueWith(t =>
+                    {
+                        Assert.That(t.IsFaulted, Is.True);
+                        Assert.That(t.Exception, Is.Not.Null);
+                        Assert.That(t.Exception.ToString(), Is.StringContaining("ApplicationException"));
+                        Assert.That(routerProxy.BrokerConn0.ProduceRequestCallCount, Is.EqualTo(1));
+                        Assert.That(routerProxy.BrokerConn1.ProduceRequestCallCount, Is.EqualTo(1));
+                    }).Wait(TimeSpan.FromSeconds(10));
             }
         }
-        #endregion
-
-        [Test]
-        public void EnsureProducerDisposesRouter()
-        {
-            var router = _kernel.GetMock<IBrokerRouter>();
-            var producer = new Producer(router.Object);
-            using (producer) { }
-            router.Verify(x => x.Dispose(), Times.Once());
-        }
     }
-}
+}
\ No newline at end of file

From b26136a0be0e24ba39821c6fa15f2295b862979a Mon Sep 17 00:00:00 2001
From: Yury Pliner <yury.pliner@gmail.com>
Date: Tue, 29 Jul 2014 22:14:19 +0600
Subject: [PATCH 2/2] Pluggable container and tests for #12

Here is a second attempt to #12:
1) refactored IContainer and its default implementation. Tests included.
2) Bus factory is not static anymore and can use abstact IContainer
3) Windsor adapter of IContainer
---
 src/TestHarness/Program.cs                    |  2 +-
 src/kafka-net.sln                             | 10 ++-
 .../Properties/AssemblyInfo.cs                | 36 ++++++++
 src/kafka-net.windsor/WindsorAdapter.cs       | 40 +++++++++
 .../kafka-net.windsor.csproj                  | 82 +++++++++++++++++++
 src/kafka-net.windsor/packages.config         |  5 ++
 src/kafka-net/Configuration/Bus.cs            |  4 +-
 src/kafka-net/Configuration/BusFactory.cs     | 20 +++--
 .../Configuration/DefaultContainer.cs         | 29 +++----
 .../DefaultServiceRegistrator.cs              | 14 ++--
 .../Configuration/IServiceRegistrator.cs      |  3 +-
 .../ServiceImplementationNotFound.cs          |  8 --
 .../Configuration/ServiceNotFound.cs          | 16 ++++
 src/kafka-net/kafka-net.csproj                |  2 +-
 .../Configuration/AnotherService.cs           |  6 ++
 .../Configuration/DefaultContainerTest.cs     | 42 ++++++++++
 src/kafka-tests/Configuration/IService.cs     |  6 ++
 src/kafka-tests/Configuration/OneService.cs   |  6 ++
 .../Configuration/WindsorAdapterTest.cs       | 44 ++++++++++
 src/kafka-tests/kafka-tests.csproj            | 15 ++++
 src/kafka-tests/packages.config               |  2 +
 21 files changed, 346 insertions(+), 46 deletions(-)
 create mode 100644 src/kafka-net.windsor/Properties/AssemblyInfo.cs
 create mode 100644 src/kafka-net.windsor/WindsorAdapter.cs
 create mode 100644 src/kafka-net.windsor/kafka-net.windsor.csproj
 create mode 100644 src/kafka-net.windsor/packages.config
 delete mode 100644 src/kafka-net/Configuration/ServiceImplementationNotFound.cs
 create mode 100644 src/kafka-net/Configuration/ServiceNotFound.cs
 create mode 100644 src/kafka-tests/Configuration/AnotherService.cs
 create mode 100644 src/kafka-tests/Configuration/DefaultContainerTest.cs
 create mode 100644 src/kafka-tests/Configuration/IService.cs
 create mode 100644 src/kafka-tests/Configuration/OneService.cs
 create mode 100644 src/kafka-tests/Configuration/WindsorAdapterTest.cs

diff --git a/src/TestHarness/Program.cs b/src/TestHarness/Program.cs
index 440f6612..228a8c1a 100644
--- a/src/TestHarness/Program.cs
+++ b/src/TestHarness/Program.cs
@@ -10,7 +10,7 @@ class Program
     {
         static void Main(string[] args)
         {
-            var bus = BusFactory.Create(new KafkaOptions
+            var bus = new BusFactory().Create(new KafkaOptions
                 {
                     Hosts = new[] {new Uri("http://CSDKAFKA01:9092"), new Uri("http://CSDKAFKA02:9092")}
                 }, x => { });
diff --git a/src/kafka-net.sln b/src/kafka-net.sln
index e73809e7..835f34ee 100644
--- a/src/kafka-net.sln
+++ b/src/kafka-net.sln
@@ -1,8 +1,6 @@
 
 Microsoft Visual Studio Solution File, Format Version 12.00
-# Visual Studio 2013
-VisualStudioVersion = 12.0.30110.0
-MinimumVisualStudioVersion = 10.0.40219.1
+# Visual Studio 2012
 Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "kafka-net", "kafka-net\kafka-net.csproj", "{1343EB68-55CB-4452-8386-24A9989DE1C0}"
 EndProject
 Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "kafka-tests", "kafka-tests\kafka-tests.csproj", "{D80AE407-BB81-4C11-BFDC-5DD463F8B1BF}"
@@ -28,6 +26,8 @@ Project("{2150E333-8FDC-42A3-9474-1A3956D46DE8}") = "Solution Items", "Solution
 		..\version = ..\version
 	EndProjectSection
 EndProject
+Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "kafka-net.windsor", "kafka-net.windsor\kafka-net.windsor.csproj", "{29ED0088-40F5-452F-95D4-9F8CD33156D3}"
+EndProject
 Global
 	GlobalSection(SolutionConfigurationPlatforms) = preSolution
 		Debug|Any CPU = Debug|Any CPU
@@ -50,6 +50,10 @@ Global
 		{FF7E1490-7453-47B0-A12E-64FED927825A}.Debug|Any CPU.Build.0 = Debug|Any CPU
 		{FF7E1490-7453-47B0-A12E-64FED927825A}.Release|Any CPU.ActiveCfg = Release|Any CPU
 		{FF7E1490-7453-47B0-A12E-64FED927825A}.Release|Any CPU.Build.0 = Release|Any CPU
+		{29ED0088-40F5-452F-95D4-9F8CD33156D3}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
+		{29ED0088-40F5-452F-95D4-9F8CD33156D3}.Debug|Any CPU.Build.0 = Debug|Any CPU
+		{29ED0088-40F5-452F-95D4-9F8CD33156D3}.Release|Any CPU.ActiveCfg = Release|Any CPU
+		{29ED0088-40F5-452F-95D4-9F8CD33156D3}.Release|Any CPU.Build.0 = Release|Any CPU
 	EndGlobalSection
 	GlobalSection(SolutionProperties) = preSolution
 		HideSolutionNode = FALSE
diff --git a/src/kafka-net.windsor/Properties/AssemblyInfo.cs b/src/kafka-net.windsor/Properties/AssemblyInfo.cs
new file mode 100644
index 00000000..d161c1db
--- /dev/null
+++ b/src/kafka-net.windsor/Properties/AssemblyInfo.cs
@@ -0,0 +1,36 @@
+using System.Reflection;
+using System.Runtime.CompilerServices;
+using System.Runtime.InteropServices;
+
+// General Information about an assembly is controlled through the following 
+// set of attributes. Change these attribute values to modify the information
+// associated with an assembly.
+[assembly: AssemblyTitle("kafka-net.windsor")]
+[assembly: AssemblyDescription("")]
+[assembly: AssemblyConfiguration("")]
+[assembly: AssemblyCompany("")]
+[assembly: AssemblyProduct("kafka-net.windsor")]
+[assembly: AssemblyCopyright("Copyright ©  2014")]
+[assembly: AssemblyTrademark("")]
+[assembly: AssemblyCulture("")]
+
+// Setting ComVisible to false makes the types in this assembly not visible 
+// to COM components.  If you need to access a type in this assembly from 
+// COM, set the ComVisible attribute to true on that type.
+[assembly: ComVisible(false)]
+
+// The following GUID is for the ID of the typelib if this project is exposed to COM
+[assembly: Guid("fdefc18c-e9e3-4150-aecf-bbe1da358dae")]
+
+// Version information for an assembly consists of the following four values:
+//
+//      Major Version
+//      Minor Version 
+//      Build Number
+//      Revision
+//
+// You can specify all the values or you can default the Build and Revision Numbers 
+// by using the '*' as shown below:
+// [assembly: AssemblyVersion("1.0.*")]
+[assembly: AssemblyVersion("1.0.0.0")]
+[assembly: AssemblyFileVersion("1.0.0.0")]
diff --git a/src/kafka-net.windsor/WindsorAdapter.cs b/src/kafka-net.windsor/WindsorAdapter.cs
new file mode 100644
index 00000000..fb1f7941
--- /dev/null
+++ b/src/kafka-net.windsor/WindsorAdapter.cs
@@ -0,0 +1,40 @@
+using Castle.MicroKernel.Registration;
+using Castle.Windsor;
+using KafkaNet.Configuration;
+
+namespace KafkaNet.Windsor
+{
+    public class WindsorAdapter : IContainer
+    {
+        private readonly IWindsorContainer _container;
+
+        public WindsorAdapter(IWindsorContainer container)
+        {
+            this._container = container;
+        }
+
+        public T Resolve<T>() where T : class
+        {
+            try
+            {
+                return _container.Resolve<T>();
+            }
+            catch (Castle.MicroKernel.ComponentNotFoundException exception)
+            {
+                throw new ServiceNotFound(string.Format("No service of type {0} has been registered", typeof(T).Name), exception);
+            }
+        }
+
+        public IServiceRegistrator Register<T>(System.Func<IServiceProvider, T> factory) where T : class
+        {
+            if (!_container.Kernel.HasComponent(typeof(T)))
+            {
+                _container.Register(
+                    Component.For<T>().UsingFactoryMethod(() => factory(this)).LifeStyle.Singleton
+                    );
+            }
+            return this;
+
+        }
+    }
+}
\ No newline at end of file
diff --git a/src/kafka-net.windsor/kafka-net.windsor.csproj b/src/kafka-net.windsor/kafka-net.windsor.csproj
new file mode 100644
index 00000000..bc764b53
--- /dev/null
+++ b/src/kafka-net.windsor/kafka-net.windsor.csproj
@@ -0,0 +1,82 @@
+<?xml version="1.0" encoding="utf-8"?>
+<Project ToolsVersion="4.0" DefaultTargets="Build" xmlns="http://schemas.microsoft.com/developer/msbuild/2003">
+  <Import Project="$(MSBuildExtensionsPath)\$(MSBuildToolsVersion)\Microsoft.Common.props" Condition="Exists('$(MSBuildExtensionsPath)\$(MSBuildToolsVersion)\Microsoft.Common.props')" />
+  <PropertyGroup>
+    <Configuration Condition=" '$(Configuration)' == '' ">Debug</Configuration>
+    <Platform Condition=" '$(Platform)' == '' ">AnyCPU</Platform>
+    <ProjectGuid>{29ED0088-40F5-452F-95D4-9F8CD33156D3}</ProjectGuid>
+    <OutputType>Library</OutputType>
+    <AppDesignerFolder>Properties</AppDesignerFolder>
+    <RootNamespace>KafkaNet.Windsor</RootNamespace>
+    <AssemblyName>kafka-net.windsor</AssemblyName>
+    <TargetFrameworkVersion>v4.5</TargetFrameworkVersion>
+    <FileAlignment>512</FileAlignment>
+    <SolutionDir Condition="$(SolutionDir) == '' Or $(SolutionDir) == '*Undefined*'">..\</SolutionDir>
+    <RestorePackages>true</RestorePackages>
+  </PropertyGroup>
+  <PropertyGroup Condition=" '$(Configuration)|$(Platform)' == 'Debug|AnyCPU' ">
+    <PlatformTarget>AnyCPU</PlatformTarget>
+    <DebugSymbols>true</DebugSymbols>
+    <DebugType>full</DebugType>
+    <Optimize>false</Optimize>
+    <OutputPath>bin\Debug\</OutputPath>
+    <DefineConstants>DEBUG;TRACE</DefineConstants>
+    <ErrorReport>prompt</ErrorReport>
+    <WarningLevel>4</WarningLevel>
+  </PropertyGroup>
+  <PropertyGroup Condition=" '$(Configuration)|$(Platform)' == 'Release|AnyCPU' ">
+    <PlatformTarget>AnyCPU</PlatformTarget>
+    <DebugType>pdbonly</DebugType>
+    <Optimize>true</Optimize>
+    <OutputPath>bin\Release\</OutputPath>
+    <DefineConstants>TRACE</DefineConstants>
+    <ErrorReport>prompt</ErrorReport>
+    <WarningLevel>4</WarningLevel>
+  </PropertyGroup>
+  <PropertyGroup>
+    <StartupObject />
+  </PropertyGroup>
+  <ItemGroup>
+    <Reference Include="Castle.Core">
+      <HintPath>..\packages\Castle.Core.3.3.0\lib\net45\Castle.Core.dll</HintPath>
+    </Reference>
+    <Reference Include="Castle.Windsor">
+      <HintPath>..\packages\Castle.Windsor.3.3.0\lib\net45\Castle.Windsor.dll</HintPath>
+    </Reference>
+    <Reference Include="System" />
+    <Reference Include="System.Core" />
+    <Reference Include="System.Xml.Linq" />
+    <Reference Include="System.Data.DataSetExtensions" />
+    <Reference Include="Microsoft.CSharp" />
+    <Reference Include="System.Data" />
+    <Reference Include="System.Xml" />
+  </ItemGroup>
+  <ItemGroup>
+    <Compile Include="Properties\AssemblyInfo.cs" />
+    <Compile Include="WindsorAdapter.cs" />
+  </ItemGroup>
+  <ItemGroup>
+    <None Include="packages.config" />
+  </ItemGroup>
+  <ItemGroup>
+    <ProjectReference Include="..\kafka-net\kafka-net.csproj">
+      <Project>{1343eb68-55cb-4452-8386-24a9989de1c0}</Project>
+      <Name>kafka-net</Name>
+    </ProjectReference>
+  </ItemGroup>
+  <Import Project="$(MSBuildToolsPath)\Microsoft.CSharp.targets" />
+  <Import Project="$(SolutionDir)\.nuget\NuGet.targets" Condition="Exists('$(SolutionDir)\.nuget\NuGet.targets')" />
+  <Target Name="EnsureNuGetPackageBuildImports" BeforeTargets="PrepareForBuild">
+    <PropertyGroup>
+      <ErrorText>This project references NuGet package(s) that are missing on this computer. Enable NuGet Package Restore to download them.  For more information, see http://go.microsoft.com/fwlink/?LinkID=322105. The missing file is {0}.</ErrorText>
+    </PropertyGroup>
+    <Error Condition="!Exists('$(SolutionDir)\.nuget\NuGet.targets')" Text="$([System.String]::Format('$(ErrorText)', '$(SolutionDir)\.nuget\NuGet.targets'))" />
+  </Target>
+  <!-- To modify your build process, add your task inside one of the targets below and uncomment it. 
+       Other similar extension points exist, see Microsoft.Common.targets.
+  <Target Name="BeforeBuild">
+  </Target>
+  <Target Name="AfterBuild">
+  </Target>
+  -->
+</Project>
\ No newline at end of file
diff --git a/src/kafka-net.windsor/packages.config b/src/kafka-net.windsor/packages.config
new file mode 100644
index 00000000..010612fa
--- /dev/null
+++ b/src/kafka-net.windsor/packages.config
@@ -0,0 +1,5 @@
+<?xml version="1.0" encoding="utf-8"?>
+<packages>
+  <package id="Castle.Core" version="3.3.0" targetFramework="net45" />
+  <package id="Castle.Windsor" version="3.3.0" targetFramework="net45" />
+</packages>
\ No newline at end of file
diff --git a/src/kafka-net/Configuration/Bus.cs b/src/kafka-net/Configuration/Bus.cs
index 4316ca74..2eadc7ab 100644
--- a/src/kafka-net/Configuration/Bus.cs
+++ b/src/kafka-net/Configuration/Bus.cs
@@ -1,6 +1,4 @@
-using System;
-using System.Collections.Concurrent;
-using System.Collections.Generic;
+using System.Collections.Generic;
 using System.Threading;
 using System.Threading.Tasks;
 using KafkaNet.Protocol;
diff --git a/src/kafka-net/Configuration/BusFactory.cs b/src/kafka-net/Configuration/BusFactory.cs
index be75b585..1092fdae 100644
--- a/src/kafka-net/Configuration/BusFactory.cs
+++ b/src/kafka-net/Configuration/BusFactory.cs
@@ -2,14 +2,24 @@
 
 namespace KafkaNet.Configuration
 {
-    public static class BusFactory
+    public class BusFactory
     {
-        public static IBus Create(IKafkaOptions settings, Action<IServiceRegistrator> configure)
+        private readonly IContainer container;
+
+        public BusFactory(IContainer container)
+        {
+            this.container = container;
+        }
+
+        public BusFactory() : this(new DefaultContainer())
+        {
+        }
+
+        public IBus Create(IKafkaOptions settings, Action<IServiceRegistrator> configure)
         {
-            var container = new DefaultContainer();
-            DefaultServiceRegistrator.Register(container);
-            container.Register(settings);
             configure(container);
+            container.Register(_ => settings);
+            DefaultServiceRegistrator.Register(container);
             return container.Resolve<IBus>();
         }
     }
diff --git a/src/kafka-net/Configuration/DefaultContainer.cs b/src/kafka-net/Configuration/DefaultContainer.cs
index 52fa02ca..1911e2b4 100644
--- a/src/kafka-net/Configuration/DefaultContainer.cs
+++ b/src/kafka-net/Configuration/DefaultContainer.cs
@@ -5,33 +5,30 @@ namespace KafkaNet.Configuration
 {
     public class DefaultContainer : IContainer
     {
-        private readonly Dictionary<Type, object> _implementations = new Dictionary<Type, object>();
         private readonly Dictionary<Type, object> _factories = new Dictionary<Type, object>();
+        private readonly Dictionary<Type, object> _instances = new Dictionary<Type, object>(); 
 
         public T Resolve<T>() where T: class
         {
-            object implementation;
-            if (_implementations.TryGetValue(typeof (T), out implementation))
-                return (T) implementation;
+            object instance;
+            if (_instances.TryGetValue(typeof (T), out instance))
+                return (T) instance;
             object factory;
-            if (_factories.TryGetValue(typeof(T), out factory))
+            if (_factories.TryGetValue(typeof (T), out factory))
             {
-                var serviceFactory = (Func<IServiceProvider, T>)factory;
-                var serviceImplementation = serviceFactory(this);
-                _implementations.Add(typeof (T), serviceImplementation);
-                return serviceImplementation;
+                var newInstance = ((Func<IServiceProvider, T>) factory)(this);
+                _instances.Add(typeof(T), newInstance);
+                return newInstance;
             }
-            throw new ServiceImplementationNotFound();
+            throw new ServiceNotFound(string.Format("No service of type {0} has been registered", typeof(T).Name));
         }
 
-        public void Register<T>(T implementation) where T : class
-        {
-            _implementations.Add(typeof (T), implementation);
-        }
-
-        public void Register<T>(Func<IServiceProvider, T> factory) where T : class 
+        public IServiceRegistrator Register<T>(Func<IServiceProvider, T> factory) where T : class
         {
+            if(_factories.ContainsKey(typeof(T)))
+                return this;
             _factories.Add(typeof(T), factory);
+            return this;
         }
     }
 }
\ No newline at end of file
diff --git a/src/kafka-net/Configuration/DefaultServiceRegistrator.cs b/src/kafka-net/Configuration/DefaultServiceRegistrator.cs
index 99c45bb2..a8084004 100644
--- a/src/kafka-net/Configuration/DefaultServiceRegistrator.cs
+++ b/src/kafka-net/Configuration/DefaultServiceRegistrator.cs
@@ -4,13 +4,13 @@ public static class DefaultServiceRegistrator
     {
         public static void Register(IServiceRegistrator registrator)
         {
-            registrator.Register<IKafkaLog>(new DefaultTraceLog());
-            registrator.Register<IPartitionSelector>(new DefaultPartitionSelector());
-            registrator.Register<IKafkaConnectionFactory>(x => new DefaultKafkaConnectionFactory(x.Resolve<IKafkaLog>()));
-            registrator.Register<IBrokerRouter>(x => new BrokerRouter(x.Resolve<IKafkaOptions>(), x.Resolve<IKafkaLog>(), x.Resolve<IPartitionSelector>(), x.Resolve<IKafkaConnectionFactory>()));
-            registrator.Register<IProducer>(x => new Producer(x.Resolve<IBrokerRouter>(), x.Resolve<IKafkaOptions>()));
-            registrator.Register<IConsumerFactory>(x => new ConsumerFactory(x.Resolve<IBrokerRouter>(), x.Resolve<IKafkaLog>()));
-            registrator.Register<IBus>(x => new Bus(x.Resolve<IProducer>(), x.Resolve<IConsumerFactory>()));
+            registrator.Register<IKafkaLog>(_ => new DefaultTraceLog())
+                       .Register<IPartitionSelector>(_ => new DefaultPartitionSelector())
+                       .Register<IKafkaConnectionFactory>(x => new DefaultKafkaConnectionFactory(x.Resolve<IKafkaLog>()))
+                       .Register<IBrokerRouter>(x => new BrokerRouter(x.Resolve<IKafkaOptions>(), x.Resolve<IKafkaLog>(), x.Resolve<IPartitionSelector>(), x.Resolve<IKafkaConnectionFactory>()))
+                       .Register<IProducer>(x => new Producer(x.Resolve<IBrokerRouter>(), x.Resolve<IKafkaOptions>()))
+                       .Register<IConsumerFactory>(x => new ConsumerFactory(x.Resolve<IBrokerRouter>(), x.Resolve<IKafkaLog>()))
+                       .Register<IBus>(x => new Bus(x.Resolve<IProducer>(), x.Resolve<IConsumerFactory>()));
         }
     }
 }
\ No newline at end of file
diff --git a/src/kafka-net/Configuration/IServiceRegistrator.cs b/src/kafka-net/Configuration/IServiceRegistrator.cs
index ae6c2e24..cea39368 100644
--- a/src/kafka-net/Configuration/IServiceRegistrator.cs
+++ b/src/kafka-net/Configuration/IServiceRegistrator.cs
@@ -4,7 +4,6 @@ namespace KafkaNet.Configuration
 {
     public interface IServiceRegistrator
     {
-        void Register<T>(T implementation) where T : class;
-        void Register<T>(Func<IServiceProvider, T> factory) where T : class;
+        IServiceRegistrator Register<T>(Func<IServiceProvider, T> factory) where T : class;
     }
 }
\ No newline at end of file
diff --git a/src/kafka-net/Configuration/ServiceImplementationNotFound.cs b/src/kafka-net/Configuration/ServiceImplementationNotFound.cs
deleted file mode 100644
index 3fb24aeb..00000000
--- a/src/kafka-net/Configuration/ServiceImplementationNotFound.cs
+++ /dev/null
@@ -1,8 +0,0 @@
-using System;
-
-namespace KafkaNet.Configuration
-{
-    public class ServiceImplementationNotFound : Exception
-    {
-    }
-}
\ No newline at end of file
diff --git a/src/kafka-net/Configuration/ServiceNotFound.cs b/src/kafka-net/Configuration/ServiceNotFound.cs
new file mode 100644
index 00000000..e9e8774e
--- /dev/null
+++ b/src/kafka-net/Configuration/ServiceNotFound.cs
@@ -0,0 +1,16 @@
+using System;
+
+namespace KafkaNet.Configuration
+{
+    public class ServiceNotFound : Exception
+    {
+        public ServiceNotFound(string message, Exception exception) : base(message, exception)
+        {
+        }
+
+        public ServiceNotFound(string message)
+            : base(message)
+        {
+        }
+    }
+}
\ No newline at end of file
diff --git a/src/kafka-net/kafka-net.csproj b/src/kafka-net/kafka-net.csproj
index c66a01db..6571b99e 100644
--- a/src/kafka-net/kafka-net.csproj
+++ b/src/kafka-net/kafka-net.csproj
@@ -55,7 +55,7 @@
     <Compile Include="Interfaces\IKafkaOptions.cs" />
     <Compile Include="Configuration\IServiceProvider.cs" />
     <Compile Include="Configuration\IServiceRegistrator.cs" />
-    <Compile Include="Configuration\ServiceImplementationNotFound.cs" />
+    <Compile Include="Configuration\ServiceNotFound.cs" />
     <Compile Include="Default\DefaultKafkaConnectionFactory.cs" />
     <Compile Include="Interfaces\IMetadataQueries.cs" />
     <Compile Include="Interfaces\IKafkaTcpSocket.cs" />
diff --git a/src/kafka-tests/Configuration/AnotherService.cs b/src/kafka-tests/Configuration/AnotherService.cs
new file mode 100644
index 00000000..b3fa468a
--- /dev/null
+++ b/src/kafka-tests/Configuration/AnotherService.cs
@@ -0,0 +1,6 @@
+namespace kafka_tests.Configuration
+{
+    public class AnotherService : IService
+    {
+    }
+}
\ No newline at end of file
diff --git a/src/kafka-tests/Configuration/DefaultContainerTest.cs b/src/kafka-tests/Configuration/DefaultContainerTest.cs
new file mode 100644
index 00000000..6e80de9e
--- /dev/null
+++ b/src/kafka-tests/Configuration/DefaultContainerTest.cs
@@ -0,0 +1,42 @@
+using KafkaNet.Configuration;
+using NUnit.Framework;
+
+namespace kafka_tests.Configuration
+{
+    [TestFixture]
+    public class DefaultContainerTest
+    {
+        private DefaultContainer _defaultContainer;
+
+        [SetUp]
+        public void SetUp()
+        {
+            _defaultContainer = new DefaultContainer();
+        }
+
+        [Test]
+        public void FirstRegistrationShouldWin()
+        {
+            var oneService = new OneService();
+            var anotherService = new AnotherService();
+            _defaultContainer.Register<IService>(_ => oneService);
+            _defaultContainer.Register<IService>(_ => anotherService);
+            Assert.AreEqual(oneService, _defaultContainer.Resolve<IService>());
+        }
+
+        [Test]
+        [ExpectedException(typeof (ServiceNotFound))]
+        public void ShouldThrowExceptionIfServiceNotRegistered()
+        {
+            _defaultContainer.Resolve<IService>();
+        }
+
+        [Test]
+        public void ShouldRememberCreatedInstance()
+        {
+            _defaultContainer.Register<IService>(_ => new OneService());
+            var instance = _defaultContainer.Resolve<IService>();
+            Assert.AreSame(instance, _defaultContainer.Resolve<IService>());
+        }
+    }
+}
\ No newline at end of file
diff --git a/src/kafka-tests/Configuration/IService.cs b/src/kafka-tests/Configuration/IService.cs
new file mode 100644
index 00000000..32ca517a
--- /dev/null
+++ b/src/kafka-tests/Configuration/IService.cs
@@ -0,0 +1,6 @@
+namespace kafka_tests.Configuration
+{
+    public interface IService
+    {
+    }
+}
\ No newline at end of file
diff --git a/src/kafka-tests/Configuration/OneService.cs b/src/kafka-tests/Configuration/OneService.cs
new file mode 100644
index 00000000..7fec1c1c
--- /dev/null
+++ b/src/kafka-tests/Configuration/OneService.cs
@@ -0,0 +1,6 @@
+namespace kafka_tests.Configuration
+{
+    public class OneService : IService
+    {
+    }
+}
\ No newline at end of file
diff --git a/src/kafka-tests/Configuration/WindsorAdapterTest.cs b/src/kafka-tests/Configuration/WindsorAdapterTest.cs
new file mode 100644
index 00000000..70a6cd3d
--- /dev/null
+++ b/src/kafka-tests/Configuration/WindsorAdapterTest.cs
@@ -0,0 +1,44 @@
+using Castle.Windsor;
+using KafkaNet.Configuration;
+using KafkaNet.Windsor;
+using NUnit.Framework;
+
+namespace kafka_tests.Configuration
+{
+    [TestFixture]
+    public class WindsorAdapterTest
+    {
+        private WindsorAdapter _defaultContainer;
+
+        [SetUp]
+        public void SetUp()
+        {
+            _defaultContainer = new WindsorAdapter(new WindsorContainer());
+        }
+
+        [Test]
+        public void FirstRegistrationShouldWin()
+        {
+            var oneService = new OneService();
+            var anotherService = new AnotherService();
+            _defaultContainer.Register<IService>(_ => oneService);
+            _defaultContainer.Register<IService>(_ => anotherService);
+            Assert.AreEqual(oneService, _defaultContainer.Resolve<IService>());
+        }
+
+        [Test]
+        [ExpectedException(typeof (ServiceNotFound))]
+        public void ShouldThrowExceptionIfServiceNotRegistered()
+        {
+            _defaultContainer.Resolve<IService>();
+        }
+
+        [Test]
+        public void ShouldRememberCreatedInstance()
+        {
+            _defaultContainer.Register<IService>(_ => new OneService());
+            var instance = _defaultContainer.Resolve<IService>();
+            Assert.AreSame(instance, _defaultContainer.Resolve<IService>());
+        }
+    }
+}
\ No newline at end of file
diff --git a/src/kafka-tests/kafka-tests.csproj b/src/kafka-tests/kafka-tests.csproj
index 7feec390..75b63040 100644
--- a/src/kafka-tests/kafka-tests.csproj
+++ b/src/kafka-tests/kafka-tests.csproj
@@ -32,6 +32,12 @@
     <WarningLevel>4</WarningLevel>
   </PropertyGroup>
   <ItemGroup>
+    <Reference Include="Castle.Core">
+      <HintPath>..\packages\Castle.Core.3.3.0\lib\net45\Castle.Core.dll</HintPath>
+    </Reference>
+    <Reference Include="Castle.Windsor">
+      <HintPath>..\packages\Castle.Windsor.3.3.0\lib\net45\Castle.Windsor.dll</HintPath>
+    </Reference>
     <Reference Include="Moq">
       <HintPath>..\packages\Moq.4.0.10827\lib\NET40\Moq.dll</HintPath>
     </Reference>
@@ -57,6 +63,11 @@
     <Reference Include="System.Xml" />
   </ItemGroup>
   <ItemGroup>
+    <Compile Include="Configuration\AnotherService.cs" />
+    <Compile Include="Configuration\WindsorAdapterTest.cs" />
+    <Compile Include="Configuration\DefaultContainerTest.cs" />
+    <Compile Include="Configuration\IService.cs" />
+    <Compile Include="Configuration\OneService.cs" />
     <Compile Include="Fakes\BrokerRouterProxy.cs" />
     <Compile Include="Fakes\FakeKafkaConnection.cs" />
     <Compile Include="Fakes\FakeTcpServer.cs" />
@@ -86,6 +97,10 @@
     <None Include="packages.config" />
   </ItemGroup>
   <ItemGroup>
+    <ProjectReference Include="..\kafka-net.windsor\kafka-net.windsor.csproj">
+      <Project>{29ed0088-40f5-452f-95d4-9f8cd33156d3}</Project>
+      <Name>kafka-net.windsor</Name>
+    </ProjectReference>
     <ProjectReference Include="..\kafka-net\kafka-net.csproj">
       <Project>{1343eb68-55cb-4452-8386-24a9989de1c0}</Project>
       <Name>kafka-net</Name>
diff --git a/src/kafka-tests/packages.config b/src/kafka-tests/packages.config
index 3a923378..054c8768 100644
--- a/src/kafka-tests/packages.config
+++ b/src/kafka-tests/packages.config
@@ -1,5 +1,7 @@
 <?xml version="1.0" encoding="utf-8"?>
 <packages>
+  <package id="Castle.Core" version="3.3.0" targetFramework="net45" />
+  <package id="Castle.Windsor" version="3.3.0" targetFramework="net45" />
   <package id="Moq" version="4.0.10827" targetFramework="net45" />
   <package id="Ninject" version="3.0.1.10" targetFramework="net45" />
   <package id="Ninject.MockingKernel" version="3.0.0.5" targetFramework="net45" />