Merge pull request pardahlman#140 from ritasker/expose-mandatory-option-for-publish

(pardahlman#129) Expose Mandatory Option For Publish

Author: pardahlman

.gitignore

@@ -13,6 +13,9 @@ docs/*build*/
 #VS Code
 .vscode/**
 
+#Rider
+.idea/**
+
 # User-specific files (MonoDevelop/Xamarin Studio)
 *.userprefs
 **ncrunch**

sample/RawRabbit.Messages.Sample/project.json

@@ -7,6 +7,6 @@
 
 	"frameworks": {
 		"netstandard1.5": {},
-		"net451": {},
+		"net451": {}
 	}
 }

src/RawRabbit/Configuration/Publish/IPublishConfigurationBuilder.cs

@@ -1,5 +1,6 @@
 using System;
 using RabbitMQ.Client;
+using RabbitMQ.Client.Events;
 using RawRabbit.Configuration.Exchange;
 
 namespace RawRabbit.Configuration.Publish
@@ -9,5 +10,6 @@ public interface IPublishConfigurationBuilder
 		IPublishConfigurationBuilder WithExchange(Action<IExchangeConfigurationBuilder> exchange);
 		IPublishConfigurationBuilder WithRoutingKey(string routingKey);
 		IPublishConfigurationBuilder WithProperties(Action<IBasicProperties> properties);
+		IPublishConfigurationBuilder WithMandatoryDelivery(EventHandler<BasicReturnEventArgs> basicReturn);
 	}
 }

src/RawRabbit/Configuration/Publish/PublishConfiguration.cs

@@ -1,5 +1,6 @@
 using System;
 using RabbitMQ.Client;
+using RabbitMQ.Client.Events;
 using RawRabbit.Configuration.Exchange;
 
 namespace RawRabbit.Configuration.Publish
@@ -9,5 +10,6 @@ public class PublishConfiguration
 		public ExchangeConfiguration Exchange { get; set; }
 		public string RoutingKey { get; set; }
 		public Action<IBasicProperties> PropertyModifier { get; set; }
+		public EventHandler<BasicReturnEventArgs> BasicReturn { get; set; }
 	}
 }

src/RawRabbit/Configuration/Publish/PublishConfigurationBuilder.cs

@@ -1,5 +1,6 @@
 using System;
 using RabbitMQ.Client;
+using RabbitMQ.Client.Events;
 using RawRabbit.Configuration.Exchange;
 using RawRabbit.Configuration.Request;
 
@@ -11,12 +12,14 @@ public class PublishConfigurationBuilder : IPublishConfigurationBuilder
 		private string _routingKey;
 		private Action<IBasicProperties> _properties;
 		private const string _oneOrMoreWords = "#";
+		private EventHandler<BasicReturnEventArgs> _basicReturn;
 
 		public PublishConfiguration Configuration => new PublishConfiguration
 		{
 			Exchange = _exchange.Configuration,
 			RoutingKey = _routingKey,
-			PropertyModifier = _properties ?? (b => {})
+			PropertyModifier = _properties ?? (b => {}),
+			BasicReturn = _basicReturn
 		};
 
 		public PublishConfigurationBuilder(ExchangeConfiguration defaultExchange = null, string routingKey =null)
@@ -48,5 +51,11 @@ public IPublishConfigurationBuilder WithProperties(Action<IBasicProperties> prop
 			_properties = properties;
 			return this;
 		}
+
+		public IPublishConfigurationBuilder WithMandatoryDelivery(EventHandler<BasicReturnEventArgs> basicReturn)
+		{
+			_basicReturn = basicReturn;
+			return this;
+		}
 	}
 }

src/RawRabbit/Operations/Publisher.cs

@@ -1,6 +1,5 @@
 using System;
 using System.Threading.Tasks;
-using RabbitMQ.Client;
 using RawRabbit.Channel.Abstraction;
 using RawRabbit.Common;
 using RawRabbit.Configuration;
@@ -57,16 +56,23 @@ public Task PublishAsync<TMessage>(TMessage message, Guid globalMessageId, Publi
 					{
 						throw t.Exception;
 					}
+
+					channelTask.Result.BasicReturn += config.BasicReturn;
+
 					lock (_publishLock)
 					{
 						var ackTask = _acknowledger.GetAckTask(channelTask.Result);
 						channelTask.Result.BasicPublish(
 							exchange: config.Exchange.ExchangeName,
 							routingKey: _config.RouteWithGlobalId ? $"{config.RoutingKey}.{globalMessageId}" : config.RoutingKey,
 							basicProperties: props,
-							body: _serializer.Serialize(message)
-							);
-						return ackTask;
+							body: _serializer.Serialize(message),
+							mandatory: (config.BasicReturn != null)
+						);
+						return ackTask
+						.ContinueWith(a => {
+							channelTask.Result.BasicReturn -= config.BasicReturn;
+						});
 					}
 				})
 				.Unwrap();

test/RawRabbit.IntegrationTests/SimpleUse/PublishAndSubscribeTests.cs

@@ -10,7 +10,6 @@
 using RawRabbit.Configuration;
 using RawRabbit.Exceptions;
 using RawRabbit.IntegrationTests.TestMessages;
-using RawRabbit.vNext;
 using Xunit;
 using ExchangeType = RawRabbit.Configuration.Exchange.ExchangeType;
 
@@ -389,6 +388,26 @@ public async Task Should_Be_Able_To_Publish_Message_After_Failed_Publish()
 				Assert.True(true);
 			}
 		}
+
+		[Fact]
+		public async Task Should_Run_Basic_Return_When_The_Manatory_Set_After_Failed_Publish()
+		{
+			/* Setup */
+			using (var client = TestClientFactory.CreateNormal())
+			{
+				var tcs = new TaskCompletionSource<bool>();
+				/* Test */
+				await client.PublishAsync(new SimpleMessage(),
+					configuration: cfg => cfg
+						.WithMandatoryDelivery((obj, evt) =>
+						{
+							tcs.SetResult(true);
+						}));
+
+				/* Assert */
+				Assert.True(tcs.Task.Result);
+			}
+		}
 	}
 }