Added support for setting up publisher confirms for Rabbit MQ

src/MassTransit/Context/SendContext.cs

@@ -29,7 +29,7 @@ public SendContext(T message)
         {
             _id = NewId.NextGuid();
             _message = message;
-
+            
             this.SetMessageType(typeof (T));
             DeclaringMessageType = typeof (T);
         }

src/Transports/MassTransit.Transports.RabbitMq/Configuration/Builders/RabbitMqTransportFactoryBuilder.cs

@@ -13,9 +13,12 @@
 namespace MassTransit.Transports.RabbitMq.Configuration.Builders
 {
 	using System;
+	using PublisherConfirm;
 
-	public interface RabbitMqTransportFactoryBuilder
+    public interface RabbitMqTransportFactoryBuilder
 	{
 		void AddConnectionFactoryBuilder(Uri uri, ConnectionFactoryBuilder connectionFactoryBuilder);
+
+	    void SetPublisherConfirmSettings(IPublisherConfirmSettings publisherConfirmSettings);
 	}
 }

src/Transports/MassTransit.Transports.RabbitMq/Configuration/Builders/RabbitMqTransportFactoryBuilderImpl.cs

@@ -14,11 +14,12 @@ namespace MassTransit.Transports.RabbitMq.Configuration.Builders
 {
 	using System;
 	using System.Collections.Generic;
+	using PublisherConfirm;
 
-	public class RabbitMqTransportFactoryBuilderImpl :
-		RabbitMqTransportFactoryBuilder
+    public class RabbitMqTransportFactoryBuilderImpl : RabbitMqTransportFactoryBuilder
 	{
 		readonly IDictionary<Uri, ConnectionFactoryBuilder> _connectionFactoryBuilders;
+        IPublisherConfirmSettings _publisherConfirmSettings = new PublisherConfirmSettings();
 
 		public RabbitMqTransportFactoryBuilderImpl()
 		{
@@ -30,9 +31,14 @@ public void AddConnectionFactoryBuilder(Uri uri, ConnectionFactoryBuilder connec
 			_connectionFactoryBuilders[uri] = connectionFactoryBuilder;
 		}
 
+        public void SetPublisherConfirmSettings(IPublisherConfirmSettings publisherConfirmSettings)
+        {
+            _publisherConfirmSettings = publisherConfirmSettings;
+        }
+
 		public RabbitMqTransportFactory Build()
 		{
-			var factory = new RabbitMqTransportFactory(_connectionFactoryBuilders);
+            var factory = new RabbitMqTransportFactory(_connectionFactoryBuilders, _publisherConfirmSettings);
 
 			return factory;
 		}

src/Transports/MassTransit.Transports.RabbitMq/Configuration/Configurators/PublisherConfirmFactoryConfigurator.cs

@@ -0,0 +1,24 @@
+// Copyright 2007-2011 Chris Patterson, Dru Sellers, Travis Smith, et. al.
+//  
+// Licensed under the Apache License, Version 2.0 (the "License"); you may not use 
+// this file except in compliance with the License. You may obtain a copy of the 
+// License at 
+// 
+//     http://www.apache.org/licenses/LICENSE-2.0 
+// 
+// Unless required by applicable law or agreed to in writing, software distributed 
+// under the License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR 
+// CONDITIONS OF ANY KIND, either express or implied. See the License for the 
+// specific language governing permissions and limitations under the License.
+namespace MassTransit.Transports.RabbitMq.Configuration.Configurators
+{
+    using System;
+
+    /// <summary>
+	/// Configures SSL/TLS for RabbitMQ. See http://www.rabbitmq.com/ssl.html
+	/// for details on how to set up RabbitMQ for SSL.
+	/// </summary>
+	public interface PublisherConfirmFactoryConfigurator
+	{
+	}
+}

src/Transports/MassTransit.Transports.RabbitMq/Configuration/Configurators/PublisherConfirmFactoryConfiguratorImpl.cs

@@ -0,0 +1,68 @@
+// Copyright 2007-2011 Chris Patterson, Dru Sellers, Travis Smith, et. al.
+//  
+// Licensed under the Apache License, Version 2.0 (the "License"); you may not use 
+// this file except in compliance with the License. You may obtain a copy of the 
+// License at 
+// 
+//     http://www.apache.org/licenses/LICENSE-2.0 
+// 
+// Unless required by applicable law or agreed to in writing, software distributed 
+// under the License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR 
+// CONDITIONS OF ANY KIND, either express or implied. See the License for the 
+// specific language governing permissions and limitations under the License.
+namespace MassTransit.Transports.RabbitMq.Configuration.Configurators
+{
+    using System;
+    using System.Collections.Generic;
+    using Builders;
+    using MassTransit.Configurators;
+    using PublisherConfirm;
+
+    public class PublisherConfirmFactoryConfiguratorImpl :
+        PublisherConfirmFactoryConfigurator,
+        RabbitMqTransportFactoryBuilderConfigurator
+    {
+        readonly bool _usePublisherConfirms;
+        readonly Action<ulong, string> _registerMessageAction;
+        readonly Action<ulong, bool> _acktion;
+        readonly Action<ulong, bool> _nacktion;
+
+        public PublisherConfirmFactoryConfiguratorImpl(bool usePublisherConfirms, Action<ulong, string> registerMessageAction, 
+            Action<ulong, bool> acktion, Action<ulong, bool> nacktion)
+        {
+            _usePublisherConfirms = usePublisherConfirms;
+            _registerMessageAction = registerMessageAction;
+            _acktion = acktion;
+            _nacktion = nacktion;
+        }
+
+        public RabbitMqTransportFactoryBuilder Configure(RabbitMqTransportFactoryBuilder builder)
+        {
+            builder.SetPublisherConfirmSettings(
+                new PublisherConfirmSettings
+                    {
+                        UsePublisherConfirms = _usePublisherConfirms,
+                        RegisterMessageAction = _registerMessageAction,
+                        Acktion =  _acktion,
+                        Nacktion = _nacktion
+                    }
+                );
+
+            return builder;
+        }
+
+        public IEnumerable<ValidationResult> Validate()
+        {
+            if (_usePublisherConfirms)
+            {
+                if (_registerMessageAction == null)
+                    yield return this.Failure("RegisterMessageAction", "RegisterMessageAction must be specified if publisher confirms are enabled");
+                if (_acktion == null)
+                    yield return this.Failure("Acktion", "Acktion must be specified if publisher confirms are enabled");
+                if(_nacktion == null)
+                    yield return this.Failure("Nacktion", "Nacktion must be specified if publisher confirms are enabled");
+            }
+        }
+
+    }
+}

src/Transports/MassTransit.Transports.RabbitMq/Configuration/RabbitMqTransportFactoryExtensions.cs

@@ -26,5 +26,14 @@ public static class RabbitMqTransportFactoryExtensions
 
 			configurator.AddConfigurator(hostConfigurator);
 		}
+
+        public static void UsePublisherConfirms(this RabbitMqTransportFactoryConfigurator configurator, 
+            Action<ulong, string> registerMessageAction, Action<ulong, bool> acktion, Action<ulong, bool> nacktion)
+        {
+            var hostConfigurator = new PublisherConfirmFactoryConfiguratorImpl(true, registerMessageAction, acktion, nacktion);
+
+            configurator.AddConfigurator(hostConfigurator);
+        }
+
 	}
 }

src/Transports/MassTransit.Transports.RabbitMq/MassTransit.Transports.RabbitMq.csproj

@@ -56,20 +56,22 @@
     <NoWarn>1591</NoWarn>
   </PropertyGroup>
   <ItemGroup>
-    <Reference Include="Magnum" Condition="'$(TargetFrameworkVersion)' == 'v3.5'">
-      <HintPath>..\..\..\lib\Magnum\net-3.5\Magnum.dll</HintPath>
-    </Reference>
-    <Reference Include="Magnum" Condition="'$(TargetFrameworkVersion)' == 'v4.0'">
+    <Reference Include="Magnum">
       <HintPath>..\..\..\lib\Magnum\net-4.0\Magnum.dll</HintPath>
     </Reference>
-    <Reference Include="RabbitMQ.Client">
+    <Reference Include="MassTransit">
+      <HintPath>..\..\packages\MassTransit.2.7.0\lib\net40\MassTransit.dll</HintPath>
+    </Reference>
+    <Reference Include="RabbitMQ.Client, Version=2.8.7.0, Culture=neutral, PublicKeyToken=89e7d7c5feba84ce, processorArchitecture=MSIL">
       <SpecificVersion>False</SpecificVersion>
-      <HintPath>..\..\..\lib\RabbitMQ\RabbitMQ.Client.dll</HintPath>
+      <HintPath>..\..\packages\RabbitMQ.Client.2.8.7\lib\net30\RabbitMQ.Client.dll</HintPath>
     </Reference>
     <Reference Include="System" />
+    <Reference Include="System.Configuration" />
     <Reference Include="System.Core">
       <RequiredTargetFramework>3.5</RequiredTargetFramework>
     </Reference>
+    <Reference Include="System.Data" />
   </ItemGroup>
   <ItemGroup>
     <Compile Include="..\..\SolutionVersion.cs">
@@ -82,6 +84,8 @@
     <Compile Include="Configuration\Configurators\ConnectionFactoryBuilderConfigurator.cs" />
     <Compile Include="Configuration\Configurators\ConnectionFactoryConfigurator.cs" />
     <Compile Include="Configuration\Configurators\ConnectionFactoryConfiguratorImpl.cs" />
+    <Compile Include="Configuration\Configurators\PublisherConfirmFactoryConfigurator.cs" />
+    <Compile Include="Configuration\Configurators\PublisherConfirmFactoryConfiguratorImpl.cs" />
     <Compile Include="Configuration\Configurators\SslConnectionFactoryConfigurator.cs" />
     <Compile Include="Configuration\Configurators\SslConnectionFactoryConfiguratorImpl.cs" />
     <Compile Include="Configuration\Configurators\RabbitMqTransportFactoryBuilderConfigurator.cs" />
@@ -95,6 +99,8 @@
     <Compile Include="OutboundRabbitMqTransport.cs" />
     <Compile Include="PublishEndpointInterceptor.cs" />
     <Compile Include="PublishEndpointSinkLocator.cs" />
+    <Compile Include="PublisherConfirm\IPublisherConfirmSettings.cs" />
+    <Compile Include="PublisherConfirm\PublisherConfirmSettings.cs" />
     <Compile Include="RabbitMqConnection.cs" />
     <Compile Include="RabbitMqConsumer.cs" />
     <Compile Include="RabbitMqEndpointAddress.cs" />
@@ -110,13 +116,6 @@
     <Compile Include="Testing\BusTestScenarioExtensions.cs" />
     <Compile Include="Testing\RabbitMqBusScenarioBuilder.cs" />
   </ItemGroup>
-  <ItemGroup>
-    <ProjectReference Include="..\..\MassTransit\MassTransit.csproj">
-      <Project>{6EFD69FC-CBCC-4F85-AEE0-EFBA73F4D273}</Project>
-      <Name>MassTransit</Name>
-      <Private>False</Private>
-    </ProjectReference>
-  </ItemGroup>
   <ItemGroup>
     <BootstrapperPackage Include="Microsoft.Net.Client.3.5">
       <Visible>False</Visible>
@@ -137,6 +136,9 @@
   <ItemGroup>
     <None Include="README.md" />
   </ItemGroup>
+  <ItemGroup>
+    <WCFMetadata Include="Service References\" />
+  </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.

src/Transports/MassTransit.Transports.RabbitMq/OutboundRabbitMqTransport.cs

@@ -16,6 +16,7 @@ namespace MassTransit.Transports.RabbitMq
     using System.Collections;
     using System.IO;
     using Magnum;
+    using PublisherConfirm;
     using RabbitMQ.Client;
     using RabbitMQ.Client.Exceptions;
 
@@ -25,14 +26,18 @@ public class OutboundRabbitMqTransport :
         readonly IRabbitMqEndpointAddress _address;
         readonly bool _bindToQueue;
         readonly ConnectionHandler<RabbitMqConnection> _connectionHandler;
+        readonly IPublisherConfirmSettings _publisherConfirmSettings;
         RabbitMqProducer _producer;
 
         public OutboundRabbitMqTransport(IRabbitMqEndpointAddress address,
-            ConnectionHandler<RabbitMqConnection> connectionHandler, bool bindToQueue)
+            ConnectionHandler<RabbitMqConnection> connectionHandler, 
+            IPublisherConfirmSettings publisherConfirmSettings,
+            bool bindToQueue)
         {
             _address = address;
             _connectionHandler = connectionHandler;
             _bindToQueue = bindToQueue;
+            _publisherConfirmSettings = publisherConfirmSettings;
         }
 
         public IEndpointAddress Address
@@ -52,21 +57,24 @@ public void Send(ISendContext context)
 
                         properties.SetPersistent(true);
                         properties.MessageId = context.MessageId ?? properties.MessageId ?? NewId.Next().ToString();
+
                         if (context.ExpirationTime.HasValue)
                         {
                             DateTime value = context.ExpirationTime.Value;
-                            properties.Expiration =
-                                (value.Kind == DateTimeKind.Utc
-                                     ? value - SystemUtil.UtcNow
-                                     : value - SystemUtil.Now).
-                                    ToString();
+                            properties.Expiration = (value.Kind == DateTimeKind.Utc ? value - SystemUtil.UtcNow : value - SystemUtil.Now).ToString();
+                        }
+
+                        properties.Headers = new Hashtable { { "Content-Type", context.ContentType } };
+
+                        if (context.Headers[PublisherConfirmSettings.ClientMessageId] != null)
+                        {
+                            properties.Headers.Add(PublisherConfirmSettings.ClientMessageId, context.Headers[PublisherConfirmSettings.ClientMessageId]);
                         }
 
                         using (var body = new MemoryStream())
                         {
                             context.SerializeTo(body);
-                            properties.Headers = new Hashtable {{"Content-Type", context.ContentType}};
-
+
                             _producer.Publish(_address.Name, properties, body.ToArray());
 
                             _address.LogSent(context.MessageId ?? "", context.MessageType);
@@ -97,7 +105,7 @@ void AddProducerBinding()
             if (_producer != null)
                 return;
 
-            _producer = new RabbitMqProducer(_address, _bindToQueue);
+            _producer = new RabbitMqProducer(_address, _publisherConfirmSettings, _bindToQueue);
 
             _connectionHandler.AddBinding(_producer);
         }

src/Transports/MassTransit.Transports.RabbitMq/PublisherConfirm/IPublisherConfirmSettings.cs

@@ -0,0 +1,16 @@
+namespace MassTransit.Transports.RabbitMq.PublisherConfirm
+{
+    using System;
+
+    public interface IPublisherConfirmSettings
+    {
+        bool UsePublisherConfirms { get; set; }
+
+        Action<ulong, string> RegisterMessageAction { get; set; }
+
+        Action<ulong, bool> Acktion { get; set; }
+
+        Action<ulong, bool> Nacktion { get; set; }
+
+    }
+}

src/Transports/MassTransit.Transports.RabbitMq/PublisherConfirm/PublisherConfirmSettings.cs

@@ -0,0 +1,18 @@
+namespace MassTransit.Transports.RabbitMq.PublisherConfirm
+{
+    using System;
+
+    public class PublisherConfirmSettings : IPublisherConfirmSettings
+    {
+        public const string ClientMessageId = "ClientMessageId";
+
+        public bool UsePublisherConfirms { get; set; }
+
+        public Action<ulong, string> RegisterMessageAction { get; set; }
+
+        public Action<ulong, bool> Acktion { get; set; }
+
+        public Action<ulong, bool> Nacktion { get; set; }
+
+    }
+}