Merge branch 'master' into feature-async-subscriptions

# Conflicts:
#	src/Abc.Zebus.Tests/Abc.Zebus.Tests.csproj
#	src/Abc.Zebus.Tests/packages.config
#	src/Abc.Zebus/Abc.Zebus.csproj
#	src/Abc.Zebus/Directory/PeerDirectoryClient.cs
#	src/Abc.Zebus/Dispatch/DynamicMessageHandlerInvoker.cs
#	src/Abc.Zebus/Dispatch/EventHandlerInvoker`1.cs

.editorconfig

@@ -0,0 +1,32 @@
+
+root = true
+
+[*]
+trim_trailing_whitespace = true
+insert_final_newline = true
+end_of_line = crlf
+
+[*.{cs,csx,xaml,cake}]
+indent_size = 4
+indent_style = space
+
+[*.{config,nuspec,resx}]
+indent_size = 2
+indent_style = space
+
+[*.{csproj,vcxproj,props,targets}]
+indent_size = 2
+indent_style = space
+insert_final_newline = false
+
+[*.sln]
+indent_style = tab
+
+[*.md]
+trim_trailing_whitespace = false
+
+[*.sh]
+end_of_line = lf
+
+[*.g4]
+indent_style = tab

.gitignore

@@ -36,3 +36,5 @@ lib/packages/**
 tools/cake/tools/Cake/*
 tools/cake/tools/nuget.exe
 build/tools/*
+
+src/Abc.Zebus.Tests/Properties/launchSettings.json

README.md

@@ -1,4 +1,4 @@
-# About [![AppVeyor branch](https://img.shields.io/appveyor/ci/TechAbc/Zebus/master.svg)](https://ci.appveyor.com/project/TechABC/zebus) [![NuGet](https://buildstats.info/nuget/Zebus)](http://www.nuget.org/packages/Zebus/) [![Join the chat at https://gitter.im/Abc-Arbitrage/Zebus](https://badges.gitter.im/Join%20Chat.svg)](https://gitter.im/Abc-Arbitrage/Zebus?utm_source=badge&utm_medium=badge&utm_campaign=pr-badge&utm_content=badge)
+# About [![AppVeyor branch](https://ci.appveyor.com/api/projects/status/4uskxl4dm06fk3b9?svg=true)](https://ci.appveyor.com/project/Abc-Arbitrage/zebus) [![NuGet](https://buildstats.info/nuget/Zebus)](http://www.nuget.org/packages/Zebus/) [![Join the chat at https://gitter.im/Abc-Arbitrage/Zebus](https://badges.gitter.im/Join%20Chat.svg)](https://gitter.im/Abc-Arbitrage/Zebus?utm_source=badge&utm_medium=badge&utm_campaign=pr-badge&utm_content=badge)
 
 Zebus is a lightweight peer to peer service bus, built with [CQRS](http://martinfowler.com/bliki/CQRS.html) principles in mind. It allows applications to communicate with each other in a fast and easy manner. Most of the complexity is hidden in the library and you can focus on writing code that matters to you, not debugging messaging code.
 

RELEASE_NOTES.md

@@ -1,3 +1,25 @@
+## 2.2.0
+- Features
+    - Updated Newtonsoft.Json, protobuf-net and log4net dependencies
+
+## 2.1.2
+- Features
+    - Reduce heap allocations in the receiving code path
+
+## 2.1.1
+- Features
+    - Improve performance when message logging is disabled
+    
+## 2.1.0
+- Features
+    - Inject an error reporter in various zmq outbound socket operations
+    - Allow local message dispatch when the bus is stopping
+- Bug fixes
+    - Ensure all replayed messages are acked to the persistence
+    - Support unserializable messages in dispatch error handling
+- Code cleanup
+    - Migrate to VS2017 project file
+
 ## 2.0.0
 This version introduces a major breaking change: a dedicated `TransportMessage` is now used to forward messages to the persistence service, instead of sending a `PersistMessageCommand`. An updated persistence service is still able to process commands sent by old peers, but an old persistence service is not be able to process the new TransportMessage. Thus the persistence service must be updated first.
 - Features

build/Abc.Zebus.Testing.nuspec

@@ -16,15 +16,15 @@
       <dependency id="AutoFixture" version="3.36.8" />
       <dependency id="CompareNETObjects" version="3.03.0.0" />
       <dependency id="Moq" version="4.2.1507.0118" />
-      <dependency id="Newtonsoft.Json" version="7.0.1" />
+      <dependency id="Newtonsoft.Json" version="11.0.2" />
       <dependency id="NUnit" version="2.6.4" />
       <dependency id="structuremap" version="3.1.6.186" />
     </dependencies>
   </metadata>
   <files>
-    <file src="Abc.Zebus.Testing.pdb" target="lib\net45\" />
-    <file src="Abc.Zebus.Testing.dll" target="lib\net45\" />
+    <file src="Abc.Zebus.Testing.pdb" target="lib\net46\" />
+    <file src="Abc.Zebus.Testing.dll" target="lib\net46\" />
 
     <file src=".\..\..\src\Abc.Zebus.Testing\**\*.cs" target="src" />
   </files>
-</package>
+</package>

build/Abc.Zebus.nuspec

@@ -15,18 +15,16 @@
       <reference file="Abc.Zebus.dll" />
     </references>
     <dependencies>
-      <dependency id="log4net" version="2.0.3" />
-      <dependency id="Newtonsoft.Json" version="7.0.1" />
-      <dependency id="protobuf-net" version="2.0.0.668" />
+      <dependency id="log4net" version="2.0.8" />
+      <dependency id="Newtonsoft.Json" version="11.0.2" />
+      <dependency id="protobuf-net" version="2.3.7" />
       <dependency id="structuremap" version="3.1.6.186" />
     </dependencies>
   </metadata>
   <files>
-    <file src="Abc.Zebus.dll" target="lib\net45\" />
-    <file src="Abc.Zebus.pdb" target="lib\net45\" />
-
-    <file src=".\..\..\lib\zmq\clrzmq.dll" target="lib\net45\" />
-    
-    <file src=".\..\..\src\\Abc.Zebus\**\*.cs" target="src" />
+    <file src="Abc.Zebus.dll" target="lib\net46\" />
+    <file src="Abc.Zebus.pdb" target="lib\net46\" />
+    <file src=".\..\..\lib\zmq\clrzmq.dll" target="lib\net46\" />
+    <file src=".\..\..\src\Abc.Zebus\**\*.cs" target="src" />
   </files>
-</package>
+</package>

src/Abc.Zebus.Testing/Abc.Zebus.Testing.csproj

@@ -1,75 +1,44 @@
-<?xml version="1.0" encoding="utf-8"?>
-<Project ToolsVersion="12.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')" />
+<Project Sdk="Microsoft.NET.Sdk">
   <PropertyGroup>
-    <Configuration Condition=" '$(Configuration)' == '' ">Debug</Configuration>
-    <Platform Condition=" '$(Platform)' == '' ">AnyCPU</Platform>
-    <ProjectGuid>{4CE123DF-8021-411C-929B-53E5C5FC1E04}</ProjectGuid>
-    <OutputType>Library</OutputType>
-    <AppDesignerFolder>Properties</AppDesignerFolder>
-    <RootNamespace>Abc.Zebus.Testing</RootNamespace>
-    <AssemblyName>Abc.Zebus.Testing</AssemblyName>
-    <TargetFrameworkVersion>v4.5</TargetFrameworkVersion>
-    <FileAlignment>512</FileAlignment>
+    <TargetFramework>net46</TargetFramework>
+    <GenerateAssemblyInfo>false</GenerateAssemblyInfo>
+    <ErrorReport>prompt</ErrorReport>
+    <WarningLevel>4</WarningLevel>
+    <TreatWarningsAsErrors>true</TreatWarningsAsErrors>
   </PropertyGroup>
+
   <PropertyGroup Condition=" '$(Configuration)|$(Platform)' == 'Debug|AnyCPU' ">
     <DebugSymbols>true</DebugSymbols>
     <DebugType>full</DebugType>
     <Optimize>false</Optimize>
     <OutputPath>bin\Debug\</OutputPath>
     <DefineConstants>DEBUG;TRACE</DefineConstants>
-    <ErrorReport>prompt</ErrorReport>
-    <WarningLevel>4</WarningLevel>
-    <TreatWarningsAsErrors>true</TreatWarningsAsErrors>
   </PropertyGroup>
   <PropertyGroup Condition=" '$(Configuration)|$(Platform)' == 'Release|AnyCPU' ">
     <DebugType>pdbonly</DebugType>
     <Optimize>true</Optimize>
     <OutputPath>bin\Release\</OutputPath>
     <DefineConstants>TRACE</DefineConstants>
-    <ErrorReport>prompt</ErrorReport>
-    <WarningLevel>4</WarningLevel>
-    <TreatWarningsAsErrors>true</TreatWarningsAsErrors>
   </PropertyGroup>
+
+  <ItemGroup>
+    <ProjectReference Include="..\Abc.Zebus\Abc.Zebus.csproj" />
+  </ItemGroup>
+
+  <ItemGroup>
+    <PackageReference Include="AutoFixture" Version="3.36.8" />
+    <PackageReference Include="CompareNETObjects" Version="3.03.0.0" />
+    <PackageReference Include="Moq" Version="4.2.1507.0118" />
+    <PackageReference Include="Newtonsoft.Json" Version="11.0.2" />
+    <PackageReference Include="NUnit" Version="2.6.4" />
+    <PackageReference Include="protobuf-net" Version="2.3.7" />
+    <PackageReference Include="structuremap" Version="3.1.6.186" />
+  </ItemGroup>
+
   <ItemGroup>
-    <Reference Include="KellermanSoftware.Compare-NET-Objects">
-      <HintPath>..\..\lib\packages\CompareNETObjects.3.03.0.0\lib\net45\KellermanSoftware.Compare-NET-Objects.dll</HintPath>
-      <Private>True</Private>
-    </Reference>
-    <Reference Include="Moq">
-      <HintPath>..\..\lib\packages\Moq.4.2.1507.0118\lib\net40\Moq.dll</HintPath>
-      <Private>True</Private>
-    </Reference>
-    <Reference Include="Newtonsoft.Json, Version=7.0.0.0, Culture=neutral, PublicKeyToken=30ad4fe6b2a6aeed, processorArchitecture=MSIL">
-      <SpecificVersion>False</SpecificVersion>
-      <HintPath>..\..\lib\packages\Newtonsoft.Json.7.0.1\lib\net45\Newtonsoft.Json.dll</HintPath>
-      <Private>True</Private>
-    </Reference>
-    <Reference Include="nunit.framework">
-      <HintPath>..\..\lib\packages\NUnit.2.6.4\lib\nunit.framework.dll</HintPath>
-      <Private>True</Private>
-    </Reference>
-    <Reference Include="Ploeh.AutoFixture">
-      <HintPath>..\..\lib\packages\AutoFixture.3.36.8\lib\net40\Ploeh.AutoFixture.dll</HintPath>
-      <Private>True</Private>
-    </Reference>
-    <Reference Include="protobuf-net, Version=2.0.0.668, Culture=neutral, PublicKeyToken=257b51d87d2e4d67, processorArchitecture=MSIL">
-      <SpecificVersion>False</SpecificVersion>
-      <HintPath>..\..\lib\packages\protobuf-net.2.0.0.668\lib\net40\protobuf-net.dll</HintPath>
-      <Private>True</Private>
-    </Reference>
-    <Reference Include="StructureMap">
-      <HintPath>..\..\lib\packages\structuremap.3.1.6.186\lib\net40\StructureMap.dll</HintPath>
-      <Private>True</Private>
-    </Reference>
-    <Reference Include="StructureMap.Net4">
-      <HintPath>..\..\lib\packages\structuremap.3.1.6.186\lib\net40\StructureMap.Net4.dll</HintPath>
-      <Private>True</Private>
-    </Reference>
-    <Reference Include="System" />
-    <Reference Include="System.Core" />
     <Reference Include="System.Management" />
   </ItemGroup>
+
   <ItemGroup>
     <Compile Include="..\SharedAssemblyInfo.cs">
       <Link>Properties\SharedAssemblyInfo.cs</Link>
@@ -77,52 +46,8 @@
     <Compile Include="..\SharedVersionInfo.cs">
       <Link>Properties\SharedVersionInfo.cs</Link>
     </Compile>
-    <Compile Include="Comparison\ComparisonExtensions.cs" />
-    <Compile Include="Comparison\MessageComparer.cs" />
-    <Compile Include="CurrentThreadTaskScheduler.cs" />
-    <Compile Include="Directory\TestPeerDirectory.cs" />
-    <Compile Include="Dispatch\IAsyncExecutableMessage.cs" />
-    <Compile Include="Dispatch\IExecutableMessage.cs" />
-    <Compile Include="Dispatch\TestBatchedMessageHandlerInvoker`1.cs" />
-    <Compile Include="Dispatch\TestMessageHandlerInvoker`1.cs" />
-    <Compile Include="Dispatch\TestAsyncMessageHandlerInvoker`1.cs" />
-    <Compile Include="Extensions\ExtendSystemDateTime.cs" />
-    <Compile Include="Extensions\NUnitExtensions.cs" />
-    <Compile Include="Integration\IntegrationTestFixture.cs" />
-    <Compile Include="Integration\TestService.cs" />
-    <Compile Include="Measurements\Measure.cs" />
-    <Compile Include="Measurements\MeasureConfiguration.cs" />
-    <Compile Include="MessageSerializationTester.cs" />
-    <Compile Include="Properties\AssemblyInfo.cs" />
-    <Compile Include="TestBus.cs" />
-    <Compile Include="TestExtensions.cs" />
-    <Compile Include="Transport\TestTransport.cs" />
-    <Compile Include="Transport\TransportMessageSent.cs" />
-    <Compile Include="Transport\UpdatedPeer.cs" />
-    <Compile Include="UnitTesting\MoqExtensions.cs" />
-    <Compile Include="UnitTesting\SetupSequence.cs" />
-    <Compile Include="Wait.cs" />
-  </ItemGroup>
-  <ItemGroup>
     <Content Include="Integration\Configurations\Directory-Local.config">
       <CopyToOutputDirectory>PreserveNewest</CopyToOutputDirectory>
     </Content>
   </ItemGroup>
-  <ItemGroup>
-    <ProjectReference Include="..\Abc.Zebus\Abc.Zebus.csproj">
-      <Project>{1f4c6307-6113-40d5-bf42-4b6bf5df13b2}</Project>
-      <Name>Abc.Zebus</Name>
-    </ProjectReference>
-  </ItemGroup>
-  <ItemGroup>
-    <None Include="packages.config" />
-  </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.
-  <Target Name="BeforeBuild">
-  </Target>
-  <Target Name="AfterBuild">
-  </Target>
-  -->
 </Project>

src/Abc.Zebus.Testing/Extensions/ExtendBuffer.cs

@@ -1,15 +0,0 @@
-using System.Linq;
-using Abc.Zebus.Util.GenZero;
-using NUnit.Framework;
-
-namespace Abc.Zebus.Testing.Extensions
-{
-    internal static class ExtendBuffer
-    {
-        public static void ShouldEqual(this Buffer actual, ref Buffer expected, string message = null)
-        {
-            Assert.AreEqual(expected.Length, actual.Length, message);
-            Assert.AreEqual(expected.Data.Take(expected.Length), actual.Data.Take(actual.Length), message);
-        }
-    }
-}

src/Abc.Zebus.Testing/TestBus.cs

@@ -82,6 +82,8 @@ namespace Abc.Zebus.Testing
             {
                 _events.Add(message);
             }
+            var handler = _handlers.GetValueOrDefault(new HandlerKey(message.GetType(), default(PeerId)));
+            handler?.Invoke(message);
         }
 
         public Task<CommandResult> Send(ICommand message)

src/Abc.Zebus.Testing/packages.config

@@ -1,10 +0,0 @@
-<?xml version="1.0" encoding="utf-8"?>
-<packages>
-  <package id="AutoFixture" version="3.36.8" targetFramework="net45" />
-  <package id="CompareNETObjects" version="3.03.0.0" targetFramework="net45" />
-  <package id="Moq" version="4.2.1507.0118" targetFramework="net45" />
-  <package id="Newtonsoft.Json" version="7.0.1" targetFramework="net45" />
-  <package id="NUnit" version="2.6.4" targetFramework="net45" />
-  <package id="protobuf-net" version="2.0.0.668" targetFramework="net45" />
-  <package id="structuremap" version="3.1.6.186" targetFramework="net45" />
-</packages>

src/Abc.Zebus.Tests.TestExe/Abc.Zebus.Tests.TestExe.csproj

@@ -1,17 +1,12 @@
-<?xml version="1.0" encoding="utf-8"?>
-<Project ToolsVersion="12.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')" />
+<Project Sdk="Microsoft.NET.Sdk">
   <PropertyGroup>
-    <Configuration Condition=" '$(Configuration)' == '' ">Debug</Configuration>
-    <Platform Condition=" '$(Platform)' == '' ">AnyCPU</Platform>
-    <ProjectGuid>{652017D8-EB34-4A79-B14F-DB0477E43915}</ProjectGuid>
+    <TargetFramework>net46</TargetFramework>
     <OutputType>Exe</OutputType>
-    <AppDesignerFolder>Properties</AppDesignerFolder>
-    <RootNamespace>Abc.Zebus.Tests.TestExe</RootNamespace>
-    <AssemblyName>Abc.Zebus.Tests.TestExe</AssemblyName>
-    <TargetFrameworkVersion>v4.5</TargetFrameworkVersion>
-    <FileAlignment>512</FileAlignment>
+    <GenerateAssemblyInfo>false</GenerateAssemblyInfo>
+    <ErrorReport>prompt</ErrorReport>
+    <WarningLevel>4</WarningLevel>
   </PropertyGroup>
+
   <PropertyGroup Condition=" '$(Configuration)|$(Platform)' == 'Debug|AnyCPU' ">
     <PlatformTarget>AnyCPU</PlatformTarget>
     <DebugSymbols>true</DebugSymbols>
@@ -19,8 +14,6 @@
     <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>
@@ -28,31 +21,13 @@
     <Optimize>true</Optimize>
     <OutputPath>bin\Release\</OutputPath>
     <DefineConstants>TRACE</DefineConstants>
-    <ErrorReport>prompt</ErrorReport>
-    <WarningLevel>4</WarningLevel>
   </PropertyGroup>
+
   <ItemGroup>
-    <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="Program.cs" />
-    <Compile Include="Properties\AssemblyInfo.cs" />
   </ItemGroup>
   <ItemGroup>
     <None Include="App.config" />
   </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.
-  <Target Name="BeforeBuild">
-  </Target>
-  <Target Name="AfterBuild">
-  </Target>
-  -->
 </Project>

src/Abc.Zebus.Tests/Abc.Zebus.Tests.csproj

@@ -1,256 +1,67 @@
-<?xml version="1.0" encoding="utf-8"?>
-<Project ToolsVersion="12.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')" />
+<Project Sdk="Microsoft.NET.Sdk">
   <PropertyGroup>
-    <Configuration Condition=" '$(Configuration)' == '' ">Debug</Configuration>
-    <Platform Condition=" '$(Platform)' == '' ">AnyCPU</Platform>
-    <ProjectGuid>{7BF5AC28-91DB-40EF-8CB7-024E518FA799}</ProjectGuid>
+    <TargetFramework>net46</TargetFramework>
     <OutputType>Exe</OutputType>
-    <AppDesignerFolder>Properties</AppDesignerFolder>
-    <RootNamespace>Abc.Zebus.Tests</RootNamespace>
-    <AssemblyName>Abc.Zebus.Tests</AssemblyName>
-    <TargetFrameworkVersion>v4.6</TargetFrameworkVersion>
-    <FileAlignment>512</FileAlignment>
-    <TargetFrameworkProfile />
+    <GenerateAssemblyInfo>false</GenerateAssemblyInfo>
+    <ErrorReport>prompt</ErrorReport>
+    <WarningLevel>4</WarningLevel>
+    <TreatWarningsAsErrors>true</TreatWarningsAsErrors>
+    <Prefer32Bit>false</Prefer32Bit>
   </PropertyGroup>
+
   <PropertyGroup Condition=" '$(Configuration)|$(Platform)' == 'Debug|AnyCPU' ">
     <DebugSymbols>true</DebugSymbols>
     <DebugType>full</DebugType>
     <Optimize>false</Optimize>
     <OutputPath>bin\Debug\</OutputPath>
     <DefineConstants>DEBUG;TRACE</DefineConstants>
-    <ErrorReport>prompt</ErrorReport>
-    <WarningLevel>4</WarningLevel>
-    <TreatWarningsAsErrors>true</TreatWarningsAsErrors>
     <UseVSHostingProcess>false</UseVSHostingProcess>
-    <Prefer32Bit>false</Prefer32Bit>
   </PropertyGroup>
   <PropertyGroup Condition=" '$(Configuration)|$(Platform)' == 'Release|AnyCPU' ">
     <DebugType>pdbonly</DebugType>
     <Optimize>true</Optimize>
     <OutputPath>bin\Release\</OutputPath>
     <DefineConstants>TRACE</DefineConstants>
-    <ErrorReport>prompt</ErrorReport>
-    <WarningLevel>4</WarningLevel>
-    <TreatWarningsAsErrors>true</TreatWarningsAsErrors>
-    <Prefer32Bit>false</Prefer32Bit>
-  </PropertyGroup>
-  <PropertyGroup>
-    <StartupObject />
   </PropertyGroup>
+
+  <ItemGroup>
+    <EmbeddedResource Include="Transport\transport_message_1_4_1.bin" />
+  </ItemGroup>
+
+  <ItemGroup>
+    <ProjectReference Include="..\Abc.Zebus.Testing\Abc.Zebus.Testing.csproj" />
+    <ProjectReference Include="..\Abc.Zebus.Tests.TestExe\Abc.Zebus.Tests.TestExe.csproj" />
+    <ProjectReference Include="..\Abc.Zebus\Abc.Zebus.csproj" />
+  </ItemGroup>
+
+  <ItemGroup>
+    <PackageReference Include="CompareNETObjects" Version="3.03.0.0" />
+    <PackageReference Include="log4net" Version="2.0.8" />
+    <PackageReference Include="Moq" Version="4.2.1507.0118" />
+    <PackageReference Include="Newtonsoft.Json" Version="11.0.2" />
+    <PackageReference Include="NUnit" Version="2.6.4" />
+    <PackageReference Include="protobuf-net" Version="2.3.7" />
+    <PackageReference Include="structuremap" Version="3.1.6.186" />
+  </ItemGroup>
+
   <ItemGroup>
     <Reference Include="clrzmq">
       <HintPath>..\..\lib\zmq\clrzmq.dll</HintPath>
     </Reference>
-    <Reference Include="KellermanSoftware.Compare-NET-Objects">
-      <HintPath>..\..\lib\packages\CompareNETObjects.3.03.0.0\lib\net45\KellermanSoftware.Compare-NET-Objects.dll</HintPath>
-      <Private>True</Private>
-    </Reference>
-    <Reference Include="log4net">
-      <HintPath>..\..\lib\packages\log4net.2.0.3\lib\net40-full\log4net.dll</HintPath>
-      <Private>True</Private>
-    </Reference>
-    <Reference Include="Moq">
-      <HintPath>..\..\lib\packages\Moq.4.2.1507.0118\lib\net40\Moq.dll</HintPath>
-      <Private>True</Private>
-    </Reference>
-    <Reference Include="Newtonsoft.Json, Version=7.0.0.0, Culture=neutral, PublicKeyToken=30ad4fe6b2a6aeed, processorArchitecture=MSIL">
-      <SpecificVersion>False</SpecificVersion>
-      <HintPath>..\..\lib\packages\Newtonsoft.Json.7.0.1\lib\net45\Newtonsoft.Json.dll</HintPath>
-      <Private>True</Private>
-    </Reference>
-    <Reference Include="nunit.framework">
-      <HintPath>..\..\lib\packages\NUnit.2.6.4\lib\nunit.framework.dll</HintPath>
-      <Private>True</Private>
-    </Reference>
-    <Reference Include="protobuf-net">
-      <HintPath>..\..\lib\packages\protobuf-net.2.0.0.668\lib\net40\protobuf-net.dll</HintPath>
-      <Private>True</Private>
-    </Reference>
-    <Reference Include="StructureMap">
-      <HintPath>..\..\lib\packages\structuremap.3.1.6.186\lib\net40\StructureMap.dll</HintPath>
-      <Private>True</Private>
-    </Reference>
-    <Reference Include="StructureMap.Net4">
-      <HintPath>..\..\lib\packages\structuremap.3.1.6.186\lib\net40\StructureMap.Net4.dll</HintPath>
-      <Private>True</Private>
-    </Reference>
-    <Reference Include="System" />
-    <Reference Include="System.Core" />
-    <Reference Include="System.Xml" />
     <Reference Include="VersionedLibrary_2">
       <HintPath>.\VersionedLibrary_2.dll</HintPath>
       <Aliases>senderVersion</Aliases>
     </Reference>
   </ItemGroup>
-  <ItemGroup>
-    <Compile Include="CommandResultTests.cs" />
-    <Compile Include="Comparison\ComparisonExtensionsTests.cs" />
-    <Compile Include="Core\BusFactoryTests.cs" />
-    <Compile Include="Core\BusInMemoryTests.cs" />
-    <Compile Include="Core\BusManualTests.cs" />
-    <Compile Include="Core\BusPerformanceTests.cs" />
-    <Compile Include="Core\BusTests.cs" />
-    <Compile Include="Core\BusTests.Send.cs" />
-    <Compile Include="Core\BusTests.CompletionMessages.cs" />
-    <Compile Include="Core\BusTests.DispatchErrors.cs" />
-    <Compile Include="Core\BusTests.Reply.cs" />
-    <Compile Include="Core\BusTests.Dispatch.cs" />
-    <Compile Include="Core\BusTests.Publish.cs" />
-    <Compile Include="Core\BusTests.Subscribe.cs" />
-    <Compile Include="Core\BusTests.DeserializationErrors.cs" />
-    <Compile Include="Core\BusTests.Core.cs" />
-    <Compile Include="Core\MessageContextAwareBusTests.cs" />
-    <Compile Include="Core\RoundRobinPeerSelectorTests.cs" />
-    <Compile Include="Directory\PeerDirectoryClientTests.Performance.cs" />
-    <Compile Include="Directory\PeerSubscriptionList.cs" />
-    <Compile Include="Directory\PeerSubscriptionTreeTests.Performance.cs" />
-    <Compile Include="Dispatch\AsyncMessageHandlerInvokerTests.cs" />
-    <Compile Include="Dispatch\BatchedMessageHandlerInvokerTests.cs" />
-    <Compile Include="Dispatch\DispatchMessages\AsyncCommand.cs" />
-    <Compile Include="Dispatch\DispatchMessages\AsyncDoNotStartTaskCommand.cs" />
-    <Compile Include="Dispatch\DispatchMessages\AsyncDoNotStartTaskCommandHandler.cs" />
-    <Compile Include="Dispatch\DispatchMessages\AsyncExecutableEvent.cs" />
-    <Compile Include="Dispatch\DispatchMessages\ExecutableEvent.cs" />
-    <Compile Include="Dispatch\DispatchMessages\ExecutableEventHandler.cs" />
-    <Compile Include="Dispatch\DispatchMessages\CapturingTaskSchedulerAsyncCommandHandler.cs" />
-    <Compile Include="Dispatch\DispatchMessages\CapturingTaskSchedulerSyncCommandHandler.cs" />
-    <Compile Include="Dispatch\DispatchMessages\ForwardCommand.cs" />
-    <Compile Include="Dispatch\DispatchMessages\ForwardCommandHandler.cs" />
-    <Compile Include="Dispatch\DispatchQueueTests.cs" />
-    <Compile Include="Dispatch\DispatchMessages\ManualCommand.cs" />
-    <Compile Include="Dispatch\DispatchMessages\ManualCommandHandler.cs" />
-    <Compile Include="Dispatch\DynamicMessageHandlerInvokerTests.cs" />
-    <Compile Include="Dispatch\MessageDispatcherTests.CustomQueues.cs" />
-    <Compile Include="Dispatch\DispatchMessages\Namespace1\UseOtherQueue.cs" />
-    <Compile Include="Dispatch\DispatchMessages\Namespace1\Namespace2\SyncCommandHandlerWithOtherQueueName.cs" />
-    <Compile Include="Dispatch\DispatchMessages\ReplyCommand.cs" />
-    <Compile Include="Dispatch\DispatchMessages\ReplyCommandHandler.cs" />
-    <Compile Include="Dispatch\DispatchMessages\RoutableCommand.cs" />
-    <Compile Include="Dispatch\DispatchMessages\RoutableCommandHandler.cs" />
-    <Compile Include="Dispatch\DispatchMessages\SyncCommandHandlerWithQueueName2.cs" />
-    <Compile Include="Dispatch\Pipes\AsyncPipeInvocationTests.cs" />
-    <Compile Include="DomainExceptionTests.cs" />
-    <Compile Include="Hosting\HostInitializerHelperTests.cs" />
-    <Compile Include="Hosting\PeriodicActionHostInitializerTests.cs" />
-    <Compile Include="Initialization\ZebusRegistryTests.cs" />
-    <Compile Include="Lotus\CustomProcessingFailedTests.cs" />
-    <Compile Include="Lotus\ReplayMessageHandlerTests.cs" />
-    <Compile Include="MessageContextTests.cs" />
-    <Compile Include="Messages\EmptyCommand.cs" />
-    <Compile Include="Messages\FakeCommandWithTimestamp.cs" />
-    <Compile Include="Messages\FakeInfrastructureTransientCommand.cs" />
-    <Compile Include="Messages\FakeRoutableCommandWithBoolean.cs" />
-    <Compile Include="Messages\FakeRoutableCommandWithEnum.cs">
-      <SubType>Code</SubType>
-    </Compile>
-    <Compile Include="Routing\BindingKeyPredicateBuilderTests.cs" />
-    <Compile Include="Routing\BindingKeyTests.cs" />
-    <Compile Include="Scan\MessageHandlerInvokerLoaderTests.cs" />
-    <Compile Include="Scan\StaticAsyncMessageHandlerInvokerLoaderTests.cs" />
-    <Compile Include="Scan\SyncMessageHandlerInvokerLoaderTests.cs" />
-    <Compile Include="Dispatch\DispatchMessages\SyncCommandHandlerWithQueueName1.cs" />
-    <Compile Include="Messages\FakeCommandResult.cs" />
-    <Compile Include="Directory\PeerDescriptorTests.cs" />
-    <Compile Include="Directory\PeerDirectoryClientTests.cs" />
-    <Compile Include="Dispatch\DispatchMessages\DispatchCommand.cs" />
-    <Compile Include="Dispatch\DispatchMessages\AsyncFailingCommand.cs" />
-    <Compile Include="Dispatch\DispatchMessages\AsyncFailingCommandHandler.cs" />
-    <Compile Include="Dispatch\DispatchMessages\FailingCommand.cs" />
-    <Compile Include="Dispatch\DispatchMessages\FailingCommandHandler.cs" />
-    <Compile Include="Dispatch\SyncMessageHandlerInvokerTests.cs" />
-    <Compile Include="Dispatch\DispatchMessages\AsyncCommandHandler.cs" />
-    <Compile Include="Dispatch\DispatchMessages\SyncCommandHandler.cs" />
-    <Compile Include="Messages\FakeRoutableCommand.cs" />
-    <Compile Include="Messages\FakeCommand.cs" />
-    <Compile Include="Messages\FakeEvent.cs" />
-    <Compile Include="Log4netConfigurator.cs" />
-    <Compile Include="MessageIdTests.cs" />
-    <Compile Include="Messages\FakeInfrastructureCommand.cs" />
-    <Compile Include="Messages\FakeNonPersistentCommand.cs" />
-    <Compile Include="MessageUtilTests.cs" />
-    <Compile Include="Persistence\PersistentTransportTests.cs" />
-    <Compile Include="Persistence\PersistentTransportFixture.cs" />
-    <Compile Include="Persistence\TranscientPersistentTransportTests.cs" />
-    <Compile Include="Dispatch\Pipes\AttributePipeSourceTests.cs" />
-    <Compile Include="Dispatch\Pipes\TestPipe.cs" />
-    <Compile Include="Dispatch\Pipes\FakePipeSource.cs" />
-    <Compile Include="Dispatch\Pipes\PipeInvocationTests.cs" />
-    <Compile Include="Dispatch\Pipes\PipeManagerTests.cs" />
-    <Compile Include="Dispatch\Pipes\PipeSourceTests.cs" />
-    <Compile Include="Program.cs" />
-    <Compile Include="Properties\AssemblyInfo.cs" />
-    <Compile Include="Dispatch\MessageDispatcherTests.cs" />
-    <Compile Include="Dispatch\DispatchMessages\ScanCommand1.cs" />
-    <Compile Include="Dispatch\DispatchMessages\ScanCommand2.cs" />
-    <Compile Include="Dispatch\DispatchMessages\ScanCommandHandler1.cs" />
-    <Compile Include="Dispatch\DispatchMessages\ScanCommandHandler2.cs" />
-    <Compile Include="SerializationTests.cs" />
-    <Compile Include="Directory\PeerSubscriptionTreeTests.cs" />
-    <Compile Include="Serialization\MessageSerializerTests.cs" />
-    <Compile Include="Serialization\TestMessageSerializer.cs" />
-    <Compile Include="SubscriptionTests.cs" />
-    <Compile Include="SubscriptionTests.Legacy.cs" />
-    <Compile Include="TestDataBuilder.cs" />
-    <Compile Include="Transport\BackwardCompatibilityTests.cs" />
-    <Compile Include="Transport\V1_5_0\OriginatorInfo_1_5_0.cs" />
-    <Compile Include="Transport\TransportMessageReaderTests.cs" />
-    <Compile Include="Transport\TransportMessageWriterTests.cs" />
-    <Compile Include="Transport\V1_5_0\TransportMessage_1_5_0.cs" />
-    <Compile Include="Transport\ZmqTests.cs" />
-    <Compile Include="Transport\ZmqTransportPerformanceTests.cs" />
-    <Compile Include="Transport\ZmqTransportTests.cs" />
-    <Compile Include="Util\Collections\ConcurrentSetTests.cs" />
-    <Compile Include="Util\Collections\FlushableBlockingCollectionPerformanceTests.cs" />
-    <Compile Include="Util\Collections\FlushableBlockingCollectionTests.cs" />
-    <Compile Include="Util\CultureScopeTests.cs" />
-    <Compile Include="Util\DisposableActionTests.cs" />
-    <Compile Include="Util\Extensions\ExtendDictionaryTests.cs" />
-    <Compile Include="Util\Extensions\ExtendICollectionTests.cs" />
-    <Compile Include="Util\Extensions\ExtendStringTests.cs" />
-    <Compile Include="Util\Extensions\ExtendTypeTests.cs" />
-    <Compile Include="Util\FluentTime\TimeSpanCreationTests.cs" />
-    <Compile Include="Util\GenZero\BufferTests.cs" />
-    <Compile Include="Util\SystemDateTimeTests.cs" />
-    <Compile Include="Util\TcpUtilTests.cs" />
-    <Compile Include="Util\TypeUtilTests.cs" />
-    <Compile Include="Util\UniqueTimestampProviderTests.cs" />
-  </ItemGroup>
+
   <ItemGroup>
     <None Include="app.config" />
     <None Include="log4net.config">
       <CopyToOutputDirectory>PreserveNewest</CopyToOutputDirectory>
     </None>
-    <None Include="packages.config" />
-    <EmbeddedResource Include="Transport\transport_message_1_4_1.bin" />
   </ItemGroup>
+
   <ItemGroup>
-    <Folder Include="Util\IO\" />
+    <Service Include="{82a7f48d-3b50-4b1e-b82e-3ada8210c358}" />
   </ItemGroup>
-  <ItemGroup>
-    <ProjectReference Include="..\Abc.Zebus.Testing\Abc.Zebus.Testing.csproj">
-      <Project>{4ce123df-8021-411c-929b-53e5c5fc1e04}</Project>
-      <Name>Abc.Zebus.Testing</Name>
-    </ProjectReference>
-    <ProjectReference Include="..\Abc.Zebus.Tests.TestExe\Abc.Zebus.Tests.TestExe.csproj">
-      <Project>{652017d8-eb34-4a79-b14f-db0477e43915}</Project>
-      <Name>Abc.Zebus.Tests.TestExe</Name>
-    </ProjectReference>
-    <ProjectReference Include="..\Abc.Zebus\Abc.Zebus.csproj">
-      <Project>{1f4c6307-6113-40d5-bf42-4b6bf5df13b2}</Project>
-      <Name>Abc.Zebus</Name>
-    </ProjectReference>
-  </ItemGroup>
-  <Import Project="$(MSBuildToolsPath)\Microsoft.CSharp.targets" />
-  <PropertyGroup>
-    <PreBuildEvent>
-    </PreBuildEvent>
-  </PropertyGroup>
-  <!-- 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/Abc.Zebus.Tests/Core/BusPerformanceTests.cs

@@ -19,7 +19,7 @@ namespace Abc.Zebus.Tests.Core
     public class BusPerformanceTests
     {
         // this must be a valid directory endpoint
-        private static readonly string _directoryEndPoint = Environment.GetEnvironmentVariable("ZEBUS_TEST_DIRECTORY", EnvironmentVariableTarget.User);
+        private static readonly string _directoryEndPoint = Environment.GetEnvironmentVariable("ZEBUS_TEST_DIRECTORY");
 
         [Test]
         public void MeasureCommandThroughputWithoutPersistence()

src/Abc.Zebus.Tests/Directory/PeerSubscriptionTreeTests.Performance.cs

@@ -32,7 +32,7 @@ namespace Abc.Zebus.Tests.Directory
                 subscriptionTree.Add(peerSubscription.Item1, peerSubscription.Item2.BindingKey);
             }
 
-            var bindingKey = BindingKey.Split(routingKey);
+            var bindingKey = new BindingKey(routingKey.Split('.'));
 
             Console.WriteLine("{0} test -------------", subscriptionList.GetType().Name);
             Console.WriteLine();

src/Abc.Zebus.Tests/Directory/PeerSubscriptionTreeTests.cs

@@ -3,6 +3,7 @@ using Abc.Zebus.Directory;
 using Abc.Zebus.Routing;
 using Abc.Zebus.Testing.Extensions;
 using Abc.Zebus.Tests.Messages;
+using Abc.Zebus.Tests.Routing;
 using NUnit.Framework;
 
 namespace Abc.Zebus.Tests.Directory
@@ -20,11 +21,11 @@ namespace Abc.Zebus.Tests.Directory
             // Arrange
             var peerSubscriptionTree = new PeerSubscriptionTree();
             var peer = new Peer(new PeerId("jesuistonpeer"), "endpoint");
-            var subscription = BindingKey.Split("*");
+            var subscription = BindingKeyHelper.CreateFromString("*", '.');
             peerSubscriptionTree.Add(peer, subscription);
 
             // Act
-            var matchingPeers = peerSubscriptionTree.GetPeers(BindingKey.Split(routingKey));
+            var matchingPeers = peerSubscriptionTree.GetPeers(BindingKeyHelper.CreateFromString(routingKey, '.'));
 
             // Assert
             matchingPeers.Single().ShouldEqual(peer);
@@ -64,7 +65,7 @@ namespace Abc.Zebus.Tests.Directory
             peerSubscriptionTree.Add(peer, BindingKey.Empty);
 
             // Act
-            var matchingPeers = peerSubscriptionTree.GetPeers(BindingKey.Split(routingKey));
+            var matchingPeers = peerSubscriptionTree.GetPeers(BindingKeyHelper.CreateFromString(routingKey, '.'));
 
             // Assert
             matchingPeers.Single().ShouldEqual(peer);
@@ -77,12 +78,12 @@ namespace Abc.Zebus.Tests.Directory
             // Arrange
             var peerSubscriptionTree = new PeerSubscriptionTree();
             var peer = new Peer(new PeerId("jesuistonpeer"), "endpoint");
-            var subscription = BindingKey.Split("*.*.*");
+            var subscription = BindingKeyHelper.CreateFromString("*.*.*", '.');
 
             peerSubscriptionTree.Add(peer, subscription);
 
             // Act
-            var matchingPeers = peerSubscriptionTree.GetPeers(BindingKey.Split(routingKey));
+            var matchingPeers = peerSubscriptionTree.GetPeers(BindingKeyHelper.CreateFromString(routingKey, '.'));
 
             // Assert
             matchingPeers.Single().ShouldEqual(peer);
@@ -97,7 +98,7 @@ namespace Abc.Zebus.Tests.Directory
             // Arrange
             var peerSubscriptionTree = new PeerSubscriptionTree();
             var peer = new Peer(new PeerId("jesuistonpeer"), "endpoint");
-            var subscription = BindingKey.Split(subscriptionKey);
+            var subscription = BindingKeyHelper.CreateFromString(subscriptionKey, '.');
 
             peerSubscriptionTree.Add(peer, subscription);
 
@@ -115,7 +116,7 @@ namespace Abc.Zebus.Tests.Directory
             // Arrange
             var peerSubscriptionTree = new PeerSubscriptionTree();
             var peer = new Peer(new PeerId("jesuistonpeer"), "endpoint");
-            var subscription = BindingKey.Split(subscriptionKey);
+            var subscription = BindingKeyHelper.CreateFromString(subscriptionKey, '.');
 
             peerSubscriptionTree.Add(peer, subscription);
 
@@ -138,11 +139,11 @@ namespace Abc.Zebus.Tests.Directory
             var peer = new Peer(new PeerId("1"), "endpoint");
 
             peerSubscriptionTree.IsEmpty.ShouldBeTrue();
-            peerSubscriptionTree.Add(peer, BindingKey.Split(subscriptionKey));
-            var subscription = BindingKey.Split("lol");
+            peerSubscriptionTree.Add(peer, BindingKeyHelper.CreateFromString(subscriptionKey, '.'));
+            var subscription = BindingKeyHelper.CreateFromString("lol", '.');
             peerSubscriptionTree.Add(peer, subscription);
             peerSubscriptionTree.IsEmpty.ShouldBeFalse();
-            peerSubscriptionTree.Remove(peer, BindingKey.Split(subscriptionKey));
+            peerSubscriptionTree.Remove(peer, BindingKeyHelper.CreateFromString(subscriptionKey, '.'));
             peerSubscriptionTree.IsEmpty.ShouldBeFalse();
             peerSubscriptionTree.Remove(peer, subscription);
             peerSubscriptionTree.IsEmpty.ShouldBeTrue();
@@ -188,31 +189,31 @@ namespace Abc.Zebus.Tests.Directory
             var peer9 = new Peer(new PeerId("9"), "endpoint");
             var peer0 = new Peer(new PeerId("0"), "endpoint");
 
-            peerSubscriptionTree.Add(peer1, BindingKey.Split("#"));
-            peerSubscriptionTree.Add(peer2, BindingKey.Split("a.b"));
-            peerSubscriptionTree.Add(peer3, BindingKey.Split("a.*"));
-            peerSubscriptionTree.Add(peer4, BindingKey.Split("b.*.c"));
-            peerSubscriptionTree.Add(peer5, BindingKey.Split("b.*.f"));
-            peerSubscriptionTree.Add(peer6, BindingKey.Split("d.*.c"));
-            peerSubscriptionTree.Add(peer7, BindingKey.Split("a"));
-            peerSubscriptionTree.Add(peer8, BindingKey.Split("*.*"));
-            peerSubscriptionTree.Add(peer9, BindingKey.Split("a.#"));
-            peerSubscriptionTree.Add(peer0, BindingKey.Split("*"));
+            peerSubscriptionTree.Add(peer1, BindingKeyHelper.CreateFromString("#", '.'));
+            peerSubscriptionTree.Add(peer2, BindingKeyHelper.CreateFromString("a.b", '.'));
+            peerSubscriptionTree.Add(peer3, BindingKeyHelper.CreateFromString("a.*", '.'));
+            peerSubscriptionTree.Add(peer4, BindingKeyHelper.CreateFromString("b.*.c", '.'));
+            peerSubscriptionTree.Add(peer5, BindingKeyHelper.CreateFromString("b.*.f", '.'));
+            peerSubscriptionTree.Add(peer6, BindingKeyHelper.CreateFromString("d.*.c", '.'));
+            peerSubscriptionTree.Add(peer7, BindingKeyHelper.CreateFromString("a", '.'));
+            peerSubscriptionTree.Add(peer8, BindingKeyHelper.CreateFromString("*.*", '.'));
+            peerSubscriptionTree.Add(peer9, BindingKeyHelper.CreateFromString("a.#", '.'));
+            peerSubscriptionTree.Add(peer0, BindingKeyHelper.CreateFromString("*", '.'));
 
             // Act - Assert
-            var peers = peerSubscriptionTree.GetPeers(BindingKey.Split("b.1.c"));
+            var peers = peerSubscriptionTree.GetPeers(BindingKeyHelper.CreateFromString("b.1.c", '.'));
             peers.Count.ShouldEqual(2);
             peers.ShouldContain(peer1);
             peers.ShouldContain(peer4);
 
-            peers = peerSubscriptionTree.GetPeers(BindingKey.Split("a.1"));
+            peers = peerSubscriptionTree.GetPeers(BindingKeyHelper.CreateFromString("a.1", '.'));
             peers.Count.ShouldEqual(4);
             peers.ShouldContain(peer1);
             peers.ShouldContain(peer3);
             peers.ShouldContain(peer8);
             peers.ShouldContain(peer9);
 
-            peers = peerSubscriptionTree.GetPeers(BindingKey.Split("a"));
+            peers = peerSubscriptionTree.GetPeers(BindingKeyHelper.CreateFromString("a", '.'));
             peers.Count.ShouldEqual(3);
             peers.ShouldContain(peer1);
             peers.ShouldContain(peer7);
@@ -232,12 +233,12 @@ namespace Abc.Zebus.Tests.Directory
             // Arrange
             var peerSubscriptionTree = new PeerSubscriptionTree();
             var peer = new Peer(new PeerId("jesuistonpeer"), "endpoint");
-            var subscription = BindingKey.Split(subscriptionKey);
+            var subscription = BindingKeyHelper.CreateFromString(subscriptionKey, '.');
 
             peerSubscriptionTree.Add(peer, subscription);
 
             // Act
-            var matchingPeers = peerSubscriptionTree.GetPeers(BindingKey.Split(routingKey));
+            var matchingPeers = peerSubscriptionTree.GetPeers(BindingKeyHelper.CreateFromString(routingKey, '.'));
 
             // Assert
             matchingPeers.ShouldBeEmpty();

src/Abc.Zebus.Tests/Dispatch/DispatchQueueTests.cs

@@ -60,6 +60,23 @@ namespace Abc.Zebus.Tests.Dispatch
             task.Wait(500.Milliseconds()).ShouldBeTrue();
         }
 
+        [Test]
+        public void should_continue_processing_messages_after_continuation_error()
+        {
+            _dispatchQueue.Start();
+
+            var message1 = new ExecutableEvent { Callback = x => throw new Exception("Processing error") };
+            var dispatch = new MessageDispatch(MessageContext.CreateTest(), message1, (d, r) => throw new Exception("Continuation error"));
+            dispatch.SetHandlerCount(1);
+
+            _dispatchQueue.Enqueue(dispatch, new TestMessageHandlerInvoker<ExecutableEvent>());
+
+            var message2 = new ExecutableEvent();
+            var task = EnqueueInvocation(message2);
+
+            task.Wait(500.Milliseconds()).ShouldBeTrue();
+        }
+
         [Test]
         public void should_finish_current_invocation_before_stopping()
         {

src/Abc.Zebus.Tests/Persistence/PersistentTransportTests.cs

@@ -67,6 +67,25 @@ namespace Abc.Zebus.Tests.Persistence
             forwardedTransportMessage.WasPersisted.ShouldEqual(true);
         }
 
+        [Test]
+        public void should_force_WasPersisted_for_replayed_messages_during_safety_phase()
+        {
+            Transport.Start();
+
+            InnerTransport.RaiseMessageReceived(new ReplayPhaseEnded(ReplayId).ToTransportMessage());
+
+            var sourceTransportMessage = new FakeCommand(123).ToTransportMessage();
+            sourceTransportMessage.WasPersisted = null;
+
+            var replayTransportMessage = sourceTransportMessage.ToReplayedTransportMessage(ReplayId);
+            InnerTransport.RaiseMessageReceived(replayTransportMessage);
+
+            Wait.Until(() => MessagesForwardedToBus.Count == 1, 150.Milliseconds());
+
+            var forwardedTransportMessage = MessagesForwardedToBus.ExpectedSingle();
+            forwardedTransportMessage.WasPersisted.ShouldEqual(true);
+        }
+
         [Test]
         public void should_forward_a_normal_message_after_a_back_to_live_event()
         {
@@ -78,9 +97,10 @@ namespace Abc.Zebus.Tests.Persistence
 
             InnerTransport.RaiseMessageReceived(new ReplayPhaseEnded(StartMessageReplayCommand.ReplayId).ToTransportMessage());
 
-            Thread.Sleep(50);
-            MessagesForwardedToBus.Count.ShouldEqual(1);
-            MessagesForwardedToBus.Single().ShouldEqualDeeply(transportMessageToForward);
+            Wait.Until(() => MessagesForwardedToBus.Count == 1, 150.Milliseconds());
+
+            var transportMessage = MessagesForwardedToBus.Single();
+            transportMessage.ShouldEqualDeeply(transportMessageToForward);
         }
 
         [Test, Repeat(20)]

src/Abc.Zebus.Tests/Routing/BindingKeyExtensions.cs

@@ -0,0 +1,14 @@
+using Abc.Zebus.Routing;
+using System;
+using System.Collections.Generic;
+using System.Linq;
+using System.Text;
+using System.Threading.Tasks;
+
+namespace Abc.Zebus.Tests.Routing
+{
+    public static class BindingKeyHelper
+    {
+        public static BindingKey CreateFromString(string s, char separator) => new BindingKey(s.Split(separator));
+    }
+}

src/Abc.Zebus.Tests/SubscriptionTests.Legacy.cs

@@ -1,92 +0,0 @@
-using Abc.Zebus.Routing;
-using Abc.Zebus.Testing.Extensions;
-using Abc.Zebus.Tests.Messages;
-using NUnit.Framework;
-
-namespace Abc.Zebus.Tests
-{
-    public partial class SubscriptionTests
-    {
-        [Test]
-        public void should_match_joined_routing_key_with_single_token_subscription()
-        {
-            var subscription = new Subscription(MessageUtil.TypeId<FakeRoutableCommand>(), new BindingKey("Abc.Service.0"));
-
-            var routingKey = BindingKey.Joined("Abc.Service.0");
-            subscription.Matches(routingKey).ShouldBeTrue();
-        }
-
-        [Test]
-        public void should_not_match_invalid_joined_routing_key_with_single_token_subscription()
-        {
-            var subscription = new Subscription(MessageUtil.TypeId<FakeRoutableCommand>(), new BindingKey("Abc.Service.0"));
-
-            var routingKey = BindingKey.Joined("Abc.Service.1");
-            subscription.Matches(routingKey).ShouldBeFalse();
-        }
-
-        [Test]
-        public void should_match_joined_routing_key_with_splitted_token_subscription()
-        {
-            var subscription = new Subscription(MessageUtil.TypeId<FakeRoutableCommand>(), new BindingKey("Abc", "Service", "0"));
-
-            var routingKey = BindingKey.Joined("Abc.Service.0");
-            subscription.Matches(routingKey).ShouldBeTrue();
-        }
-
-        [Test]
-        public void should_match_joined_routing_key_with_splitted_token_subscription_and_wildcard_1()
-        {
-            var subscription = new Subscription(MessageUtil.TypeId<FakeRoutableCommand>(), new BindingKey("Abc", "Service", "*", "Foo"));
-
-            var routingKey = BindingKey.Joined("Abc.Service.42.Foo");
-            subscription.Matches(routingKey).ShouldBeTrue();
-        }
-
-        [Test]
-        public void should_match_joined_routing_key_with_splitted_token_subscription_and_wildcard_2()
-        {
-            var subscription = new Subscription(MessageUtil.TypeId<FakeRoutableCommand>(), new BindingKey("Abc", "#"));
-
-            var routingKey = BindingKey.Joined("Abc.Service.42");
-            subscription.Matches(routingKey).ShouldBeTrue();
-        }
-
-        [Test]
-        public void should_not_match_invalid_joined_routing_key_with_splitted_token_subscription()
-        {
-            var subscription = new Subscription(MessageUtil.TypeId<FakeRoutableCommand>(), new BindingKey("Abc", "Service", "0"));
-
-            var routingKey = BindingKey.Joined("Abc.Service.1");
-            subscription.Matches(routingKey).ShouldBeFalse();
-        }
-
-        [Test]
-        public void should_match_joined_qpid_message_with_machine_name()
-        {
-            var bindingKey = BindingKey.Joined("machinename.Abc.Foo.0");
-            var subscription = Subscription.Matching<InstanceHeartBeat>(x => x.InstanceName == "Abc.Foo.0");
-
-            subscription.Matches(bindingKey).ShouldBeTrue();
-        }
-
-        [Test]
-        public void should_match_joined_qpid_message_without_machine_name()
-        {
-            var bindingKey = BindingKey.Joined(".Abc.Foo.0");
-            var subscription = Subscription.Matching<InstanceHeartBeat>(x => x.InstanceName == "Abc.Foo.0");
-
-            subscription.Matches(bindingKey).ShouldBeTrue();
-        }
-
-        [Routable]
-        public class InstanceHeartBeat : IEvent
-        {
-            [RoutingPosition(2)]
-            public string InstanceName;
-
-            [RoutingPosition(1)]
-            public string MachineName;
-        }
-    }
-}

src/Abc.Zebus.Tests/SubscriptionTests.cs

@@ -5,6 +5,7 @@ using Abc.Zebus.Routing;
 using Abc.Zebus.Testing.Extensions;
 using Abc.Zebus.Testing.Measurements;
 using Abc.Zebus.Tests.Messages;
+using Abc.Zebus.Tests.Routing;
 using NUnit.Framework;
 
 namespace Abc.Zebus.Tests
@@ -26,7 +27,7 @@ namespace Abc.Zebus.Tests
         public void single_star_should_always_match(string routingKey)
         {
             var subscription = CreateSubscription("*");
-            subscription.Matches(BindingKey.Split(routingKey)).ShouldBeTrue();
+            subscription.Matches(BindingKeyHelper.CreateFromString(routingKey, '.')).ShouldBeTrue();
         }
 
         [TestCase("whatever")]
@@ -35,7 +36,7 @@ namespace Abc.Zebus.Tests
         public void single_dash_should_always_match(string routingKey)
         {
             var subscription = CreateSubscription("#");
-            subscription.Matches(BindingKey.Split(routingKey)).ShouldBeTrue();
+            subscription.Matches(BindingKeyHelper.CreateFromString(routingKey, '.')).ShouldBeTrue();
         }
 
         [TestCase("whatever")]
@@ -44,7 +45,7 @@ namespace Abc.Zebus.Tests
         public void empty_bindingkey_should_always_match(string routingKey)
         {
             var subscription = new Subscription(new MessageTypeId(typeof(FakeCommand)), BindingKey.Empty);
-            subscription.Matches(BindingKey.Split(routingKey)).ShouldBeTrue();
+            subscription.Matches(BindingKeyHelper.CreateFromString(routingKey, '.')).ShouldBeTrue();
         }
 
         [TestCase("a.b.c")]
@@ -52,7 +53,7 @@ namespace Abc.Zebus.Tests
         public void stars_should_always_match_if_same_number_of_parts(string routingKey)
         {
             var subscription = CreateSubscription("*.*.*");
-            subscription.Matches(BindingKey.Split(routingKey)).ShouldBeTrue();
+            subscription.Matches(BindingKeyHelper.CreateFromString(routingKey, '.')).ShouldBeTrue();
         }
 
         [TestCase("a.b.*")]
@@ -62,7 +63,7 @@ namespace Abc.Zebus.Tests
         public void binding_key_with_star_should_match_routing_key(string bindingKey)
         {
             var subscription = CreateSubscription(bindingKey);
-            subscription.Matches(BindingKey.Split("a.b.c")).ShouldBeTrue();
+            subscription.Matches(BindingKeyHelper.CreateFromString("a.b.c", '.')).ShouldBeTrue();
         }
 
         [TestCase("a.b.#")]
@@ -70,7 +71,7 @@ namespace Abc.Zebus.Tests
         public void binding_key_with_dashr_should_match_routing_key(string bindingKey)
         {
             var subscription = CreateSubscription(bindingKey);
-            subscription.Matches(BindingKey.Split("a.b.c")).ShouldBeTrue();
+            subscription.Matches(BindingKeyHelper.CreateFromString("a.b.c", '.')).ShouldBeTrue();
         }
 
         [TestCase("a.b", "a.b.c.d")]
@@ -79,14 +80,14 @@ namespace Abc.Zebus.Tests
         public void should_not_match_binding_key(string routingKey, string bindingKey)
         {
             var subscription = CreateSubscription(bindingKey);
-            subscription.Matches(BindingKey.Split(routingKey)).ShouldBeFalse();
+            subscription.Matches(BindingKeyHelper.CreateFromString(routingKey, '.')).ShouldBeFalse();
         }
 
         [Test]
         public void exact_same_routing_key_should_match_binding_key()
         {
             var subscription = CreateSubscription("a.b.c");
-            subscription.Matches(BindingKey.Split("a.b.c")).ShouldBeTrue();
+            subscription.Matches(BindingKeyHelper.CreateFromString("a.b.c", '.')).ShouldBeTrue();
         }
 
         [Test]
@@ -94,7 +95,7 @@ namespace Abc.Zebus.Tests
         {
             var subscription = Subscription.ByExample(x => new FakeRoutableCommand(12, "name"));
             subscription.MessageTypeId.ShouldEqual(new MessageTypeId(typeof(FakeRoutableCommand)));
-            subscription.BindingKey.ShouldEqual(BindingKey.Split("12.name.*"));
+            subscription.BindingKey.ShouldEqual(BindingKeyHelper.CreateFromString("12.name.*", '.'));
         }
 
         [Test]
@@ -118,7 +119,7 @@ namespace Abc.Zebus.Tests
         {
             var subscription = Subscription.ByExample(x => new FakeRoutableCommand(x.Any<decimal>(), "name"));
             subscription.MessageTypeId.ShouldEqual(new MessageTypeId(typeof(FakeRoutableCommand)));
-            subscription.BindingKey.ShouldEqual(BindingKey.Split("*.name.*"));
+            subscription.BindingKey.ShouldEqual(BindingKeyHelper.CreateFromString("*.name.*", '.'));
         }
 
         [Test]
@@ -241,7 +242,7 @@ namespace Abc.Zebus.Tests
         public void MeasurePerformance(string routingKey, string bindingKey)
         {
             var subscription = CreateSubscription(bindingKey);
-            var key = BindingKey.Split(routingKey);
+            var key = BindingKeyHelper.CreateFromString(routingKey, '.');
 
             Measure.Execution(x =>
             {
@@ -253,7 +254,7 @@ namespace Abc.Zebus.Tests
 
         private Subscription CreateSubscription(string bindingKey)
         {
-            return new Subscription(new MessageTypeId(typeof(FakeCommand)), BindingKey.Split(bindingKey));
+            return new Subscription(new MessageTypeId(typeof(FakeCommand)), BindingKeyHelper.CreateFromString(bindingKey, '.'));
         }
 
         private int GetFieldValue()

src/Abc.Zebus.Tests/Testing/TestBusTests.cs

@@ -0,0 +1,23 @@
+using Abc.Zebus.Testing;
+using Abc.Zebus.Tests.Messages;
+using NUnit.Framework;
+
+namespace Abc.Zebus.Tests.Testing
+{
+    [TestFixture]
+    public class TestBusTests
+    {
+        [Test]
+        public void should_notify_explicit_handlers()
+        {
+            var bus = new TestBus();
+
+            bool received = false;
+            bus.Subscribe<FakeEvent>(e => received = true);
+
+            bus.Publish(new FakeEvent(1));
+
+            Assert.That(received);
+        }
+    }
+}

src/Abc.Zebus.Tests/Transport/TransportMessageReaderTests.cs

@@ -1,6 +1,5 @@
 using System.Collections.Generic;
 using System.IO;
-using Abc.Zebus.Serialization;
 using Abc.Zebus.Serialization.Protobuf;
 using Abc.Zebus.Testing;
 using Abc.Zebus.Testing.Extensions;
@@ -8,7 +7,7 @@ using Abc.Zebus.Testing.Measurements;
 using Abc.Zebus.Tests.Messages;
 using Abc.Zebus.Transport;
 using NUnit.Framework;
-using Serializer = ProtoBuf.Serializer;
+using ProtoBuf;
 
 namespace Abc.Zebus.Tests.Transport
 {
@@ -108,14 +107,15 @@ namespace Abc.Zebus.Tests.Transport
             var inputStream = new CodedInputStream(outputStream.Buffer, 0, outputStream.Position);
             inputStream.ReadTransportMessage();
 
-            const int count = 1000 * 1000 * 1000;
+            const int count = 100_000_000;
             using (Measure.Throughput(count))
             {
                 for (var i = 0; i < count; i++)
                 {
+                    inputStream.Position = 0;
                     inputStream.ReadTransportMessage();
                 }
             }
         }
     }
-}
+}

src/Abc.Zebus.Tests/Transport/ZmqTransportPerformanceTests.cs

@@ -61,7 +61,7 @@ namespace Abc.Zebus.Tests.Transport
         private ZmqTransport CreateAndStartZmqTransport(string peerId, Action<TransportMessage> onMessageReceived = null)
         {
             var configurationMock = new Mock<IZmqTransportConfiguration>();
-            var transport = new ZmqTransport(configurationMock.Object, new ZmqSocketOptions());
+            var transport = new ZmqTransport(configurationMock.Object, new ZmqSocketOptions(), new DefaultZmqOutboundSocketErrorHandler());
             transport.Configure(new PeerId(peerId), "test");
             transport.SocketOptions.SendTimeout = 5.Seconds();
 

src/Abc.Zebus.Tests/Transport/ZmqTransportTests.cs

@@ -51,7 +51,7 @@ namespace Abc.Zebus.Tests.Transport
         {
             var configurationMock = new Mock<IZmqTransportConfiguration>();
             configurationMock.SetupGet(x => x.WaitForEndOfStreamAckTimeout).Returns(100.Milliseconds());
-            var transport = new ZmqTransport(configurationMock.Object, new ZmqSocketOptions());
+            var transport = new ZmqTransport(configurationMock.Object, new ZmqSocketOptions(), new DefaultZmqOutboundSocketErrorHandler());
 
             Assert.That(transport.Stop, Throws.Nothing);
         }
@@ -145,6 +145,7 @@ namespace Abc.Zebus.Tests.Transport
         public void should_send_message_to_peer_and_persistence()
         {
             // standard case: the message is forwarded to the persistence through SendContext.PersistencePeer
+            // the target peer is up
 
             var senderTransport = CreateAndStartZmqTransport();
 
@@ -172,6 +173,30 @@ namespace Abc.Zebus.Tests.Transport
             messageFromPersistence.PersistentPeerIds.ShouldBeEquivalentTo(new[] { receiverPeer.Id });
         }
 
+        [Test]
+        public void should_send_message_to_persistence()
+        {
+            // standard case: the message is forwarded to the persistence through SendContext.PersistencePeer
+            // the target peer is down
+
+            var senderTransport = CreateAndStartZmqTransport();
+
+            var receiverPeerId = new PeerId("Abc.R.0");
+
+            var persistenceMessages = new ConcurrentBag<TransportMessage>();
+            var persistenceTransport = CreateAndStartZmqTransport(onMessageReceived: persistenceMessages.Add);
+            var persistencePeer = new Peer(persistenceTransport.PeerId, persistenceTransport.InboundEndPoint);
+
+            var message = new FakeCommand(999).ToTransportMessage();
+            senderTransport.Send(message, Enumerable.Empty<Peer>(), new SendContext { PersistentPeerIds = { receiverPeerId }, PersistencePeer = persistencePeer });
+
+            Wait.Until(() => persistenceMessages.Count == 1, 500.Milliseconds());
+            var messageFromPersistence = persistenceMessages.ExpectedSingle();
+            messageFromPersistence.ShouldHaveSamePropertiesAs(message, "Environment", "WasPersisted", "PersistentPeerIds", "IsPersistTransportMessage");
+            messageFromPersistence.Environment.ShouldEqual("Test");
+            messageFromPersistence.PersistentPeerIds.ShouldBeEquivalentTo(new[] { receiverPeerId });
+        }
+
         [Test]
         public void should_send_persist_transport_message_to_persistence()
         {
@@ -498,31 +523,49 @@ namespace Abc.Zebus.Tests.Transport
         [Test]
         public void should_process_all_messages_in_buffer_on_stop()
         {
-            var receivedMessages = new List<TransportMessage>();
+            var state = new should_process_all_messages_in_buffer_on_stop_state { ShouldSend = true };
 
-            var receivingTransport = CreateAndStartZmqTransport(onMessageReceived: receivedMessages.Add);
+            var receivingTransport = CreateAndStartZmqTransport(onMessageReceived: x => state.ReceivedMessageCount++);
             var sendingTransport = CreateAndStartZmqTransport();
             var receivingPeer = new Peer(sendingTransport.PeerId, receivingTransport.InboundEndPoint);
-            var count = 0;
-            var shouldSendMessages = true;
+
             var senderTask = new Thread(() =>
             {
-                while (shouldSendMessages)
-                    sendingTransport.Send(new FakeCommand(count++).ToTransportMessage(), new[] { receivingPeer });
+                Log($"Send loop started");
+
+                while (state.ShouldSend)
+                {
+                    sendingTransport.Send(new FakeCommand(state.SentMessageCount++).ToTransportMessage(), new[] { receivingPeer });
+                }
+
+                Log($"Send loop terminated, Count: {state.SentMessageCount}");
 
                 sendingTransport.Stop();
+
+                Log($"Sender stopped");
             });
+
             senderTask.Start();
-            Wait.Until(() => receivedMessages.Count != 0, 2.Seconds());
+            Wait.Until(() => state.ReceivedMessageCount != 0, 2.Seconds());
 
-            Console.WriteLine("Stopping the sender for the end\r\n\r\n");
-            shouldSendMessages = false;
+            Log($"Stopping the sender");
+            state.ShouldSend = false;
             senderTask.Join();
-            Console.WriteLine("Stopping the receiver for the end\r\n\r\n");
+
+            Log($"Stopping the receiver");
             receivingTransport.Stop();
-            Console.WriteLine("Receiver stopped\r\n\r\n");
+            Log($"Receiver stopped");
 
-            receivedMessages.Count.ShouldEqual(count);
+            state.ReceivedMessageCount.ShouldEqual(state.SentMessageCount);
+
+            void Log(string text) => Console.WriteLine(DateTime.Now.TimeOfDay + " " + text + Environment.NewLine + Environment.NewLine);
+        }
+
+        private class should_process_all_messages_in_buffer_on_stop_state
+        {
+            public volatile int ReceivedMessageCount;
+            public volatile int SentMessageCount;
+            public volatile bool ShouldSend;
         }
 
         [Test]
@@ -557,7 +600,7 @@ namespace Abc.Zebus.Tests.Transport
             if (peerId == null)
                 peerId = "Abc.Testing." + _transports.Count;
 
-            var transport = transportFactory == null ? new ZmqTransport(configurationMock.Object, new ZmqSocketOptions()) : transportFactory(configurationMock.Object);
+            var transport = transportFactory == null ? new ZmqTransport(configurationMock.Object, new ZmqSocketOptions(), new DefaultZmqOutboundSocketErrorHandler()) : transportFactory(configurationMock.Object);
             transport.SetLogId(_transports.Count);
 
             transport.SocketOptions.SendTimeout = 10.Milliseconds();
@@ -597,7 +640,7 @@ namespace Abc.Zebus.Tests.Transport
             }
 
 
-            public CapturingIsListeningTimeZmqTransport(IZmqTransportConfiguration configuration, Stopwatch stopwatch) : base(configuration, new ZmqSocketOptions())
+            public CapturingIsListeningTimeZmqTransport(IZmqTransportConfiguration configuration, Stopwatch stopwatch) : base(configuration, new ZmqSocketOptions(), new DefaultZmqOutboundSocketErrorHandler())
             {
                 _stopwatch = stopwatch;
                 IsListeningSwitchTimestamp = TimeSpan.MaxValue;

src/Abc.Zebus.Tests/Util/GenZero/BufferTests.cs

@@ -1,119 +0,0 @@
-using System;
-using System.Linq;
-using Abc.Zebus.Testing.Extensions;
-using NUnit.Framework;
-using Buffer = Abc.Zebus.Util.Buffer;
-
-namespace Abc.Zebus.Tests.Util.GenZero
-{
-    [TestFixture]
-    public class BufferTests
-    {
-        [Test]
-        public void should_copy_from_buffer()
-        {
-            var src = new Buffer(new byte[] { 0, 1, 2, 3, 4 });
-            var dest = new Buffer(10);
-
-            dest.CopyFrom(ref src);
-
-            dest.Length.ShouldEqual(src.Length);
-            dest.Data.Take(dest.Length).ShouldEqual(src.Data.Take(src.Length));
-        }
-
-        [Test]
-        public void should_copy_from_byte_array()
-        {
-            var buffer = new Buffer(10);
-            var bytes = new byte[] { 0, 1, 2, 3, 4 };
-
-            buffer.CopyFrom(bytes);
-
-            buffer.Length.ShouldEqual(bytes.Length);
-            buffer.Data.Take(buffer.Length).ShouldEqual(bytes);
-        }
-
-        [Test]
-        public void should_copy_to_buffer()
-        {
-            var src = new Buffer(new byte[] { 0, 1, 2, 3, 4 });
-            var dest = new Buffer(10);
-
-            src.CopyTo(ref dest);
-
-            dest.Length.ShouldEqual(src.Length);
-            dest.Data.Take(dest.Length).ShouldEqual(src.Data.Take(src.Length));
-        }
-
-        [Test]
-        public void should_throw_when_assigned_length_is_greater_than_the_available_byte_count()
-        {
-            var buffer = new Buffer(new byte[] { 0, 1, 2, 3, 4 });
-
-            Assert.DoesNotThrow(() => buffer.Length = 2);
-            Assert.DoesNotThrow(() => buffer.Length = 5);
-            Assert.Throws<ArgumentOutOfRangeException>(() => buffer.Length = 10);
-        }
-
-        [Test]
-        public void should_throw_when_assigned_negative_length()
-        {
-            var buffer = new Buffer(new byte[] { 0, 1, 2, 3, 4 });
-
-            Assert.Throws<ArgumentOutOfRangeException>(() => buffer.Length = -1);
-        }
-
-        [Test]
-        public void should_only_take_length_bytes_into_account_for_hashcode()
-        {
-            var b1 = new Buffer(new byte[] { 0, 1, 2, 3, 4 });
-            var b2 = new Buffer(new byte[] { 0, 1 });
-
-            b1.Length = 2;
-
-            b1.GetHashCode().ShouldEqual(b2.GetHashCode());
-        }
-
-        [Test]
-        public void should_throw_when_destination_buffer_is_too_small()
-        {
-            var src = new Buffer(new byte[] { 0, 1, 2, 3, 4 });
-            var dest = new Buffer(2);
-
-            Assert.Throws<ArgumentException>(() => src.CopyTo(ref dest));
-        }
-
-        [Test]
-        public void should_to_byte_array_return_correct_bytes()
-        {
-            var byteArray = new byte[] { 0, 1, 2, 3, 4 };
-            var buffer = new Buffer(byteArray);
-
-            var otherByteArray = buffer.ToByteArray();
-
-            otherByteArray.ShouldEqual(byteArray);
-        }
-
-        [Test]
-        public void should_two_equal_buffers_have_the_same_hashcode()
-        {
-            var b1 = new Buffer(new byte[] { 0, 1, 2, 3, 4 });
-            var b2 = new Buffer(new byte[] { 0, 1, 2, 3, 4 });
-
-            b1.GetHashCode().ShouldEqual(b2.GetHashCode());
-        }
-
-
-        [Test]
-        public void should_copy_from_byte_array_with_offset_and_length()
-        {
-            var buffer = new Buffer(10);
-            var bytes = new byte[] { 0, 1, 2, 3, 4 };
-
-            buffer.CopyFrom(bytes, 2, 1);
-
-            buffer.Length.ShouldEqual(1);
-            buffer.Data.Take(buffer.Length).ShouldEqual(new byte[] { 2 });
-        }
-    }
-}

src/Abc.Zebus.Tests/app.config

@@ -1,14 +1,6 @@
 <?xml version="1.0" encoding="utf-8"?>
 <configuration>
-  <runtime>
-    <assemblyBinding xmlns="urn:schemas-microsoft-com:asm.v1">
-      <dependentAssembly>
-        <assemblyIdentity name="protobuf-net" publicKeyToken="257b51d87d2e4d67" culture="neutral" />
-        <bindingRedirect oldVersion="0.0.0.0-2.0.0.668" newVersion="2.0.0.668" />
-      </dependentAssembly>
-    </assemblyBinding>
-  </runtime>
   <startup>
     <supportedRuntime version="v4.0" sku=".NETFramework,Version=v4.6" />
   </startup>
-</configuration>
+</configuration>

src/Abc.Zebus.Tests/packages.config

@@ -1,10 +0,0 @@
-<?xml version="1.0" encoding="utf-8"?>
-<packages>
-  <package id="CompareNETObjects" version="3.03.0.0" targetFramework="net45" requireReinstallation="true" />
-  <package id="log4net" version="2.0.3" targetFramework="net45" />
-  <package id="Moq" version="4.2.1507.0118" targetFramework="net45" />
-  <package id="Newtonsoft.Json" version="7.0.1" targetFramework="net45" />
-  <package id="NUnit" version="2.6.4" targetFramework="net45" />
-  <package id="protobuf-net" version="2.0.0.668" targetFramework="net45" />
-  <package id="structuremap" version="3.1.6.186" targetFramework="net45" />
-</packages>

src/Abc.Zebus.sln

@@ -1,44 +1,26 @@
 
 Microsoft Visual Studio Solution File, Format Version 12.00
-# Visual Studio 14
-VisualStudioVersion = 14.0.24720.0
+# Visual Studio 15
+VisualStudioVersion = 15.0.27428.2037
 MinimumVisualStudioVersion = 10.0.40219.1
-Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Abc.Zebus", "Abc.Zebus\Abc.Zebus.csproj", "{1F4C6307-6113-40D5-BF42-4B6BF5DF13B2}"
+Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "Abc.Zebus", "Abc.Zebus\Abc.Zebus.csproj", "{1F4C6307-6113-40D5-BF42-4B6BF5DF13B2}"
 EndProject
-Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Abc.Zebus.Tests", "Abc.Zebus.Tests\Abc.Zebus.Tests.csproj", "{7BF5AC28-91DB-40EF-8CB7-024E518FA799}"
+Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "Abc.Zebus.Tests", "Abc.Zebus.Tests\Abc.Zebus.Tests.csproj", "{7BF5AC28-91DB-40EF-8CB7-024E518FA799}"
 EndProject
-Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Abc.Zebus.Testing", "Abc.Zebus.Testing\Abc.Zebus.Testing.csproj", "{4CE123DF-8021-411C-929B-53E5C5FC1E04}"
+Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "Abc.Zebus.Testing", "Abc.Zebus.Testing\Abc.Zebus.Testing.csproj", "{4CE123DF-8021-411C-929B-53E5C5FC1E04}"
 EndProject
 Project("{2150E333-8FDC-42A3-9474-1A3956D46DE8}") = "@ Solution Items", "@ Solution Items", "{A85342C3-7BA4-4A6B-B690-B1D0ACE14C76}"
 	ProjectSection(SolutionItems) = preProject
+		..\build\Abc.Zebus.nuspec = ..\build\Abc.Zebus.nuspec
+		..\build\Abc.Zebus.Testing.nuspec = ..\build\Abc.Zebus.Testing.nuspec
+		..\build\build.cake = ..\build\build.cake
+		..\RELEASE_NOTES.md = ..\RELEASE_NOTES.md
 		SharedAssemblyInfo.cs = SharedAssemblyInfo.cs
 		SharedVersionInfo.cs = SharedVersionInfo.cs
+		..\version.yml = ..\version.yml
 	EndProjectSection
 EndProject
-Project("{2150E333-8FDC-42A3-9474-1A3956D46DE8}") = "build", "build", "{7933BD78-DFD0-4F8E-BFE5-1660B14E7533}"
-	ProjectSection(SolutionItems) = preProject
-		..\build\Zebus.build = ..\build\Zebus.build
-	EndProjectSection
-EndProject
-Project("{2150E333-8FDC-42A3-9474-1A3956D46DE8}") = "nuget", "nuget", "{041FD6B0-49C1-4C7B-B6CC-6BA106FE734B}"
-	ProjectSection(SolutionItems) = preProject
-		..\build\nuget\BuildNuget.ps1 = ..\build\nuget\BuildNuget.ps1
-	EndProjectSection
-EndProject
-Project("{2150E333-8FDC-42A3-9474-1A3956D46DE8}") = "nuspecs", "nuspecs", "{CF60551C-BB7F-4053-93BE-3E2B53222DF7}"
-	ProjectSection(SolutionItems) = preProject
-		..\build\nuget\nuspecs\Abc.Zebus.nuspec = ..\build\nuget\nuspecs\Abc.Zebus.nuspec
-		..\build\nuget\nuspecs\Abc.Zebus.Testing.nuspec = ..\build\nuget\nuspecs\Abc.Zebus.Testing.nuspec
-	EndProjectSection
-EndProject
-Project("{2150E333-8FDC-42A3-9474-1A3956D46DE8}") = "tools", "tools", "{7218C17B-72FC-4BED-AD81-F4BC6196D3AA}"
-EndProject
-Project("{2150E333-8FDC-42A3-9474-1A3956D46DE8}") = "Zebus.Directory", "Zebus.Directory", "{0232BC2F-5907-4957-8B7D-0DD1CCC6428F}"
-	ProjectSection(SolutionItems) = preProject
-		..\build\nuget\tools\Zebus.Directory\readme.txt = ..\build\nuget\tools\Zebus.Directory\readme.txt
-	EndProjectSection
-EndProject
-Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Abc.Zebus.Tests.TestExe", "Abc.Zebus.Tests.TestExe\Abc.Zebus.Tests.TestExe.csproj", "{652017D8-EB34-4A79-B14F-DB0477E43915}"
+Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "Abc.Zebus.Tests.TestExe", "Abc.Zebus.Tests.TestExe\Abc.Zebus.Tests.TestExe.csproj", "{652017D8-EB34-4A79-B14F-DB0477E43915}"
 EndProject
 Global
 	GlobalSection(SolutionConfigurationPlatforms) = preSolution
@@ -107,11 +89,7 @@ Global
 	GlobalSection(SolutionProperties) = preSolution
 		HideSolutionNode = FALSE
 	EndGlobalSection
-	GlobalSection(NestedProjects) = preSolution
-		{7933BD78-DFD0-4F8E-BFE5-1660B14E7533} = {A85342C3-7BA4-4A6B-B690-B1D0ACE14C76}
-		{041FD6B0-49C1-4C7B-B6CC-6BA106FE734B} = {7933BD78-DFD0-4F8E-BFE5-1660B14E7533}
-		{CF60551C-BB7F-4053-93BE-3E2B53222DF7} = {041FD6B0-49C1-4C7B-B6CC-6BA106FE734B}
-		{7218C17B-72FC-4BED-AD81-F4BC6196D3AA} = {041FD6B0-49C1-4C7B-B6CC-6BA106FE734B}
-		{0232BC2F-5907-4957-8B7D-0DD1CCC6428F} = {7218C17B-72FC-4BED-AD81-F4BC6196D3AA}
+	GlobalSection(ExtensibilityGlobals) = postSolution
+		SolutionGuid = {A5A62CA9-98D0-4C95-BC83-F0F57B76CC44}
 	EndGlobalSection
 EndGlobal

src/Abc.Zebus/Abc.Zebus.csproj

@@ -1,71 +1,41 @@
-<?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')" />
+<Project Sdk="Microsoft.NET.Sdk">
   <PropertyGroup>
-    <Configuration Condition=" '$(Configuration)' == '' ">Debug</Configuration>
-    <Platform Condition=" '$(Platform)' == '' ">AnyCPU</Platform>
-    <ProjectGuid>{1F4C6307-6113-40D5-BF42-4B6BF5DF13B2}</ProjectGuid>
-    <OutputType>Library</OutputType>
-    <AppDesignerFolder>Properties</AppDesignerFolder>
-    <RootNamespace>Abc.Zebus</RootNamespace>
-    <AssemblyName>Abc.Zebus</AssemblyName>
-    <TargetFrameworkVersion>v4.5</TargetFrameworkVersion>
-    <FileAlignment>512</FileAlignment>
-    <TargetFrameworkProfile />
+    <TargetFramework>net46</TargetFramework>
+    <GenerateAssemblyInfo>false</GenerateAssemblyInfo>
+    <ErrorReport>prompt</ErrorReport>
+    <WarningLevel>4</WarningLevel>
+    <TreatWarningsAsErrors>true</TreatWarningsAsErrors>
+    <AllowUnsafeBlocks>true</AllowUnsafeBlocks>
   </PropertyGroup>
+
   <PropertyGroup Condition=" '$(Configuration)|$(Platform)' == 'Debug|AnyCPU' ">
     <DebugSymbols>true</DebugSymbols>
     <DebugType>full</DebugType>
     <Optimize>false</Optimize>
     <OutputPath>bin\Debug\</OutputPath>
     <DefineConstants>DEBUG;TRACE</DefineConstants>
-    <ErrorReport>prompt</ErrorReport>
-    <WarningLevel>4</WarningLevel>
-    <TreatWarningsAsErrors>true</TreatWarningsAsErrors>
-    <UseVSHostingProcess>false</UseVSHostingProcess>
-    <AllowUnsafeBlocks>true</AllowUnsafeBlocks>
   </PropertyGroup>
   <PropertyGroup Condition=" '$(Configuration)|$(Platform)' == 'Release|AnyCPU' ">
     <DebugType>pdbonly</DebugType>
     <Optimize>true</Optimize>
     <OutputPath>bin\Release\</OutputPath>
     <DefineConstants>TRACE</DefineConstants>
-    <ErrorReport>prompt</ErrorReport>
-    <WarningLevel>4</WarningLevel>
-    <TreatWarningsAsErrors>true</TreatWarningsAsErrors>
-    <AllowUnsafeBlocks>true</AllowUnsafeBlocks>
   </PropertyGroup>
+
+  <ItemGroup>
+    <PackageReference Include="log4net" Version="2.0.8" />
+    <PackageReference Include="Newtonsoft.Json" Version="11.0.2" />
+    <PackageReference Include="protobuf-net" Version="2.3.7" />
+    <PackageReference Include="structuremap" Version="3.1.6.186" />
+  </ItemGroup>
+
   <ItemGroup>
     <Reference Include="clrzmq">
       <HintPath>..\..\lib\zmq\clrzmq.dll</HintPath>
     </Reference>
-    <Reference Include="log4net">
-      <HintPath>..\..\lib\packages\log4net.2.0.3\lib\net40-full\log4net.dll</HintPath>
-      <Private>True</Private>
-    </Reference>
     <Reference Include="Microsoft.CSharp" />
-    <Reference Include="Newtonsoft.Json, Version=7.0.0.0, Culture=neutral, PublicKeyToken=30ad4fe6b2a6aeed, processorArchitecture=MSIL">
-      <SpecificVersion>False</SpecificVersion>
-      <HintPath>..\..\lib\packages\Newtonsoft.Json.7.0.1\lib\net45\Newtonsoft.Json.dll</HintPath>
-      <Private>True</Private>
-    </Reference>
-    <Reference Include="protobuf-net">
-      <HintPath>..\..\lib\packages\protobuf-net.2.0.0.668\lib\net40\protobuf-net.dll</HintPath>
-      <Private>True</Private>
-    </Reference>
-    <Reference Include="StructureMap">
-      <HintPath>..\..\lib\packages\structuremap.3.1.6.186\lib\net40\StructureMap.dll</HintPath>
-      <Private>True</Private>
-    </Reference>
-    <Reference Include="StructureMap.Net4">
-      <HintPath>..\..\lib\packages\structuremap.3.1.6.186\lib\net40\StructureMap.Net4.dll</HintPath>
-      <Private>True</Private>
-    </Reference>
-    <Reference Include="System" />
-    <Reference Include="System.Core" />
-    <Reference Include="System.Runtime.Serialization" />
-    <Reference Include="System.Xml" />
   </ItemGroup>
+
   <ItemGroup>
     <Compile Include="..\SharedAssemblyInfo.cs">
       <Link>Properties\SharedAssemblyInfo.cs</Link>
@@ -73,220 +43,12 @@
     <Compile Include="..\SharedVersionInfo.cs">
       <Link>Properties\SharedVersionInfo.cs</Link>
     </Compile>
-    <Compile Include="Core\Bus.cs" />
-    <Compile Include="CommandResult.cs" />
-    <Compile Include="Core\BusFactory.cs" />
-    <Compile Include="Core\DefaultMessageSendingStrategy.cs" />
-    <Compile Include="Core\DefaultStoppingStrategy.cs" />
-    <Compile Include="Core\IMessageSendingStrategy.cs" />
-    <Compile Include="Directory\MarkPeerAsNotRespondingCommand.cs" />
-    <Compile Include="Directory\MarkPeerAsRespondingCommand.cs" />
-    <Compile Include="Dispatch\DispatchQueue.cs" />
-    <Compile Include="Dispatch\DynamicMessageHandlerInvoker.cs" />
-    <Compile Include="Core\IStoppingStrategy.cs" />
-    <Compile Include="Core\RoundRobinPeerSelector.cs" />
-    <Compile Include="Directory\DecommissionPeerCommand.cs" />
-    <Compile Include="Directory\DirectoryErrorCodes.cs" />
-    <Compile Include="Directory\IPeerDirectory.cs" />
-    <Compile Include="Directory\MessageBinding.cs" />
-    <Compile Include="Directory\PeerDecommissioned.cs" />
-    <Compile Include="Directory\PeerDirectoryClient.PeerEntry.cs" />
-    <Compile Include="Directory\PeerNotResponding.cs" />
-    <Compile Include="Directory\PeerResponding.cs" />
-    <Compile Include="Directory\PeerSubscriptionsForTypesUpdated.cs" />
-    <Compile Include="Directory\PeerSubscriptionsUpdated.cs" />
-    <Compile Include="Directory\PeerSubscriptionTree.cs" />
-    <Compile Include="Directory\PingPeerCommand.cs" />
-    <Compile Include="Directory\PeerUpdateAction.cs" />
-    <Compile Include="Directory\SubscriptionsForType.cs" />
-    <Compile Include="Directory\UpdatePeerSubscriptionsCommand.cs" />
-    <Compile Include="Directory\UpdatePeerSubscriptionsForTypesCommand.cs" />
-    <Compile Include="Dispatch\AsyncMessageHandlerInvoker.cs" />
-    <Compile Include="Dispatch\DispatchQueueFactory.cs" />
-    <Compile Include="Dispatch\IDispatchQueueFactory.cs" />
-    <Compile Include="Dispatch\IMessageDispatchFactory.cs" />
-    <Compile Include="Dispatch\IMessageHandlerInvocation.cs" />
-    <Compile Include="BusExtensions.cs" />
-    <Compile Include="Dispatch\MessageHandlerInvokerMode.cs" />
-    <Compile Include="ErrorStatus.cs" />
-    <Compile Include="Dispatch\BatchedMessageHandlerInvoker.cs" />
-    <Compile Include="Hosting\HostInitializer.cs" />
-    <Compile Include="Hosting\HostInitializerHelper.cs" />
-    <Compile Include="Hosting\PeriodicActionHostInitializer.cs" />
-    <Compile Include="IBatchedMessageHandler.cs" />
-    <Compile Include="ICommandHandler`1.cs" />
-    <Compile Include="IMessageHandler`1.cs" />
-    <Compile Include="Lotus\ReplayMessageCommand.cs" />
-    <Compile Include="Lotus\ReplayMessageHandler.cs" />
-    <Compile Include="MessageTypeDescriptor.cs" />
-    <Compile Include="Monitoring\IProvideQueueLength.cs" />
-    <Compile Include="Scan\BatchedMessageHandlerInvokerLoader.cs" />
-    <Compile Include="Serialization\Protobuf\ByteArray.cs" />
-    <Compile Include="Serialization\Protobuf\CodedInputStream.cs" />
-    <Compile Include="Serialization\Protobuf\CodedOutputStream.ComputeSize.cs" />
-    <Compile Include="Serialization\Protobuf\CodedOutputStream.cs" />
-    <Compile Include="Serialization\Protobuf\InvalidProtocolBufferException.cs" />
-    <Compile Include="Routing\BindingKeyPredicateBuilder.cs" />
-    <Compile Include="Routing\BindingKeyToken.cs" />
-    <Compile Include="Routing\IBindingKeyPredicateBuilder.cs" />
-    <Compile Include="Serialization\PeerIdConverter.cs" />
-    <Compile Include="Persistence\PersistentTransport.cs" />
-    <Compile Include="Dispatch\Pipes\AfterInvokeArgs.cs" />
-    <Compile Include="Dispatch\Pipes\BeforeInvokeArgs.cs" />
-    <Compile Include="Scan\DispatchQueueNameScanner.cs" />
-    <Compile Include="Dispatch\DispatchQueueNameAttribute.cs" />
-    <Compile Include="Dispatch\IMessageHandlerInvoker.cs" />
-    <Compile Include="Dispatch\IProvideDispatchQueueNameForCurrentNamespace.cs" />
-    <Compile Include="Core\MessageContextAwareBus.cs" />
-    <Compile Include="Dispatch\MessageDispatch.cs" />
-    <Compile Include="Dispatch\MessageHandlerInvoker.cs" />
-    <Compile Include="IBusConfiguration.cs" />
-    <Compile Include="Directory\PeerDescriptor.cs" />
-    <Compile Include="Directory\PeerDirectoryClient.cs" />
-    <Compile Include="Directory\PeerStarted.cs" />
-    <Compile Include="Directory\PeerStopped.cs" />
-    <Compile Include="Directory\RegisterPeerCommand.cs" />
-    <Compile Include="Directory\RegisterPeerResponse.cs" />
-    <Compile Include="Directory\UnregisterPeerCommand.cs" />
-    <Compile Include="Dispatch\DispatchResult.cs" />
-    <Compile Include="Dispatch\IMessageDispatcher.cs" />
-    <Compile Include="Dispatch\MessageDispatcher.cs" />
-    <Compile Include="DomainException.cs" />
-    <Compile Include="IBus.cs" />
-    <Compile Include="IMessageContextAware.cs" />
-    <Compile Include="Initialization\ZebusRegistry.cs" />
-    <Compile Include="Dispatch\SyncMessageHandlerInvoker.cs" />
-    <Compile Include="Dispatch\LocalDispatch.cs" />
-    <Compile Include="Lotus\CustomProcessingFailed.cs" />
-    <Compile Include="Lotus\MessageProcessingFailed.cs" />
-    <Compile Include="MessageContext.cs" />
-    <Compile Include="Core\BusMessageLogger.cs" />
-    <Compile Include="MessageId.cs" />
-    <Compile Include="MessageTypeId.cs" />
-    <Compile Include="MessageUtil.cs" />
-    <Compile Include="PeerId.cs" />
-    <Compile Include="Persistence\IReplayEvent.cs" />
-    <Compile Include="Persistence\StartMessageReplayCommand.cs" />
-    <Compile Include="Persistence\SafetyPhaseEnded.cs" />
-    <Compile Include="Persistence\ReplayPhaseEnded.cs" />
-    <Compile Include="Persistence\PersistenceUnreachableException.cs" />
-    <Compile Include="Dispatch\Pipes\AttributePipeSource.cs" />
-    <Compile Include="Dispatch\Pipes\IPipe.cs" />
-    <Compile Include="Dispatch\Pipes\IPipeManager.cs" />
-    <Compile Include="Dispatch\Pipes\IPipeSource.cs" />
-    <Compile Include="Dispatch\Pipes\PipeAttribute.cs" />
-    <Compile Include="Dispatch\Pipes\PipeInvocation.cs" />
-    <Compile Include="Dispatch\Pipes\PipeManager.cs" />
-    <Compile Include="Dispatch\Pipes\PipeSource.cs" />
-    <Compile Include="Persistence\MessageHandled.cs" />
-    <Compile Include="Persistence\MessageReplayed.cs" />
-    <Compile Include="Persistence\PersistMessageCommand.cs" />
-    <Compile Include="Scan\IMessageHandlerInvokerLoader.cs" />
-    <Compile Include="Scan\AsyncMessageHandlerInvokerLoader.cs" />
-    <Compile Include="Scan\SyncMessageHandlerInvokerLoader.cs" />
-    <Compile Include="Scan\MessageHandlerInvokerLoader.cs" />
-    <Compile Include="Scan\TypeSource.cs" />
-    <Compile Include="Serialization\ProtocolBufferSerializationException.cs" />
-    <Compile Include="EventSourcing\AggregateRootIdAttribute.cs" />
-    <Compile Include="EventSourcing\DomainEventSourcing.cs" />
-    <Compile Include="IAsyncMessageHandler.cs" />
-    <Compile Include="ICommand.cs" />
-    <Compile Include="ICommandHandler.cs" />
-    <Compile Include="EventSourcing\IDomainEvent.cs" />
-    <Compile Include="IEvent.cs" />
-    <Compile Include="ICustomInvokerMessageHandler.cs" />
-    <Compile Include="EventSourcing\IExtraTypesProvider.cs" />
-    <Compile Include="IMessage.cs" />
-    <Compile Include="IMessageHandler.cs" />
-    <Compile Include="InfrastructureAttribute.cs" />
-    <Compile Include="MessageTypeIdAttribute.cs" />
-    <Compile Include="NoScanAttribute.cs" />
-    <Compile Include="Routing\Routable.cs" />
-    <Compile Include="Routing\RoutingPositionAttribute.cs" />
-    <Compile Include="EventSourcing\SerializationIdAttribute.cs" />
-    <Compile Include="Serialization\Protobuf\WireFormat.cs" />
-    <Compile Include="SubscriptionMode.cs" />
-    <Compile Include="SubscriptionModeAttribute.cs" />
-    <Compile Include="SubscriptionRequest.cs" />
-    <Compile Include="TransientAttribute.cs" />
-    <Compile Include="Subscription.cs" />
-    <Compile Include="Serialization\IMessageSerializer.cs" />
-    <Compile Include="Serialization\MessageSerializer.cs" />
-    <Compile Include="Core\MessageExecutionCompleted.cs" />
-    <Compile Include="Properties\AssemblyInfo.cs" />
-    <Compile Include="Serialization\MessageSerializerExtensions.cs" />
-    <Compile Include="Routing\BindingKey.cs" />
-    <Compile Include="Timeout\RequestTimeoutCommand.cs" />
-    <Compile Include="Timeout\Timeout.cs" />
-    <Compile Include="Timeout\TimeoutCommand.cs" />
-    <Compile Include="Persistence\IPersistentTransport.cs" />
-    <Compile Include="Transport\ITransport.cs" />
-    <Compile Include="Peer.cs" />
-    <Compile Include="Transport\IZmqTransportConfiguration.cs" />
-    <Compile Include="Transport\SendContext.cs" />
-    <Compile Include="Transport\TransportMessageReader.cs" />
-    <Compile Include="Transport\TransportMessageWriter.cs" />
-    <Compile Include="Transport\ZmqInboundSocket.cs" />
-    <Compile Include="Transport\ZmqOutboundSocket.cs" />
-    <Compile Include="Transport\ZmqEndPoint.cs" />
-    <Compile Include="Transport\OriginatorInfo.cs" />
-    <Compile Include="Transport\ZmqTransport.cs" />
-    <Compile Include="Transport\TransportMessage.cs" />
-    <Compile Include="Transport\ZmqSocketOptions.cs" />
-    <Compile Include="Transport\ZmqUtil.cs" />
-    <Compile Include="Serialization\Serializer.cs" />
-    <Compile Include="Util\Annotations\JetBrains.Annotations.cs" />
-    <Compile Include="Util\ArrayUtil.cs" />
-    <Compile Include="Util\BackgroundThread.cs" />
-    <Compile Include="Util\Collections\ConcurrentSet.cs" />
-    <Compile Include="Util\Collections\FlushableBlockingCollection.cs" />
-    <Compile Include="Util\CultureScope.cs" />
-    <Compile Include="Util\DisposableAction.cs" />
-    <Compile Include="Util\Extensions\ExtendDictionary.cs" />
-    <Compile Include="Util\Extensions\ExtendICollection.cs" />
-    <Compile Include="Util\Extensions\ExtendEnum.cs" />
-    <Compile Include="Util\Extensions\ExtendIEnumerable.cs" />
-    <Compile Include="Util\Extensions\ExtendString.cs" />
-    <Compile Include="Util\Extensions\ExtendTask.cs" />
-    <Compile Include="Util\Extensions\ExtendType.cs" />
-    <Compile Include="Util\TimeSpanCreation.cs" />
-    <Compile Include="Util\Buffer.cs" />
-    <Compile Include="Util\PathUtil.cs" />
-    <Compile Include="Util\SystemDateTime.cs" />
-    <Compile Include="Util\TcpUtil.cs" />
-    <Compile Include="Util\TaskUtil.cs" />
-    <Compile Include="Util\TypeUtil.cs" />
-    <Compile Include="Util\UniqueTimestampProvider.cs" />
-  </ItemGroup>
-  <ItemGroup>
-    <Compile Include="Persistence\PersistentTransport.Phases.cs" />
-  </ItemGroup>
-  <ItemGroup>
+
+    <None Include="Transport\libzmq-4.0.5-v140-mt" />
     <EmbeddedResource Include="Transport\libzmq-x64-0.0.0.0.dll" />
     <EmbeddedResource Include="Transport\libzmq-x86-0.0.0.0.dll" />
+    <EmbeddedResource Include="Transport\libzmq-x64-0.0.0.0.pdb" />
+    <EmbeddedResource Include="Transport\libzmq-x86-0.0.0.0.pdb" />
   </ItemGroup>
-  <ItemGroup>
-    <None Include="packages.config">
-      <SubType>Designer</SubType>
-    </None>
-    <None Include="Transport\libzmq-4.0.5-v140-mt" />
-  </ItemGroup>
-  <ItemGroup>
-    <None Include="Transport\libzmq-x64-0.0.0.0.pdb" />
-  </ItemGroup>
-  <ItemGroup>
-    <None Include="Transport\libzmq-x86-0.0.0.0.pdb" />
-  </ItemGroup>
-  <Import Project="$(MSBuildToolsPath)\Microsoft.CSharp.targets" />
-  <PropertyGroup>
-    <PreBuildEvent>
-    </PreBuildEvent>
-  </PropertyGroup>
-  <!-- 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/Abc.Zebus/Core/Bus.cs

@@ -21,11 +21,12 @@ namespace Abc.Zebus.Core
 {
     public class Bus : IBus, IMessageDispatchFactory
     {
+        private static readonly BusMessageLogger _messageLogger = new BusMessageLogger(typeof(Bus));
+        private static readonly ILog _logger = LogManager.GetLogger(typeof(Bus));
+
         private readonly ConcurrentDictionary<MessageId, TaskCompletionSource<CommandResult>> _messageIdToTaskCompletionSources = new ConcurrentDictionary<MessageId, TaskCompletionSource<CommandResult>>();
         private readonly UniqueTimestampProvider _deserializationFailureTimestampProvider = new UniqueTimestampProvider();
         private readonly Dictionary<Subscription, int> _subscriptions = new Dictionary<Subscription, int>();
-        private readonly BusMessageLogger _messageLogger = new BusMessageLogger(typeof(Bus));
-        private readonly ILog _logger = LogManager.GetLogger(typeof(Bus));
         private readonly ITransport _transport;
         private readonly IPeerDirectory _directory;
         private readonly IMessageSerializer _serializer;

src/Abc.Zebus/Core/BusMessageLogger.cs

@@ -2,6 +2,7 @@
 using System.Collections.Concurrent;
 using System.Collections.Generic;
 using System.Reflection;
+using Abc.Zebus.Scan;
 using Abc.Zebus.Util.Annotations;
 using Abc.Zebus.Util.Extensions;
 using log4net;
@@ -12,16 +13,14 @@ namespace Abc.Zebus.Core
     public class BusMessageLogger
     {
         private static readonly ConcurrentDictionary<Type, MessageTypeLogInfo> _logInfos = new ConcurrentDictionary<Type, MessageTypeLogInfo>();
-        private static readonly Func<Type, MessageTypeLogInfo> _logInfoFactory;
+        private static readonly Func<Type, MessageTypeLogInfo> _logInfoFactory = CreateLogger;
         private readonly Type _loggerType;
         private readonly ILog _logger;
+        private bool _logDebugEnabled;
+        private bool _logInfoEnabled;
 
-        static BusMessageLogger()
-        {
-            _logInfoFactory = CreateLogger;
-        }
-
-        public BusMessageLogger(Type loggerType) : this(loggerType, loggerType.FullName)
+        public BusMessageLogger(Type loggerType)
+            : this(loggerType, loggerType.FullName)
         {
         }
 
@@ -29,28 +28,44 @@ namespace Abc.Zebus.Core
         {
             _loggerType = loggerType;
             _logger = LogManager.GetLogger(loggerFullName);
+
+            // Instances of BusMessageLogger are static, no need to unsubscribe from these events
+            _logger.Logger.Repository.ConfigurationChanged += (sender, args) => UpdateLogConfig();
+            _logger.Logger.Repository.ConfigurationReset += (sender, args) => UpdateLogConfig();
+            UpdateLogConfig();
+
+            void UpdateLogConfig()
+            {
+                _logDebugEnabled = _logger.IsDebugEnabled;
+                _logInfoEnabled = _logger.IsInfoEnabled;
+            }
         }
 
         public bool IsInfoEnabled(IMessage message)
-        {
-            var logInfo = GetLogInfo(message);
-            return logInfo.Logger.IsInfoEnabled;
-        }
+            => _logInfoEnabled && GetLogInfo(message).Logger.IsInfoEnabled;
 
         [StringFormatMethod("format")]
-        public void InfoFormat(string format, IMessage message, MessageId? messageId = null, long messageSize = 0, PeerId peerId = default(PeerId))
+        public void InfoFormat(string format, IMessage message, string dispatchQueueName = null, MessageId? messageId = null, long messageSize = 0, PeerId peerId = default(PeerId))
         {
+            if (!_logInfoEnabled)
+                return;
+
             var logInfo = GetLogInfo(message);
             if (!logInfo.Logger.IsInfoEnabled)
                 return;
 
             var messageText = logInfo.GetMessageText(message);
-            _logger.InfoFormat(format, messageText, messageId, messageSize, peerId);
+            dispatchQueueName = string.IsNullOrEmpty(dispatchQueueName) || dispatchQueueName == DispatchQueueNameScanner.DefaultQueueName ? string.Empty : $" [{dispatchQueueName}]";
+
+            _logger.InfoFormat(format, messageText, dispatchQueueName, messageId, messageSize, peerId);
         }
 
         [StringFormatMethod("format")]
         public void DebugFormat(string format, IMessage message, MessageId? messageId = null, long messageSize = 0, PeerId peerId = default(PeerId))
         {
+            if (!_logDebugEnabled)
+                return;
+
             var logInfo = GetLogInfo(message);
             if (!logInfo.Logger.IsDebugEnabled)
                 return;
@@ -62,15 +77,18 @@ namespace Abc.Zebus.Core
         [StringFormatMethod("format")]
         public void InfoFormat(string format, IMessage message, MessageId messageId, long messageSize, IList<Peer> peers, Level logLevel = null)
         {
-            if (peers.Count == 0)
-            {
-                InfoFormat(format, message, messageId, messageSize);
+            if (!_logInfoEnabled)
                 return;
-            }
-            if (peers.Count == 1)
+
+            switch (peers.Count)
             {
-                InfoFormat(format, message, messageId, messageSize, peerId: peers[0].Id);
-                return;
+                case 0:
+                    InfoFormat(format, message, messageId: messageId, messageSize: messageSize);
+                    return;
+
+                case 1:
+                    InfoFormat(format, message, messageId: messageId, messageSize:messageSize, peerId:peers[0].Id);
+                    return;
             }
 
             var logInfo = GetLogInfo(message);
@@ -86,12 +104,11 @@ namespace Abc.Zebus.Core
             _logger.Logger.Log(_loggerType, logLevel ?? Level.Info, string.Format(format, messageText, messageId, messageSize, peerIdText), null);
         }
 
-        public static string ToString(IMessage message) => GetLogInfo(message).GetMessageText(message);
+        public static string ToString(IMessage message)
+            => GetLogInfo(message).GetMessageText(message);
 
         private static MessageTypeLogInfo GetLogInfo(IMessage message)
-        {
-            return _logInfos.GetOrAdd(message.GetType(), _logInfoFactory);
-        }
+            => _logInfos.GetOrAdd(message.GetType(), _logInfoFactory);
 
         private static MessageTypeLogInfo CreateLogger(Type messageType)
         {
@@ -126,4 +143,4 @@ namespace Abc.Zebus.Core
             }
         }
     }
-}
+}

src/Abc.Zebus/Directory/MessageBinding.cs

@@ -15,26 +15,17 @@ namespace Abc.Zebus.Directory
         }
 
         public static MessageBinding FromMessage(IMessage message)
-        {
-            return new MessageBinding(message.TypeId(), BindingKey.Create(message));
-        }
+            => new MessageBinding(message.TypeId(), BindingKey.Create(message));
 
-        public static MessageBinding Default<T>() where T : IMessage
-        {
-            return new MessageBinding(MessageUtil.TypeId<T>(), BindingKey.Empty);
-        }
+        public static MessageBinding Default<T>()
+            where T : IMessage
+            => new MessageBinding(MessageUtil.TypeId<T>(), BindingKey.Empty);
 
         public bool Equals(MessageBinding other)
-        {
-            return Equals(MessageTypeId, other.MessageTypeId) && RoutingKey.Equals(other.RoutingKey);
-        }
+            => MessageTypeId == other.MessageTypeId && RoutingKey.Equals(other.RoutingKey);
 
         public override bool Equals(object obj)
-        {
-            if (ReferenceEquals(null, obj))
-                return false;
-            return obj is MessageBinding && Equals((MessageBinding)obj);
-        }
+            => obj is MessageBinding binding && Equals(binding);
 
         public override int GetHashCode()
         {

src/Abc.Zebus/Directory/PeerDirectoryClient.cs

@@ -288,8 +288,7 @@ namespace Abc.Zebus.Directory
             if (EnqueueIfRegistering(message))
                 return;
 
-            PeerEntry removedPeer;
-            if (!_peers.TryRemove(message.PeerId, out removedPeer))
+            if (!_peers.TryRemove(message.PeerId, out var removedPeer))
                 return;
 
             removedPeer.RemoveSubscriptions();

src/Abc.Zebus/Directory/SubscriptionsForType.cs

@@ -27,22 +27,19 @@ namespace Abc.Zebus.Directory
         {
         }
 
-        public static SubscriptionsForType Create<TMessage>(params BindingKey[] bindingKeys) where TMessage : IMessage
-        {
-            return new SubscriptionsForType(MessageUtil.TypeId<TMessage>(), bindingKeys);
-        }
+        public static SubscriptionsForType Create<TMessage>(params BindingKey[] bindingKeys)
+            where TMessage : IMessage
+            => new SubscriptionsForType(MessageUtil.TypeId<TMessage>(), bindingKeys);
 
         public static Dictionary<MessageTypeId, SubscriptionsForType> CreateDictionary(IEnumerable<Subscription> subscriptions)
-        {
-            return subscriptions.GroupBy(sub => sub.MessageTypeId)
-                                .ToDictionary(grp => grp.Key, grp => new SubscriptionsForType(grp.Key, grp.Select(sub => sub.BindingKey).ToArray()));
-        }
+            => subscriptions.GroupBy(sub => sub.MessageTypeId)
+                            .ToDictionary(grp => grp.Key, grp => new SubscriptionsForType(grp.Key, grp.Select(sub => sub.BindingKey).ToArray()));
 
         public Subscription[] ToSubscriptions()
             => BindingKeys?.Select(bindingKey => new Subscription(MessageTypeId, bindingKey)).ToArray() ?? new Subscription[0];
 
         public bool Equals(SubscriptionsForType other)
-            => other != null && Equals(MessageTypeId, other.MessageTypeId) && BindingKeys.SequenceEqual(other.BindingKeys);
+            => other != null && MessageTypeId == other.MessageTypeId && BindingKeys.SequenceEqual(other.BindingKeys);
 
         public override bool Equals(object obj)
         {

src/Abc.Zebus/Dispatch/DispatchQueue.cs

@@ -14,7 +14,8 @@ namespace Abc.Zebus.Dispatch
         [ThreadStatic]
         private static string _currentDispatchQueueName;
 
-        private readonly ILog _logger = LogManager.GetLogger(typeof(DispatchQueue));
+        private static readonly ILog _logger = LogManager.GetLogger(typeof(DispatchQueue));
+
         private readonly IPipeManager _pipeManager;
         private readonly int _batchSize;
         private FlushableBlockingCollection<Entry> _queue = new FlushableBlockingCollection<Entry>();
@@ -328,7 +329,16 @@ namespace Abc.Zebus.Dispatch
             public void SetHandled(Exception error)
             {
                 foreach (var entry in Entries)
-                    entry.Dispatch.SetHandled(entry.Invoker, error);
+                {
+                    try
+                    {
+                        entry.Dispatch.SetHandled(entry.Invoker, error);
+                    }
+                    catch (Exception ex)
+                    {
+                        _logger.Error($"Unable to run dispatch continuation, MessageType: {entry.Invoker.MessageType.Name}, HandlerType: {entry.Invoker.MessageHandlerType.Name}, HandlerError: {error}, ContinuationError: {ex}");
+                    }
+                }
             }
 
             public void Clear()

src/Abc.Zebus/Dispatch/DynamicMessageHandlerInvoker.cs

@@ -35,6 +35,14 @@ namespace Abc.Zebus.Dispatch
         }
 
         public override bool ShouldHandle(IMessage message)
-            => _predicates.Any(predicate => predicate(message));
+        {
+            foreach (var predicate in _predicates)
+            {
+                if (predicate(message))
+                    return true;
+            }
+
+            return false;
+        }
     }
 }

src/Abc.Zebus/Dispatch/MessageDispatch.cs

@@ -6,8 +6,9 @@ namespace Abc.Zebus.Dispatch
 {
     public class MessageDispatch
     {
+        private static readonly object _exceptionsLock = new object();
+
         private readonly Action<MessageDispatch, DispatchResult> _continuation;
-        private readonly object _exceptionsLock = new object();
         private Dictionary<Type, Exception> _exceptions;
         private int _remainingHandlerCount;
 
@@ -55,4 +56,4 @@ namespace Abc.Zebus.Dispatch
             _remainingHandlerCount = handlerCount;
         }
     }
-}
+}

src/Abc.Zebus/Dispatch/Pipes/PipeInvocation.cs

@@ -10,6 +10,7 @@ namespace Abc.Zebus.Dispatch.Pipes
     public class PipeInvocation : IMessageHandlerInvocation
     {
         private static readonly BusMessageLogger _messageLogger = new BusMessageLogger(typeof(PipeInvocation), "Abc.Zebus.Dispatch");
+
         private readonly List<Action<object>> _handlerMutations = new List<Action<object>>();
         private readonly IMessageHandlerInvoker _invoker;
         private readonly IList<IMessage> _messages;
@@ -56,6 +57,9 @@ namespace Abc.Zebus.Dispatch.Pipes
 
         private object[] BeforeInvoke()
         {
+            if (_pipes.Count == 0)
+                return ArrayUtil.Empty<object>();
+
             var stateRef = new BeforeInvokeArgs.StateRef();
             var pipeStates = new object[_pipes.Count];
             for (var pipeIndex = 0; pipeIndex < _pipes.Count; ++pipeIndex)
@@ -64,6 +68,7 @@ namespace Abc.Zebus.Dispatch.Pipes
                 _pipes[pipeIndex].BeforeInvoke(beforeInvokeArgs);
                 pipeStates[pipeIndex] = beforeInvokeArgs.State;
             }
+
             return pipeStates;
         }
 
@@ -95,14 +100,14 @@ namespace Abc.Zebus.Dispatch.Pipes
 
         IDisposable IMessageHandlerInvocation.SetupForInvocation()
         {
-            _messageLogger.InfoFormat("HANDLE: {0} [{1}]", _messages[0], _messageContext.MessageId);
+            _messageLogger.InfoFormat("HANDLE{1}: {0} [{2}]", _messages[0], _invoker.DispatchQueueName, _messageContext.MessageId);
 
             return MessageContext.SetCurrent(_messageContext);
         }
 
         IDisposable IMessageHandlerInvocation.SetupForInvocation(object messageHandler)
         {
-            _messageLogger.InfoFormat("HANDLE: {0} [{1}]", _messages[0], _messageContext.MessageId);
+            _messageLogger.InfoFormat("HANDLE{1}: {0} [{2}]", _messages[0], _invoker.DispatchQueueName, _messageContext.MessageId);
 
             ApplyMutations(messageHandler);
 
@@ -111,14 +116,11 @@ namespace Abc.Zebus.Dispatch.Pipes
 
         private void ApplyMutations(object messageHandler)
         {
-            var messageContextAwareHandler = messageHandler as IMessageContextAware;
-            if (messageContextAwareHandler != null)
+            if (messageHandler is IMessageContextAware messageContextAwareHandler)
                 messageContextAwareHandler.Context = Context;
 
             foreach (var messageHandlerMutation in _handlerMutations)
-            {
                 messageHandlerMutation(messageHandler);
-            }
         }
     }
 }

src/Abc.Zebus/Dispatch/Pipes/PipeManager.cs

@@ -13,11 +13,13 @@ namespace Abc.Zebus.Dispatch.Pipes
         private readonly ConcurrentDictionary<Type, PipeList> _pipesByMessageType = new ConcurrentDictionary<Type, PipeList>();
         private readonly ConcurrentSet<string> _enabledPipeNames = new ConcurrentSet<string>();
         private readonly ConcurrentSet<string> _disabledPipeNames = new ConcurrentSet<string>();
+        private readonly Func<Type, PipeList> _createPipeList;
         private readonly IPipeSource[] _pipeSources;
 
         public PipeManager(IPipeSource[] pipeSources)
         {
             _pipeSources = pipeSources;
+            _createPipeList = CreatePipeList;
         }
 
         public void EnablePipe(string pipeName)
@@ -48,26 +50,18 @@ namespace Abc.Zebus.Dispatch.Pipes
             return new PipeInvocation(messageHandlerInvoker, messages, messageContext, pipes);
         }
 
-        public IEnumerable<IPipe> GetEnabledPipes(Type messageHandlerType)
-        {
-            return GetPipeList(messageHandlerType).EnabledPipes;
-        }
+        public IEnumerable<IPipe> GetEnabledPipes(Type messageHandlerType) 
+            => GetPipeList(messageHandlerType).EnabledPipes;
 
-        private PipeList GetPipeList(Type messageHandlerType)
-        {
-            return _pipesByMessageType.GetOrAdd(messageHandlerType, CreatePipeListEntry);
-        }
+        private PipeList GetPipeList(Type messageHandlerType) 
+            => _pipesByMessageType.GetOrAdd(messageHandlerType, _createPipeList);
 
-        private PipeList CreatePipeListEntry(Type handlerType)
-        {
-            var pipes = _pipeSources.SelectMany(x => x.GetPipes(handlerType));
-            return new PipeList(this, pipes);
-        }
+        private PipeList CreatePipeList(Type handlerType) 
+            => new PipeList(this, _pipeSources.SelectMany(x => x.GetPipes(handlerType)));
 
-        private bool IsPipeEnabled(IPipe pipe)
-        {
-            return !_disabledPipeNames.Contains(pipe.Name) && (pipe.IsAutoEnabled || _enabledPipeNames.Contains(pipe.Name));
-        }
+        private bool IsPipeEnabled(IPipe pipe) 
+            => !_disabledPipeNames.Contains(pipe.Name)
+               && (pipe.IsAutoEnabled || _enabledPipeNames.Contains(pipe.Name));
 
         private class PipeList
         {
@@ -90,4 +84,4 @@ namespace Abc.Zebus.Dispatch.Pipes
             }
         }
     }
-}
+}

src/Abc.Zebus/InfrastructureAttribute.cs

@@ -3,7 +3,7 @@
 namespace Abc.Zebus
 {
     [AttributeUsage(AttributeTargets.Class)]
-    public sealed class InfrastructureAttribute : Attribute
+    internal sealed class InfrastructureAttribute : Attribute
     {
     }
 }

src/Abc.Zebus/Initialization/ZebusRegistry.cs

@@ -41,6 +41,8 @@ namespace Abc.Zebus.Initialization
             ForSingletonOf<IStoppingStrategy>().Use<DefaultStoppingStrategy>();
 
             ForSingletonOf<IBindingKeyPredicateBuilder>().Use<BindingKeyPredicateBuilder>();
+
+            ForSingletonOf<IZmqOutboundSocketErrorHandler>().Use<DefaultZmqOutboundSocketErrorHandler>();
         }
     }
 }

src/Abc.Zebus/MessageContext.cs

@@ -1,4 +1,5 @@
 using System;
+using System.Threading;
 using Abc.Zebus.Transport;
 using Abc.Zebus.Util;
 
@@ -33,11 +34,7 @@ namespace Abc.Zebus
         public static MessageContext Current => _current;
 
         public static IDisposable SetCurrent(MessageContext context)
-        {
-            var previous = _current;
-            _current = context;
-            return new DisposableAction(() => _current = previous);
-        }
+            => new MessageContextScope(context);
 
         public static IDisposable OverrideInitiatorUsername(string username)
         {
@@ -111,5 +108,23 @@ namespace Abc.Zebus
 
             return new ErrorStatus(ReplyCode, ReplyMessage);
         }
+
+        private class MessageContextScope : IDisposable
+        {
+            private readonly MessageContext _previous;
+            private int _disposed;
+
+            public MessageContextScope(MessageContext context)
+            {
+                _previous = _current;
+                _current = context;
+            }
+
+            public void Dispose()
+            {
+                if (Interlocked.Exchange(ref _disposed, 1) == 0)
+                    _current = _previous;
+            }
+        }
     }
 }

src/Abc.Zebus/MessageTypeId.cs

@@ -27,8 +27,8 @@ namespace Abc.Zebus
         [ProtoMember(1, IsRequired = true)]
         public string FullName
         {
-            get { return _descriptor?.FullName; }
-            private set { _descriptor = MessageUtil.GetMessageTypeDescriptor(value); }
+            get => _descriptor?.FullName;
+            private set => _descriptor = MessageUtil.GetMessageTypeDescriptor(value);
         }
 
         public Type GetMessageType() => _descriptor?.MessageType;
@@ -42,12 +42,12 @@ namespace Abc.Zebus
         }
 
         public bool Equals(MessageTypeId other) => _descriptor == other._descriptor;
-        public override bool Equals(object obj) => obj is MessageTypeId && Equals((MessageTypeId)obj);
+        public override bool Equals(object obj) => obj is MessageTypeId messageTypeId && Equals(messageTypeId);
 
         [SuppressMessage("ReSharper", "NonReadonlyMemberInGetHashCode")]
         public override int GetHashCode() => _descriptor?.GetHashCode() ?? 0;
 
-        public static bool operator ==(MessageTypeId left, MessageTypeId right) => Equals(left, right);
-        public static bool operator !=(MessageTypeId left, MessageTypeId right) => !Equals(left, right);
+        public static bool operator ==(MessageTypeId left, MessageTypeId right) => left.Equals(right);
+        public static bool operator !=(MessageTypeId left, MessageTypeId right) => !left.Equals(right);
     }
 }

src/Abc.Zebus/Persistence/PersistentTransport.Phases.cs

@@ -32,6 +32,11 @@ namespace Abc.Zebus.Persistence
                 var messageReplayed = replayEvent as MessageReplayed;
                 if (messageReplayed != null)
                 {
+                    // the message was persisted because it comes from the persistence
+                    // but previous Zebus versions do not specify the WasPersisted field
+                    // => force WasPersisted to support previous Zebus version and make sure the message will be acked
+                    messageReplayed.Message.WasPersisted = true;
+
                     OnMessageReplayed(messageReplayed);
                     return;
                 }
@@ -120,11 +125,6 @@ namespace Abc.Zebus.Persistence
             {
                 Transport._logger.DebugFormat("REPLAY: {0} {1}", messageReplayed.Message.MessageTypeId, messageReplayed.Message.Id);
 
-                // the message was persisted because it comes from the persistence
-                // but previous Zebus versions do not specify the WasPersisted field
-                // => force WasPersisted to support previous Zebus version and make sure the message will be acked
-                messageReplayed.Message.WasPersisted = true;
-
                 Transport.TriggerMessageReceived(messageReplayed.Message);
                 Transport._receivedMessagesIds.TryAdd(messageReplayed.Message.Id, true);
 

src/Abc.Zebus/Persistence/StartMessageReplayCommand.cs

@@ -3,7 +3,7 @@ using ProtoBuf;
 
 namespace Abc.Zebus.Persistence
 {
-    [ProtoContract, Transient, Infrastructure]
+    [ProtoContract, Transient]
     public class StartMessageReplayCommand : ICommand
     {
         [ProtoMember(1, IsRequired = true)]
@@ -14,4 +14,4 @@ namespace Abc.Zebus.Persistence
             ReplayId = replayId;
         }
     }
-}
+}

src/Abc.Zebus/Routing/BindingKey.cs

@@ -19,24 +19,16 @@ namespace Abc.Zebus.Routing
 
         private static readonly ConcurrentDictionary<Type, BindingKeyBuilder> _builders = new ConcurrentDictionary<Type, BindingKeyBuilder>();
         private static readonly Func<Type, BindingKeyBuilder> _bindingKeyBuilderFactory = CreateBuilder;
-        private static readonly char[] _separator = { '.' };
 
         [ProtoMember(1, IsRequired = true)]
         private readonly string[] _parts;
 
         public BindingKey(params string[] parts)
-            : this(parts, false)
-        {
-        }
-
-        private BindingKey(string[] parts, bool isJoined)
         {
             if (parts == null || parts.Length == 0)
                 _parts = null;
             else
                 _parts = parts;
-
-            IsJoined = isJoined;
         }
 
         public int PartCount => _parts?.Length ?? 0;
@@ -53,8 +45,6 @@ namespace Abc.Zebus.Routing
             return _parts[index] == _star;
         }
 
-        internal bool IsJoined { get; }
-
         [Pure]
         public string GetPart(int index) => index < PartCount ? _parts[index] : null;
 
@@ -105,13 +95,6 @@ namespace Abc.Zebus.Routing
             return string.Join(".", _parts);
         }
 
-        /// <summary>
-        ///     For Qpid compatibility only, do not use in Zebus code.
-        /// </summary>
-        internal static BindingKey Joined(string s) => new BindingKey(new[] { s }, true);
-
-        internal static BindingKey Split(string s) => new BindingKey(s.Split(_separator));
-
         internal static BindingKey Create(IMessage message)
             => GetBindingKeyBuilder(message.GetType())?.BuildKey(message) ?? Empty;
 

src/Abc.Zebus/Serialization/Protobuf/CodedInputStream.cs

@@ -97,7 +97,11 @@ namespace Abc.Zebus.Serialization.Protobuf
         /// <summary>
         /// Returns the current position in the input stream, or the position in the input buffer
         /// </summary>
-        public long Position { get { return bufferPos; } }
+        public int Position
+        {
+            get => bufferPos;
+            set => bufferPos = value;
+        }
 
         #region Reading of tags etc
 
@@ -640,4 +644,4 @@ namespace Abc.Zebus.Serialization.Protobuf
 
         #endregion
     }
-}
+}

src/Abc.Zebus/Subscription.cs

@@ -52,9 +52,6 @@ namespace Abc.Zebus
             if (BindingKey.IsEmpty)
                 return true;
 
-            if (routingKey.IsJoined && BindingKey.PartCount != 1)
-                return MatchesJoinedRoutingKey(routingKey.GetPart(0));
-
             for (var i = 0; i < routingKey.PartCount; i++)
             {
                 var evaluatedPart = BindingKey.GetPart(i);
@@ -68,20 +65,6 @@ namespace Abc.Zebus
             return routingKey.PartCount == BindingKey.PartCount;
         }
 
-        private bool MatchesJoinedRoutingKey(string routingKey)
-        {
-            // TODO: Remove code when the gateway is decommissioned
-            // slow and ugly, this code is only used in the gateway for Qpid compatibility
-
-            var bindingKey = BindingKey.ToString();
-            var wildCardIndex = bindingKey.IndexOf('#');
-            if (wildCardIndex != -1)
-                bindingKey = bindingKey.Substring(0, wildCardIndex);
-
-            var pattern = "^" + bindingKey.Replace("*", ".*");
-            return Regex.IsMatch(routingKey, pattern);
-        }
-
         public bool Equals(Subscription other)
         {
             if (other == null)

src/Abc.Zebus/Transport/DefaultZmqOutboundSocketErrorHandler.cs

@@ -0,0 +1,19 @@
+using System;
+
+namespace Abc.Zebus.Transport
+{
+    public class DefaultZmqOutboundSocketErrorHandler : IZmqOutboundSocketErrorHandler
+    {
+        public void OnConnectException(PeerId peerId, string endPoint, Exception exception)
+        {
+        }
+
+        public void OnDisconnectException(PeerId peerId, string endPoint, Exception exception)
+        {
+        }
+
+        public void OnSendFailed(PeerId peerId, string endPoint, MessageTypeId messageTypeId, MessageId id)
+        {
+        }
+    }
+}

src/Abc.Zebus/Transport/IZmqOutboundSocketErrorHandler.cs

@@ -0,0 +1,11 @@
+using System;
+
+namespace Abc.Zebus.Transport
+{
+    public interface IZmqOutboundSocketErrorHandler
+    {
+        void OnConnectException(PeerId peerId, string endPoint, Exception exception);
+        void OnDisconnectException(PeerId peerId, string endPoint, Exception exception);
+        void OnSendFailed(PeerId peerId, string endPoint, MessageTypeId messageTypeId, MessageId id);
+    }
+}

src/Abc.Zebus/Transport/TransportMessageReader.cs

@@ -12,9 +12,7 @@ namespace Abc.Zebus.Transport
         {
             var transportMessage = new TransportMessage { Content = Stream.Null };
 
-            uint number;
-            WireType wireType;
-            while (!input.IsAtEnd && input.TryReadTag(out number, out wireType))
+            while (!input.IsAtEnd && input.TryReadTag(out var number, out var wireType))
             {
                 switch (number)
                 {
@@ -53,13 +51,11 @@ namespace Abc.Zebus.Transport
             var length = input.ReadLength();
             var endPosition = input.Position + length;
 
-            uint number;
-            WireType wireType;
             var senderId = new PeerId();
             string senderEndPoint = null;
             string initiatorUserName = null;
 
-            while (input.Position < endPosition && input.TryReadTag(out number, out wireType))
+            while (input.Position < endPosition && input.TryReadTag(out var number, out var wireType))
             {
                 switch (number)
                 {
@@ -89,7 +85,7 @@ namespace Abc.Zebus.Transport
 
         private static MessageId ReadMessageId(CodedInputStream input)
         {
-            var guid = ReadSingleField(input, x => ReadGuid(input));
+            var guid = ReadSingleField(input, x => x.ReadGuid());
             return new MessageId(guid);
         }
 
@@ -99,11 +95,6 @@ namespace Abc.Zebus.Transport
             return new MessageTypeId(fullName);
         }
 
-        private static Guid ReadGuid(CodedInputStream input)
-        {
-            return input.ReadGuid();
-        }
-
         private static Stream ReadStream(CodedInputStream input)
         {
             var length = input.ReadLength();
@@ -115,11 +106,9 @@ namespace Abc.Zebus.Transport
             var length = input.ReadLength();
             var endPosition = input.Position + length;
 
-            uint number;
-            WireType wireType;
             var value = default(T);
 
-            while (input.Position < endPosition && input.TryReadTag(out number, out wireType))
+            while (input.Position < endPosition && input.TryReadTag(out var number, out var wireType))
             {
                 switch (number)
                 {
@@ -173,4 +162,4 @@ namespace Abc.Zebus.Transport
             }
         }
     }
-}
+}

src/Abc.Zebus/Transport/ZmqInboundSocket.cs

@@ -17,6 +17,7 @@ namespace Abc.Zebus.Transport
         private byte[] _readBuffer = new byte[0];
         private ZmqSocket _socket;
         private ZmqEndPoint _endPoint;
+        private TimeSpan _lastReceiveTimeout = TimeSpan.MinValue;
 
         public ZmqInboundSocket(ZmqContext context, PeerId peerId, ZmqEndPoint originalEndpoint, ZmqSocketOptions options, string environment)
         {
@@ -50,10 +51,14 @@ namespace Abc.Zebus.Transport
 
         public CodedInputStream Receive(TimeSpan? timeout = null)
         {
-            int size;
-
-            _socket.ReceiveTimeout = timeout ?? _options.ReadTimeout;
-            _readBuffer = _socket.Receive(_readBuffer, TimeSpan.MaxValue, out size);
+            var effectiveTimeout = timeout ?? _options.ReadTimeout;
+            if (effectiveTimeout != _lastReceiveTimeout)
+            {
+                _socket.ReceiveTimeout = effectiveTimeout; // This stuff allocates
+                _lastReceiveTimeout = effectiveTimeout;
+            }
+            
+            _readBuffer = _socket.Receive(_readBuffer, TimeSpan.MaxValue, out var size);
 
             if (size <= 0)
                 return null;
@@ -79,4 +84,4 @@ namespace Abc.Zebus.Transport
             _socket.Disconnect(_endPoint.Value);
         }
     }
-}
+}

src/Abc.Zebus/Transport/ZmqOutboundSocket.cs

@@ -1,6 +1,5 @@
 using System;
 using System.Diagnostics;
-using System.IO;
 using log4net;
 using ZeroMQ;
 
@@ -12,15 +11,17 @@ namespace Abc.Zebus.Transport
         private readonly Stopwatch _closedStateStopwatch = new Stopwatch();
         private readonly ZmqContext _context;
         private readonly ZmqSocketOptions _options;
+        private readonly IZmqOutboundSocketErrorHandler _errorHandler;
         private ZmqSocket _socket;
         private int _failedSendCount;
         private bool _isInClosedState;
         private TimeSpan _closedStateDuration;
 
-        public ZmqOutboundSocket(ZmqContext context, PeerId peerId, string endPoint, ZmqSocketOptions options)
+        public ZmqOutboundSocket(ZmqContext context, PeerId peerId, string endPoint, ZmqSocketOptions options, IZmqOutboundSocketErrorHandler errorHandler)
         {
             _context = context;
             _options = options;
+            _errorHandler = errorHandler;
             PeerId = peerId;
             EndPoint = endPoint;
         }
@@ -56,6 +57,7 @@ namespace Abc.Zebus.Transport
                 IsConnected = false;
 
                 _logger.ErrorFormat("Unable to connect socket, Peer: {0}, EndPoint: {1}, Exception: {2}", PeerId, EndPoint, ex);
+                _errorHandler.OnConnectException(PeerId, EndPoint, ex);
 
                 SwitchToClosedState(_options.ClosedStateDurationAfterConnectFailure);
             }
@@ -83,6 +85,7 @@ namespace Abc.Zebus.Transport
             catch (Exception ex)
             {
                 _logger.ErrorFormat("Unable to disconnect socket, Peer: {0}, Exception: {1}", PeerId, ex);
+                _errorHandler.OnDisconnectException(PeerId, EndPoint, ex);
             }
 
             IsConnected = false;
@@ -100,6 +103,7 @@ namespace Abc.Zebus.Transport
             }
 
             _logger.ErrorFormat("Unable to send message, destination peer: {0}, MessageTypeId: {1}, MessageId: {2}", PeerId, message.MessageTypeId, message.Id);
+            _errorHandler.OnSendFailed(PeerId, EndPoint, message.MessageTypeId, message.Id);
 
             if (_failedSendCount >= _options.SendRetriesBeforeSwitchingToClosedState)
                 SwitchToClosedState(_options.ClosedStateDurationAfterSendFailure);
@@ -113,17 +117,19 @@ namespace Abc.Zebus.Transport
             {
                 if (_closedStateStopwatch.Elapsed < _closedStateDuration)
                 {
-                    _logger.DebugFormat("Send or connect ignored in closed state, Peer: {0}, MessageTypeId: {1}, MessageId: {2}", PeerId, message.MessageTypeId, message.Id);
+                    _logger.WarnFormat("Send or connect ignored in closed state, Peer: {0}, MessageTypeId: {1}, MessageId: {2}", PeerId, message.MessageTypeId, message.Id);
                     return false;
                 }
+
                 SwitchToOpenState();
             }
+
             return true;
         }
 
         private void SwitchToClosedState(TimeSpan duration)
         {
-            _logger.InfoFormat("Switching to closed state, Peer: {0}, Duration: {1}", PeerId, duration);
+            _logger.ErrorFormat("Switching to closed state, Peer: {0}, Duration: {1}", PeerId, duration);
 
             _closedStateStopwatch.Start();
             _closedStateDuration = duration;
@@ -138,4 +144,4 @@ namespace Abc.Zebus.Transport
             _closedStateStopwatch.Reset();
         }
     }
-}
+}

src/Abc.Zebus/Transport/ZmqTransport.cs

@@ -15,6 +15,7 @@ namespace Abc.Zebus.Transport
     public class ZmqTransport : ITransport
     {
         private readonly IZmqTransportConfiguration _configuration;
+        private readonly IZmqOutboundSocketErrorHandler _errorHandler;
         private readonly ZmqEndPoint _configuredInboundEndPoint;
         private ILog _logger = LogManager.GetLogger(typeof(ZmqTransport));
         private ConcurrentDictionary<PeerId, ZmqOutboundSocket> _outboundSockets;
@@ -57,9 +58,10 @@ namespace Abc.Zebus.Transport
             }
         }
 
-        public ZmqTransport(IZmqTransportConfiguration configuration, ZmqSocketOptions socketOptions)
+        public ZmqTransport(IZmqTransportConfiguration configuration, ZmqSocketOptions socketOptions, IZmqOutboundSocketErrorHandler errorHandler)
         {
             _configuration = configuration;
+            _errorHandler = errorHandler;
             _configuredInboundEndPoint = new ZmqEndPoint(configuration.InboundEndPoint);
             SocketOptions = socketOptions;
         }
@@ -414,7 +416,7 @@ namespace Abc.Zebus.Transport
             ZmqOutboundSocket outboundSocket;
             if (!_outboundSockets.TryGetValue(peer.Id, out outboundSocket))
             {
-                outboundSocket = new ZmqOutboundSocket(_context, peer.Id, peer.EndPoint, SocketOptions);
+                outboundSocket = new ZmqOutboundSocket(_context, peer.Id, peer.EndPoint, SocketOptions, _errorHandler);
                 outboundSocket.ConnectFor(transportMessage);
 
                 _outboundSockets.TryAdd(peer.Id, outboundSocket);

src/Abc.Zebus/Util/Buffer.cs

@@ -1,122 +0,0 @@
-using System;
-using System.IO;
-using Abc.Zebus.Util.Annotations;
-
-namespace Abc.Zebus.Util
-{
-    internal struct Buffer
-    {
-        private byte[] _data;
-        private int _length;
-
-        public Buffer(byte[] data) :
-            this(data, data.Length)
-        {
-        }
-
-        public Buffer(byte[] data, int length)
-        {
-            _data = data;
-            _length = length;
-        }
-
-        public Buffer(int byteCount)
-        {
-            _data = new byte[byteCount];
-            _length = 0;
-        }
-
-        public byte[] Data
-        {
-            get { return _data; }
-            private set { _data = value; }
-        }
-
-        public int Length
-        {
-            get { return _length; }
-            set
-            {
-                if(value < 0 || value > _data.Length)
-                    throw new ArgumentOutOfRangeException("Length");
-                _length = value;
-            }
-        }
-
-        public byte[] ToByteArray()
-        {
-            var data = new Byte[_length];
-            System.Buffer.BlockCopy(_data, 0, data, 0, _length);
-            return data; 
-        }
-
-        public void CopyTo(ref Buffer buffer)
-        {
-            buffer._length = _length;
-            System.Buffer.BlockCopy(_data, 0, buffer._data, 0, _length);
-        }
-
-        public void CopyFrom(ref Buffer buffer)
-        {
-            _length = buffer._length;
-            System.Buffer.BlockCopy(buffer._data, 0, _data, 0, _length);
-        }
-        
-        public void CopyFrom([NotNull] byte[] bytes)
-        {
-            if (bytes == null) 
-                throw new ArgumentNullException(nameof(bytes));
-            
-            _length = bytes.Length;
-            System.Buffer.BlockCopy(bytes, 0, _data, 0, _length);
-        }
-        
-        public Stream GetStream()
-        {
-            return new MemoryStream(_data, 0, _length);
-        }
-        
-        public void Reset()
-        {
-            _length = 0;
-        }
-        
-        public void WriteTo(BinaryWriter writer)
-        {
-            writer.Write(_length);
-            writer.Write(_data, 0, _length);
-        }
-
-        public void ReadFrom(BinaryReader binaryReader)
-        {
-            var length = binaryReader.ReadInt32();
-            binaryReader.Read(_data, 0, length);
-            _length = length;
-        }
-
-        public override int GetHashCode()
-        {
-            unchecked
-            {
-                const int p = 16777619;
-                int hash = (int)2166136261;
-
-                for (int i = 0; i < _length; i++)
-                    hash = (hash ^ _data[i]) * p;
-
-                hash += hash << 13;
-                hash ^= hash >> 7;
-                hash += hash << 3;
-                hash ^= hash >> 17;
-                hash += hash << 5;
-                return hash;
-            }
-        }
-
-        public void CopyFrom(byte[] buffer, int offset, int length)
-        {
-            _length = length;
-            System.Buffer.BlockCopy(buffer, offset, _data, 0, _length);
-        }
-    }
-}

src/Abc.Zebus/Util/Extensions/ExtendIEnumerable.cs

@@ -34,14 +34,8 @@ namespace Abc.Zebus.Util.Extensions
         }
 
         [Pure]
-        public static IList<T> AsList<T>([InstantHandle] this IEnumerable<T> collection)
-        {
-            var list = collection as IList<T>;
-            if (list == null || list.IsReadOnly || list is T[])
-                return collection.ToList();
-
-            return list;
-        }
+        public static IList<T> AsList<T>([InstantHandle] this IEnumerable<T> collection) 
+            => collection is IList<T> list ? list : collection.ToList();
 
         [Pure]
         public static IEnumerable<TSource> DistinctBy<TSource, TKey>(this IEnumerable<TSource> source, Func<TSource, TKey> keySelector)

src/Abc.Zebus/packages.config

@@ -1,7 +0,0 @@
-<?xml version="1.0" encoding="utf-8"?>
-<packages>
-  <package id="log4net" version="2.0.3" targetFramework="net45" />
-  <package id="Newtonsoft.Json" version="7.0.1" targetFramework="net45" />
-  <package id="protobuf-net" version="2.0.0.668" targetFramework="net45" />
-  <package id="structuremap" version="3.1.6.186" targetFramework="net45" />
-</packages>

version.yml

@@ -1,2 +1,2 @@
-NugetVersion: 2.0.5
-AssemblyVersion: 2.0.5
+NugetVersion: 2.2.1
+AssemblyVersion: 2.2.1