Merge branch '2.0' into add-messagepack

Author: LordMike

RawRabbit.sln

@@ -74,6 +74,8 @@ Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "RawRabbit.Enrichers.Polly.T
 EndProject
 Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "RawRabbit.Enrichers.RetryLater", "src\RawRabbit.Enrichers.RetryLater\RawRabbit.Enrichers.RetryLater.csproj", "{E1816B3D-9C4B-4D08-9537-ACB6806AF690}"
 EndProject
+Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "RawRabbit.Enrichers.ZeroFormatter", "src\RawRabbit.Enrichers.ZeroFormatter\RawRabbit.Enrichers.ZeroFormatter.csproj", "{B5B01F7F-4592-4F58-9E7B-3C689E76590A}"
+EndProject
 Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "RawRabbit.Enrichers.MessagePack", "src\RawRabbit.Enrichers.MessagePack\RawRabbit.Enrichers.MessagePack.csproj", "{56BD6C7D-5AE4-4E2F-BF9D-3DCBB54D2552}"
 EndProject
 Global
@@ -202,6 +204,10 @@ Global
 		{E1816B3D-9C4B-4D08-9537-ACB6806AF690}.Debug|Any CPU.Build.0 = Debug|Any CPU
 		{E1816B3D-9C4B-4D08-9537-ACB6806AF690}.Release|Any CPU.ActiveCfg = Release|Any CPU
 		{E1816B3D-9C4B-4D08-9537-ACB6806AF690}.Release|Any CPU.Build.0 = Release|Any CPU
+		{B5B01F7F-4592-4F58-9E7B-3C689E76590A}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
+		{B5B01F7F-4592-4F58-9E7B-3C689E76590A}.Debug|Any CPU.Build.0 = Debug|Any CPU
+		{B5B01F7F-4592-4F58-9E7B-3C689E76590A}.Release|Any CPU.ActiveCfg = Release|Any CPU
+		{B5B01F7F-4592-4F58-9E7B-3C689E76590A}.Release|Any CPU.Build.0 = Release|Any CPU
 		{56BD6C7D-5AE4-4E2F-BF9D-3DCBB54D2552}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
 		{56BD6C7D-5AE4-4E2F-BF9D-3DCBB54D2552}.Debug|Any CPU.Build.0 = Debug|Any CPU
 		{56BD6C7D-5AE4-4E2F-BF9D-3DCBB54D2552}.Release|Any CPU.ActiveCfg = Release|Any CPU
@@ -241,6 +247,7 @@ Global
 		{8D45F8AC-B65F-4A2B-9153-8A7F3D423575} = {7FCF8D3B-BA55-4C47-AC60-5CEF75418BEB}
 		{4B4C5936-D61E-4FD8-AEB7-154CEAF84E15} = {2F91E22A-AEBA-4BEF-9A03-C8232830F697}
 		{E1816B3D-9C4B-4D08-9537-ACB6806AF690} = {7FCF8D3B-BA55-4C47-AC60-5CEF75418BEB}
+		{B5B01F7F-4592-4F58-9E7B-3C689E76590A} = {7FCF8D3B-BA55-4C47-AC60-5CEF75418BEB}
 		{56BD6C7D-5AE4-4E2F-BF9D-3DCBB54D2552} = {7FCF8D3B-BA55-4C47-AC60-5CEF75418BEB}
 	EndGlobalSection
 	GlobalSection(ExtensibilityGlobals) = postSolution

src/RawRabbit.Enrichers.ZeroFormatter/RawRabbit.Enrichers.ZeroFormatter.csproj

@@ -0,0 +1,16 @@
+<Project Sdk="Microsoft.NET.Sdk">
+
+  <PropertyGroup>
+    <TargetFrameworks>netstandard1.6;net451</TargetFrameworks>
+    <Authors>LordMike</Authors>
+  </PropertyGroup>
+
+  <ItemGroup>
+    <PackageReference Include="ZeroFormatter" Version="1.6.4" />
+  </ItemGroup>
+
+  <ItemGroup>
+    <ProjectReference Include="..\RawRabbit\RawRabbit.csproj" />
+  </ItemGroup>
+
+</Project>

src/RawRabbit.Enrichers.ZeroFormatter/ZeroFormatterPlugin.cs

@@ -0,0 +1,19 @@
+using RawRabbit.Instantiation;
+using RawRabbit.Serialization;
+
+namespace RawRabbit.Enrichers.ZeroFormatter
+{
+    public static class ZeroFormatterPlugin
+    {
+        /// <summary>
+        /// Replaces the default serializer with ZeroFormatter.
+        /// </summary>
+        public static IClientBuilder UseZeroFormatter(this IClientBuilder builder)
+        {
+            builder.Register(
+                pipe: p => { },
+                ioc: di => di.AddSingleton<ISerializer, ZeroFormatterSerializerWorker>());
+            return builder;
+        }
+    }
+}

src/RawRabbit.Enrichers.ZeroFormatter/ZeroFormatterSerializerWorker.cs

@@ -0,0 +1,45 @@
+using System;
+using System.Linq;
+using System.Reflection;
+using RawRabbit.Serialization;
+using ZeroFormatter;
+
+namespace RawRabbit.Enrichers.ZeroFormatter
+{
+    internal class ZeroFormatterSerializerWorker : ISerializer
+    {
+        public string ContentType => "application/x-zeroformatter";
+        private readonly MethodInfo _deserializeType;
+        private readonly MethodInfo _serializeType;
+
+        public ZeroFormatterSerializerWorker()
+        {
+            _deserializeType = typeof(ZeroFormatterSerializer)
+                .GetMethod(nameof(ZeroFormatterSerializer.Deserialize), new[] { typeof(byte[]) });
+            _serializeType = typeof(ZeroFormatterSerializer)
+                .GetMethods()
+                .FirstOrDefault(s => s.Name == nameof(ZeroFormatterSerializer.Serialize) && s.ReturnType == typeof(byte[]));
+        }
+
+        public byte[] Serialize(object obj)
+        {
+            if (obj == null)
+                throw new ArgumentNullException();
+
+            return (byte[])_serializeType
+                .MakeGenericMethod(obj.GetType())
+                .Invoke(null, new[] { obj });
+        }
+
+        public object Deserialize(Type type, byte[] bytes)
+        {
+            return _deserializeType.MakeGenericMethod(type)
+                .Invoke(null, new object[] { bytes });
+        }
+
+        public TType Deserialize<TType>(byte[] bytes)
+        {
+            return ZeroFormatterSerializer.Deserialize<TType>(bytes);
+        }
+    }
+}

src/RawRabbit.Operations.Request/Middleware/ResponseConsumeMiddleware.cs

@@ -23,10 +23,12 @@ public class ResponseConsumerOptions
 
 	public class ResponseConsumeMiddleware : Pipe.Middleware.Middleware
 	{
+		protected static readonly ConcurrentDictionary<IBasicConsumer, ConcurrentDictionary<string, TaskCompletionSource<BasicDeliverEventArgs>>> AllResponses =
+			new ConcurrentDictionary<IBasicConsumer, ConcurrentDictionary<string, TaskCompletionSource<BasicDeliverEventArgs>>>();
+
 		protected readonly IConsumerFactory ConsumerFactory;
 		protected readonly Pipe.Middleware.Middleware ResponsePipe;
 		private readonly ILog _logger = LogProvider.For<ResponseConsumeMiddleware>();
-		protected readonly ConcurrentDictionary<IBasicConsumer, ConcurrentDictionary<string, TaskCompletionSource<BasicDeliverEventArgs>>> AllResponses;
 		protected Func<IPipeContext, ConsumerConfiguration> ResponseConfigFunc;
 		protected Func<IPipeContext, string> CorrelationidFunc;
 		protected Func<IPipeContext, bool> DedicatedConsumerFunc;
@@ -38,7 +40,6 @@ public ResponseConsumeMiddleware(IConsumerFactory consumerFactory, IPipeBuilderF
 			DedicatedConsumerFunc = options?.UseDedicatedConsumer ?? (context => context.GetDedicatedResponseConsumer());
 			ConsumerFactory = consumerFactory;
 			ResponsePipe = factory.Create(options.ResponseRecieved);
-			AllResponses = new ConcurrentDictionary<IBasicConsumer, ConcurrentDictionary<string, TaskCompletionSource<BasicDeliverEventArgs>>>();
 		}
 
 		public override async Task InvokeAsync(IPipeContext context, CancellationToken token)

src/RawRabbit/Channel/ChannelFactory.cs

@@ -82,7 +82,7 @@ public ChannelFactory(IConnectionFactory connectionFactory, RawRabbitConfigurati
 
 			_logger.Debug("Connection is recoverable. Waiting for 'Recovery' event to be triggered. ");
 			var recoverTcs = new TaskCompletionSource<IConnection>();
-			token.Register(() => recoverTcs.SetCanceled());
+			token.Register(() => recoverTcs.TrySetCanceled());
 
 			EventHandler<EventArgs> completeTask = null;
 			completeTask = (sender, args) =>

src/RawRabbit/Channel/StaticChannelPool.cs

@@ -53,10 +53,17 @@ private void StartServeChannels()
 			do
 			{
 				_current = _current?.Next ?? Pool.First;
+				if (_current == null)
+				{
+					_logger.Debug("Unable to server channels. Pool empty.");
+					Monitor.Exit(_workLock);
+					return;
+				}
 				if (_current.Value.IsClosed)
 				{
 					Pool.Remove(_current);
 					if (Pool.Count != 0) continue;
+					Monitor.Exit(_workLock);
 					if (Recoverables.Count == 0)
 					{
 						throw new ChannelAvailabilityException("No open channels in pool and no recoverable channels");