IoC Issue #12

src/TestHarness/Program.cs

@@ -1,27 +1,24 @@
 using System;
 using System.Threading.Tasks;
-using KafkaNet;
+using KafkaNet.Configuration;
 using KafkaNet.Model;
 using KafkaNet.Protocol;
-using System.Collections.Generic;
 
 namespace TestHarness
 {
     class Program
     {
         static void Main(string[] args)
         {
-            var options = new KafkaOptions(new Uri("http://CSDKAFKA01:9092"), new Uri("http://CSDKAFKA02:9092"))
+            var bus = new 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)
             {
-
             }
         }
     }

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()

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 

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

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")]

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;
+
+        }
+    }
+}

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>

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>

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); }
                     });
             }